feat(reliability-dashboard): add control-plane reconciliation metrics (Hypershell 286 )

components/control-plane/cmd/hypershell-controller/main.go

@@ -260,43 +260,6 @@ func main() {
 	}
 
 	var gatewayReconciler watcher.Handler[*pb.Gateway]
-
-	if clientset != nil && dynamicClient != nil {
-		gr, grErr := reconciler.NewGatewayReconciler(
-			dynamicClient,
-			clientset,
-			conn,
-			helmClient,
-			cfg.Namespace,
-			keycloakConfig,
-			exposurePort,
-			cfg.ExternalCAIssuerName,
-			cfg.ExternalCAIssuerKind,
-			databaseConfig,
-		)
-		if grErr != nil {
-			log.Printf("WARN gateway reconciler disabled: %v", grErr)
-			gatewayReconciler = reconciler.NewStubGatewayReconciler()
-		} else {
-			gatewayReconciler = gr
-		}
-	} else {
-		log.Printf("WARN no kubernetes client available, using stub gateway reconciler")
-		gatewayReconciler = reconciler.NewStubGatewayReconciler()
-	}
-
-	// The gateway reconcile queue is shared: the gateway watch stream drives it,
-	// and the GatewayRelease reconciler enqueues referencing gateways into it when
-	// a release image changes. It is created here (not inside WatchGateways) so the
-	// release reconciler can hold the same instance.
-	gatewayQueue := watcher.NewGatewayReconcileQueue(ctx, gatewayReconciler, cfg.GatewayReconcileWorkers)
-	defer gatewayQueue.Stop()
-
-	watchCount := 4 // managed clusters, gateway releases, gateways, networks
-	if kcClient != nil {
-		watchCount++ // role bindings
-	}
-
 	// Each background component below runs under supervisor.Run on its own
 	// goroutine: a failure in one (a dropped watch stream, a reconciler's Run
 	// loop returning, the service-account provisioner's listener dying) is
@@ -341,6 +304,43 @@ func main() {
 	}
 	log.Printf("INFO registered as cluster_id=%s (name=%s); scoping gateway watch, seed, health, sandbox counts, backfill, release fan-out, and namespace GC to it", clusterID, cfg.ManagedClusterName)
 
