From 2b72e6483d96db7ea5356db040617ce5fd368fea Mon Sep 17 00:00:00 2001 From: krisztianfekete Date: Fri, 24 Jul 2026 11:18:27 +0200 Subject: [PATCH 1/2] add actor lifecycle + scheduler duration metrics --- .../internal/controlapi/create_actor.go | 18 +- .../internal/controlapi/delete_actor.go | 21 +- .../internal/controlapi/functional_test.go | 9 +- cmd/ateapi/internal/controlapi/metrics.go | 104 +++++++++- .../internal/controlapi/metrics_test.go | 180 ++++++++++++++++++ cmd/ateapi/internal/controlapi/service.go | 7 +- cmd/ateapi/internal/controlapi/workflow.go | 47 +++-- .../internal/controlapi/workflow_resume.go | 27 ++- .../controlapi/workflow_resume_test.go | 25 +++ .../controlapi/workflow_testutil_test.go | 2 +- cmd/ateapi/main.go | 7 +- internal/ateattr/ateattr.go | 48 ++++- internal/ateattr/ateattr_test.go | 36 ++++ internal/e2e/collector_metrics.go | 5 +- 14 files changed, 504 insertions(+), 32 deletions(-) diff --git a/cmd/ateapi/internal/controlapi/create_actor.go b/cmd/ateapi/internal/controlapi/create_actor.go index 14af0cdc4..43ef4f557 100644 --- a/cmd/ateapi/internal/controlapi/create_actor.go +++ b/cmd/ateapi/internal/controlapi/create_actor.go @@ -18,8 +18,10 @@ import ( "context" "errors" "fmt" + "time" "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" + "github.com/agent-substrate/substrate/internal/ateattr" "github.com/agent-substrate/substrate/internal/resources" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" "google.golang.org/grpc/codes" @@ -29,17 +31,25 @@ import ( "k8s.io/apimachinery/pkg/util/validation/field" ) -func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequest) (*ateapipb.Actor, error) { - if err := validateCreateActorRequest(req); err != nil { +func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequest) (created *ateapipb.Actor, err error) { + start := time.Now() + in := req.GetActor() + defer func() { + s.instruments.recordLifecycleOp(ctx, ateattr.OperationCreate, start, err, + ateattr.TemplateNameKey.String(in.GetActorTemplateName()), + ateattr.TemplateNamespaceKey.String(in.GetActorTemplateNamespace()), + ) + }() + + if err = validateCreateActorRequest(req); err != nil { return nil, err } - in := req.GetActor() templateNamespace := in.GetActorTemplateNamespace() templateName := in.GetActorTemplateName() setSpanActorRefAttributes(ctx, in.GetMetadata().GetAtespace(), in.GetMetadata().GetName()) - _, err := s.actorTemplateLister.ActorTemplates(templateNamespace).Get(templateName) + _, err = s.actorTemplateLister.ActorTemplates(templateNamespace).Get(templateName) if err != nil { if k8serrors.IsNotFound(err) { return nil, status.Errorf(codes.FailedPrecondition, "ActorTemplate %s/%s not found", templateNamespace, templateName) diff --git a/cmd/ateapi/internal/controlapi/delete_actor.go b/cmd/ateapi/internal/controlapi/delete_actor.go index 016a99203..1d14e6023 100644 --- a/cmd/ateapi/internal/controlapi/delete_actor.go +++ b/cmd/ateapi/internal/controlapi/delete_actor.go @@ -18,8 +18,10 @@ import ( "context" "errors" "fmt" + "time" "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" + "github.com/agent-substrate/substrate/internal/ateattr" "github.com/agent-substrate/substrate/internal/resources" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" "google.golang.org/grpc/codes" @@ -27,13 +29,24 @@ import ( "k8s.io/apimachinery/pkg/util/validation/field" ) -func (s *Service) DeleteActor(ctx context.Context, req *ateapipb.DeleteActorRequest) (*ateapipb.Actor, error) { - if err := validateDeleteActorRequest(req); err != nil { +func (s *Service) DeleteActor(ctx context.Context, req *ateapipb.DeleteActorRequest) (deleted *ateapipb.Actor, err error) { + start := time.Now() + // The request only names the actor, so tmpl (for the metric's template dims) + // is filled from whichever record we manage to load. + var tmpl *ateapipb.Actor + defer func() { + s.instruments.recordLifecycleOp(ctx, ateattr.OperationDelete, start, err, + ateattr.TemplateNameKey.String(tmpl.GetActorTemplateName()), + ateattr.TemplateNamespaceKey.String(tmpl.GetActorTemplateNamespace()), + ) + }() + + if err = validateDeleteActorRequest(req); err != nil { return nil, err } setSpanActorRefAttributes(ctx, req.GetActor().GetAtespace(), req.GetActor().GetName()) - deleted, err := s.persistence.DeleteActor(ctx, req.GetActor().GetAtespace(), req.GetActor().GetName()) + deleted, err = s.persistence.DeleteActor(ctx, req.GetActor().GetAtespace(), req.GetActor().GetName()) if err != nil { if errors.Is(err, store.ErrNotFound) { return nil, status.Errorf(codes.NotFound, "Actor %s not found", req.GetActor().GetName()) @@ -41,6 +54,7 @@ func (s *Service) DeleteActor(ctx context.Context, req *ateapipb.DeleteActorRequ if errors.Is(err, store.ErrFailedPrecondition) { current, getErr := s.persistence.GetActor(ctx, req.GetActor().GetAtespace(), req.GetActor().GetName()) if getErr == nil { + tmpl = current return nil, status.Errorf(codes.FailedPrecondition, "Actor %s is not suspended (status: %v)", req.GetActor().GetName(), current.GetStatus()) } return nil, status.Errorf(codes.FailedPrecondition, "Actor %s is not suspended", req.GetActor().GetName()) @@ -51,6 +65,7 @@ func (s *Service) DeleteActor(ctx context.Context, req *ateapipb.DeleteActorRequ return nil, fmt.Errorf("while deleting actor from DB: %w", err) } + tmpl = deleted return deleted, nil } diff --git a/cmd/ateapi/internal/controlapi/functional_test.go b/cmd/ateapi/internal/controlapi/functional_test.go index 55429e6aa..38f450d65 100644 --- a/cmd/ateapi/internal/controlapi/functional_test.go +++ b/cmd/ateapi/internal/controlapi/functional_test.go @@ -40,6 +40,7 @@ import ( "github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp/cmpopts" "github.com/redis/go-redis/v9" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials/insecure" @@ -313,7 +314,13 @@ func setupTest(t *testing.T, ns string) *testContext { } dialer := NewAteletDialer(workerInformer.GetIndexer(), ateletInformer.GetIndexer()) - service := NewService(persistence, wc, actorTemplateLister, workerPoolLister, sandboxConfigLister, dialer, k8sClient) + instruments, err := NewInstruments(sdkmetric.NewMeterProvider(sdkmetric.WithReader(sdkmetric.NewManualReader())).Meter("ateapi")) + if err != nil { + cancel() + mr.Close() + t.Fatalf("failed to create metric instruments: %v", err) + } + service := NewService(persistence, wc, actorTemplateLister, workerPoolLister, sandboxConfigLister, dialer, k8sClient, instruments) // 5. Start REAL gRPC Server for ATE API grpcServer := grpc.NewServer(grpc.UnaryInterceptor(ateinterceptors.ServerUnaryInterceptor)) diff --git a/cmd/ateapi/internal/controlapi/metrics.go b/cmd/ateapi/internal/controlapi/metrics.go index 7ee58302c..4d37babd7 100644 --- a/cmd/ateapi/internal/controlapi/metrics.go +++ b/cmd/ateapi/internal/controlapi/metrics.go @@ -17,15 +17,22 @@ package controlapi import ( "context" "fmt" + "time" "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/attribute" "go.opentelemetry.io/otel/metric" + "google.golang.org/grpc/status" "k8s.io/apimachinery/pkg/labels" ) -const workerpoolWorkersMetric = "ate.workerpool.workers" +const ( + workerpoolWorkersMetric = "ate.workerpool.workers" + lifecycleOpDurationMetric = "ate.actor.lifecycle.operation.duration" + schedulerAssignmentMetric = "ate.scheduler.assignment.duration" +) // RegisterWorkerCount wires the ate.workerpool.workers observable against workers // (workercache.Cache.Workers in prod) and listPools (a WorkerPool lister's List, @@ -84,3 +91,98 @@ func RegisterWorkerCount(meter metric.Meter, workers func() ([]*ateapipb.Worker, } return nil } + +// Instruments holds ateapi's actor-lifecycle and scheduler duration histograms. +// A nil *Instruments is a valid no-op, so call sites need no guard. Worker-count +// is registered separately (RegisterWorkerCount): a callback-driven observable, +// not a synchronous instrument. +type Instruments struct { + lifecycleOpDuration metric.Float64Histogram + schedulerAssignmentDuration metric.Float64Histogram +} + +// NewInstruments builds the two histograms against meter. Assignment buckets are +// an order of magnitude finer than lifecycle's: it is a sub-second in-memory pick, +// not a multi-second restore. +func NewInstruments(meter metric.Meter) (*Instruments, error) { + lifecycleOpDuration, err := meter.Float64Histogram( + lifecycleOpDurationMetric, + metric.WithUnit("s"), + metric.WithDescription("Duration of an actor lifecycle operation (create, resume, suspend, pause, delete) handled by ateapi."), + metric.WithExplicitBucketBoundaries(0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.15, 0.25, 0.5, 1, 2.5, 5, 10, 30), + ) + if err != nil { + return nil, fmt.Errorf("create %s histogram: %w", lifecycleOpDurationMetric, err) + } + + schedulerAssignmentDuration, err := meter.Float64Histogram( + schedulerAssignmentMetric, + metric.WithUnit("s"), + metric.WithDescription("Duration of the scheduler's worker-assignment step during actor resume."), + metric.WithExplicitBucketBoundaries(0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1), + ) + if err != nil { + return nil, fmt.Errorf("create %s histogram: %w", schedulerAssignmentMetric, err) + } + + return &Instruments{ + lifecycleOpDuration: lifecycleOpDuration, + schedulerAssignmentDuration: schedulerAssignmentDuration, + }, nil +} + +// recordLifecycleOp records op's duration. A non-nil err is classified onto +// error.type via its gRPC status code; error.type's absence marks success, so +// there is no parallel failure counter. extraAttrs carries the per-operation +// dimensions (template, pool, class, snapshot kind). +func (i *Instruments) recordLifecycleOp(ctx context.Context, op string, start time.Time, err error, extraAttrs ...attribute.KeyValue) { + if i == nil || i.lifecycleOpDuration == nil { + return + } + attrs := make([]attribute.KeyValue, 0, len(extraAttrs)+2) + attrs = append(attrs, ateattr.ActorOperationNameKey.String(op)) + attrs = append(attrs, extraAttrs...) + if err != nil { + attrs = append(attrs, ateattr.ErrorTypeKey.String(status.Code(err).String())) + } + i.lifecycleOpDuration.Record(ctx, time.Since(start).Seconds(), metric.WithAttributes(attrs...)) +} + +// lifecycleOpAttrs builds the resume/suspend/pause dimensions from workflow +// state. Nil-safe, and omits the pool and snapshot-kind labels until they are +// known so a failure before the assign/restore steps never emits an empty-string +// series. snapshotKind is empty for suspend/pause, which do not restore. +func lifecycleOpAttrs(actor *ateapipb.Actor, template *atev1alpha1.ActorTemplate, snapshotKind string) []attribute.KeyValue { + attrs := []attribute.KeyValue{ + ateattr.TemplateNameKey.String(actor.GetActorTemplateName()), + ateattr.TemplateNamespaceKey.String(actor.GetActorTemplateNamespace()), + } + if pool := actor.GetWorkerPoolName(); pool != "" { + attrs = append(attrs, ateattr.WorkerPoolNameKey.String(pool)) + } + if template != nil { + attrs = append(attrs, ateattr.SandboxClassKey.String(string(template.Spec.SandboxClass))) + } + if snapshotKind != "" { + attrs = append(attrs, ateattr.SnapshotKindKey.String(snapshotKind)) + } + return attrs +} + +// recordSchedulerAssignment records one assignment attempt. pool is stamped only +// when a worker was assigned and error.type only for the Error outcome, so +// no_free_worker (a capacity signal, not a failure) carries neither. +func (i *Instruments) recordSchedulerAssignment(ctx context.Context, start time.Time, outcome, pool string, err error) { + if i == nil || i.schedulerAssignmentDuration == nil { + return + } + attrs := make([]attribute.KeyValue, 0, 3) + attrs = append(attrs, ateattr.SchedulerOutcomeKey.String(outcome)) + if pool != "" { + attrs = append(attrs, ateattr.WorkerPoolNameKey.String(pool)) + } + if outcome == ateattr.SchedulerOutcomeError && err != nil { + attrs = append(attrs, ateattr.ErrorTypeKey.String(status.Code(err).String())) + } + i.schedulerAssignmentDuration.Record(ctx, time.Since(start).Seconds(), metric.WithAttributes(attrs...)) +} diff --git a/cmd/ateapi/internal/controlapi/metrics_test.go b/cmd/ateapi/internal/controlapi/metrics_test.go index 736389149..08ff5f73e 100644 --- a/cmd/ateapi/internal/controlapi/metrics_test.go +++ b/cmd/ateapi/internal/controlapi/metrics_test.go @@ -17,12 +17,16 @@ package controlapi import ( "context" "testing" + "time" "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/attribute" sdkmetric "go.opentelemetry.io/otel/sdk/metric" "go.opentelemetry.io/otel/sdk/metric/metricdata" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" ) @@ -134,6 +138,182 @@ func TestWorkerCountSkipsWhenCacheNotReady(t *testing.T) { } } +// newTestInstruments builds the lifecycle/scheduler histograms on a local +// ManualReader-backed provider so tests stay parallel-safe and never touch the +// global meter provider. +func newTestInstruments(t *testing.T) (*Instruments, *sdkmetric.ManualReader) { + t.Helper() + reader := sdkmetric.NewManualReader() + mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + inst, err := NewInstruments(mp.Meter("ateapi")) + if err != nil { + t.Fatalf("NewInstruments: %v", err) + } + return inst, reader +} + +// singleHistogramDP asserts name is a float histogram in seconds with exactly one +// datapoint and returns it. +func singleHistogramDP(t *testing.T, reader *sdkmetric.ManualReader, name string) metricdata.HistogramDataPoint[float64] { + t.Helper() + m := mustMetric(t, reader, name) + if m.Unit != "s" { + t.Errorf("%s unit = %q, want s", name, m.Unit) + } + hist, ok := m.Data.(metricdata.Histogram[float64]) + if !ok { + t.Fatalf("%s data type = %T, want Histogram[float64]", name, m.Data) + } + if len(hist.DataPoints) != 1 { + t.Fatalf("%s: got %d datapoints, want 1", name, len(hist.DataPoints)) + } + return hist.DataPoints[0] +} + +// assertAttrKeys asserts the datapoint carries exactly want, order-independent. +func assertAttrKeys(t *testing.T, dp metricdata.HistogramDataPoint[float64], want ...attribute.Key) { + t.Helper() + got := make(map[attribute.Key]bool, dp.Attributes.Len()) + for _, kv := range dp.Attributes.ToSlice() { + got[kv.Key] = true + } + if len(got) != len(want) { + t.Errorf("attribute keys = %v, want %v", dp.Attributes.ToSlice(), want) + } + for _, k := range want { + if !got[k] { + t.Errorf("missing attribute key %s; got %v", k, dp.Attributes.ToSlice()) + } + } +} + +// attrString returns the string value of key k and whether it is present. +func attrString(dp metricdata.HistogramDataPoint[float64], k attribute.Key) (string, bool) { + v, ok := dp.Attributes.Value(k) + return v.AsString(), ok +} + +func TestLifecycleOpDurationShape(t *testing.T) { + inst, reader := newTestInstruments(t) + + actor := &ateapipb.Actor{ + ActorTemplateName: "support-agent", + ActorTemplateNamespace: "ate-agents", + WorkerPoolName: "pool-a", + } + template := &atev1alpha1.ActorTemplate{ + Spec: atev1alpha1.ActorTemplateSpec{SandboxClass: atev1alpha1.SandboxClassGvisor}, + } + inst.recordLifecycleOp(context.Background(), ateattr.OperationResume, time.Now(), nil, + lifecycleOpAttrs(actor, template, ateattr.SnapshotKindLatest)...) + + dp := singleHistogramDP(t, reader, lifecycleOpDurationMetric) + assertAttrKeys(t, dp, + ateattr.ActorOperationNameKey, + ateattr.TemplateNameKey, + ateattr.TemplateNamespaceKey, + ateattr.WorkerPoolNameKey, + ateattr.SandboxClassKey, + ateattr.SnapshotKindKey, + ) + if op, _ := attrString(dp, ateattr.ActorOperationNameKey); op != ateattr.OperationResume { + t.Errorf("operation = %q, want %q", op, ateattr.OperationResume) + } +} + +// TestRecordLifecycleOp_OutcomeClassification asserts success omits error.type and +// each gRPC failure maps onto its status-code string; the absence of error.type is +// the success signal, so there is no separate failure counter. +func TestRecordLifecycleOp_OutcomeClassification(t *testing.T) { + tests := []struct { + name string + op string + err error + wantErrorType string // empty means error.type must be absent + }{ + {name: "create success", op: ateattr.OperationCreate, err: nil, wantErrorType: ""}, + {name: "create not found", op: ateattr.OperationCreate, err: status.Error(codes.NotFound, "missing"), wantErrorType: "NotFound"}, + {name: "resume aborted", op: ateattr.OperationResume, err: status.Error(codes.Aborted, "conflict"), wantErrorType: "Aborted"}, + {name: "resume crash", op: ateattr.OperationResume, err: status.Error(codes.DataLoss, "crashed"), wantErrorType: "DataLoss"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + inst, reader := newTestInstruments(t) + inst.recordLifecycleOp(context.Background(), tt.op, time.Now(), tt.err, + ateattr.TemplateNameKey.String("support-agent"), + ateattr.TemplateNamespaceKey.String("ate-agents"), + ) + + dp := singleHistogramDP(t, reader, lifecycleOpDurationMetric) + if op, _ := attrString(dp, ateattr.ActorOperationNameKey); op != tt.op { + t.Errorf("operation = %q, want %q", op, tt.op) + } + gotErrType, hasErrType := attrString(dp, ateattr.ErrorTypeKey) + if tt.wantErrorType == "" { + if hasErrType { + t.Errorf("error.type = %q, want absent", gotErrType) + } + } else if !hasErrType || gotErrType != tt.wantErrorType { + t.Errorf("error.type = %q (present=%v), want %q", gotErrType, hasErrType, tt.wantErrorType) + } + }) + } +} + +// TestSchedulerAssignmentShapeAndOutcomes asserts the assignment histogram stamps +// pool only when a worker was assigned and error.type only for the error outcome, +// so no_free_worker (a capacity signal) carries neither. +func TestSchedulerAssignmentShapeAndOutcomes(t *testing.T) { + tests := []struct { + name string + outcome string + pool string + err error + wantKeys []attribute.Key + wantErrorType string + }{ + { + name: "assigned stamps pool, no error.type", + outcome: ateattr.SchedulerOutcomeAssigned, + pool: "pool-a", + err: nil, + wantKeys: []attribute.Key{ateattr.SchedulerOutcomeKey, ateattr.WorkerPoolNameKey}, + }, + { + name: "no_free_worker carries neither pool nor error.type", + outcome: ateattr.SchedulerOutcomeNoFreeWorker, + pool: "", + err: status.Error(codes.FailedPrecondition, "no free workers available"), + wantKeys: []attribute.Key{ateattr.SchedulerOutcomeKey}, + }, + { + name: "error carries error.type, no pool", + outcome: ateattr.SchedulerOutcomeError, + pool: "", + err: status.Error(codes.Internal, "boom"), + wantKeys: []attribute.Key{ateattr.SchedulerOutcomeKey, ateattr.ErrorTypeKey}, + wantErrorType: "Internal", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + inst, reader := newTestInstruments(t) + inst.recordSchedulerAssignment(context.Background(), time.Now(), tt.outcome, tt.pool, tt.err) + + dp := singleHistogramDP(t, reader, schedulerAssignmentMetric) + assertAttrKeys(t, dp, tt.wantKeys...) + if o, _ := attrString(dp, ateattr.SchedulerOutcomeKey); o != tt.outcome { + t.Errorf("outcome = %q, want %q", o, tt.outcome) + } + if tt.wantErrorType != "" { + if et, ok := attrString(dp, ateattr.ErrorTypeKey); !ok || et != tt.wantErrorType { + t.Errorf("error.type = %q (present=%v), want %q", et, ok, tt.wantErrorType) + } + } + }) + } +} + func workerPool(name string, class atev1alpha1.SandboxClass) *atev1alpha1.WorkerPool { return &atev1alpha1.WorkerPool{ ObjectMeta: metav1.ObjectMeta{Name: name}, diff --git a/cmd/ateapi/internal/controlapi/service.go b/cmd/ateapi/internal/controlapi/service.go index 7ec695b00..75c846023 100644 --- a/cmd/ateapi/internal/controlapi/service.go +++ b/cmd/ateapi/internal/controlapi/service.go @@ -30,11 +30,12 @@ type Service struct { actorTemplateLister listersv1alpha1.ActorTemplateLister workerPoolLister listersv1alpha1.WorkerPoolLister actorWorkflow *ActorWorkflow + instruments *Instruments } var _ ateapipb.ControlServer = (*Service)(nil) -// NewService creates a service. +// NewService creates a service. instruments may be nil; the record helpers no-op. func NewService( persistence store.Interface, workerCache *workercache.Cache, @@ -43,13 +44,15 @@ func NewService( sandboxConfigLister listersv1alpha1.SandboxConfigLister, dialer *AteletDialer, kubeClient kubernetes.Interface, + instruments *Instruments, ) *Service { s := &Service{ persistence: persistence, actorTemplateLister: actorTemplateLister, workerPoolLister: workerPoolLister, dialer: dialer, - actorWorkflow: NewActorWorkflow(persistence, workerCache, dialer, actorTemplateLister, workerPoolLister, sandboxConfigLister, kubeClient), + actorWorkflow: NewActorWorkflow(persistence, workerCache, dialer, actorTemplateLister, workerPoolLister, sandboxConfigLister, kubeClient, instruments), + instruments: instruments, } return s diff --git a/cmd/ateapi/internal/controlapi/workflow.go b/cmd/ateapi/internal/controlapi/workflow.go index 73f4a17dc..1e258d22f 100644 --- a/cmd/ateapi/internal/controlapi/workflow.go +++ b/cmd/ateapi/internal/controlapi/workflow.go @@ -18,9 +18,11 @@ import ( "context" "errors" "fmt" + "time" "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" "github.com/agent-substrate/substrate/cmd/ateapi/internal/workercache" + "github.com/agent-substrate/substrate/internal/ateattr" listersv1alpha1 "github.com/agent-substrate/substrate/pkg/client/listers/api/v1alpha1" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" "go.opentelemetry.io/otel" @@ -136,9 +138,10 @@ type ActorWorkflow struct { sandboxConfigLister listersv1alpha1.SandboxConfigLister kubeClient kubernetes.Interface secretCache *envSecretCache + instruments *Instruments } -// NewActorWorkflow creates a new ActorWorkflow. +// NewActorWorkflow creates a new ActorWorkflow. instruments may be nil. func NewActorWorkflow( store store.Interface, workerCache *workercache.Cache, @@ -147,6 +150,7 @@ func NewActorWorkflow( workerPoolLister listersv1alpha1.WorkerPoolLister, sandboxConfigLister listersv1alpha1.SandboxConfigLister, kubeClient kubernetes.Interface, + instruments *Instruments, ) *ActorWorkflow { return &ActorWorkflow{ store: store, @@ -157,11 +161,13 @@ func NewActorWorkflow( sandboxConfigLister: sandboxConfigLister, kubeClient: kubeClient, secretCache: newEnvSecretCache(envSecretCacheTTL), + instruments: instruments, } } // ResumeActor executes the workflow to resume a suspended actor. Idempotent. -func (w *ActorWorkflow) ResumeActor(ctx context.Context, atespace, name string, boot bool) (*ateapipb.Actor, error) { +func (w *ActorWorkflow) ResumeActor(ctx context.Context, atespace, name string, boot bool) (actor *ateapipb.Actor, err error) { + start := time.Now() input := &ResumeInput{ ActorName: name, Atespace: atespace, @@ -169,7 +175,14 @@ func (w *ActorWorkflow) ResumeActor(ctx context.Context, atespace, name string, } state := &ResumeState{} - ctx, lock, err := w.acquireActorLock(ctx, atespace, name) + // Recorded before the lock so lock contention still counts as an attempt; the + // incoming ctx stays valid where acquireActorLock returns nil on failure. + defer func() { + w.instruments.recordLifecycleOp(ctx, ateattr.OperationResume, start, err, + lifecycleOpAttrs(state.Actor, state.ActorTemplate, state.SnapshotKind)...) + }() + + lockCtx, lock, err := w.acquireActorLock(ctx, atespace, name) if err != nil { return nil, err } @@ -177,12 +190,12 @@ func (w *ActorWorkflow) ResumeActor(ctx context.Context, atespace, name string, steps := []WorkflowStep[*ResumeInput, *ResumeState]{ &LoadActorForResumeStep{store: w.store, actorTemplateLister: w.actorTemplateLister}, - &AssignWorkerStep{store: w.store, workerCache: w.workerCache}, + &AssignWorkerStep{store: w.store, workerCache: w.workerCache, instruments: w.instruments}, &CallAteletRestoreStep{store: w.store, dialer: w.dialer, kubeClient: w.kubeClient, secretCache: w.secretCache, workerPoolLister: w.workerPoolLister, sandboxConfigLister: w.sandboxConfigLister}, &FinalizeRunningStep{store: w.store}, } - if err := RunWorkflow(ctx, input, state, steps); err != nil { + if err = RunWorkflow(lockCtx, input, state, steps); err != nil { return nil, err } @@ -190,14 +203,20 @@ func (w *ActorWorkflow) ResumeActor(ctx context.Context, atespace, name string, } // SuspendActor executes the workflow to suspend a running actor. Idempotent. -func (w *ActorWorkflow) SuspendActor(ctx context.Context, atespace, name string) (*ateapipb.Actor, error) { +func (w *ActorWorkflow) SuspendActor(ctx context.Context, atespace, name string) (actor *ateapipb.Actor, err error) { + start := time.Now() input := &SuspendInput{ ActorName: name, Atespace: atespace, } state := &SuspendState{} - ctx, lock, err := w.acquireActorLock(ctx, atespace, name) + defer func() { + w.instruments.recordLifecycleOp(ctx, ateattr.OperationSuspend, start, err, + lifecycleOpAttrs(state.Actor, state.ActorTemplate, "")...) + }() + + lockCtx, lock, err := w.acquireActorLock(ctx, atespace, name) if err != nil { return nil, err } @@ -210,7 +229,7 @@ func (w *ActorWorkflow) SuspendActor(ctx context.Context, atespace, name string) &FinalizeSuspendedStep{store: w.store}, } - if err := RunWorkflow(ctx, input, state, steps); err != nil { + if err = RunWorkflow(lockCtx, input, state, steps); err != nil { return nil, err } @@ -218,14 +237,20 @@ func (w *ActorWorkflow) SuspendActor(ctx context.Context, atespace, name string) } // PauseActor executes the workflow to pause a running actor. Idempotent. -func (w *ActorWorkflow) PauseActor(ctx context.Context, atespace, name string) (*ateapipb.Actor, error) { +func (w *ActorWorkflow) PauseActor(ctx context.Context, atespace, name string) (actor *ateapipb.Actor, err error) { + start := time.Now() input := &PauseInput{ ActorName: name, Atespace: atespace, } state := &PauseState{} - ctx, lock, err := w.acquireActorLock(ctx, atespace, name) + defer func() { + w.instruments.recordLifecycleOp(ctx, ateattr.OperationPause, start, err, + lifecycleOpAttrs(state.Actor, state.ActorTemplate, "")...) + }() + + lockCtx, lock, err := w.acquireActorLock(ctx, atespace, name) if err != nil { return nil, err } @@ -238,7 +263,7 @@ func (w *ActorWorkflow) PauseActor(ctx context.Context, atespace, name string) ( &FinalizePausedStep{store: w.store}, } - if err := RunWorkflow(ctx, input, state, steps); err != nil { + if err = RunWorkflow(lockCtx, input, state, steps); err != nil { return nil, err } diff --git a/cmd/ateapi/internal/controlapi/workflow_resume.go b/cmd/ateapi/internal/controlapi/workflow_resume.go index 3720cd59b..8a1c9b3aa 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume.go @@ -25,6 +25,7 @@ import ( "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" "github.com/agent-substrate/substrate/cmd/ateapi/internal/workercache" + "github.com/agent-substrate/substrate/internal/ateattr" "github.com/agent-substrate/substrate/internal/proto/ateletpb" atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" listersv1alpha1 "github.com/agent-substrate/substrate/pkg/client/listers/api/v1alpha1" @@ -50,6 +51,7 @@ type ResumeState struct { Actor *ateapipb.Actor Worker *ateapipb.Worker ActorTemplate *atev1alpha1.ActorTemplate + SnapshotKind string } type LoadActorForResumeStep struct { @@ -142,6 +144,7 @@ func isWorkerEligibleForActor(worker *ateapipb.Worker, templateClass atev1alpha1 type AssignWorkerStep struct { store store.Interface workerCache *workercache.Cache + instruments *Instruments } func (s *AssignWorkerStep) Name() string { return "AssignWorker" } @@ -160,7 +163,23 @@ func (s *AssignWorkerStep) CheckPrerequisite(ctx context.Context, input *ResumeI } } -func (s *AssignWorkerStep) Execute(ctx context.Context, input *ResumeInput, state *ResumeState) error { +// schedulerRecordable excludes retried persistence conflicts: runStep re-runs +// Execute transparently on store.ErrPersistenceRetry, so counting those attempts +// would inflate the error rate and double-count the eventual success. +func schedulerRecordable(err error) bool { + return !errors.Is(err, store.ErrPersistenceRetry) +} + +func (s *AssignWorkerStep) Execute(ctx context.Context, input *ResumeInput, state *ResumeState) (err error) { + start := time.Now() + outcome := ateattr.SchedulerOutcomeError + pool := "" + defer func() { + if schedulerRecordable(err) { + s.instruments.recordSchedulerAssignment(ctx, start, outcome, pool, err) + } + }() + workers, err := s.workerCache.Workers() if err != nil { return fmt.Errorf("while listing workers: %w", err) @@ -210,6 +229,7 @@ func (s *AssignWorkerStep) Execute(ctx context.Context, input *ResumeInput, stat return err } if pickedWorker == nil { + outcome = ateattr.SchedulerOutcomeNoFreeWorker return status.Errorf(codes.FailedPrecondition, "no free workers available") } @@ -248,6 +268,8 @@ func (s *AssignWorkerStep) Execute(ctx context.Context, input *ResumeInput, stat } state.Actor = updatedActor state.Worker = assignedWorker + pool = assignedWorker.GetWorkerPool() + outcome = ateattr.SchedulerOutcomeAssigned return nil } @@ -360,6 +382,7 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput, if data := state.Actor.GetLatestSnapshotInfo().GetData(); data != nil { slog.InfoContext(ctx, "Actor has snapshot; Restoring from snapshot") + state.SnapshotKind = ateattr.SnapshotKindLatest req := &ateletpb.RestoreRequest{ TargetAteomUid: state.Actor.GetAteomPodUid(), @@ -395,6 +418,7 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput, return maybeCrashActor(ctx, s.store, input.Atespace, input.ActorName, err, "while restoring workload") } else if state.ActorTemplate.Status.GoldenSnapshot != "" && !input.Boot { slog.InfoContext(ctx, "Actor has no snapshot; ActorTemplate has golden snapshot; Restoring from golden snapshot") + state.SnapshotKind = ateattr.SnapshotKindGolden snapshot := state.ActorTemplate.Status.GoldenSnapshot @@ -418,6 +442,7 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput, return maybeCrashActor(ctx, s.store, input.Atespace, input.ActorName, err, "while creating workload from golden snapshot") } else { slog.InfoContext(ctx, "Actor has no snapshot; ActorTemplate has no golden snapshot; Booting from ActorTemplate spec") + state.SnapshotKind = ateattr.SnapshotKindBoot // Booting from scratch: resolve the sandbox binaries from the pool's // SandboxConfig and send them so atelet can fetch and record them. diff --git a/cmd/ateapi/internal/controlapi/workflow_resume_test.go b/cmd/ateapi/internal/controlapi/workflow_resume_test.go index 2593a4e20..3b2cd445b 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume_test.go @@ -16,9 +16,11 @@ package controlapi import ( "context" + "fmt" "testing" "time" + "github.com/agent-substrate/substrate/cmd/ateapi/internal/store" "github.com/agent-substrate/substrate/cmd/ateapi/internal/store/storetest" "github.com/agent-substrate/substrate/cmd/ateapi/internal/workercache" atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" @@ -29,6 +31,29 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) +// TestSchedulerRecordable guards the retry-dedup rule: runStep re-runs Execute on +// store.ErrPersistenceRetry, and those attempts (raw or wrapped) must not be +// recorded, while the terminal success or real error must be. +func TestSchedulerRecordable(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + {name: "success is recorded", err: nil, want: true}, + {name: "retry conflict is skipped", err: store.ErrPersistenceRetry, want: false}, + {name: "wrapped retry conflict is skipped", err: fmt.Errorf("update worker: %w", store.ErrPersistenceRetry), want: false}, + {name: "real error is recorded", err: status.Error(codes.Internal, "boom"), want: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := schedulerRecordable(tt.err); got != tt.want { + t.Errorf("schedulerRecordable(%v) = %v, want %v", tt.err, got, tt.want) + } + }) + } +} + func TestIsWorkerEligibleForActor(t *testing.T) { tests := []struct { name string diff --git a/cmd/ateapi/internal/controlapi/workflow_testutil_test.go b/cmd/ateapi/internal/controlapi/workflow_testutil_test.go index c6962694a..c98cd4e69 100644 --- a/cmd/ateapi/internal/controlapi/workflow_testutil_test.go +++ b/cmd/ateapi/internal/controlapi/workflow_testutil_test.go @@ -41,7 +41,7 @@ func newTestActorWorkflow(t *testing.T, st store.Interface, tmplNamespace, tmplN }); err != nil { t.Fatalf("add template to indexer: %v", err) } - return NewActorWorkflow(st, nil, nil, listersv1alpha1.NewActorTemplateLister(indexer), nil, nil, nil) + return NewActorWorkflow(st, nil, nil, listersv1alpha1.NewActorTemplateLister(indexer), nil, nil, nil, nil) } // seedWorkflowActor stores an actor with the given status, bound to the given diff --git a/cmd/ateapi/main.go b/cmd/ateapi/main.go index 575924b21..58d0be407 100644 --- a/cmd/ateapi/main.go +++ b/cmd/ateapi/main.go @@ -159,8 +159,13 @@ func main() { serverboot.Fatal(ctx, "Failed to register worker-count metric", err) } + instruments, err := controlapi.NewInstruments(otel.Meter("ateapi")) + if err != nil { + serverboot.Fatal(ctx, "Failed to create metric instruments", err) + } + dialer := controlapi.NewAteletDialer(workerPodInformer.GetIndexer(), ateletPodInformer.GetIndexer()) - sm := controlapi.NewService(redisPersistence, workerCache, actorTemplateLister, workerPoolLister, sandboxConfigLister, dialer, clientset) + sm := controlapi.NewService(redisPersistence, workerCache, actorTemplateLister, workerPoolLister, sandboxConfigLister, dialer, clientset, instruments) jwtIssuerDiscoveryClient := buildK8sServiceAccountIssuerDiscoveryClient(ctx, *clientJWTCAFile, *clientJWTIssuer) if authModeParsed == ateapiauth.ModeJWT && jwtIssuerDiscoveryClient == nil { diff --git a/internal/ateattr/ateattr.go b/internal/ateattr/ateattr.go index a2cd067ba..c2d298fa5 100644 --- a/internal/ateattr/ateattr.go +++ b/internal/ateattr/ateattr.go @@ -46,14 +46,23 @@ 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. -// WorkerStateKey stays worker-rooted rather than nesting under the pool so it -// can grow siblings. +// 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 ( - WorkerPoolNameKey = attribute.Key("ate.workerpool.name") - WorkerStateKey = attribute.Key("ate.worker.state") - SandboxClassKey = attribute.Key("ate.sandbox.class") + 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") + 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 WorkerStateKey. Only idle and assigned are representable today; // starting and unhealthy workers are not modeled in the cache. const ( @@ -61,6 +70,35 @@ const ( WorkerStateAssigned = "assigned" ) +// Values for ActorOperationNameKey: the five actor lifecycle operations ateapi +// serves. +const ( + OperationCreate = "create" + OperationResume = "resume" + OperationSuspend = "suspend" + OperationPause = "pause" + OperationDelete = "delete" +) + +// Values for SchedulerOutcomeKey. NoFreeWorker is a capacity signal, not a +// failure, so it is a distinct outcome rather than an error.type value; only the +// Error outcome carries an error.type. +const ( + SchedulerOutcomeAssigned = "assigned" + SchedulerOutcomeNoFreeWorker = "no_free_worker" + SchedulerOutcomeError = "error" +) + +// Values for SnapshotKindKey, stamped by ateapi from its own resume branching, so +// the label is bounded at the producer: Latest restores the actor's most recent +// snapshot, Golden restores the template's golden image, Boot is a from-scratch +// start (not a restore). +const ( + SnapshotKindGolden = "golden" + SnapshotKindLatest = "latest" + SnapshotKindBoot = "boot" +) + // 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 794d3881c..088585042 100644 --- a/internal/ateattr/ateattr_test.go +++ b/internal/ateattr/ateattr_test.go @@ -150,9 +150,13 @@ 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"}, + {SchedulerOutcomeKey, "ate.scheduler.outcome"}, + {ErrorTypeKey, "error.type"}, } for _, tt := range tests { t.Run(tt.want, func(t *testing.T) { @@ -162,3 +166,35 @@ func TestKeySpellings(t *testing.T) { }) } } + +// TestMetricLabelValues pins the wire value of every bounded metric-label +// constant. These are the exact strings dashboards and alerts group by, so a +// typo or rename must fail here rather than silently fork a time series in +// production. +func TestMetricLabelValues(t *testing.T) { + tests := []struct { + got string + want string + }{ + {WorkerStateIdle, "idle"}, + {WorkerStateAssigned, "assigned"}, + {OperationCreate, "create"}, + {OperationResume, "resume"}, + {OperationSuspend, "suspend"}, + {OperationPause, "pause"}, + {OperationDelete, "delete"}, + {SchedulerOutcomeAssigned, "assigned"}, + {SchedulerOutcomeNoFreeWorker, "no_free_worker"}, + {SchedulerOutcomeError, "error"}, + {SnapshotKindGolden, "golden"}, + {SnapshotKindLatest, "latest"}, + {SnapshotKindBoot, "boot"}, + } + for _, tt := range tests { + t.Run(tt.want, func(t *testing.T) { + if tt.got != tt.want { + t.Errorf("value = %q, want %q", tt.got, tt.want) + } + }) + } +} diff --git a/internal/e2e/collector_metrics.go b/internal/e2e/collector_metrics.go index 6249631a5..4c7e4056c 100644 --- a/internal/e2e/collector_metrics.go +++ b/internal/e2e/collector_metrics.go @@ -36,10 +36,11 @@ const ( // 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. +// slice lands and as more components are wired to push to the collector. var PlatformMetricPrefixes = []string{ "ate_workerpool_workers", + "ate_actor_lifecycle_operation_duration", + "ate_scheduler_assignment_duration", } // ScrapeCollectorMetrics port-forwards the kind stack's OTel Collector and reads From ecf0aa2496b584b3dab5e5c36c5bd6bfb09a8008 Mon Sep 17 00:00:00 2001 From: krisztianfekete Date: Fri, 24 Jul 2026 13:02:50 +0200 Subject: [PATCH 2/2] update o11y docs --- docs/observability.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/docs/observability.md b/docs/observability.md index 303f054a8..add4f108a 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -109,9 +109,14 @@ 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 | | `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`) | +| `ate.workerpool.workers` | ateapi | up/down counter | live worker count per pool, split by state (`idle`/`assigned`) and sandbox class to provide fleet capacity and saturation at a glance | +| `ate.actor.lifecycle.operation.duration` | ateapi | histogram | how long each actor operation (create/resume/suspend/pause/delete) takes and whether it failed (`error.type` present = failure, absent = success); labeled by operation, template, pool, sandbox class, and snapshot kind on resume | +| `ate.scheduler.assignment.duration` | ateapi | histogram | time it takes for an actor to be assigned to a worker, with the outcome (`assigned` / `no_free_worker` / `error`) to catch scheduling latency and capacity starvation problems | The table lists the OpenTelemetry instrument names. How a name appears in a query depends on the backend (Cloud Monitoring (GMP) / Kind collector). +The `ate.*` control-plane metrics all have the same bounded labels (operation, outcome, pool, sandbox class, snapshot kind) so they aggregate cleanly, and they deliberately keep high-cardinality actor identity (name/uid/atespace) outside of metrics. These live on logs and traces instead. + ### 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.