diff --git a/cmd/ateapi/internal/controlapi/workflow.go b/cmd/ateapi/internal/controlapi/workflow.go index 8f591fc5a..6b1a5c7c3 100644 --- a/cmd/ateapi/internal/controlapi/workflow.go +++ b/cmd/ateapi/internal/controlapi/workflow.go @@ -154,7 +154,7 @@ func NewActorWorkflow( return &ActorWorkflow{ store: store, workerCache: workerCache, - scheduler: scheduling.New(workerCache), + scheduler: scheduling.New(workerCache, scheduling.WithMeter(otel.Meter("ateapi"))), dialer: dialer, actorTemplateLister: actorTemplateLister, workerPoolLister: workerPoolLister, diff --git a/cmd/ateapi/internal/scheduling/scheduling.go b/cmd/ateapi/internal/scheduling/scheduling.go index a68b6d7c9..5918cd173 100644 --- a/cmd/ateapi/internal/scheduling/scheduling.go +++ b/cmd/ateapi/internal/scheduling/scheduling.go @@ -19,13 +19,20 @@ import ( "context" "errors" "fmt" + "log/slog" "math/rand" "slices" + "github.com/agent-substrate/substrate/internal/ateattr" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" "k8s.io/apimachinery/pkg/labels" ) +// metric identifier for tracking eligible worker counts. +const eligibleWorkersMetric = "ate.scheduler.eligible_workers" + // Constraints describes what a worker must satisfy to host an actor. type Constraints struct { // SandboxClass must equal the worker's sandbox class. Snapshots are not @@ -64,8 +71,10 @@ type WorkerSource interface { type scheduler struct { source WorkerSource // intn returns a uniformly distributed random value in [0,n). - // Defaults to the global math/rand source + // Defaults to the global math/rand source. intn func(n int) int + // eligibleWorkers records the number of eligible workers available during scheduling. + eligibleWorkers metric.Int64Histogram } // Option configures the Scheduler returned by New. @@ -77,6 +86,25 @@ func WithIntn(intn func(n int) int) Option { return func(s *scheduler) { s.intn = intn } } +// WithMeter configures the meter used to create telemetry instruments for the scheduler. +func WithMeter(meter metric.Meter) Option { + return func(s *scheduler) { + if meter == nil { + return + } + h, err := meter.Int64Histogram( + eligibleWorkersMetric, + metric.WithUnit("{worker}"), + metric.WithDescription("Number of eligible workers available during scheduling."), + ) + if err != nil { + slog.Error("Failed to create ate.scheduler.eligible_workers histogram", "error", err) + return + } + s.eligibleWorkers = h + } +} + // New returns a Scheduler placing onto workers reported by source. func New(source WorkerSource, opts ...Option) Scheduler { s := &scheduler{source: source, intn: rand.Intn} @@ -86,12 +114,15 @@ func New(source WorkerSource, opts ...Option) Scheduler { return s } +// Schedule filters the current worker fleet to find unassigned candidates matching the given constraints, +// records the ate.scheduler.eligible_workers metric, and picks a random candidate if available. func (s *scheduler) Schedule(ctx context.Context, constraints Constraints) (*ateapipb.Worker, error) { workers, err := s.source.Workers() if err != nil { return nil, fmt.Errorf("while listing workers: %w", err) } + // Filter for candidate workers that are unassigned and meet all scheduling constraints var candidates []*ateapipb.Worker for _, worker := range workers { if worker.GetAssignment() == nil && s.Applies(worker, constraints) { @@ -99,12 +130,94 @@ func (s *scheduler) Schedule(ctx context.Context, constraints Constraints) (*ate } } + // Record telemetry on the number of eligible workers per pool/namespace before returning + s.recordEligibleWorkers(ctx, workers, candidates, constraints) + if len(candidates) == 0 { return nil, ErrNoCapacity } + return candidates[s.intn(len(candidates))], nil } +// recordEligibleWorkers records candidate worker counts grouped by WorkerPool namespace, WorkerPool name, +// SandboxClass, and SchedulingConstraint, and records histogram datapoints. +func (s *scheduler) recordEligibleWorkers(ctx context.Context, allWorkers []*ateapipb.Worker, candidates []*ateapipb.Worker, constraints Constraints) { + if s.eligibleWorkers == nil { + return + } + + constraintStr := classifyConstraint(constraints) + + type key struct { + namespace string + pool string + sandboxClass string + constraint string + } + eligibleByPool := make(map[key]int64) + + // Seed key counts at 0 for all worker pools matching constraints, + // ensures saturated pools report 0 eligible workers rather than missing series. + for _, w := range allWorkers { + if s.Applies(w, constraints) { + eligibleByPool[key{ + namespace: w.GetWorkerNamespace(), + pool: w.GetWorkerPool(), + sandboxClass: w.GetSandboxClass(), + constraint: constraintStr, + }] = 0 + } + } + + // Records unassigned/eligible candidate workers for each pool + for _, w := range candidates { + eligibleByPool[key{ + namespace: w.GetWorkerNamespace(), + pool: w.GetWorkerPool(), + sandboxClass: w.GetSandboxClass(), + constraint: constraintStr, + }]++ + } + + // Handle when no worker pools match constraints + if len(eligibleByPool) == 0 { + attrs := []attribute.KeyValue{ + ateattr.SchedulingConstraintKey.String(constraintStr), + } + if constraints.SandboxClass != "" { + attrs = append(attrs, ateattr.SandboxClassKey.String(constraints.SandboxClass)) + } + s.eligibleWorkers.Record(ctx, 0, metric.WithAttributes(attrs...)) + return + } + + // Emit histogram observation for each worker pool key using standard ateattr keys + for k, count := range eligibleByPool { + s.eligibleWorkers.Record(ctx, count, metric.WithAttributes( + ateattr.WorkerPoolNamespaceKey.String(k.namespace), + ateattr.WorkerPoolNameKey.String(k.pool), + ateattr.SandboxClassKey.String(k.sandboxClass), + ateattr.SchedulingConstraintKey.String(k.constraint), + )) + } +} + +func classifyConstraint(c Constraints) string { + if len(c.RequiredNodes) > 0 { + return ateattr.ConstraintRequiredNodes + } + if (c.TemplateSelector != nil && !c.TemplateSelector.Empty()) || (c.ActorSelector != nil && !c.ActorSelector.Empty()) { + return ateattr.ConstraintSelector + } + return ateattr.ConstraintNone +} + +// Applies evaluates whether a single worker satisfies all requested scheduling constraints: +// 1. SandboxClass match (hard requirement for snapshot compatibility). +// 2. Active worker state (draining/unspecified workers are excluded). +// 3. Template and Actor label selectors match worker labels. +// 4. Node VM locality constraints (if required for local snapshot restoration). func (s *scheduler) Applies(worker *ateapipb.Worker, constraints Constraints) bool { if worker.GetSandboxClass() != constraints.SandboxClass { return false diff --git a/cmd/ateapi/internal/scheduling/scheduling_test.go b/cmd/ateapi/internal/scheduling/scheduling_test.go index 52062e47c..68a8000f0 100644 --- a/cmd/ateapi/internal/scheduling/scheduling_test.go +++ b/cmd/ateapi/internal/scheduling/scheduling_test.go @@ -19,7 +19,10 @@ import ( "errors" "testing" + "github.com/agent-substrate/substrate/internal/ateattr" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" "k8s.io/apimachinery/pkg/labels" ) @@ -267,3 +270,301 @@ func assigned(atespace, name string) func(*ateapipb.Worker) { // firstIntn always picks the first candidate, making Schedule deterministic. func firstIntn(int) int { return 0 } + +func workerWithPool(pod, ns, pool, class, node string, lbls map[string]string, opts ...func(*ateapipb.Worker)) *ateapipb.Worker { + w := worker(pod, class, node, lbls, opts...) + w.WorkerNamespace = ns + w.WorkerPool = pool + return w +} + +func TestSchedule_EligibleWorkersMetric(t *testing.T) { + t.Run("records histogram metric with namespaced attributes for candidates", func(t *testing.T) { + reader := sdkmetric.NewManualReader() + provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + meter := provider.Meter("test") + + flt := fleet{ + workerWithPool("w-1", "ns-a", "pool-1", "gvisor", "node-a", nil), + workerWithPool("w-2", "ns-a", "pool-1", "gvisor", "node-a", nil, assigned("demo", "a")), + workerWithPool("w-3", "ns-b", "pool-2", "gvisor", "node-b", nil), + } + + s := New(flt, WithIntn(firstIntn), WithMeter(meter)) + _, err := s.Schedule(context.Background(), Constraints{SandboxClass: "gvisor"}) + if err != nil { + t.Fatalf("Schedule() error = %v", err) + } + + var rm metricdata.ResourceMetrics + if err := reader.Collect(context.Background(), &rm); err != nil { + t.Fatalf("reader.Collect() error = %v", err) + } + + foundMetric := false + for _, sm := range rm.ScopeMetrics { + for _, m := range sm.Metrics { + if m.Name == "ate.scheduler.eligible_workers" { + foundMetric = true + histogram, ok := m.Data.(metricdata.Histogram[int64]) + if !ok { + t.Fatalf("metric Data is %T, want metricdata.Histogram[int64]", m.Data) + } + if len(histogram.DataPoints) == 0 { + t.Fatalf("got 0 DataPoints for ate.scheduler.eligible_workers") + } + for _, dp := range histogram.DataPoints { + attrs := dp.Attributes + ns, _ := attrs.Value(ateattr.WorkerPoolNamespaceKey) + pool, _ := attrs.Value(ateattr.WorkerPoolNameKey) + class, _ := attrs.Value(ateattr.SandboxClassKey) + constraint, _ := attrs.Value(ateattr.SchedulingConstraintKey) + if class.AsString() != "gvisor" { + t.Errorf("got sandbox class %q, want %q", class.AsString(), "gvisor") + } + if constraint.AsString() != "none" { + t.Errorf("got constraint %q, want %q", constraint.AsString(), "none") + } + if ns.AsString() == "ns-a" && pool.AsString() == "pool-1" { + if dp.Count != 1 || dp.Sum != 1 { + t.Errorf("pool-1 datapoint count=%d sum=%d, want count=1 sum=1", dp.Count, dp.Sum) + } + } + if ns.AsString() == "ns-b" && pool.AsString() == "pool-2" { + if dp.Count != 1 || dp.Sum != 1 { + t.Errorf("pool-2 datapoint count=%d sum=%d, want count=1 sum=1", dp.Count, dp.Sum) + } + } + } + } + } + } + if !foundMetric { + t.Fatalf("ate.scheduler.eligible_workers metric not found") + } + }) + + t.Run("records 0 eligible workers when fleet has no capacity", func(t *testing.T) { + reader := sdkmetric.NewManualReader() + provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + meter := provider.Meter("test") + + flt := fleet{ + workerWithPool("w-busy", "ns-a", "pool-1", "gvisor", "node-a", nil, assigned("demo", "a")), + } + + s := New(flt, WithIntn(firstIntn), WithMeter(meter)) + _, err := s.Schedule(context.Background(), Constraints{SandboxClass: "gvisor"}) + if !errors.Is(err, ErrNoCapacity) { + t.Fatalf("Schedule() error = %v, want ErrNoCapacity", err) + } + + var rm metricdata.ResourceMetrics + if err := reader.Collect(context.Background(), &rm); err != nil { + t.Fatalf("reader.Collect() error = %v", err) + } + + foundMetric := false + for _, sm := range rm.ScopeMetrics { + for _, m := range sm.Metrics { + if m.Name == "ate.scheduler.eligible_workers" { + foundMetric = true + histogram, ok := m.Data.(metricdata.Histogram[int64]) + if !ok { + t.Fatalf("metric Data is %T, want metricdata.Histogram[int64]", m.Data) + } + if len(histogram.DataPoints) == 0 { + t.Fatalf("got 0 DataPoints") + } + dp := histogram.DataPoints[0] + if dp.Sum != 0 { + t.Errorf("datapoint sum = %d, want 0", dp.Sum) + } + attrs := dp.Attributes + ns, _ := attrs.Value(ateattr.WorkerPoolNamespaceKey) + pool, _ := attrs.Value(ateattr.WorkerPoolNameKey) + constraint, _ := attrs.Value(ateattr.SchedulingConstraintKey) + if ns.AsString() != "ns-a" || pool.AsString() != "pool-1" { + t.Errorf("got namespace=%q pool=%q, want ns-a / pool-1", ns.AsString(), pool.AsString()) + } + if constraint.AsString() != "none" { + t.Errorf("got constraint=%q, want none", constraint.AsString()) + } + } + } + } + if !foundMetric { + t.Fatalf("ate.scheduler.eligible_workers metric not found") + } + }) + + t.Run("records constraint classification attributes correctly", func(t *testing.T) { + sel, _ := labels.Parse("env=prod") + tests := []struct { + name string + constraints Constraints + wantConstraint string + }{ + { + name: "none", + constraints: Constraints{SandboxClass: "gvisor"}, + wantConstraint: ateattr.ConstraintNone, + }, + { + name: "selector", + constraints: Constraints{SandboxClass: "gvisor", ActorSelector: sel}, + wantConstraint: ateattr.ConstraintSelector, + }, + { + name: "required_nodes", + constraints: Constraints{SandboxClass: "gvisor", ActorSelector: sel, RequiredNodes: []string{"node-a"}}, + wantConstraint: ateattr.ConstraintRequiredNodes, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + reader := sdkmetric.NewManualReader() + provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + meter := provider.Meter("test") + + flt := fleet{ + workerWithPool("w-1", "ns-a", "pool-1", "gvisor", "node-a", map[string]string{"env": "prod"}), + } + + s := New(flt, WithIntn(firstIntn), WithMeter(meter)) + _, err := s.Schedule(context.Background(), tc.constraints) + if err != nil { + t.Fatalf("Schedule() error = %v", err) + } + + var rm metricdata.ResourceMetrics + _ = reader.Collect(context.Background(), &rm) + for _, sm := range rm.ScopeMetrics { + for _, m := range sm.Metrics { + if m.Name == "ate.scheduler.eligible_workers" { + histogram := m.Data.(metricdata.Histogram[int64]) + dp := histogram.DataPoints[0] + constraint, _ := dp.Attributes.Value(ateattr.SchedulingConstraintKey) + if constraint.AsString() != tc.wantConstraint { + t.Errorf("got constraint=%q, want %q", constraint.AsString(), tc.wantConstraint) + } + } + } + } + }) + } + }) + + t.Run("records 0 eligible workers when fleet is completely empty", func(t *testing.T) { + reader := sdkmetric.NewManualReader() + provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + meter := provider.Meter("test") + + flt := fleet{} + + s := New(flt, WithIntn(firstIntn), WithMeter(meter)) + _, err := s.Schedule(context.Background(), Constraints{SandboxClass: "gvisor"}) + if !errors.Is(err, ErrNoCapacity) { + t.Fatalf("Schedule() error = %v, want ErrNoCapacity", err) + } + + var rm metricdata.ResourceMetrics + _ = reader.Collect(context.Background(), &rm) + foundMetric := false + for _, sm := range rm.ScopeMetrics { + for _, m := range sm.Metrics { + if m.Name == "ate.scheduler.eligible_workers" { + foundMetric = true + histogram := m.Data.(metricdata.Histogram[int64]) + dp := histogram.DataPoints[0] + if dp.Sum != 0 { + t.Errorf("datapoint sum = %d, want 0", dp.Sum) + } + class, _ := dp.Attributes.Value(ateattr.SandboxClassKey) + if class.AsString() != "gvisor" { + t.Errorf("got sandbox class %q, want gvisor", class.AsString()) + } + } + } + } + if !foundMetric { + t.Fatalf("ate.scheduler.eligible_workers metric not found") + } + }) + + t.Run("records 0 eligible workers on sandbox class mismatch", func(t *testing.T) { + reader := sdkmetric.NewManualReader() + provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + meter := provider.Meter("test") + + flt := fleet{ + workerWithPool("w-1", "ns-a", "pool-1", "gvisor", "node-a", nil), + } + + s := New(flt, WithIntn(firstIntn), WithMeter(meter)) + _, err := s.Schedule(context.Background(), Constraints{SandboxClass: "kata"}) + if !errors.Is(err, ErrNoCapacity) { + t.Fatalf("Schedule() error = %v, want ErrNoCapacity", err) + } + + var rm metricdata.ResourceMetrics + _ = reader.Collect(context.Background(), &rm) + foundMetric := false + for _, sm := range rm.ScopeMetrics { + for _, m := range sm.Metrics { + if m.Name == "ate.scheduler.eligible_workers" { + foundMetric = true + histogram := m.Data.(metricdata.Histogram[int64]) + dp := histogram.DataPoints[0] + if dp.Sum != 0 { + t.Errorf("datapoint sum = %d, want 0", dp.Sum) + } + class, _ := dp.Attributes.Value(ateattr.SandboxClassKey) + if class.AsString() != "kata" { + t.Errorf("got sandbox class %q, want kata", class.AsString()) + } + } + } + } + if !foundMetric { + t.Fatalf("ate.scheduler.eligible_workers metric not found") + } + }) + + t.Run("excludes draining and inactive workers from eligible counts", func(t *testing.T) { + reader := sdkmetric.NewManualReader() + provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + meter := provider.Meter("test") + + drainingWorker := workerWithPool("w-draining", "ns-a", "pool-1", "gvisor", "node-a", nil) + drainingWorker.State = ateapipb.Worker_STATE_DRAINING + + flt := fleet{drainingWorker} + + s := New(flt, WithIntn(firstIntn), WithMeter(meter)) + _, err := s.Schedule(context.Background(), Constraints{SandboxClass: "gvisor"}) + if !errors.Is(err, ErrNoCapacity) { + t.Fatalf("Schedule() error = %v, want ErrNoCapacity", err) + } + + var rm metricdata.ResourceMetrics + _ = reader.Collect(context.Background(), &rm) + foundMetric := false + for _, sm := range rm.ScopeMetrics { + for _, m := range sm.Metrics { + if m.Name == "ate.scheduler.eligible_workers" { + foundMetric = true + histogram := m.Data.(metricdata.Histogram[int64]) + dp := histogram.DataPoints[0] + if dp.Sum != 0 { + t.Errorf("datapoint sum = %d, want 0", dp.Sum) + } + } + } + } + if !foundMetric { + t.Fatalf("ate.scheduler.eligible_workers metric not found") + } + }) +} diff --git a/docs/observability.md b/docs/observability.md index 78ab8e922..3a7eafe2a 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -111,6 +111,7 @@ Agent Substrate emits foundational OpenTelemetry system and server metrics to mo |--------|------------|------|----------| | `rpc.server.call.duration` | ateapi & atelet (gRPC servers, via `otelgrpc`) | histogram | per-method gRPC latency, request rate, and errors (labels `rpc.method`, `rpc.response.status_code`) | | `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 (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 `kind`, `actor_template_namespace`, `actor_template_name`) | The table lists the OpenTelemetry instrument names. How a name appears in a query depends on the backend (Cloud Monitoring (GMP) / Kind collector). @@ -119,6 +120,9 @@ 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). +For `ate.scheduler.eligible_workers`: +* `ate.scheduling.constraint` categorizes the scheduling request constraint type: `none` (unconstrained), `selector` (actor or template label selectors specified), or `required_nodes` (pinned to specific node VMs). + ### Local Metrics with Prometheus (Kind Cluster) For local development inside a `kind` cluster, Agent Substrate automatically provisions a Prometheus server in the `otel-system` namespace. diff --git a/internal/ateattr/ateattr.go b/internal/ateattr/ateattr.go index e316384b1..8488e2bdf 100644 --- a/internal/ateattr/ateattr.go +++ b/internal/ateattr/ateattr.go @@ -52,12 +52,20 @@ const ( // WorkerPoolNamespaceKey pairs with WorkerPoolNameKey: a WorkerPool is // namespaced, so the name alone does not identify one. const ( - WorkerPoolNamespaceKey = attribute.Key("ate.workerpool.namespace") - WorkerPoolNameKey = attribute.Key("ate.workerpool.name") - WorkerStateKey = attribute.Key("ate.worker.state") - SandboxClassKey = attribute.Key("ate.sandbox.class") - RouterResumeKey = attribute.Key("ate.router.resume") - RouterOutcomeKey = attribute.Key("ate.router.outcome") + WorkerPoolNamespaceKey = attribute.Key("ate.workerpool.namespace") + WorkerPoolNameKey = attribute.Key("ate.workerpool.name") + WorkerStateKey = attribute.Key("ate.worker.state") + SandboxClassKey = attribute.Key("ate.sandbox.class") + SchedulingConstraintKey = attribute.Key("ate.scheduling.constraint") + RouterResumeKey = attribute.Key("ate.router.resume") + RouterOutcomeKey = attribute.Key("ate.router.outcome") +) + +// Values for SchedulingConstraintKey. +const ( + ConstraintNone = "none" + ConstraintRequiredNodes = "required_nodes" + ConstraintSelector = "selector" ) // Values for RouterResumeKey. diff --git a/internal/e2e/collector_metrics.go b/internal/e2e/collector_metrics.go index c6dbd9034..06ba6137f 100644 --- a/internal/e2e/collector_metrics.go +++ b/internal/e2e/collector_metrics.go @@ -42,6 +42,7 @@ const ( var PlatformMetricPrefixes = []string{ "ate_workerpool_workers", "atenet_router_route_duration", + "ate_scheduler_eligible_workers", } // 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 46a912eee..78c86ae35 100644 --- a/internal/e2e/suites/metrics/metrics_test.go +++ b/internal/e2e/suites/metrics/metrics_test.go @@ -22,10 +22,13 @@ package metrics import ( "context" + "fmt" "os" + "strings" "testing" "time" + "github.com/agent-substrate/substrate/internal/ateattr" "github.com/agent-substrate/substrate/internal/e2e" "github.com/agent-substrate/substrate/internal/resources" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" @@ -48,7 +51,7 @@ func TestPlatformMetricsEmitted(t *testing.T) { ctx := context.Background() clients := e2e.GetClients() tmplNS, tmplName := templateRef() - actorID := "metrics-probe" + actorID := fmt.Sprintf("metrics-probe-%d", time.Now().UnixNano()) // CreateActor requires the atespace to exist first; ignore AlreadyExists. _, _ = clients.SubstrateAPI.CreateAtespace(ctx, &ateapipb.CreateAtespaceRequest{ @@ -95,6 +98,28 @@ func TestPlatformMetricsEmitted(t *testing.T) { missing = e2e.MissingPlatformMetrics(scrape, e2e.PlatformMetricPrefixes) ateomSeen = e2e.CollectorHasService(scrape, "ateom-gvisor", "ateom-microvm") if len(missing) == 0 && ateomSeen { + // Verify ate_scheduler_eligible_workers metric carries valid, non-empty namespaced attributes. + hasMetricWithValidLabels := false + for _, line := range strings.Split(scrape, "\n") { + if strings.HasPrefix(line, "ate_scheduler_eligible_workers") && + strings.Contains(line, `ate_workerpool_namespace=`) && + strings.Contains(line, `ate_workerpool_name=`) && + strings.Contains(line, `ate_sandbox_class=`) && + strings.Contains(line, `ate_scheduling_constraint=`) { + nsVal := extractPrometheusLabelValue(line, "ate_workerpool_namespace") + poolVal := extractPrometheusLabelValue(line, "ate_workerpool_name") + classVal := extractPrometheusLabelValue(line, "ate_sandbox_class") + constraintVal := extractPrometheusLabelValue(line, "ate_scheduling_constraint") + if nsVal != "" && poolVal != "" && classVal != "" && + (constraintVal == ateattr.ConstraintNone || constraintVal == ateattr.ConstraintRequiredNodes || constraintVal == ateattr.ConstraintSelector) { + hasMetricWithValidLabels = true + break + } + } + } + if !hasMetricWithValidLabels { + t.Fatalf("ate_scheduler_eligible_workers metric missing valid namespaced attributes in collector scrape output:\n%s", scrape) + } return } time.Sleep(3 * time.Second) @@ -126,3 +151,17 @@ func waitForStatus(t *testing.T, ctx context.Context, clients *e2e.Clients, acto } t.Fatalf("actor %q never reached %v", actorID, want) } + +func extractPrometheusLabelValue(line, labelName string) string { + key := labelName + `="` + idx := strings.Index(line, key) + if idx == -1 { + return "" + } + start := idx + len(key) + end := strings.IndexByte(line[start:], '"') + if end == -1 { + return "" + } + return line[start : start+end] +}