feat(otel): add actor lifecycle + scheduler duration metrics

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)

cmd/ateapi/internal/controlapi/delete_actor.go

@@ -18,29 +18,43 @@ 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"
 	"google.golang.org/grpc/status"
 	"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())
 		}
 		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
 }
 

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))

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...))
+}