feat: Add control plane telemetry - workerpool desired and ready workers gauges

cmd/atecontroller/internal/controllers/workerpool_controller.go

@@ -18,6 +18,8 @@ import (
 	"context"
 	"fmt"
 
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/metric"
 	appsv1 "k8s.io/api/apps/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
 	k8errors "k8s.io/apimachinery/pkg/api/errors"
@@ -28,6 +30,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
+	"github.com/agent-substrate/substrate/internal/ateattr"
 	atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1"
 )
 
@@ -49,6 +52,9 @@ type WorkerPoolReconciler struct {
 	// OTelTracesSamplerArg is the OTEL_TRACES_SAMPLER_ARG propagated to ateom
 	// pods. Ignored unless OTelTracesSampler is set.
 	OTelTracesSamplerArg string
+
+	desiredWorkers metric.Int64ObservableUpDownCounter
+	readyWorkers   metric.Int64ObservableUpDownCounter
 }
 
 //+kubebuilder:rbac:groups=ate.dev,resources=workerpools,verbs=get;list;watch;create;update;patch;delete
@@ -124,8 +130,9 @@ func (r *WorkerPoolReconciler) syncStatus(ctx context.Context, wp *atev1alpha1.W
 	}
 
 	want := atev1alpha1.WorkerPoolStatus{
-		Replicas: dep.Status.Replicas,
-		Selector: selector.String(),
+		Replicas:      dep.Status.Replicas,
+		ReadyReplicas: dep.Status.ReadyReplicas,
+		Selector:      selector.String(),
 	}
 	if equality.Semantic.DeepEqual(wp.Status, want) {
 		return nil
@@ -139,8 +146,61 @@ func (r *WorkerPoolReconciler) syncStatus(ctx context.Context, wp *atev1alpha1.W
 	return nil
 }
 
+// InitMetrics initializes the OpenTelemetry instruments for ate.workerpool.desired_workers
+// and ate.workerpool.ready_workers and registers the asynchronous callback.
+func (r *WorkerPoolReconciler) InitMetrics(meter metric.Meter) error {
+	desiredWorkers, err := meter.Int64ObservableUpDownCounter(
+		"ate.workerpool.desired_workers",
+		metric.WithUnit("{worker}"),
+		metric.WithDescription("number of worker pods requested for a WorkerPool (spec.replicas)"),
+	)
+	if err != nil {
+		return fmt.Errorf("create ate.workerpool.desired_workers instrument: %w", err)
+	}
+	r.desiredWorkers = desiredWorkers
+
+	readyWorkers, err := meter.Int64ObservableUpDownCounter(
+		"ate.workerpool.ready_workers",
+		metric.WithUnit("{worker}"),
+		metric.WithDescription("number of worker pods currently ready for a WorkerPool (status.readyReplicas)"),
+	)
+	if err != nil {
+		return fmt.Errorf("create ate.workerpool.ready_workers instrument: %w", err)
+	}
+	r.readyWorkers = readyWorkers
+
+	_, err = meter.RegisterCallback(
+		func(ctx context.Context, obs metric.Observer) error {
+			var list atev1alpha1.WorkerPoolList
+			if err := r.List(ctx, &list); err != nil {
+				log.FromContext(ctx).Error(err, "failed to list worker pools to observe ate.workerpool.desired_workers and ate.workerpool.ready_workers")
+				return nil
+			}
+			for _, wp := range list.Items {
+				attrs := metric.WithAttributes(
+					ateattr.WorkerPoolNamespaceKey.String(wp.Namespace),
+					ateattr.WorkerPoolNameKey.String(wp.Name),
+				)
+				obs.ObserveInt64(r.desiredWorkers, int64(wp.Spec.Replicas), attrs)
+				obs.ObserveInt64(r.readyWorkers, int64(wp.Status.ReadyReplicas), attrs)
+			}
+			return nil
+		},
+		r.desiredWorkers,
+		r.readyWorkers,
+	)
+	if err != nil {
+		return fmt.Errorf("register workerpool metrics callback: %w", err)
+	}
+
+	return nil
+}
+
 // SetupWithManager sets up the controller with the Manager.
 func (r *WorkerPoolReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	if err := r.InitMetrics(otel.Meter("atecontroller")); err != nil {
+		return fmt.Errorf("failed to initialize workerpool metrics: %w", err)
+	}
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&atev1alpha1.WorkerPool{}).
 		Owns(&appsv1.Deployment{}).

cmd/atecontroller/internal/controllers/workerpool_controller_test.go

@@ -18,9 +18,13 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"reflect"
 	"testing"
 	"time"
 
+	"go.opentelemetry.io/otel/attribute"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	k8errors "k8s.io/apimachinery/pkg/api/errors"
@@ -35,8 +39,10 @@ import (
 	"k8s.io/client-go/util/retry"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 
+	"github.com/agent-substrate/substrate/internal/ateattr"
 	"github.com/agent-substrate/substrate/internal/testenv"
 	atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1"
 )
@@ -652,3 +658,81 @@ func eventually(t *testing.T, condition func(ctx context.Context) (bool, error))
 		t.Fatalf("condition not met within timeout: %v", err)
 	}
 }
