feat(otel): add ate.workerpool.workers metric with e2e suite

cmd/ateapi/internal/controlapi/metrics.go

@@ -0,0 +1,87 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package controlapi
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/agent-substrate/substrate/internal/ateattr"
+	atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1"
+	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
+	"go.opentelemetry.io/otel/metric"
+	"k8s.io/apimachinery/pkg/labels"
+)
+
+const workerpoolWorkersMetric = "ate.workerpool.workers"
+
+// RegisterWorkerCount wires the ate.workerpool.workers observable against workers
+// (workercache.Cache.Workers in prod) and listPools (a WorkerPool lister's List,
+// used to seed zero-valued series). Worker counts are spatially summable (over
+// states = pool size, over pools = fleet), which is the UpDownCounter contract; a
+// gauge would be wrong for a value meant to be summed.
+func RegisterWorkerCount(meter metric.Meter, workers func() ([]*ateapipb.Worker, error), listPools func(labels.Selector) ([]*atev1alpha1.WorkerPool, error)) error {
+	counter, err := meter.Int64ObservableUpDownCounter(
+		workerpoolWorkersMetric,
+		metric.WithUnit("{worker}"),
+		metric.WithDescription("Number of workers by pool, worker state, and sandbox class."),
+	)
+	if err != nil {
+		return fmt.Errorf("create %s updowncounter: %w", workerpoolWorkersMetric, err)
+	}
+
+	_, err = meter.RegisterCallback(func(_ context.Context, o metric.Observer) error {
+		ws, err := workers()
+		if err != nil {
+			// Worker cache unavailable (warmup/reconnect): skip the whole observation.
+			return nil
+		}
+		type key struct{ pool, state, class string }
+		tally := make(map[key]int64)
+		// Seed both states at 0 for every known pool so a saturated or empty pool
+		// reports 0, not an absent series that breaks idle==0 alerts.
+		if listPools != nil {
+			if pools, err := listPools(labels.Everything()); err == nil {
+				for _, p := range pools {
+					class := string(p.Spec.SandboxClass)
+					if class == "" {
+						class = string(atev1alpha1.SandboxClassGvisor)
+					}
+					tally[key{p.Name, ateattr.WorkerStateIdle, class}] = 0
+					tally[key{p.Name, ateattr.WorkerStateAssigned, class}] = 0
+				}
+			}
+		}
+		for _, w := range ws {
+			state := ateattr.WorkerStateIdle
+			if w.GetAssignment() != nil {
+				state = ateattr.WorkerStateAssigned
+			}
+			tally[key{w.GetWorkerPool(), state, w.GetSandboxClass()}]++
+		}
+		for k, n := range tally {
+			o.ObserveInt64(counter, n, metric.WithAttributes(
+				ateattr.WorkerPoolNameKey.String(k.pool),
+				ateattr.WorkerStateKey.String(k.state),
+				ateattr.SandboxClassKey.String(k.class),
+			))
+		}
+		return nil
+	}, counter)
+	if err != nil {
+		return fmt.Errorf("register %s callback: %w", workerpoolWorkersMetric, err)
+	}
+	return nil
+}

cmd/ateapi/internal/controlapi/metrics_test.go

