From ae9b3a08ef85b19b94fb3b5d4790ce53ed4b1963 Mon Sep 17 00:00:00 2001 From: krisztianfekete Date: Wed, 22 Jul 2026 17:44:42 +0200 Subject: [PATCH 1/5] feat(otel): add platform-metric label keys to the shared registry --- internal/ateattr/ateattr.go | 83 ++++++++++++++++++++++++++++++-- internal/ateattr/ateattr_test.go | 51 ++++++++++++++++++++ 2 files changed, 130 insertions(+), 4 deletions(-) diff --git a/internal/ateattr/ateattr.go b/internal/ateattr/ateattr.go index eb5eb152e..345d3baca 100644 --- a/internal/ateattr/ateattr.go +++ b/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,81 @@ 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. +// ActorOperationNameKey follows the registry's *.operation.name pattern +// (db.operation.name, gen_ai.operation.name). WorkerStateKey stays worker-rooted +// rather than nesting under the pool so it can grow siblings. +const ( + ActorOperationNameKey = attribute.Key("ate.actor.operation.name") + WorkerPoolNameKey = attribute.Key("ate.workerpool.name") + WorkerStateKey = attribute.Key("ate.worker.state") + SandboxClassKey = attribute.Key("ate.sandbox.class") + SnapshotKindKey = attribute.Key("ate.snapshot.kind") + SnapshotPhaseKey = attribute.Key("ate.snapshot.phase") + SchedulerOutcomeKey = attribute.Key("ate.scheduler.outcome") +) + +// ErrorTypeKey is the OTel registry attribute, reused verbatim (not aliased into +// ate.*): failures are reported on the same instrument via this key, its absence +// meaning success, never as a parallel _failures counter. +const ErrorTypeKey = attribute.Key("error.type") + +// Values for ActorOperationNameKey. +const ( + OperationCreate = "create" + OperationResume = "resume" + OperationSuspend = "suspend" + OperationPause = "pause" + OperationDelete = "delete" +) + +// 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" +) + +// Values for SchedulerOutcomeKey. NoFreeWorker is a capacity signal, not an +// error, so it is a distinct outcome rather than an error.type value. +const ( + SchedulerOutcomeAssigned = "assigned" + SchedulerOutcomeNoFreeWorker = "no_free_worker" + SchedulerOutcomeError = "error" +) + +// Values for SnapshotKindKey. Boot is not a restore, so it appears only on the +// ateapi lifecycle histogram, never on the atelet restore histogram. +const ( + SnapshotKindGolden = "golden" + SnapshotKindLatest = "latest" + SnapshotKindBoot = "boot" + SnapshotKindUnknown = "unknown" +) + +// ClampRestoreSnapshotKind bounds an untrusted kind before it becomes a metric +// label, so a caller cannot inflate cardinality. +func ClampRestoreSnapshotKind(kind string) string { + switch kind { + case SnapshotKindGolden, SnapshotKindLatest: + return kind + default: + return SnapshotKindUnknown + } +} + +// Values for SnapshotPhaseKey: restore phases (download, oci_unpack, +// ateom_restore) and checkpoint phases (checkpoint, upload) share the key. +const ( + SnapshotPhaseDownload = "download" + SnapshotPhaseOCIUnpack = "oci_unpack" + SnapshotPhaseAteomRestore = "ateom_restore" + SnapshotPhaseCheckpoint = "checkpoint" + SnapshotPhaseUpload = "upload" +) + // 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. diff --git a/internal/ateattr/ateattr_test.go b/internal/ateattr/ateattr_test.go index 6daa3d521..01cda4a97 100644 --- a/internal/ateattr/ateattr_test.go +++ b/internal/ateattr/ateattr_test.go @@ -135,3 +135,54 @@ 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"}, + {ActorOperationNameKey, "ate.actor.operation.name"}, + {WorkerPoolNameKey, "ate.workerpool.name"}, + {WorkerStateKey, "ate.worker.state"}, + {SandboxClassKey, "ate.sandbox.class"}, + {SnapshotKindKey, "ate.snapshot.kind"}, + {SnapshotPhaseKey, "ate.snapshot.phase"}, + {SchedulerOutcomeKey, "ate.scheduler.outcome"}, + {ErrorTypeKey, "error.type"}, + } + 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) + } + }) + } +} + +func TestClampRestoreSnapshotKind(t *testing.T) { + tests := []struct { + in, want string + }{ + {SnapshotKindGolden, SnapshotKindGolden}, + {SnapshotKindLatest, SnapshotKindLatest}, + {SnapshotKindBoot, SnapshotKindUnknown}, + {"", SnapshotKindUnknown}, + {"$(rm -rf)", SnapshotKindUnknown}, + } + for _, tt := range tests { + t.Run(tt.in, func(t *testing.T) { + if got := ClampRestoreSnapshotKind(tt.in); got != tt.want { + t.Errorf("ClampRestoreSnapshotKind(%q) = %q, want %q", tt.in, got, tt.want) + } + }) + } +} From 77968c88d3c682eff73f202cc96cd5f424d46580 Mon Sep 17 00:00:00 2001 From: krisztianfekete Date: Thu, 23 Jul 2026 11:31:24 +0200 Subject: [PATCH 2/5] feat(otel): rework into worker-pool saturation metric with e2e suite --- cmd/ateapi/internal/controlapi/metrics.go | 87 +++++++++ .../internal/controlapi/metrics_test.go | 184 ++++++++++++++++++ cmd/ateapi/main.go | 5 + internal/ateattr/ateattr.go | 67 +------ internal/ateattr/ateattr_test.go | 24 --- internal/e2e/collector_metrics.go | 135 +++++++++++++ internal/e2e/collector_metrics_test.go | 111 +++++++++++ internal/e2e/portforward.go | 100 ++++++++++ internal/e2e/router_client.go | 78 +------- internal/e2e/suites/metrics/metrics_test.go | 141 ++++++++++++++ internal/e2e/suites/metrics/testmain_test.go | 24 +++ 11 files changed, 800 insertions(+), 156 deletions(-) create mode 100644 cmd/ateapi/internal/controlapi/metrics.go create mode 100644 cmd/ateapi/internal/controlapi/metrics_test.go create mode 100644 internal/e2e/collector_metrics.go create mode 100644 internal/e2e/collector_metrics_test.go create mode 100644 internal/e2e/portforward.go create mode 100644 internal/e2e/suites/metrics/metrics_test.go create mode 100644 internal/e2e/suites/metrics/testmain_test.go diff --git a/cmd/ateapi/internal/controlapi/metrics.go b/cmd/ateapi/internal/controlapi/metrics.go new file mode 100644 index 000000000..e60a44145 --- /dev/null +++ b/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 +} diff --git a/cmd/ateapi/internal/controlapi/metrics_test.go b/cmd/ateapi/internal/controlapi/metrics_test.go new file mode 100644 index 000000000..fefe006dc --- /dev/null +++ b/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) + } + } +} diff --git a/cmd/ateapi/main.go b/cmd/ateapi/main.go index c08feb51c..8843c9216 100644 --- a/cmd/ateapi/main.go +++ b/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) diff --git a/internal/ateattr/ateattr.go b/internal/ateattr/ateattr.go index 345d3baca..a2cd067ba 100644 --- a/internal/ateattr/ateattr.go +++ b/internal/ateattr/ateattr.go @@ -46,31 +46,12 @@ const ( // 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. -// ActorOperationNameKey follows the registry's *.operation.name pattern -// (db.operation.name, gen_ai.operation.name). WorkerStateKey stays worker-rooted -// rather than nesting under the pool so it can grow siblings. +// WorkerStateKey stays worker-rooted rather than nesting under the pool so it +// can grow siblings. const ( - ActorOperationNameKey = attribute.Key("ate.actor.operation.name") - WorkerPoolNameKey = attribute.Key("ate.workerpool.name") - WorkerStateKey = attribute.Key("ate.worker.state") - SandboxClassKey = attribute.Key("ate.sandbox.class") - SnapshotKindKey = attribute.Key("ate.snapshot.kind") - SnapshotPhaseKey = attribute.Key("ate.snapshot.phase") - SchedulerOutcomeKey = attribute.Key("ate.scheduler.outcome") -) - -// ErrorTypeKey is the OTel registry attribute, reused verbatim (not aliased into -// ate.*): failures are reported on the same instrument via this key, its absence -// meaning success, never as a parallel _failures counter. -const ErrorTypeKey = attribute.Key("error.type") - -// Values for ActorOperationNameKey. -const ( - OperationCreate = "create" - OperationResume = "resume" - OperationSuspend = "suspend" - OperationPause = "pause" - OperationDelete = "delete" + 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; @@ -80,44 +61,6 @@ const ( WorkerStateAssigned = "assigned" ) -// Values for SchedulerOutcomeKey. NoFreeWorker is a capacity signal, not an -// error, so it is a distinct outcome rather than an error.type value. -const ( - SchedulerOutcomeAssigned = "assigned" - SchedulerOutcomeNoFreeWorker = "no_free_worker" - SchedulerOutcomeError = "error" -) - -// Values for SnapshotKindKey. Boot is not a restore, so it appears only on the -// ateapi lifecycle histogram, never on the atelet restore histogram. -const ( - SnapshotKindGolden = "golden" - SnapshotKindLatest = "latest" - SnapshotKindBoot = "boot" - SnapshotKindUnknown = "unknown" -) - -// ClampRestoreSnapshotKind bounds an untrusted kind before it becomes a metric -// label, so a caller cannot inflate cardinality. -func ClampRestoreSnapshotKind(kind string) string { - switch kind { - case SnapshotKindGolden, SnapshotKindLatest: - return kind - default: - return SnapshotKindUnknown - } -} - -// Values for SnapshotPhaseKey: restore phases (download, oci_unpack, -// ateom_restore) and checkpoint phases (checkpoint, upload) share the key. -const ( - SnapshotPhaseDownload = "download" - SnapshotPhaseOCIUnpack = "oci_unpack" - SnapshotPhaseAteomRestore = "ateom_restore" - SnapshotPhaseCheckpoint = "checkpoint" - SnapshotPhaseUpload = "upload" -) - // 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. diff --git a/internal/ateattr/ateattr_test.go b/internal/ateattr/ateattr_test.go index 01cda4a97..794d3881c 100644 --- a/internal/ateattr/ateattr_test.go +++ b/internal/ateattr/ateattr_test.go @@ -150,14 +150,9 @@ func TestKeySpellings(t *testing.T) { {TemplateNameKey, "ate.template.name"}, {TemplateNamespaceKey, "ate.template.namespace"}, {ActorVersionKey, "ate.actor.version"}, - {ActorOperationNameKey, "ate.actor.operation.name"}, {WorkerPoolNameKey, "ate.workerpool.name"}, {WorkerStateKey, "ate.worker.state"}, {SandboxClassKey, "ate.sandbox.class"}, - {SnapshotKindKey, "ate.snapshot.kind"}, - {SnapshotPhaseKey, "ate.snapshot.phase"}, - {SchedulerOutcomeKey, "ate.scheduler.outcome"}, - {ErrorTypeKey, "error.type"}, } for _, tt := range tests { t.Run(tt.want, func(t *testing.T) { @@ -167,22 +162,3 @@ func TestKeySpellings(t *testing.T) { }) } } - -func TestClampRestoreSnapshotKind(t *testing.T) { - tests := []struct { - in, want string - }{ - {SnapshotKindGolden, SnapshotKindGolden}, - {SnapshotKindLatest, SnapshotKindLatest}, - {SnapshotKindBoot, SnapshotKindUnknown}, - {"", SnapshotKindUnknown}, - {"$(rm -rf)", SnapshotKindUnknown}, - } - for _, tt := range tests { - t.Run(tt.in, func(t *testing.T) { - if got := ClampRestoreSnapshotKind(tt.in); got != tt.want { - t.Errorf("ClampRestoreSnapshotKind(%q) = %q, want %q", tt.in, got, tt.want) - } - }) - } -} diff --git a/internal/e2e/collector_metrics.go b/internal/e2e/collector_metrics.go new file mode 100644 index 000000000..d50980c46 --- /dev/null +++ b/internal/e2e/collector_metrics.go @@ -0,0 +1,135 @@ +// 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 e2e + +import ( + "context" + "fmt" + "io" + "net/http" + "strings" + "time" + + "github.com/agent-substrate/substrate/internal/ateclient" + "k8s.io/client-go/kubernetes" +) + +const ( + collectorNamespace = "otel-system" + collectorService = "opentelemetry-collector" + collectorPromPort = 8889 +) + +// PlatformMetricPrefixes are the Prometheus metric-name prefixes (OTLP dots +// mapped to underscores) the substrate platform must emit once a full actor +// lifecycle has run. The Collector's prometheus exporter appends unit and type +// suffixes (e.g. _seconds_bucket, _bytes_count), so matching is by prefix. +// This slice grows as each platform-metric slice lands; today it pins the +// worker-count instrument plus the two hand-written metrics that predate this +// work, guarding all three against silent regression. +var PlatformMetricPrefixes = []string{ + "ate_workerpool_workers", + "atenet_router_route_duration", + "atelet_snapshot_size", +} + +// ScrapeCollectorMetrics port-forwards the kind stack's OTel Collector and reads +// its Prometheus exporter surface, returning the raw exposition text. +func ScrapeCollectorMetrics(ctx context.Context) (string, error) { + config, err := ateclient.LoadConfig(KubeConfig, KubeContext) + if err != nil { + return "", fmt.Errorf("loading kubeconfig: %w", err) + } + clientset, err := kubernetes.NewForConfig(config) + if err != nil { + return "", fmt.Errorf("creating k8s client: %w", err) + } + + pod, _, err := firstReadyPodForService(ctx, clientset, collectorNamespace, collectorService) + if err != nil { + return "", err + } + localPort, stop, err := podPortForward(ctx, config, clientset, collectorNamespace, pod.Name, collectorPromPort) + if err != nil { + return "", err + } + defer stop() + + url := fmt.Sprintf("http://127.0.0.1:%d/metrics", localPort) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return "", err + } + resp, err := (&http.Client{Timeout: 30 * time.Second}).Do(req) + if err != nil { + return "", fmt.Errorf("scraping collector metrics: %w", err) + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + return "", fmt.Errorf("reading collector metrics: %w", err) + } + if resp.StatusCode != http.StatusOK { + return "", fmt.Errorf("collector metrics returned %d: %s", resp.StatusCode, body) + } + return string(body), nil +} + +// MissingPlatformMetrics returns the prefixes with no matching series in the +// Prometheus exposition text. A metric matches when its name equals a prefix or +// begins with prefix+"_"; the underscore boundary stops "ate_actor_restore" from +// matching an unrelated "ate_actor_restored". +func MissingPlatformMetrics(scrape string, prefixes []string) []string { + present := make(map[string]bool, len(prefixes)) + for _, line := range strings.Split(scrape, "\n") { + name := metricNameFromLine(line) + if name == "" { + continue + } + for _, p := range prefixes { + if name == p || strings.HasPrefix(name, p+"_") { + present[p] = true + } + } + } + var missing []string + for _, p := range prefixes { + if !present[p] { + missing = append(missing, p) + } + } + return missing +} + +// metricNameFromLine extracts the metric name from one exposition line, handling +// the "# HELP name ...", "# TYPE name type", and "name{labels} value" forms. +// It returns "" for blank lines and other comments. +func metricNameFromLine(line string) string { + line = strings.TrimSpace(line) + if line == "" { + return "" + } + if strings.HasPrefix(line, "#") { + fields := strings.Fields(line) + if len(fields) >= 3 && (fields[1] == "HELP" || fields[1] == "TYPE") { + return fields[2] + } + return "" + } + if i := strings.IndexAny(line, "{ \t"); i >= 0 { + return line[:i] + } + return line +} diff --git a/internal/e2e/collector_metrics_test.go b/internal/e2e/collector_metrics_test.go new file mode 100644 index 000000000..11ac71cae --- /dev/null +++ b/internal/e2e/collector_metrics_test.go @@ -0,0 +1,111 @@ +// 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 e2e + +import ( + "sort" + "testing" +) + +// A trimmed sample of what the Collector's prometheus exporter emits: HELP/TYPE +// comments plus suffixed series lines, so the matcher is exercised against the +// real exposition shape. +const sampleScrape = `# HELP ate_actor_lifecycle_operation_duration_seconds Duration of an actor lifecycle operation. +# TYPE ate_actor_lifecycle_operation_duration_seconds histogram +ate_actor_lifecycle_operation_duration_seconds_bucket{ate_actor_operation_name="resume",le="0.1"} 2 +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 atenet_router_route_duration_seconds histogram +atenet_router_route_duration_seconds_count 1 +# TYPE atelet_snapshot_size_bytes histogram +atelet_snapshot_size_bytes_count 1 +# TYPE ate_actor_restored_total counter +ate_actor_restored_total 5 +` + +func TestMissingPlatformMetrics(t *testing.T) { + tests := []struct { + name string + scrape string + prefixes []string + want []string + }{ + { + name: "all present via suffix, exact, and comment forms", + scrape: sampleScrape, + prefixes: []string{ + "ate_actor_lifecycle_operation_duration", + "ate_workerpool_workers", + "atenet_router_route_duration", + "atelet_snapshot_size", + }, + want: nil, + }, + { + name: "absent prefix is reported", + scrape: sampleScrape, + prefixes: []string{"ate_scheduler_assignment_duration"}, + want: []string{"ate_scheduler_assignment_duration"}, + }, + { + name: "underscore boundary avoids restore matching restored", + scrape: sampleScrape, + prefixes: []string{"ate_actor_restore_duration"}, + want: []string{"ate_actor_restore_duration"}, + }, + { + name: "empty scrape misses everything", + scrape: "", + prefixes: []string{"ate_workerpool_workers", "atelet_snapshot_size"}, + want: []string{"ate_workerpool_workers", "atelet_snapshot_size"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := MissingPlatformMetrics(tt.scrape, tt.prefixes) + sort.Strings(got) + sort.Strings(tt.want) + if len(got) != len(tt.want) { + t.Fatalf("missing = %v, want %v", got, tt.want) + } + for i := range got { + if got[i] != tt.want[i] { + t.Fatalf("missing = %v, want %v", got, tt.want) + } + } + }) + } +} + +func TestMetricNameFromLine(t *testing.T) { + tests := []struct { + line string + want string + }{ + {"ate_workerpool_workers{pool=\"a\"} 3", "ate_workerpool_workers"}, + {"ate_scheduler_assignment_duration_seconds_count 7", "ate_scheduler_assignment_duration_seconds_count"}, + {"# TYPE ate_workerpool_workers gauge", "ate_workerpool_workers"}, + {"# HELP ate_workerpool_workers Number of workers.", "ate_workerpool_workers"}, + {" ", ""}, + {"# some other comment", ""}, + } + for _, tt := range tests { + if got := metricNameFromLine(tt.line); got != tt.want { + t.Errorf("metricNameFromLine(%q) = %q, want %q", tt.line, got, tt.want) + } + } +} diff --git a/internal/e2e/portforward.go b/internal/e2e/portforward.go new file mode 100644 index 000000000..50aa01073 --- /dev/null +++ b/internal/e2e/portforward.go @@ -0,0 +1,100 @@ +// 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 e2e + +import ( + "context" + "fmt" + "io" + "net/http" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" + "k8s.io/client-go/tools/portforward" + "k8s.io/client-go/transport/spdy" +) + +// firstReadyPodForService returns a ready pod backing service, plus the Service +// itself (callers need its port mapping). It refuses a selectorless Service +// rather than forward to an arbitrary pod in the namespace. +func firstReadyPodForService(ctx context.Context, clientset kubernetes.Interface, namespace, service string) (*corev1.Pod, *corev1.Service, error) { + svc, err := clientset.CoreV1().Services(namespace).Get(ctx, service, metav1.GetOptions{}) + if err != nil { + return nil, nil, fmt.Errorf("getting %s service: %w", service, err) + } + if len(svc.Spec.Selector) == 0 { + return nil, nil, fmt.Errorf("service %s has no selector", service) + } + selector := labels.SelectorFromSet(svc.Spec.Selector).String() + pods, err := clientset.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{LabelSelector: selector}) + if err != nil { + return nil, nil, fmt.Errorf("listing %s pods: %w", service, err) + } + for i := range pods.Items { + if isPodReady(&pods.Items[i]) { + return &pods.Items[i], svc, nil + } + } + return nil, nil, fmt.Errorf("no ready %s pods in %s", service, namespace) +} + +// podPortForward forwards a random local port to targetPort on the pod (local +// port 0 asks the OS for a free one), returning the chosen local port and a stop +// func the caller must invoke to tear the tunnel down. +func podPortForward(ctx context.Context, config *rest.Config, clientset kubernetes.Interface, namespace, podName string, targetPort int32) (int, func(), error) { + req := clientset.CoreV1().RESTClient().Post(). + Resource("pods"). + Namespace(namespace). + Name(podName). + SubResource("portforward") + + transport, upgrader, err := spdy.RoundTripperFor(config) + if err != nil { + return 0, nil, fmt.Errorf("creating SPDY transport: %w", err) + } + dialer := spdy.NewDialer(upgrader, &http.Client{Transport: transport}, http.MethodPost, req.URL()) + + stopCh := make(chan struct{}) + readyCh := make(chan struct{}) + fw, err := portforward.New(dialer, []string{fmt.Sprintf("0:%d", targetPort)}, stopCh, readyCh, io.Discard, io.Discard) + if err != nil { + return 0, nil, fmt.Errorf("creating port forwarder: %w", err) + } + + errCh := make(chan error, 1) + go func() { + if err := fw.ForwardPorts(); err != nil { + errCh <- err + } + }() + select { + case <-readyCh: + case err := <-errCh: + return 0, nil, fmt.Errorf("port forwarding: %w", err) + case <-ctx.Done(): + close(stopCh) + return 0, nil, ctx.Err() + } + + ports, err := fw.GetPorts() + if err != nil || len(ports) == 0 { + close(stopCh) + return 0, nil, fmt.Errorf("getting forwarded ports: %w", err) + } + return int(ports[0].Local), func() { close(stopCh) }, nil +} diff --git a/internal/e2e/router_client.go b/internal/e2e/router_client.go index cd5f0d67a..0f16457d3 100644 --- a/internal/e2e/router_client.go +++ b/internal/e2e/router_client.go @@ -17,19 +17,14 @@ package e2e import ( "context" "fmt" - "io" "net/http" "time" "github.com/agent-substrate/substrate/internal/ateclient" "github.com/agent-substrate/substrate/internal/resources" corev1 "k8s.io/api/core/v1" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/util/intstr" "k8s.io/client-go/kubernetes" - "k8s.io/client-go/tools/portforward" - "k8s.io/client-go/transport/spdy" ) const ( @@ -44,7 +39,7 @@ const ( type RouterClient struct { baseURL string http *http.Client - stopCh chan struct{} + stop func() } // NewRouterClient establishes a port-forward to the atenet router. Call Close @@ -59,30 +54,9 @@ func NewRouterClient(ctx context.Context) (*RouterClient, error) { return nil, fmt.Errorf("creating k8s client: %w", err) } - // Resolve the router Service to one of its backing pods. - svc, err := clientset.CoreV1().Services(routerNamespace).Get(ctx, routerService, metav1.GetOptions{}) + targetPod, svc, err := firstReadyPodForService(ctx, clientset, routerNamespace, routerService) if err != nil { - return nil, fmt.Errorf("getting %s service: %w", routerService, err) - } - // A headless/selectorless Service would make SelectorFromSet match every - // pod in the namespace; refuse rather than forward to an arbitrary pod. - if len(svc.Spec.Selector) == 0 { - return nil, fmt.Errorf("service %s has no selector", routerService) - } - selector := labels.SelectorFromSet(svc.Spec.Selector).String() - pods, err := clientset.CoreV1().Pods(routerNamespace).List(ctx, metav1.ListOptions{LabelSelector: selector}) - if err != nil { - return nil, fmt.Errorf("listing %s pods: %w", routerService, err) - } - var targetPod *corev1.Pod - for i := range pods.Items { - if isPodReady(&pods.Items[i]) { - targetPod = &pods.Items[i] - break - } - } - if targetPod == nil { - return nil, fmt.Errorf("no ready %s pods in %s", routerService, routerNamespace) + return nil, err } // Port-forward targets a pod's container port, so resolve the Service's @@ -93,51 +67,15 @@ func NewRouterClient(ctx context.Context) (*RouterClient, error) { return nil, err } - req := clientset.CoreV1().RESTClient().Post(). - Resource("pods"). - Namespace(routerNamespace). - Name(targetPod.Name). - SubResource("portforward") - - transport, upgrader, err := spdy.RoundTripperFor(config) + localPort, stop, err := podPortForward(ctx, config, clientset, routerNamespace, targetPod.Name, targetPort) if err != nil { - return nil, fmt.Errorf("creating SPDY transport: %w", err) - } - dialer := spdy.NewDialer(upgrader, &http.Client{Transport: transport}, http.MethodPost, req.URL()) - - stopCh := make(chan struct{}) - readyCh := make(chan struct{}) - // Port 0 asks the OS for a random available local port. - fw, err := portforward.New(dialer, []string{fmt.Sprintf("0:%d", targetPort)}, stopCh, readyCh, io.Discard, io.Discard) - if err != nil { - return nil, fmt.Errorf("creating port forwarder: %w", err) - } - - errCh := make(chan error, 1) - go func() { - if err := fw.ForwardPorts(); err != nil { - errCh <- err - } - }() - select { - case <-readyCh: - case err := <-errCh: - return nil, fmt.Errorf("port forwarding: %w", err) - case <-ctx.Done(): - close(stopCh) - return nil, ctx.Err() - } - - ports, err := fw.GetPorts() - if err != nil || len(ports) == 0 { - close(stopCh) - return nil, fmt.Errorf("getting forwarded ports: %w", err) + return nil, err } return &RouterClient{ - baseURL: fmt.Sprintf("http://127.0.0.1:%d", ports[0].Local), + baseURL: fmt.Sprintf("http://127.0.0.1:%d", localPort), http: &http.Client{Timeout: 30 * time.Second}, - stopCh: stopCh, + stop: stop, }, nil } @@ -190,7 +128,7 @@ func resolveHTTPTargetPort(svc *corev1.Service, pod *corev1.Pod) (int32, error) // Close stops the port-forward tunnel. func (c *RouterClient) Close() { - close(c.stopCh) + c.stop() } // Get issues GET path to (atespace, actorName) through the router, setting the diff --git a/internal/e2e/suites/metrics/metrics_test.go b/internal/e2e/suites/metrics/metrics_test.go new file mode 100644 index 000000000..624505f4d --- /dev/null +++ b/internal/e2e/suites/metrics/metrics_test.go @@ -0,0 +1,141 @@ +// 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 metrics is an e2e suite that drives one full actor lifecycle and then +// asserts every platform metric in e2e.PlatformMetricPrefixes reaches the kind +// stack's OTel Collector. It closes the "silent regression" gap: a renamed or +// dropped instrument fails here rather than surfacing as an empty dashboard. +// The prefix set grows as each metric slice lands. Requires the demo counter +// template to be installed (override with E2E_TEMPLATE_NAMESPACE / _NAME). +package metrics + +import ( + "context" + "os" + "testing" + "time" + + "github.com/agent-substrate/substrate/internal/e2e" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" +) + +const metricsAtespace = "ate-metrics-e2e" + +func templateRef() (namespace, name string) { + namespace, name = "ate-demo-counter", "counter" + if v := os.Getenv("E2E_TEMPLATE_NAMESPACE"); v != "" { + namespace = v + } + if v := os.Getenv("E2E_TEMPLATE_NAME"); v != "" { + name = v + } + return namespace, name +} + +func TestPlatformMetricsEmitted(t *testing.T) { + ctx := context.Background() + clients := e2e.GetClients() + tmplNS, tmplName := templateRef() + actorID := "metrics-probe" + + // CreateActor requires the atespace to exist first; ignore AlreadyExists. + _, _ = clients.SubstrateAPI.CreateAtespace(ctx, &ateapipb.CreateAtespaceRequest{ + Atespace: &ateapipb.Atespace{Metadata: &ateapipb.ResourceMetadata{Name: metricsAtespace}}, + }) + + if _, err := clients.SubstrateAPI.CreateActor(ctx, &ateapipb.CreateActorRequest{Actor: &ateapipb.Actor{ + Metadata: &ateapipb.ResourceMetadata{Atespace: metricsAtespace, Name: actorID}, + ActorTemplateNamespace: tmplNS, + ActorTemplateName: tmplName, + }}); err != nil { + t.Fatalf("CreateActor: %v", err) + } + t.Cleanup(func() { + _, _ = clients.SubstrateAPI.SuspendActor(ctx, &ateapipb.SuspendActorRequest{Actor: &ateapipb.ObjectRef{Atespace: metricsAtespace, Name: actorID}}) + _, _ = clients.SubstrateAPI.DeleteActor(ctx, &ateapipb.DeleteActorRequest{Actor: &ateapipb.ObjectRef{Atespace: metricsAtespace, Name: actorID}}) + }) + + // Drive a full lifecycle so the platform instruments fire: resume brings up + // workers (worker-count), a routed request records the router duration, and + // suspend writes a checkpoint (snapshot size). A second resume exercises the + // restore path. Later metric slices add their own instruments to + // e2e.PlatformMetricPrefixes; the same drive already triggers them. + resume(t, ctx, clients, actorID) + routeRequest(t, ctx, actorID) + suspend(t, ctx, clients, actorID) + resume(t, ctx, clients, actorID) + + deadline := time.Now().Add(2 * time.Minute) + var missing []string + for time.Now().Before(deadline) { + scrape, err := e2e.ScrapeCollectorMetrics(ctx) + if err != nil { + t.Fatalf("ScrapeCollectorMetrics: %v", err) + } + if missing = e2e.MissingPlatformMetrics(scrape, e2e.PlatformMetricPrefixes); len(missing) == 0 { + return + } + time.Sleep(3 * time.Second) + } + t.Fatalf("platform metrics never reached the collector: missing %v", missing) +} + +func resume(t *testing.T, ctx context.Context, clients *e2e.Clients, actorID string) { + t.Helper() + if _, err := clients.SubstrateAPI.ResumeActor(ctx, &ateapipb.ResumeActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: metricsAtespace, Name: actorID}, + }); err != nil { + t.Fatalf("ResumeActor: %v", err) + } + waitForStatus(t, ctx, clients, actorID, ateapipb.Actor_STATUS_RUNNING) +} + +func suspend(t *testing.T, ctx context.Context, clients *e2e.Clients, actorID string) { + t.Helper() + if _, err := clients.SubstrateAPI.SuspendActor(ctx, &ateapipb.SuspendActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: metricsAtespace, Name: actorID}, + }); err != nil { + t.Fatalf("SuspendActor: %v", err) + } + waitForStatus(t, ctx, clients, actorID, ateapipb.Actor_STATUS_SUSPENDED) +} + +func routeRequest(t *testing.T, ctx context.Context, actorID string) { + t.Helper() + rc, err := e2e.NewRouterClient(ctx) + if err != nil { + t.Fatalf("NewRouterClient: %v", err) + } + defer rc.Close() + // Any routed request records atenet.router.route.duration; the actor's own + // response is irrelevant here, so tolerate a non-2xx or transport error. + if resp, err := rc.Get(ctx, metricsAtespace, actorID, "/"); err == nil { + _ = resp.Body.Close() + } +} + +func waitForStatus(t *testing.T, ctx context.Context, clients *e2e.Clients, actorID string, want ateapipb.Actor_Status) { + t.Helper() + deadline := time.Now().Add(2 * time.Minute) + for time.Now().Before(deadline) { + resp, err := clients.SubstrateAPI.GetActor(ctx, &ateapipb.GetActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: metricsAtespace, Name: actorID}, + }) + if err == nil && resp.GetStatus() == want { + return + } + time.Sleep(2 * time.Second) + } + t.Fatalf("actor %q never reached %v", actorID, want) +} diff --git a/internal/e2e/suites/metrics/testmain_test.go b/internal/e2e/suites/metrics/testmain_test.go new file mode 100644 index 000000000..ba011905f --- /dev/null +++ b/internal/e2e/suites/metrics/testmain_test.go @@ -0,0 +1,24 @@ +// 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 metrics + +import ( + "os" + "testing" + + "github.com/agent-substrate/substrate/internal/e2e" +) + +func TestMain(m *testing.M) { os.Exit(e2e.RunTestMain(m)) } From 1823f3f5b7b669448789e20538ed90f3a0668559 Mon Sep 17 00:00:00 2001 From: krisztianfekete Date: Thu, 23 Jul 2026 17:41:10 +0200 Subject: [PATCH 3/5] address review comments --- cmd/ateapi/internal/controlapi/metrics.go | 19 +++++++++---------- 1 file changed, 9 insertions(+), 10 deletions(-) diff --git a/cmd/ateapi/internal/controlapi/metrics.go b/cmd/ateapi/internal/controlapi/metrics.go index e60a44145..7ee58302c 100644 --- a/cmd/ateapi/internal/controlapi/metrics.go +++ b/cmd/ateapi/internal/controlapi/metrics.go @@ -51,17 +51,16 @@ func RegisterWorkerCount(meter metric.Meter, workers func() ([]*ateapipb.Worker, 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 + // reports 0, not an absent series that breaks idle==0 alerts. A failed + // list just means no seeding this cycle, not a broken observation. + 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 { From c2597818b13bd08aff7ecc223945302dec051d1d Mon Sep 17 00:00:00 2001 From: krisztianfekete Date: Thu, 23 Jul 2026 18:10:11 +0200 Subject: [PATCH 4/5] remove unused helper --- cmd/ateapi/internal/controlapi/metrics_test.go | 2 -- 1 file changed, 2 deletions(-) diff --git a/cmd/ateapi/internal/controlapi/metrics_test.go b/cmd/ateapi/internal/controlapi/metrics_test.go index fefe006dc..736389149 100644 --- a/cmd/ateapi/internal/controlapi/metrics_test.go +++ b/cmd/ateapi/internal/controlapi/metrics_test.go @@ -40,8 +40,6 @@ func newWorkerCountReader(t *testing.T, workers func() ([]*ateapipb.Worker, erro 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) { From ef8a3c2dd089043f33bb6a01e58da816761a8787 Mon Sep 17 00:00:00 2001 From: krisztianfekete Date: Thu, 23 Jul 2026 19:12:08 +0200 Subject: [PATCH 5/5] scope metrics assertions to ate.workerpool.workers in e2e tests --- internal/e2e/collector_metrics.go | 13 +++--- internal/e2e/suites/metrics/metrics_test.go | 47 ++++----------------- 2 files changed, 14 insertions(+), 46 deletions(-) diff --git a/internal/e2e/collector_metrics.go b/internal/e2e/collector_metrics.go index d50980c46..6249631a5 100644 --- a/internal/e2e/collector_metrics.go +++ b/internal/e2e/collector_metrics.go @@ -33,16 +33,13 @@ const ( ) // PlatformMetricPrefixes are the Prometheus metric-name prefixes (OTLP dots -// mapped to underscores) the substrate platform must emit once a full actor -// lifecycle has run. The Collector's prometheus exporter appends unit and type -// suffixes (e.g. _seconds_bucket, _bytes_count), so matching is by prefix. -// This slice grows as each platform-metric slice lands; today it pins the -// worker-count instrument plus the two hand-written metrics that predate this -// work, guarding all three against silent regression. +// mapped to underscores) the substrate platform must emit. The Collector's +// prometheus exporter appends unit and type suffixes (e.g. _seconds_bucket, +// _bytes_count), so matching is by prefix. This slice grows as each metric +// slice lands and as more components are wired to push to the collector; today +// it pins the worker-count instrument introduced alongside this harness. var PlatformMetricPrefixes = []string{ "ate_workerpool_workers", - "atenet_router_route_duration", - "atelet_snapshot_size", } // ScrapeCollectorMetrics port-forwards the kind stack's OTel Collector and reads diff --git a/internal/e2e/suites/metrics/metrics_test.go b/internal/e2e/suites/metrics/metrics_test.go index 624505f4d..b3c3cbb71 100644 --- a/internal/e2e/suites/metrics/metrics_test.go +++ b/internal/e2e/suites/metrics/metrics_test.go @@ -12,12 +12,12 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Package metrics is an e2e suite that drives one full actor lifecycle and then -// asserts every platform metric in e2e.PlatformMetricPrefixes reaches the kind -// stack's OTel Collector. It closes the "silent regression" gap: a renamed or -// dropped instrument fails here rather than surfacing as an empty dashboard. -// The prefix set grows as each metric slice lands. Requires the demo counter -// template to be installed (override with E2E_TEMPLATE_NAMESPACE / _NAME). +// Package metrics is an e2e suite that drives an actor lifecycle and then asserts +// the platform metrics in e2e.PlatformMetricPrefixes reach the kind stack's OTel +// Collector. It closes the "silent regression" gap: a renamed or dropped +// instrument fails here rather than surfacing as an empty dashboard. The prefix +// set grows as each metric slice lands. Requires the demo counter template to be +// installed (override with E2E_TEMPLATE_NAMESPACE / _NAME). package metrics import ( @@ -66,14 +66,9 @@ func TestPlatformMetricsEmitted(t *testing.T) { _, _ = clients.SubstrateAPI.DeleteActor(ctx, &ateapipb.DeleteActorRequest{Actor: &ateapipb.ObjectRef{Atespace: metricsAtespace, Name: actorID}}) }) - // Drive a full lifecycle so the platform instruments fire: resume brings up - // workers (worker-count), a routed request records the router duration, and - // suspend writes a checkpoint (snapshot size). A second resume exercises the - // restore path. Later metric slices add their own instruments to - // e2e.PlatformMetricPrefixes; the same drive already triggers them. - resume(t, ctx, clients, actorID) - routeRequest(t, ctx, actorID) - suspend(t, ctx, clients, actorID) + // Resume so the pool has an assigned worker, which the ateapi worker-count + // observable reports. Later metric slices extend e2e.PlatformMetricPrefixes; + // they add the drive steps their instruments need. resume(t, ctx, clients, actorID) deadline := time.Now().Add(2 * time.Minute) @@ -101,30 +96,6 @@ func resume(t *testing.T, ctx context.Context, clients *e2e.Clients, actorID str waitForStatus(t, ctx, clients, actorID, ateapipb.Actor_STATUS_RUNNING) } -func suspend(t *testing.T, ctx context.Context, clients *e2e.Clients, actorID string) { - t.Helper() - if _, err := clients.SubstrateAPI.SuspendActor(ctx, &ateapipb.SuspendActorRequest{ - Actor: &ateapipb.ObjectRef{Atespace: metricsAtespace, Name: actorID}, - }); err != nil { - t.Fatalf("SuspendActor: %v", err) - } - waitForStatus(t, ctx, clients, actorID, ateapipb.Actor_STATUS_SUSPENDED) -} - -func routeRequest(t *testing.T, ctx context.Context, actorID string) { - t.Helper() - rc, err := e2e.NewRouterClient(ctx) - if err != nil { - t.Fatalf("NewRouterClient: %v", err) - } - defer rc.Close() - // Any routed request records atenet.router.route.duration; the actor's own - // response is irrelevant here, so tolerate a non-2xx or transport error. - if resp, err := rc.Get(ctx, metricsAtespace, actorID, "/"); err == nil { - _ = resp.Body.Close() - } -} - func waitForStatus(t *testing.T, ctx context.Context, clients *e2e.Clients, actorID string, want ateapipb.Actor_Status) { t.Helper() deadline := time.Now().Add(2 * time.Minute)