+
+// TestSyncStatus_ReadyReplicas verifies that a Deployment reporting 3 replicas
+// with 2 ready propagates to WorkerPool.status.
+func TestSyncStatus_ReadyReplicas(t *testing.T) {
+	t.Parallel()
+	ctx := t.Context()
+	wp := makeWorkerPool("test-sync-ready", "default", 3, "ateom:v1")
+	if err := k8sClient.Create(ctx, wp); err != nil {
+		t.Fatalf("create WorkerPool: %v", err)
+	}
+	deleteOnCleanup(t, wp)
+	eventually(t, func(ctx context.Context) (bool, error) {
+		_, err := getDeployment(ctx, wp)
+		return err == nil, nil
+	})
+	// Simulate the deployment controller reporting 3 pods, 2 of them ready.
+	updateDeploymentStatus(t, ctx, wp, "patch Deployment status", func(dep *appsv1.Deployment) {
+		dep.Status.Replicas = 3
+		dep.Status.ReadyReplicas = 2
+	})
+	eventually(t, func(ctx context.Context) (bool, error) {
+		current := &atev1alpha1.WorkerPool{}
+		if err := k8sClient.Get(ctx, types.NamespacedName{Name: wp.Name, Namespace: wp.Namespace}, current); err != nil {
+			return false, nil
+		}
+		return current.Status.Replicas == 3 && current.Status.ReadyReplicas == 2, nil
+	})
+}
+
+// TestWorkerPoolMetrics verifies that the registered callback observes
+// spec.replicas and status.readyReplicas per WorkerPool, labeled with the pool
+// namespace and name. A fake client keeps this off the envtest reconciler,
+// which would otherwise resync status out from under the assertion.
+func TestWorkerPoolMetrics(t *testing.T) {
+	t.Parallel()
+	scheme := runtime.NewScheme()
+	if err := atev1alpha1.AddToScheme(scheme); err != nil {
+		t.Fatalf("add scheme: %v", err)
+	}
+	wp := makeWorkerPool("test-metrics", "default", 4, "ateom:v1")
+	wp.Status = atev1alpha1.WorkerPoolStatus{Replicas: 4, ReadyReplicas: 2}
+	reader := sdkmetric.NewManualReader()
+	r := &WorkerPoolReconciler{
+		Client: fake.NewClientBuilder().WithScheme(scheme).WithObjects(wp).Build(),
+	}
+	if err := r.InitMetrics(sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)).Meter("test")); err != nil {
+		t.Fatalf("InitMetrics: %v", err)
+	}
+	var rm metricdata.ResourceMetrics
+	if err := reader.Collect(t.Context(), &rm); err != nil {
+		t.Fatalf("collect metrics: %v", err)
+	}
+	wantAttrs := attribute.NewSet(
+		ateattr.WorkerPoolNamespaceKey.String(wp.Namespace),
+		ateattr.WorkerPoolNameKey.String(wp.Name),
+	)
+	got := map[string]int64{}
+	for _, sm := range rm.ScopeMetrics {
+		for _, m := range sm.Metrics {
+			sum, ok := m.Data.(metricdata.Sum[int64])
+			if !ok {
+				continue
+			}
+			for _, dp := range sum.DataPoints {
+				if dp.Attributes.Equals(&wantAttrs) {
+					got[m.Name] = dp.Value
+				}
+			}
+		}
+	}
+	want := map[string]int64{
+		"ate.workerpool.desired_workers": 4,
+		"ate.workerpool.ready_workers":   2,
+	}
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("observed %v, want %v", got, want)
+	}
+}

