perf(operator): parallelize secret reads, remove monitorPod goroutines, use informer cache

- Replace sequential secret/config reads with errgroup-based parallel
  fetches, reducing API round-trips when creating sessions under load
- Remove monitorPod goroutine-per-session pattern; controller-runtime
  reconciler handles pod status via its watch/requeue loop
- Eliminate redundant CR re-GET in ReconcilePendingSession; use the
  object already provided by the controller-runtime informer cache
- Switch gauge callbacks in otel_metrics.go to use the informer cache
  instead of a cluster-wide LIST against the API server every 30s
- Replace clearAnnotation GET+UPDATE (2 API calls) with a single
  JSON merge patch

Validated with load-test-sessions.sh: 30 sessions reach Running in
~21s with 20 concurrent reconcilers, zero errors.

Author: syntaxsdev

components/operator/go.mod

@@ -10,6 +10,7 @@ require (
 	go.opentelemetry.io/otel/metric v1.33.0
 	go.opentelemetry.io/otel/sdk v1.33.0
 	go.opentelemetry.io/otel/sdk/metric v1.33.0
+	golang.org/x/sync v0.12.0
 	k8s.io/api v0.34.0
 	k8s.io/apimachinery v0.34.0
 	k8s.io/client-go v0.34.0
@@ -60,7 +61,6 @@ require (
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/net v0.38.0 // indirect
 	golang.org/x/oauth2 v0.27.0 // indirect
-	golang.org/x/sync v0.12.0 // indirect
 	golang.org/x/sys v0.31.0 // indirect
 	golang.org/x/term v0.30.0 // indirect
 	golang.org/x/text v0.23.0 // indirect

components/operator/internal/controller/otel_metrics.go

@@ -17,13 +17,18 @@ import (
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"ambient-code-operator/internal/config"
 )
 
 var (
 	meter metric.Meter
 
+	// cachedClient uses the controller-runtime informer cache for reads,
+	// avoiding direct API server hits every 30s for gauge callbacks.
+	cachedClient client.Client
+
 	// Session lifecycle metrics (histograms)
 	sessionStartupDuration metric.Float64Histogram
 	sessionTotalDuration   metric.Float64Histogram
@@ -48,7 +53,12 @@ var (
 // InitMetrics initializes OpenTelemetry metrics.
 // Set OTEL_EXPORTER_OTLP_ENDPOINT to configure the collector address.
 // Leave unset or empty to disable metrics export (no-op).
-func InitMetrics(ctx context.Context) (func(), error) {
+// If c is non-nil, gauge callbacks will use the informer cache instead of
+// hitting the API server directly (saves a cluster-wide LIST every 30s).
+func InitMetrics(ctx context.Context, c ...client.Client) (func(), error) {
+	if len(c) > 0 && c[0] != nil {
+		cachedClient = c[0]
+	}
 	// Get OTLP endpoint from environment; skip if not configured
 	endpoint := os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT")
 	if endpoint == "" {
@@ -316,33 +326,49 @@ func initGauges() error {
 	return nil
 }
 
-// countSessionsByPhase counts sessions in the given phases, grouped by namespace
+// countSessionsByPhase counts sessions in the given phases, grouped by namespace.
+// When cachedClient is set (controller-runtime), reads from the informer cache
+// instead of making a cluster-wide LIST against the API server every 30s.
 func countSessionsByPhase(ctx context.Context, phases ...string) map[string]int64 {
 	counts := make(map[string]int64)
 
-	// Use config.DynamicClient to list sessions
-	if config.DynamicClient == nil {
-		return counts
-	}
-
-	gvr := schema.GroupVersionResource{
-		Group:    "vteam.ambient-code",
-		Version:  "v1alpha1",
-		Resource: "agenticsessions",
-	}
-
-	list, err := config.DynamicClient.Resource(gvr).List(ctx, v1.ListOptions{})
-	if err != nil {
-		log.Printf("Failed to list sessions for metrics: %v", err)
-		return counts
-	}
-
 	phaseSet := make(map[string]bool)
 	for _, p := range phases {
 		phaseSet[p] = true
 	}
 
-	for _, item := range list.Items {
+	sessionList := &unstructured.UnstructuredList{}
+	sessionList.SetGroupVersionKind(schema.GroupVersionKind{
+		Group:   "vteam.ambient-code",
+		Version: "v1alpha1",
+		Kind:    "AgenticSessionList",
+	})
+
+	if cachedClient != nil {
+		// Use informer cache - no API server round-trip
+		if err := cachedClient.List(ctx, sessionList); err != nil {
+			log.Printf("Failed to list sessions for metrics (cached): %v", err)
+			return counts
+		}
+	} else {
+		// Fallback to direct API call if cached client not available
+		if config.DynamicClient == nil {
+			return counts
+		}
+		gvr := schema.GroupVersionResource{
+			Group:    "vteam.ambient-code",
+			Version:  "v1alpha1",
+			Resource: "agenticsessions",
+		}
+		list, err := config.DynamicClient.Resource(gvr).List(ctx, v1.ListOptions{})
+		if err != nil {
+			log.Printf("Failed to list sessions for metrics: %v", err)
+			return counts
+		}
+		sessionList.Items = list.Items
+	}
+
+	for _, item := range sessionList.Items {
 		ns := item.GetNamespace()
 		status, found, _ := unstructured.NestedMap(item.Object, "status")
 		if !found {

components/operator/internal/handlers/helpers.go

@@ -14,6 +14,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/errors"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	k8stypes "k8s.io/apimachinery/pkg/types"
 )
 
 const (
@@ -221,31 +222,17 @@ func updateAnnotations(sessionNamespace, name string, annotations map[string]str
 	return nil
 }
 
-// clearAnnotation removes a specific annotation from the AgenticSession CR.
+// clearAnnotation removes a specific annotation from the AgenticSession CR
+// using a single JSON merge patch (1 API call) instead of GET + full UPDATE (2 API calls).
 func clearAnnotation(sessionNamespace, name, annotationKey string) error {
 	gvr := types.GetAgenticSessionResource()
 
-	obj, err := config.DynamicClient.Resource(gvr).Namespace(sessionNamespace).Get(context.TODO(), name, v1.GetOptions{})
-	if err != nil {
-		if errors.IsNotFound(err) {
-			return nil
-		}
-		return fmt.Errorf("failed to get AgenticSession %s: %w", name, err)
-	}
-
-	annotations := obj.GetAnnotations()
-	if annotations == nil {
-		return nil
-	}
-
-	if _, exists := annotations[annotationKey]; !exists {
-		return nil
-	}
-
-	delete(annotations, annotationKey)
-	obj.SetAnnotations(annotations)
+	// JSON merge patch: setting a key to null removes it
+	patch := []byte(fmt.Sprintf(`{"metadata":{"annotations":{%q:null}}}`, annotationKey))
 
-	_, err = config.DynamicClient.Resource(gvr).Namespace(sessionNamespace).Update(context.TODO(), obj, v1.UpdateOptions{})
+	_, err := config.DynamicClient.Resource(gvr).Namespace(sessionNamespace).Patch(
+		context.TODO(), name, k8stypes.MergePatchType, patch, v1.PatchOptions{},
+	)
 	if err != nil && !errors.IsNotFound(err) {
 		return fmt.Errorf("failed to clear annotation %s for %s: %w", annotationKey, name, err)
 	}