feat: Add control plane telemetry - scheduler eligible workers histogram

[Angela (Angelawork)]

Fixes #564 (Part 3)

• Tests pass
• Appropriate changes to documentation are included in the PR

Description

This PR implements Part 3 of #564 by adding telemetry histogram instrumentation for ate.scheduler.eligible_workers in ateapi. It measures unassigned free worker capacity remaining after all scheduling constraint filters are applied, sampled at every scheduling decision.

Key Changes:

• Updated Scheduler.Schedule() to record eligible candidate workers per pool (recordEligibleWorkers).
  Defined SchedulingConstraintKey (ate.scheduling.constraint) and constraint classification values (none, required_nodes, selector) in internal/ateattr/ateattr.go.
• Added unit tests in scheduling_test.go covering candidate counts, namespaced attributes, zero-capacity fleet states, empty fleets, sandbox class mismatches, draining workers, and constraint classifications.

Testing

• go test -buildvcs=false ./cmd/ateapi/internal/controlapi/...
• go test -buildvcs=false ./cmd/atenet/internal/router/...
• make test

E2E Test

• ./hack/create-kind-cluster.sh
• ./hack/install-ate-kind.sh --deploy-ate-system --deploy-demo-counter
• ./hack/run-e2e.sh ./internal/e2e/suites/metrics/...

[Jeff Luo (JeffLuoo)]

Please rebase and fix e2e tests

[Jeff Luo (JeffLuoo)]

Let's rename it to extractLabelValue as we are not always using Prometheus in the stack. We are using OTel SDK.

[Jeff Luo (JeffLuoo)]

Every other histogram in this repo sets buckets explicitly. The SDK default is [0, 5, 10, 25, 50, 75, 100, 250, ...], so 1 vs 5 eligible workers is one bucket and every percentile above p50 is noise for a typical fleet.

Let's define our own bucket here like:

metric.WithExplicitBucketBoundaries(0, 1, 2, 3, 5, 10, 20, 50, 100, 250)

to start with.

[Jeff Luo (JeffLuoo)]

Let's avoid being too specific in function comment. In the case of future refactoring, if we don't record ate.scheduler.eligible_workers metric here, we may forget to update the comment. Let's keep the comment generic like:

Schedule filters the current worker fleet to find unassigned candidates matching the given constraints

[Jeff Luo (JeffLuoo)]

Suggested change

	// 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) {
	candidates = append(candidates, worker)
	}
	}
	// Record telemetry on the number of eligible workers per pool/namespace before returning
	s.recordEligibleWorkers(ctx, workers, candidates, constraints)
	// Filter for candidate workers that are unassigned and meet all scheduling constraints
	matching := make([]*ateapipb.Worker, 0, len(workers))
	var candidates []*ateapipb.Worker
	for _, worker := range workers {
	if !s.Applies(worker, constraints) {
	continue
	}
	matching = append(matching, worker)
	if worker.GetAssignment() == nil {
	candidates = append(candidates, worker)
	}
	}
	// Record telemetry on the number of eligible workers per pool/namespace before returning
	s.recordEligibleWorkers(ctx, matching, constraints)

this loop already computed Applies for every worker and discarded it for the assigned ones. Keeping both subsets from one pass removes a second round of computation.

[Jeff Luo (JeffLuoo)]

Suggested change

	// 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 (s *scheduler) recordEligibleWorkers(ctx context.Context, matching []*ateapipb.Worker, constraints Constraints) {
	if s.eligibleWorkers == nil {
	return
	}
	// Sandbox class and constraint are constant across every key: Applies requires
	// an exact class match, and the classification is per call. They belong on the
	// Record call, not in the key.
	type key struct{ namespace, pool string }
	eligibleByPool := make(map[key]int64)
	for _, w := range matching {
	k := key{w.GetWorkerNamespace(), w.GetWorkerPool()}
	if _, ok := eligibleByPool[k]; !ok {
	eligibleByPool[k] = 0
	}
	if w.GetAssignment() == nil {
	eligibleByPool[k]++
	}
	}
	// No pool matched the constraints at all. Emit a single zero-valued series so
	// "nothing is schedulable" stays visible; empty namespace/pool marks it. The
	// label set matches the per-pool series, so dashboards need no special case.
	if len(eligibleByPool) == 0 {
	eligibleByPool[key{}] = 0
	}
	constraintStr := classifyConstraint(constraints)
	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(constraints.SandboxClass),
	ateattr.SchedulingConstraintKey.String(constraintStr),
	))
	}
	}

The suggested changes suggest two improvement:

1. Avoid calling the apply on constraint again.
2. When the len(eligibleByPool) == 0 condition is met, not all attributes are emitted in metrics. This will be confusing when we plot the chart in the dashboard. Suggestion is to set the value to 0, and let the emitting logic below to handle the emitting with all 4 labels. A general rule for metric labels (attributes) is never set empty string if possible.

[Jeff Luo (JeffLuoo)]

WithMeter will silently drop the error when instrument-creation errors are seen.

You can reference https://sourcegraph.com/r/github.com/agent-substrate/substrate@main/-/blob/cmd/ateapi/internal/controlapi/metrics.go?L30-35 and follow how RegisterWorkerCount registers the metric, it will explicitly print out errors when seen.

[Jeff Luo (JeffLuoo)]

Also, consider creating a metrics.go file in scheduling folder to align with other metrics implementation. You can move all metrics-related logic to the new metrics.go file.