docs/observability.md

@@ -114,6 +114,10 @@ Agent Substrate emits foundational OpenTelemetry system and server metrics to mo
 | `atenet.router.route.duration` | atenet-router | histogram | Substrate E2E — Envoy receiving a request to Envoy forwarding it to the resolved worker, excluding actor compute and the response (labels `ate.template.namespace`, `ate.template.name`, `ate.router.outcome`, `ate.router.resume`) |
 | `ate.scheduler.eligible_workers` | ateapi | histogram | number of eligible unassigned workers available during scheduling given the constraint filters (labels `ate.workerpool.namespace`, `ate.workerpool.name`, `ate.sandbox.class`, `ate.scheduling.constraint`) |
 | `atelet.snapshot.size` | atelet | histogram | uncompressed size in bytes of each gVisor snapshot image written during checkpoint (labels `file.name`, `ate.template.namespace`, `ate.template.name`) |
+| `ate.workerpool.desired_workers` | atecontroller | up/down counter | number of worker pods requested for a WorkerPool, from `spec.replicas` (labels
+`ate.workerpool.namespace`, `ate.workerpool.name`) |
+| `ate.workerpool.ready_workers` | atecontroller | up/down counter | number of worker pods currently ready for a WorkerPool, from `status.readyReplicas` (labels
+`ate.workerpool.namespace`, `ate.workerpool.name`) |
 | `ate.workerpool.workers` | ateapi | up/down counter | live worker count per pool, split by state (`idle`/`assigned`) and sandbox class to provide fleet capacity and saturation at a glance |
 | `ate.actor.lifecycle.operation.duration` | ateapi | histogram | how long each actor operation (create/resume/suspend/pause/delete) takes and whether it failed (`error.type` present = failure, absent = success); labeled by operation, template, pool, sandbox class, and snapshot kind and scope on resume; already-running resume no-ops are not recorded so the histogram tracks actual activations, not router traffic |
 | `ate.scheduler.assignment.duration` | ateapi | histogram | time it takes for an actor to be assigned to a worker, per attempt (version-conflict retries record only the final attempt), with the outcome (`assigned` / `no_free_worker` / `error`) and sandbox class to catch scheduling latency and capacity starvation problems |
@@ -122,6 +126,10 @@ Agent Substrate emits foundational OpenTelemetry system and server metrics to mo
 
 The table lists the OpenTelemetry instrument names. How a name appears in a query depends on the backend (Cloud Monitoring (GMP) / Kind collector).
 