@@ -0,0 +1,184 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package controlapi
+
+import (
+	"context"
+	"testing"
+
+	"github.com/agent-substrate/substrate/internal/ateattr"
+	atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1"
+	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+)
+
+// newWorkerCountReader registers ate.workerpool.workers against a local
+// ManualReader-backed provider so tests stay parallel-safe and never touch the
+// global meter provider.
+func newWorkerCountReader(t *testing.T, workers func() ([]*ateapipb.Worker, error), listPools func(labels.Selector) ([]*atev1alpha1.WorkerPool, error)) *sdkmetric.ManualReader {
+	t.Helper()
+	reader := sdkmetric.NewManualReader()
+	mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
+	if err := RegisterWorkerCount(mp.Meter("ateapi"), workers, listPools); err != nil {
+		t.Fatalf("RegisterWorkerCount: %v", err)
+	}
+	return reader
+}
+
+func noWorkers() ([]*ateapipb.Worker, error) { return nil, nil }
+
+func noPools(labels.Selector) ([]*atev1alpha1.WorkerPool, error) { return nil, nil }
+
+func collectMetric(t *testing.T, reader *sdkmetric.ManualReader, name string) (metricdata.Metrics, bool) {
+	t.Helper()
+	var rm metricdata.ResourceMetrics
+	if err := reader.Collect(context.Background(), &rm); err != nil {
+		t.Fatalf("collect: %v", err)
+	}
+	for _, sm := range rm.ScopeMetrics {
+		for _, m := range sm.Metrics {
+			if m.Name == name {
+				return m, true
+			}
+		}
+	}
+	return metricdata.Metrics{}, false
+}
+
+func mustMetric(t *testing.T, reader *sdkmetric.ManualReader, name string) metricdata.Metrics {
+	t.Helper()
+	m, ok := collectMetric(t, reader, name)
+	if !ok {
+		t.Fatalf("metric %q not collected", name)
+	}
+	return m
+}
+
+func worker(pool, class string, assigned bool) *ateapipb.Worker {
+	w := &ateapipb.Worker{WorkerPool: pool, SandboxClass: class}
+	if assigned {
+		w.Assignment = &ateapipb.Assignment{}
+	}
+	return w
+}
+
+func TestWorkerCountTally(t *testing.T) {
+	workers := func() ([]*ateapipb.Worker, error) {
+		return []*ateapipb.Worker{
+			worker("pool-a", "gvisor", false),
+			worker("pool-a", "gvisor", false),
+			worker("pool-a", "gvisor", true),
+			worker("pool-b", "microvm", false),
+		}, nil
+	}
+	reader := newWorkerCountReader(t, workers, noPools)
+
+	m := mustMetric(t, reader, workerpoolWorkersMetric)
+	if m.Unit != "{worker}" {
+		t.Errorf("unit = %q, want {worker}", m.Unit)
+	}
+	sum, ok := m.Data.(metricdata.Sum[int64])
+	if !ok {
+		t.Fatalf("data type = %T, want Sum[int64]", m.Data)
+	}
+	if sum.IsMonotonic {
+		t.Errorf("IsMonotonic = true, want false (updowncounter, not counter)")
+	}
+
+	type key struct{ pool, state, class string }
+	got := make(map[key]int64)
+	for _, dp := range sum.DataPoints {
+		pool, _ := dp.Attributes.Value(ateattr.WorkerPoolNameKey)
+		state, _ := dp.Attributes.Value(ateattr.WorkerStateKey)
+		class, _ := dp.Attributes.Value(ateattr.SandboxClassKey)
+		got[key{pool.AsString(), state.AsString(), class.AsString()}] = dp.Value
+	}
+	want := map[key]int64{
+		{"pool-a", ateattr.WorkerStateIdle, "gvisor"}:     2,
+		{"pool-a", ateattr.WorkerStateAssigned, "gvisor"}: 1,
+		{"pool-b", ateattr.WorkerStateIdle, "microvm"}:    1,
+	}
+	if len(got) != len(want) {
+		t.Fatalf("got %d series, want %d: %v", len(got), len(want), got)
+	}
+	for k, v := range want {
+		if got[k] != v {
+			t.Errorf("series %v = %d, want %d", k, got[k], v)
+		}
+	}
+}
+
+// TestWorkerCountSkipsWhenCacheNotReady asserts the callback emits nothing while
+// the cache is warming up, so we never publish misleading zero-valued points.
+func TestWorkerCountSkipsWhenCacheNotReady(t *testing.T) {
+	notReady := func() ([]*ateapipb.Worker, error) {
+		return nil, context.DeadlineExceeded
+	}
+	reader := newWorkerCountReader(t, notReady, noPools)
+
+	if _, ok := collectMetric(t, reader, workerpoolWorkersMetric); ok {
+		t.Errorf("%s was collected, want no datapoints while cache not ready", workerpoolWorkersMetric)
+	}
+}
+
+func workerPool(name string, class atev1alpha1.SandboxClass) *atev1alpha1.WorkerPool {
+	return &atev1alpha1.WorkerPool{
+		ObjectMeta: metav1.ObjectMeta{Name: name},
+		Spec:       atev1alpha1.WorkerPoolSpec{SandboxClass: class},
+	}
+}
+
+// TestWorkerCountSeedsZeroForKnownPools covers the saturation cases: a pool whose
+// only state has no workers, and a pool with no workers at all, both report 0
+// rather than an absent series. Empty pool class defaults to gvisor.
+func TestWorkerCountSeedsZeroForKnownPools(t *testing.T) {
+	pools := func(labels.Selector) ([]*atev1alpha1.WorkerPool, error) {
+		return []*atev1alpha1.WorkerPool{
+			workerPool("pool-a", ""),
+			workerPool("pool-c", atev1alpha1.SandboxClassMicroVM),
+		}, nil
+	}
+	workers := func() ([]*ateapipb.Worker, error) {
+		return []*ateapipb.Worker{worker("pool-a", "gvisor", true)}, nil
+	}
+	reader := newWorkerCountReader(t, workers, pools)
+
+	sum := mustMetric(t, reader, workerpoolWorkersMetric).Data.(metricdata.Sum[int64])
+	type key struct{ pool, state, class string }
+	got := make(map[key]int64)
+	for _, dp := range sum.DataPoints {
+		pool, _ := dp.Attributes.Value(ateattr.WorkerPoolNameKey)
+		state, _ := dp.Attributes.Value(ateattr.WorkerStateKey)
+		class, _ := dp.Attributes.Value(ateattr.SandboxClassKey)
+		got[key{pool.AsString(), state.AsString(), class.AsString()}] = dp.Value
+	}
+	want := map[key]int64{
+		{"pool-a", ateattr.WorkerStateIdle, "gvisor"}:      0,
+		{"pool-a", ateattr.WorkerStateAssigned, "gvisor"}:  1,
+		{"pool-c", ateattr.WorkerStateIdle, "microvm"}:     0,
+		{"pool-c", ateattr.WorkerStateAssigned, "microvm"}: 0,
+	}
+	if len(got) != len(want) {
+		t.Fatalf("got %d series, want %d: %v", len(got), len(want), got)
+	}
+	for k, v := range want {
+		if gv, ok := got[k]; !ok || gv != v {
+			t.Errorf("series %v = %d (present=%v), want %d", k, gv, ok, v)
+		}
+	}
+}