+	if clientset != nil && dynamicClient != nil {
+		gr, grErr := reconciler.NewGatewayReconciler(
+			dynamicClient,
+			clientset,
+			conn,
+			helmClient,
+			cfg.Namespace,
+			keycloakConfig,
+			exposurePort,
+			cfg.ExternalCAIssuerName,
+			cfg.ExternalCAIssuerKind,
+			databaseConfig,
+			clusterID,
+		)
+		if grErr != nil {
+			log.Printf("WARN gateway reconciler disabled: %v", grErr)
+			gatewayReconciler = reconciler.NewStubGatewayReconciler()
+		} else {
+			gatewayReconciler = gr
+		}
+	} else {
+		log.Printf("WARN no kubernetes client available, using stub gateway reconciler")
+		gatewayReconciler = reconciler.NewStubGatewayReconciler()
+	}
+
+	// The gateway reconcile queue is shared: the gateway watch stream drives it,
+	// and the GatewayRelease reconciler enqueues referencing gateways into it when
+	// a release image changes. It is created after registration so the reconciler
+	// and all queued work use the registered cluster identity.
+	gatewayQueue := watcher.NewGatewayReconcileQueue(ctx, gatewayReconciler, cfg.GatewayReconcileWorkers)
+	defer gatewayQueue.Stop()
+
+	watchCount := 4 // managed clusters, gateway releases, gateways, networks
+	if kcClient != nil {
+		watchCount++ // role bindings
+	}
+
 	// Heartbeat: re-register every 60s to update last_seen_at on the hub.
 	go func() {
 		ticker := time.NewTicker(60 * time.Second)

components/control-plane/internal/otel/metrics.go

@@ -3,6 +3,7 @@ package otel
 import (
 	"context"
 	"fmt"
+	"sync"
 	"time"
 
 	"go.opentelemetry.io/otel"
@@ -20,9 +21,16 @@ var (
 	gatewaySandboxExpiring     metric.Int64Gauge
 	gatewaySandboxIdle         metric.Int64Gauge
 	reconcileErrors            metric.Int64Counter
+	reconciliationFailures     metric.Int64Counter
+	reconciliationRetries      metric.Int64Counter
+	reconciliationLag          metric.Float64Histogram
+	staleResourceStatusCount   metric.Int64Gauge
 	watchReconnects            metric.Int64Counter
 )
 
+var staleResources sync.Map
+var staleResourcesMu sync.Mutex
+
 func registerMetrics() error {
 	meter := otel.Meter(TracerName)
 	var err error
@@ -35,7 +43,6 @@ func registerMetrics() error {
 	if err != nil {
 		return err
 	}
-
 	reconcileQueueDepth, err = meter.Int64ObservableGauge(
 		"reconcile.queue.depth",
 		metric.WithUnit("{item}"),
@@ -110,6 +117,48 @@ func registerMetrics() error {
 		return err
 	}
 
+	reconciliationFailures, err = meter.Int64Counter(
+		"hypershell.reconciliation.failures",
+		metric.WithUnit("{failure}"),
+		metric.WithDescription("Count of failed control-plane reconciliation attempts"),
+	)
+	if err != nil {
+		return err
+	}
+	reconciliationFailures.Add(context.Background(), 0)
+
+	reconciliationRetries, err = meter.Int64Counter(
+		"hypershell.reconciliation.retries",
+		metric.WithUnit("{retry}"),
+		metric.WithDescription("Count of control-plane reconciliation retries"),
+	)
+	if err != nil {
+		return err
+	}
+	reconciliationRetries.Add(context.Background(), 0)
+
+	reconciliationLag, err = meter.Float64Histogram(
+		"hypershell.reconciliation.lag",
+		metric.WithUnit("s"),
+		metric.WithDescription("Time spent processing a control-plane reconciliation"),
+		metric.WithExplicitBucketBoundaries(0.001, 0.01, 0.1, 1, 5, 10, 30, 60, 300),
+	)
+	if err != nil {
+		return err
+	}
+
+	staleResourceStatusCount, err = meter.Int64Gauge(
+		"hypershell.stale.resource.status.count",
+		metric.WithUnit("{resource}"),
+		metric.WithDescription("Resources whose observed status has not caught up with desired state"),
+	)
+	if err != nil {
+		return err
+	}
+	staleResourceStatusCount.Record(context.Background(), 0, metric.WithAttributes(
+		attribute.String("hypershell.cluster_id", DefaultAttentionClusterID),
+	))
+
 	watchReconnects, err = meter.Int64Counter(
 		"watch.reconnects",
 		metric.WithUnit("{reconnect}"),
@@ -221,6 +270,53 @@ func RecordReconcileError(ctx context.Context, kind string) {
 	reconcileErrors.Add(ctx, 1, metric.WithAttributes(
 		attribute.String("resource.kind", kind),
 	))
+	if reconciliationFailures != nil {
+		reconciliationFailures.Add(ctx, 1, metric.WithAttributes(
+			attribute.String("resource.kind", kind),
+		))
+	}
+}
+
+// RecordReconciliationRetry records a retry scheduled for a reconciliation.
+func RecordReconciliationRetry(ctx context.Context, kind string) {
+	if reconciliationRetries == nil {
+		return
+	}
+	reconciliationRetries.Add(ctx, 1, metric.WithAttributes(
+		attribute.String("resource.kind", kind),
+	))
+}
+
+// RecordReconciliationLag records the elapsed reconciliation processing time.
+func RecordReconciliationLag(ctx context.Context, kind string, duration time.Duration) {
+	if reconciliationLag == nil || duration < 0 {
+		return
+	}
+	reconciliationLag.Record(ctx, duration.Seconds(), metric.WithAttributes(
+		attribute.String("resource.kind", kind),
+	))
+}
+
+// SetResourceStatusStale records whether a resource's observed status is
+// behind its desired state. Resource IDs remain in process memory only and are
+// never exported as metric labels.
+func SetResourceStatusStale(ctx context.Context, clusterID, kind, resourceID string, stale bool) {
+	if staleResourceStatusCount == nil || kind == "" || resourceID == "" {
+		return
+	}
+	staleResourcesMu.Lock()
+	defer staleResourcesMu.Unlock()
+	key := kind + "\x00" + resourceID
+	if stale {
+		staleResources.Store(key, struct{}{})
+	} else {
+		staleResources.Delete(key)
+	}
+	count := int64(0)
+	staleResources.Range(func(_, _ any) bool { count++; return true })
+	staleResourceStatusCount.Record(ctx, count, metric.WithAttributes(
+		attribute.String("hypershell.cluster_id", AttentionClusterID(clusterID)),
+	))
 }
 
 // RecordWatchReconnect increments the watch reconnect counter.

components/control-plane/internal/otel/metrics_test.go

@@ -24,6 +24,10 @@ func testMetricsReader(t *testing.T) *sdkmetric.ManualReader {
 	previousGatewaySandboxExpiring := gatewaySandboxExpiring
 	previousGatewaySandboxIdle := gatewaySandboxIdle
 	previousReconcileErrors := reconcileErrors
+	previousReconciliationFailures := reconciliationFailures
+	previousReconciliationRetries := reconciliationRetries
+	previousReconciliationLag := reconciliationLag
+	previousStaleResourceStatusCount := staleResourceStatusCount
 	previousWatchReconnects := watchReconnects
 
 	reader := sdkmetric.NewManualReader()
@@ -38,6 +42,10 @@ func testMetricsReader(t *testing.T) *sdkmetric.ManualReader {
 	gatewaySandboxExpiring = nil
 	gatewaySandboxIdle = nil
 	reconcileErrors = nil
+	reconciliationFailures = nil
+	reconciliationRetries = nil
+	reconciliationLag = nil
+	staleResourceStatusCount = nil
 	watchReconnects = nil
 	t.Cleanup(func() {
 		otel.SetMeterProvider(previousProvider)
@@ -50,6 +58,10 @@ func testMetricsReader(t *testing.T) *sdkmetric.ManualReader {
 		gatewaySandboxExpiring = previousGatewaySandboxExpiring
 		gatewaySandboxIdle = previousGatewaySandboxIdle
 		reconcileErrors = previousReconcileErrors
+		reconciliationFailures = previousReconciliationFailures
+		reconciliationRetries = previousReconciliationRetries
+		reconciliationLag = previousReconciliationLag
+		staleResourceStatusCount = previousStaleResourceStatusCount
 		watchReconnects = previousWatchReconnects
 		_ = provider.Shutdown(context.Background())
 	})
@@ -60,6 +72,57 @@ func testMetricsReader(t *testing.T) *sdkmetric.ManualReader {
 	return reader
 }
 
+func TestReconciliationMetrics(t *testing.T) {
+	reader := testMetricsReader(t)
+	RecordReconcileError(context.Background(), "Gateway")
+	RecordReconciliationRetry(context.Background(), "Gateway")
+	RecordReconciliationLag(context.Background(), "Gateway", 2*time.Second)
+	SetResourceStatusStale(context.Background(), "cluster-a", "Gateway", "one", true)
+	SetResourceStatusStale(context.Background(), "cluster-a", "Gateway", "one", true)
+	SetResourceStatusStale(context.Background(), "cluster-a", "Gateway", "two", true)
+	SetResourceStatusStale(context.Background(), "cluster-a", "Gateway", "one", false)
+
+	var collected metricdata.ResourceMetrics
+	if err := reader.Collect(context.Background(), &collected); err != nil {
+		t.Fatalf("Collect() returned an error: %v", err)
+	}
+	for _, scope := range collected.ScopeMetrics {
+		for _, got := range scope.Metrics {
+			switch got.Name {
+			case "hypershell.reconciliation.failures":
+				if _, ok := got.Data.(metricdata.Sum[int64]); !ok {
+					t.Fatalf("failures type = %T", got.Data)
+				}
+			case "hypershell.reconciliation.retries":
+				if _, ok := got.Data.(metricdata.Sum[int64]); !ok {
+					t.Fatalf("retries type = %T", got.Data)
+				}
+			case "hypershell.reconciliation.lag":
+				h := got.Data.(metricdata.Histogram[float64])
+				want := []float64{0.001, 0.01, 0.1, 1, 5, 10, 30, 60, 300}
+				if !reflect.DeepEqual(h.DataPoints[0].Bounds, want) {
+					t.Fatalf("lag bounds = %v", h.DataPoints[0].Bounds)
+				}
+			case "hypershell.stale.resource.status.count":
+				g := got.Data.(metricdata.Gauge[int64])
+				found := false
+				for _, point := range g.DataPoints {
+					v, ok := point.Attributes.Value(attribute.Key("hypershell.cluster_id"))
+					if ok && v.AsString() == "cluster-a" {
+						found = true
+						if point.Value != 1 {
+							t.Fatalf("stale gauge = %d, want 1", point.Value)
+						}
+					}
+				}
+				if !found {
+					t.Fatal("stale cluster label was not exported")
+				}
+			}
+		}
+	}
+}
+
 func TestRecordGatewayProvisionDuration(t *testing.T) {
 	reader := testMetricsReader(t)
 

components/control-plane/internal/otel/reconcile.go

@@ -54,6 +54,7 @@ func StartReconcileSpan(ctx context.Context, kind, eventType, traceparent string
 			span.SetStatus(codes.Ok, "")
 		}
 		RecordReconcileDuration(ctx, kind, eventType, start)
+		RecordReconciliationLag(ctx, kind, time.Since(start))
 		span.End()
 	}
 }

components/control-plane/internal/reconciler/reconciler.go

@@ -342,7 +342,8 @@ type GatewayReconciler struct {
 	ingressBaseDomain     string
 	// database locates the mounted admin credentials every gateway database is
 	// provisioned with; validated once at controller startup.
-	database gateway.DatabaseConfig
+	database  gateway.DatabaseConfig
+	clusterID string
 }
 
 func NewGatewayReconciler(
@@ -356,6 +357,7 @@ func NewGatewayReconciler(
 	externalCAIssuerName string,
 	externalCAIssuerKind string,
 	database gateway.DatabaseConfig,
+	clusterID string,
 ) (*GatewayReconciler, error) {
 	if database.AdminCredentialsDir == "" {
 		return nil, fmt.Errorf("gateway database admin credentials directory is required")
@@ -410,6 +412,7 @@ func NewGatewayReconciler(
 		externalCAIssuerKind:  externalCAIssuerKind,
 		ingressBaseDomain:     ingressBaseDomain,
 		database:              database,
+		clusterID:             clusterID,
 	}, nil
 }
 
@@ -437,8 +440,9 @@ func (r *GatewayReconciler) Handle(ctx context.Context, event watcher.Event[*pb.
 		previousPhase = event.PhaseBeforeRetry
 	}
 	suppressGatewayProvisionObservation(event.ResourceID, previousPhase)
-
 	ctx, endSpan := cpotel.StartReconcileSpan(ctx, "Gateway", event.Type.String(), gw.GetMetadata().GetTraceparent())
+	cpotel.SetResourceStatusStale(ctx, r.clusterID, "Gateway", event.ResourceID,
+		event.Type != watcher.EventDeleted && gw.GetReleaseId() != gw.GetObservedReleaseId())
 	span := trace.SpanFromContext(ctx)
 	span.SetAttributes(attribute.String("hypershell.resource_id", event.ResourceID))
 	var reconcileErr error

components/control-plane/internal/watcher/requeue.go

@@ -589,6 +589,7 @@ func (q *reconcileQueue[T]) processNext() bool {
 		}
 		q.mu.Unlock()
 		q.queue.AddAfter(id, delay)
+		cpotel.RecordReconciliationRetry(q.baseCtx, q.kind)
 		log.Printf("WARN %s %s reconcile failed; retrying in %s: %v", q.kind, id, delay, err)
 		return true
 	}