From fed8cfa03507dd0e510254140a4dea02ae9a3bb9 Mon Sep 17 00:00:00 2001 From: Angela Hu Date: Mon, 10 Aug 2026 10:48:29 -0700 Subject: [PATCH 1/4] feat(atecontroller): emit ate.workerpool.desired_workers and ready_workers metrics --- .../controllers/workerpool_controller.go | 70 ++++++++++++++++++- .../controllers/workerpool_controller_test.go | 50 +++++++++++++ .../generated/ate.dev_workerpools.yaml | 5 ++ pkg/api/v1alpha1/workerpool_types.go | 6 ++ 4 files changed, 129 insertions(+), 2 deletions(-) diff --git a/cmd/atecontroller/internal/controllers/workerpool_controller.go b/cmd/atecontroller/internal/controllers/workerpool_controller.go index c01068d58..77b5b5e34 100644 --- a/cmd/atecontroller/internal/controllers/workerpool_controller.go +++ b/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,67 @@ 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 { + if meter == nil { + meter = otel.Meter("atecontroller") + } + + 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 { + if r.Client == nil { + return nil + } + var list atev1alpha1.WorkerPoolList + if err := r.List(ctx, &list); err != nil { + 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{}). diff --git a/cmd/atecontroller/internal/controllers/workerpool_controller_test.go b/cmd/atecontroller/internal/controllers/workerpool_controller_test.go index 22ea5caf2..e91243242 100644 --- a/cmd/atecontroller/internal/controllers/workerpool_controller_test.go +++ b/cmd/atecontroller/internal/controllers/workerpool_controller_test.go @@ -652,3 +652,53 @@ 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 syncStatus correctly copies +// dep.Status.ReadyReplicas and dep.Status.Replicas into WorkerPool.status. +func TestSyncStatus_ReadyReplicas(t *testing.T) { + r := &WorkerPoolReconciler{ + Client: k8sClient, + } + wp := makeWorkerPool("test-sync-ready", "default", 3, "ateom:v1") + if err := k8sClient.Create(t.Context(), wp); err != nil { + t.Fatalf("create WorkerPool: %v", err) + } + deleteOnCleanup(t, wp) + + dep := &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{Name: wp.Name, Namespace: wp.Namespace}, + Spec: appsv1.DeploymentSpec{ + Selector: &metav1.LabelSelector{MatchLabels: map[string]string{"ate.dev/worker-pool": wp.Name}}, + }, + Status: appsv1.DeploymentStatus{ + Replicas: 3, + ReadyReplicas: 2, + }, + } + + if err := r.syncStatus(t.Context(), wp, dep); err != nil { + t.Fatalf("syncStatus failed: %v", err) + } + + current := &atev1alpha1.WorkerPool{} + if err := k8sClient.Get(t.Context(), types.NamespacedName{Name: wp.Name, Namespace: wp.Namespace}, current); err != nil { + t.Fatalf("get WorkerPool failed: %v", err) + } + if current.Status.ReadyReplicas != 2 { + t.Errorf("got ReadyReplicas=%d, want 2", current.Status.ReadyReplicas) + } + if current.Status.Replicas != 3 { + t.Errorf("got Replicas=%d, want 3", current.Status.Replicas) + } +} + +// TestWorkerPoolMetricsInitialization verifies that InitMetrics registers +// the desired_workers and ready_workers instruments without error. +func TestWorkerPoolMetricsInitialization(t *testing.T) { + r := &WorkerPoolReconciler{ + Client: k8sClient, + } + if err := r.InitMetrics(nil); err != nil { + t.Fatalf("InitMetrics failed: %v", err) + } +} diff --git a/manifests/ate-install/generated/ate.dev_workerpools.yaml b/manifests/ate-install/generated/ate.dev_workerpools.yaml index f19a40894..044317b9a 100644 --- a/manifests/ate-install/generated/ate.dev_workerpools.yaml +++ b/manifests/ate-install/generated/ate.dev_workerpools.yaml @@ -430,6 +430,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 diff --git a/pkg/api/v1alpha1/workerpool_types.go b/pkg/api/v1alpha1/workerpool_types.go index d7a52b485..bbd3134e8 100644 --- a/pkg/api/v1alpha1/workerpool_types.go +++ b/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"` From e7f5232a41d092632608f1c3685097aa5d28170a Mon Sep 17 00:00:00 2001 From: Angela Hu Date: Mon, 10 Aug 2026 10:48:46 -0700 Subject: [PATCH 2/4] test(e2e): assert ate.workerpool.desired_workers and ready_workers in metrics suite --- .../controllers/workerpool_controller.go | 3 - .../controllers/workerpool_controller_test.go | 110 ++++++++++++------ docs/observability.md | 6 + internal/e2e/collector_metrics.go | 2 + internal/e2e/collector_metrics_test.go | 6 + internal/e2e/suites/metrics/metrics_test.go | 50 +++++++- .../generated/ate.dev_workerpools.yaml | 3 + 7 files changed, 138 insertions(+), 42 deletions(-) diff --git a/cmd/atecontroller/internal/controllers/workerpool_controller.go b/cmd/atecontroller/internal/controllers/workerpool_controller.go index 77b5b5e34..dc0d6f655 100644 --- a/cmd/atecontroller/internal/controllers/workerpool_controller.go +++ b/cmd/atecontroller/internal/controllers/workerpool_controller.go @@ -175,9 +175,6 @@ func (r *WorkerPoolReconciler) InitMetrics(meter metric.Meter) error { _, err = meter.RegisterCallback( func(ctx context.Context, obs metric.Observer) error { - if r.Client == nil { - return nil - } var list atev1alpha1.WorkerPoolList if err := r.List(ctx, &list); err != nil { return nil diff --git a/cmd/atecontroller/internal/controllers/workerpool_controller_test.go b/cmd/atecontroller/internal/controllers/workerpool_controller_test.go index e91243242..cb1e8a917 100644 --- a/cmd/atecontroller/internal/controllers/workerpool_controller_test.go +++ b/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" ) @@ -653,52 +659,80 @@ func eventually(t *testing.T, condition func(ctx context.Context) (bool, error)) } } -// TestSyncStatus_ReadyReplicas verifies that syncStatus correctly copies -// dep.Status.ReadyReplicas and dep.Status.Replicas into WorkerPool.status. +// TestSyncStatus_ReadyReplicas verifies that a Deployment reporting 3 replicas +// with 2 ready propagates to WorkerPool.status. func TestSyncStatus_ReadyReplicas(t *testing.T) { - r := &WorkerPoolReconciler{ - Client: k8sClient, - } + t.Parallel() + ctx := t.Context() wp := makeWorkerPool("test-sync-ready", "default", 3, "ateom:v1") - if err := k8sClient.Create(t.Context(), wp); err != nil { + if err := k8sClient.Create(ctx, wp); err != nil { t.Fatalf("create WorkerPool: %v", err) } deleteOnCleanup(t, wp) - - dep := &appsv1.Deployment{ - ObjectMeta: metav1.ObjectMeta{Name: wp.Name, Namespace: wp.Namespace}, - Spec: appsv1.DeploymentSpec{ - Selector: &metav1.LabelSelector{MatchLabels: map[string]string{"ate.dev/worker-pool": wp.Name}}, - }, - Status: appsv1.DeploymentStatus{ - Replicas: 3, - ReadyReplicas: 2, - }, - } - - if err := r.syncStatus(t.Context(), wp, dep); err != nil { - t.Fatalf("syncStatus failed: %v", err) - } - - current := &atev1alpha1.WorkerPool{} - if err := k8sClient.Get(t.Context(), types.NamespacedName{Name: wp.Name, Namespace: wp.Namespace}, current); err != nil { - t.Fatalf("get WorkerPool failed: %v", err) - } - if current.Status.ReadyReplicas != 2 { - t.Errorf("got ReadyReplicas=%d, want 2", current.Status.ReadyReplicas) - } - if current.Status.Replicas != 3 { - t.Errorf("got Replicas=%d, want 3", current.Status.Replicas) - } + 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 + }) } -// TestWorkerPoolMetricsInitialization verifies that InitMetrics registers -// the desired_workers and ready_workers instruments without error. -func TestWorkerPoolMetricsInitialization(t *testing.T) { +// 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: k8sClient, + 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 err := r.InitMetrics(nil); err != nil { - t.Fatalf("InitMetrics failed: %v", err) + if !reflect.DeepEqual(got, want) { + t.Errorf("observed %v, want %v", got, want) } } diff --git a/docs/observability.md b/docs/observability.md index 231c4c4d0..bfffb5a87 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -114,6 +114,8 @@ 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 | observable updowncounter | commanded worker pod capacity (`spec.replicas`) per WorkerPool (labels `ate.workerpool.namespace`, `ate.workerpool.name`, unit `{worker}`) | +| `ate.workerpool.ready_workers` | atecontroller | observable updowncounter | delivered ready worker pod capacity (`status.readyReplicas`) per WorkerPool (labels `ate.workerpool.namespace`, `ate.workerpool.name`, unit `{worker}`) | | `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 +124,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). diff --git a/internal/e2e/collector_metrics.go b/internal/e2e/collector_metrics.go index cae679bc3..48be77f18 100644 --- a/internal/e2e/collector_metrics.go +++ b/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", diff --git a/internal/e2e/collector_metrics_test.go b/internal/e2e/collector_metrics_test.go index 11ac71cae..85973e5e8 100644 --- a/internal/e2e/collector_metrics_test.go +++ b/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", }, diff --git a/internal/e2e/suites/metrics/metrics_test.go b/internal/e2e/suites/metrics/metrics_test.go index 344c44ce5..0a27834fa 100644 --- a/internal/e2e/suites/metrics/metrics_test.go +++ b/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. diff --git a/manifests/ate-install/generated/ate.dev_workerpools.yaml b/manifests/ate-install/generated/ate.dev_workerpools.yaml index 044317b9a..5b2047e31 100644 --- a/manifests/ate-install/generated/ate.dev_workerpools.yaml +++ b/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 From c39e29f922ce5944207b1c030d975efd33c81f01 Mon Sep 17 00:00:00 2001 From: Angela Hu Date: Tue, 11 Aug 2026 07:11:29 -0700 Subject: [PATCH 3/4] amend error msg of workerpool metrics --- .../internal/controllers/workerpool_controller.go | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/cmd/atecontroller/internal/controllers/workerpool_controller.go b/cmd/atecontroller/internal/controllers/workerpool_controller.go index dc0d6f655..7c962d62d 100644 --- a/cmd/atecontroller/internal/controllers/workerpool_controller.go +++ b/cmd/atecontroller/internal/controllers/workerpool_controller.go @@ -149,10 +149,6 @@ func (r *WorkerPoolReconciler) syncStatus(ctx context.Context, wp *atev1alpha1.W // 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 { - if meter == nil { - meter = otel.Meter("atecontroller") - } - desiredWorkers, err := meter.Int64ObservableUpDownCounter( "ate.workerpool.desired_workers", metric.WithUnit("{worker}"), @@ -177,6 +173,7 @@ func (r *WorkerPoolReconciler) InitMetrics(meter metric.Meter) error { 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 { From 901b6166a891a592856114cfeb6eda21b915b763 Mon Sep 17 00:00:00 2001 From: Angela <113480613+Angelawork@users.noreply.github.com> Date: Tue, 11 Aug 2026 11:36:54 -0400 Subject: [PATCH 4/4] Update docs/observability.md to align with word choice Co-authored-by: Jeff Luo --- docs/observability.md | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/docs/observability.md b/docs/observability.md index bfffb5a87..5e1c93d9f 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -114,8 +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 | observable updowncounter | commanded worker pod capacity (`spec.replicas`) per WorkerPool (labels `ate.workerpool.namespace`, `ate.workerpool.name`, unit `{worker}`) | -| `ate.workerpool.ready_workers` | atecontroller | observable updowncounter | delivered ready worker pod capacity (`status.readyReplicas`) per WorkerPool (labels `ate.workerpool.namespace`, `ate.workerpool.name`, unit `{worker}`) | +| `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 |