cmd/ateapi/main.go

@@ -44,6 +44,7 @@ import (
 	"github.com/redis/go-redis/v9"
 	"github.com/spf13/pflag"
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
+	"go.opentelemetry.io/otel"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 	"golang.org/x/oauth2/google"
 	"google.golang.org/grpc"
@@ -151,6 +152,10 @@ func main() {
 	ateletPodInformerFactory.WaitForCacheSync(stopCh)
 	ateFactory.WaitForCacheSync(stopCh)
 
+	if err := controlapi.RegisterWorkerCount(otel.Meter("ateapi"), workerCache.Workers, workerPoolLister.List); err != nil {
+		serverboot.Fatal(ctx, "Failed to register worker-count metric", err)
+	}
+
 	dialer := controlapi.NewAteletDialer(workerPodInformer.GetIndexer(), ateletPodInformer.GetIndexer())
 	sm := controlapi.NewService(redisPersistence, workerCache, actorTemplateLister, workerPoolLister, sandboxConfigLister, dialer, clientset)
 

internal/ateattr/ateattr.go

@@ -12,10 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-// Package ateattr projects an Actor onto substrate's ate.* span attributes.
-// Identity is a span-level subject attribute (the producer is the substrate
-// component, the actor is the subject), so it belongs on spans rather than the
-// resource, and uses substrate's own ate.* namespace rather than service.*.
+// Package ateattr is the single source of truth for substrate's ate.* telemetry
+// attributes: the identity keys stamped on spans/logs, and the bounded value
+// sets used as metric labels. Centralizing them keeps a key (and value) meaning
+// the same thing across every signal and binary.
 package ateattr
 
 import (
@@ -43,6 +43,24 @@ const (
 	ActorVersionKey      = attribute.Key("ate.actor.version")
 )
 
+// Metric-label keys: the only ate.* attributes allowed on metric datapoints,
+// each with a small bounded value set. High-cardinality identity (actor
+// name/uid, atespace) is absent by design; it belongs on spans and logs.
+// WorkerStateKey stays worker-rooted rather than nesting under the pool so it
+// can grow siblings.
+const (
+	WorkerPoolNameKey = attribute.Key("ate.workerpool.name")
+	WorkerStateKey    = attribute.Key("ate.worker.state")
+	SandboxClassKey   = attribute.Key("ate.sandbox.class")
+)
+
+// Values for WorkerStateKey. Only idle and assigned are representable today;
+// starting and unhealthy workers are not modeled in the cache.
+const (
+	WorkerStateIdle     = "idle"
+	WorkerStateAssigned = "assigned"
+)
+
 // ActorRefAttributes returns the subset knowable before the Actor record
 // resolves: only the (atespace, name) the request addresses. The uid and version
 // are server-assigned and unknown until the record loads, so they are omitted.

internal/ateattr/ateattr_test.go

@@ -135,3 +135,30 @@ func TestActorRefAttributes(t *testing.T) {
 		})
 	}
 }
+
+// TestKeySpellings pins the wire spelling of every key. Renaming one silently
+// breaks dashboards, alerts, and the contract between ateapi and atelet, so a
+// drift must fail here rather than in production.
+func TestKeySpellings(t *testing.T) {
+	tests := []struct {
+		key  attribute.Key
+		want string
+	}{
+		{AtespaceKey, "ate.atespace"},
+		{ActorNameKey, "ate.actor.name"},
+		{ActorUIDKey, "ate.actor.uid"},
+		{TemplateNameKey, "ate.template.name"},
+		{TemplateNamespaceKey, "ate.template.namespace"},
+		{ActorVersionKey, "ate.actor.version"},
+		{WorkerPoolNameKey, "ate.workerpool.name"},
+		{WorkerStateKey, "ate.worker.state"},
+		{SandboxClassKey, "ate.sandbox.class"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.want, func(t *testing.T) {
+			if string(tt.key) != tt.want {
+				t.Errorf("key = %q, want %q", string(tt.key), tt.want)
+			}
+		})
+	}
+}