+For `ate.workerpool.desired_workers` and `ate.workerpool.ready_workers`:
+* **Supply-Side Saturation Golden Signal**: Measures whether commanded capacity was delivered. Dedicated instruments match Kubernetes semantic conventions (`k8s.deployment.desired_pods` / `k8s.deployment.available_pods`) because desired + ready is not a disjoint sum.
+* **Autoscaling Control Loop & Anti-Windup**: `desired - ready > 0` sustained beyond a few minutes indicates undelivered capacity due to node pool exhaustion, quota limits, or stuck worker pods, serving as anti-windup input for demand-reactive capacity scaling.
+
 For `atenet.router.route.duration`:
 * `ate.router.outcome` categorizes the route attempt result: `ok`, `cancelled`, `timeout`, `no_capacity`, `lock_conflict`, `not_found`, `unavailable`, `rate_limited`, or `resume_error`.
 * `ate.router.resume` indicates the singleflight execution state of actor resumption: `none` (actor already running), `triggered` (initiated cold activation), or `joined` (parked on in-flight activation).

internal/e2e/collector_metrics.go

@@ -40,6 +40,8 @@ const (
 // slice lands and as more components are wired to push to the collector.
 var PlatformMetricPrefixes = []string{
 	"ate_workerpool_workers",
+	"ate_workerpool_desired_workers",
+	"ate_workerpool_ready_workers",
 	"ate_actor_crashes",
 	"ate_actor_lifecycle_operation_duration",
 	"ate_scheduler_assignment_duration",

internal/e2e/collector_metrics_test.go

@@ -28,6 +28,10 @@ ate_actor_lifecycle_operation_duration_seconds_bucket{ate_actor_operation_name="
 ate_actor_lifecycle_operation_duration_seconds_count{ate_actor_operation_name="resume"} 2
 # TYPE ate_workerpool_workers gauge
 ate_workerpool_workers{ate_workerpool_name="pool-a",ate_worker_state="idle"} 3
+# TYPE ate_workerpool_desired_workers gauge
+ate_workerpool_desired_workers{ate_workerpool_name="pool-a",ate_workerpool_namespace="default"} 5
+# TYPE ate_workerpool_ready_workers gauge
+ate_workerpool_ready_workers{ate_workerpool_name="pool-a",ate_workerpool_namespace="default"} 5
 # TYPE atenet_router_route_duration_seconds histogram
 atenet_router_route_duration_seconds_count 1
 # TYPE atelet_snapshot_size_bytes histogram
@@ -49,6 +53,8 @@ func TestMissingPlatformMetrics(t *testing.T) {
 			prefixes: []string{
 				"ate_actor_lifecycle_operation_duration",
 				"ate_workerpool_workers",
+				"ate_workerpool_desired_workers",
+				"ate_workerpool_ready_workers",
 				"atenet_router_route_duration",
 				"atelet_snapshot_size",
 			},

internal/e2e/suites/metrics/metrics_test.go

@@ -118,9 +118,57 @@ func TestPlatformMetricsEmitted(t *testing.T) {
 		controllerSeen = e2e.CollectorHasService(scrape, "atecontroller") &&
 			strings.Contains(scrape, "controller_runtime_")
 
-		if len(missing) == 0 && ateomSeen {
+		if len(missing) == 0 && ateomSeen && controllerSeen {
 			var errs []string
 
+			// Verify ate_workerpool_desired_workers carries required namespaced attributes.
+			foundDesiredLine := false
+			for _, line := range strings.Split(scrape, "\n") {
+				if strings.HasPrefix(line, "ate_workerpool_desired_workers") {
+					foundDesiredLine = true
+					nsVal := extractLabelValue(line, "ate_workerpool_namespace")
+					poolVal := extractLabelValue(line, "ate_workerpool_name")
+					var lineErrs []string
+					if nsVal == "" {
+						lineErrs = append(lineErrs, "ate_workerpool_namespace label is missing or empty")
+					}
+					if poolVal == "" {
+						lineErrs = append(lineErrs, "ate_workerpool_name label is missing or empty")
+					}
+					if len(lineErrs) > 0 {
+						errs = append(errs, fmt.Sprintf("ate_workerpool_desired_workers validation failed on line %q: %s (Extracted labels: ate_workerpool_namespace=%q, ate_workerpool_name=%q)",
+							line, strings.Join(lineErrs, "; "), nsVal, poolVal))
+					}
+				}
+			}
+			if !foundDesiredLine {
+				errs = append(errs, "ate_workerpool_desired_workers validation failed: metric line not found in collector scrape text (no time series emitted by atecontroller callback)")
+			}
+
+			// Verify ate_workerpool_ready_workers carries required namespaced attributes.
+			foundReadyLine := false
+			for _, line := range strings.Split(scrape, "\n") {
+				if strings.HasPrefix(line, "ate_workerpool_ready_workers") {
+					foundReadyLine = true
+					nsVal := extractLabelValue(line, "ate_workerpool_namespace")
+					poolVal := extractLabelValue(line, "ate_workerpool_name")
+					var lineErrs []string
+					if nsVal == "" {
+						lineErrs = append(lineErrs, "ate_workerpool_namespace label is missing or empty")
+					}
+					if poolVal == "" {
+						lineErrs = append(lineErrs, "ate_workerpool_name label is missing or empty")
+					}
+					if len(lineErrs) > 0 {
+						errs = append(errs, fmt.Sprintf("ate_workerpool_ready_workers validation failed on line %q: %s (Extracted labels: ate_workerpool_namespace=%q, ate_workerpool_name=%q)",
+							line, strings.Join(lineErrs, "; "), nsVal, poolVal))
+					}
+				}
+			}
+			if !foundReadyLine {
+				errs = append(errs, "ate_workerpool_ready_workers validation failed: metric line not found in collector scrape text (no time series emitted by atecontroller callback)")
+			}
+
 			// Verify ate_scheduler_eligible_workers metric carries valid attributes:
 			// - Full labels (namespace, pool, class, constraint) for per-pool candidate lines.
 			// - Necessary base labels (class, constraint) for edge cases when no worker pools match.

manifests/ate-install/generated/ate.dev_workerpools.yaml

@@ -37,6 +37,9 @@ spec:
     - jsonPath: .status.replicas
       name: Replicas
       type: integer
+    - jsonPath: .status.readyReplicas
+      name: Ready
+      type: integer
     - jsonPath: .metadata.creationTimestamp
       name: Age
       type: date
@@ -430,6 +433,11 @@ spec:
           status:
             description: status is the observed state of WorkerPool
             properties:
+              readyReplicas:
+                description: ReadyReplicas is the number of ready worker pods.
+                format: int32
+                minimum: 0
+                type: integer
               replicas:
                 description: Replicas is the total number of worker pods.
                 format: int32

pkg/api/v1alpha1/workerpool_types.go

@@ -96,6 +96,11 @@ type WorkerPoolStatus struct {
 	// +optional
 	Replicas int32 `json:"replicas"`
 
+	// ReadyReplicas is the number of ready worker pods.
+	// +kubebuilder:validation:Minimum=0
+	// +optional
+	ReadyReplicas int32 `json:"readyReplicas,omitempty"`
+
 	// Selector is the label selector for the worker pods.
 	// +optional
 	Selector string `json:"selector,omitempty"`
@@ -110,6 +115,7 @@ type WorkerPoolStatus struct {
 // +kubebuilder:subresource:scale:specpath=.spec.replicas,statuspath=.status.replicas,selectorpath=.status.selector
 // +kubebuilder:printcolumn:name="Desired",type=integer,JSONPath=`.spec.replicas`
 // +kubebuilder:printcolumn:name="Replicas",type=integer,JSONPath=`.status.replicas`
+// +kubebuilder:printcolumn:name="Ready",type=integer,JSONPath=`.status.readyReplicas`
 // +kubebuilder:printcolumn:name="Age",type=date,JSONPath=`.metadata.creationTimestamp`
 type WorkerPool struct {
 	metav1.TypeMeta `json:",inline"`