feat: Add control plane telemetry - actor crashes counter and operation labels

cmd/ateapi/internal/controlapi/crash.go

@@ -21,23 +21,40 @@ import (
 	"log/slog"
 
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
+	"github.com/agent-substrate/substrate/internal/ateattr"
 	"github.com/agent-substrate/substrate/internal/ateerrors"
 	"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"
 )
 
-// maybeCrashActor inspects err returned by an atelet RPC, it crashes
-// the actor if the err carries the actorCrashed=true metadata directive.
-func maybeCrashActor(ctx context.Context, st store.Interface, actorRef resources.ActorRef, err error, wrapMsg string) error {
+// maybeCrashActor inspects err returned by an atelet RPC and crashes the actor
+// if err carries the actorCrashed=true metadata directive.
+//
+// opNames is variadic (...string) to provide backward compatibility for generic callers
+// where the operation context is not explicitly supplied. If provided, opNames[0] is
+// normalized against the bounded operations {resume, suspend, pause, unknown}.
+func maybeCrashActor(ctx context.Context, st store.Interface, actorRef resources.ActorRef, err error, wrapMsg string, opNames ...string) error {
 	if err == nil {
 		return nil
 	}
 
 	if ateerrors.ActorCrashRequested(err) {
 		slog.ErrorContext(ctx, "Setting Actor to crashed due to error", slog.Any("error", err))
-		if cerr := crashActor(ctx, st, actorRef); cerr != nil {
+		// Extract AIP-193 ErrorInfo reason enum from the RPC error detail. If unclassified
+		// or unlisted, default to ateattr.ReasonUnknown to protect metric low-cardinality.
+		reason := ateerrors.ExtractReason(err)
+		if reason == "" {
+			reason = ateattr.ReasonUnknown
+		}
+
+		opName := ateattr.OperationNameUnknown
+		if len(opNames) > 0 {
+			opName = ateattr.NormalizeOperationName(opNames[0])
+		}
+
+		if cerr := crashActor(ctx, st, actorRef, opName, reason); cerr != nil {
 			slog.ErrorContext(ctx, "Failed to crash actor", slog.Any("cerr", cerr))
 			return cerr
 		}
@@ -48,17 +65,28 @@ func maybeCrashActor(ctx context.Context, st store.Interface, actorRef resources
 
 // crashActor moves the actor to CRASHED state and frees the worker it was
 // assigned to, if any, so the worker can host other actors.
-func crashActor(ctx context.Context, st store.Interface, actorRef resources.ActorRef) error {
+func crashActor(ctx context.Context, st store.Interface, actorRef resources.ActorRef, opName, reason string) error {
 	actor, err := st.GetActor(ctx, actorRef)
 	if err != nil {
 		return fmt.Errorf("while loading actor to crash: %w", err)
 	}
 
+	wasAlreadyCrashed := actor.GetStatus() == ateapipb.Actor_STATUS_CRASHED
+	opName = ateattr.NormalizeOperationName(opName)
+	if reason == "" {
+		reason = ateattr.ReasonUnknown
+	}
+
 	var errCollected []error
-	if err := releaseWorker(ctx, st, actor); err != nil {
+	sandboxClass, err := releaseWorker(ctx, st, actor)
+	if err != nil {
 		errCollected = append(errCollected, err)
 	}
 
+	// Snapshot crash attributes before pod and pool pointers are cleared below;
+	// the counter itself is emitted only after the transition commits.
+	crashAttrs := ateattr.ActorMetricAttributes(actor, sandboxClass, opName, reason)
+
 	actor.Status = ateapipb.Actor_STATUS_CRASHED
 
 	// InProgressSnapshot is kept for debugging; failed workflow
@@ -69,48 +97,59 @@ func crashActor(ctx context.Context, st store.Interface, actorRef resources.Acto
 	actor.AteomPodUid = ""
 	actor.WorkerPoolName = ""
 
-	if _, err := st.UpdateActor(ctx, actor, actor.GetMetadata().GetVersion()); err != nil {
+	_, err = st.UpdateActor(ctx, actor, actor.GetMetadata().GetVersion())
+	if err != nil {
 		errCollected = append(errCollected, fmt.Errorf("while marking actor crashed: %w", err))
+		return errors.Join(errCollected...)
+	}
+
+	// Increment metric only after a successful UpdateActor, and only if the actor was not already crashed.
+	if !wasAlreadyCrashed {
+		recordActorCrash(ctx, crashAttrs)
 	}
+
 	return errors.Join(errCollected...)
 }
 
 // releaseWorker clears the worker's assignment if it still points at the given
 // actor. A missing worker or an already-cleared assignment is not an error.
-func releaseWorker(ctx context.Context, st store.Interface, actor *ateapipb.Actor) error {
+// It returns the worker's sandboxClass if found.
+func releaseWorker(ctx context.Context, st store.Interface, actor *ateapipb.Actor) (string, error) {
 	podNamespace := actor.GetAteomPodNamespace()
 	podName := actor.GetAteomPodName()
 	podUid := actor.GetAteomPodUid()
 	poolName := actor.GetWorkerPoolName()
 
 	if podNamespace == "" || podName == "" || poolName == "" {
 		slog.WarnContext(ctx, "Actor's worker assignment is already cleared")
-		return nil
+		return "", nil
 	}
 
 	worker, err := st.GetWorker(ctx, podNamespace, poolName, podName)
 	if errors.Is(err, store.ErrNotFound) {
 		// No need to release if the worker is not found.
 		slog.WarnContext(ctx, "Worker already gone while crashing actor, skipping release", slog.String("worker", podUid))
-		return nil
+		return "", nil
 	}
 	if err != nil {
-		return fmt.Errorf("while getting worker to release: %w", err)
+		return "", fmt.Errorf("while getting worker to release: %w", err)
 	}
+
+	sandboxClass := worker.GetSandboxClass()
 	wass := worker.GetAssignment()
 	if wass == nil {
 		slog.WarnContext(ctx, "Worker's assignment is already nil, skipping release", slog.String("worker", podUid))
-		return nil
+		return sandboxClass, nil
 	}
 	// Only free it if it still belongs to us
 	if resources.ActorRefFromObjectRef(wass.GetActor()) != resources.ActorRefFromActor(actor) {
 		slog.WarnContext(ctx, "Worker already assigned to another Actor", slog.String("worker", podUid))
-		return nil
+		return sandboxClass, nil
 	}
 
 	worker.Assignment = nil
 	if err := st.UpdateWorker(ctx, worker, worker.GetVersion()); err != nil {
-		return fmt.Errorf("while releasing worker: %w", err)
+		return sandboxClass, fmt.Errorf("while releasing worker: %w", err)
 	}
-	return nil
+	return sandboxClass, nil
 }

cmd/ateapi/internal/controlapi/crash_test.go

@@ -22,9 +22,12 @@ import (
 
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store/storetest"
+	"github.com/agent-substrate/substrate/internal/ateattr"
 	"github.com/agent-substrate/substrate/internal/ateerrors"
 	"github.com/agent-substrate/substrate/internal/resources"
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
@@ -222,7 +225,8 @@ func TestCrashActor(t *testing.T) {
 				tt.setup(t, ctx, st)
 			}
 
-			err := crashActor(ctx, st, actorRef)
+			err := crashActor(ctx, st, actorRef, ateattr.OperationNameUnknown, ateattr.ReasonUnknown)
+
 			tt.check(t, ctx, st, err)
 		})
 	}
@@ -350,3 +354,95 @@ func TestMaybeCrashActor(t *testing.T) {
 		})
 	}
 }
+
+func TestCrashActor_Metrics(t *testing.T) {
+	reader := sdkmetric.NewManualReader()
+	mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
+	if err := RegisterActorCrashes(mp.Meter("ateapi")); err != nil {
+		t.Fatalf("RegisterActorCrashes: %v", err)
+	}
+
+	ctx := context.Background()
+	st, cleanup := storetest.SetupTestStore(t)
+	defer cleanup()
+
+	actorRef := resources.ActorRef{Atespace: "team-a", Name: "actor-crash-test"}
+	worker := &ateapipb.Worker{
+		WorkerNamespace: "demo-ns",
+		WorkerPool:      "pool-1",
+		WorkerPod:       "pod-1",
+		SandboxClass:    "gvisor",
+		Assignment: &ateapipb.Assignment{
+			Actor: &ateapipb.ObjectRef{Atespace: actorRef.Atespace, Name: actorRef.Name},
+		},
+	}
+	if err := st.CreateWorker(ctx, worker); err != nil {
+		t.Fatalf("CreateWorker: %v", err)
+	}
+
+	actor := &ateapipb.Actor{
+		Metadata: &ateapipb.ResourceMetadata{
+			Atespace: actorRef.Atespace,
+			Name:     actorRef.Name,
+			Uid:      "uid-123",
+		},
+		ActorTemplateNamespace: "demo-ns",
+		ActorTemplateName:      "counter-template",
+		WorkerPoolName:         "pool-1",
+		AteomPodNamespace:      "demo-ns",
+		AteomPodName:           "pod-1",
+		AteomPodUid:            "pod-uid-1",
+		Status:                 ateapipb.Actor_STATUS_RUNNING,
+	}
+	if _, err := st.CreateActor(ctx, actor); err != nil {
+		t.Fatalf("CreateActor: %v", err)
+	}
+
+	if err := crashActor(ctx, st, actorRef, ateattr.OperationNameResume, ateattr.ReasonCorruptedAssignment); err != nil {
+		t.Fatalf("crashActor: %v", err)
+	}
+
+	assertCrashMetricDatapoint(t, reader, ateattr.OperationNameResume, ateattr.ReasonCorruptedAssignment, "demo-ns", "counter-template", "pool-1", "gvisor", 1)
+}
+
+func assertCrashMetricDatapoint(t *testing.T, reader *sdkmetric.ManualReader, wantOpName, wantReason, wantTmplNS, wantTmplName, wantWorkerPool, wantSandboxClass string, wantValue int64) {
+	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 != "ate.actor.crashes" {
+				continue
+			}
+			sum, ok := m.Data.(metricdata.Sum[int64])
+			if !ok {
+				continue
+			}
+			for _, dp := range sum.DataPoints {
+				op, _ := dp.Attributes.Value(ateattr.OperationNameKey)
+				r, _ := dp.Attributes.Value(ateattr.FailureReasonKey)
+				tNS, _ := dp.Attributes.Value(ateattr.TemplateNamespaceKey)
+				tName, _ := dp.Attributes.Value(ateattr.TemplateNameKey)
+				wp, _ := dp.Attributes.Value(ateattr.WorkerPoolNameKey)
+				sc, _ := dp.Attributes.Value(ateattr.SandboxClassKey)
+
+				if op.AsString() == wantOpName &&
+					r.AsString() == wantReason &&
+					tNS.AsString() == wantTmplNS &&
+					tName.AsString() == wantTmplName &&
+					wp.AsString() == wantWorkerPool &&
+					sc.AsString() == wantSandboxClass {
+					if dp.Value != wantValue {
+						t.Errorf("metric value = %d, want %d", dp.Value, wantValue)
+					}
+					return
+				}
+			}
+		}
+	}
+	t.Errorf("did not find ate.actor.crashes metric with attrs: opName=%q, reason=%q, tmplNS=%q, tmplName=%q, workerPool=%q, sandboxClass=%q",
+		wantOpName, wantReason, wantTmplNS, wantTmplName, wantWorkerPool, wantSandboxClass)
+}

cmd/ateapi/internal/controlapi/metrics.go

@@ -21,11 +21,39 @@ import (
 	"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"
 	"k8s.io/apimachinery/pkg/labels"
 )
 
-const workerpoolWorkersMetric = "ate.workerpool.workers"
+const (
+	workerpoolWorkersMetric = "ate.workerpool.workers"
+	actorCrashesMetric      = "ate.actor.crashes"
+)
+
+var actorCrashesCounter metric.Int64Counter
+
+// RegisterActorCrashes initializes the ate.actor.crashes counter instrument.
+func RegisterActorCrashes(meter metric.Meter) error {
+	counter, err := meter.Int64Counter(
+		actorCrashesMetric,
+		metric.WithUnit("{actor}"),
+		metric.WithDescription("Total number of actors that transitioned to STATUS_CRASHED."),
+	)
+	if err != nil {
+		return fmt.Errorf("create %s counter: %w", actorCrashesMetric, err)
+	}
+	actorCrashesCounter = counter
+	return nil
+}
+
+// recordActorCrash records a crash event on ate.actor.crashes with low-cardinality attributes.
+func recordActorCrash(ctx context.Context, attrs []attribute.KeyValue) {
+	if actorCrashesCounter == nil || len(attrs) == 0 {
+		return
+	}
+	actorCrashesCounter.Add(ctx, 1, metric.WithAttributes(attrs...))
+}
 
 // RegisterWorkerCount wires the ate.workerpool.workers observable against workers
 // (workercache.Cache.Workers in prod) and listPools (a WorkerPool lister's List,

cmd/ateapi/internal/controlapi/syncer.go

@@ -21,6 +21,7 @@ import (
 	"maps"
 
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
+	"github.com/agent-substrate/substrate/internal/ateattr"
 	"github.com/agent-substrate/substrate/internal/resources"
 	listersv1alpha1 "github.com/agent-substrate/substrate/pkg/client/listers/api/v1alpha1"
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
@@ -308,6 +309,20 @@ func (s *WorkerPoolSyncer) releaseActorOnDeadWorker(ctx context.Context, namespa
 	if actor.Status == ateapipb.Actor_STATUS_SUSPENDED {
 		return nil
 	}
+	opName := ateattr.OperationNameUnknown
+	switch actor.Status {
+	case ateapipb.Actor_STATUS_RESUMING:
+		opName = ateattr.OperationNameResume
+	case ateapipb.Actor_STATUS_SUSPENDING:
+		opName = ateattr.OperationNameSuspend
+	case ateapipb.Actor_STATUS_PAUSING:
+		opName = ateattr.OperationNamePause
+	}
+
+	wasAlreadyCrashed := actor.GetStatus() == ateapipb.Actor_STATUS_CRASHED
+
+	// Snapshot crash attributes before pod and pool pointers are cleared on actor.
+	crashAttrs := ateattr.ActorMetricAttributes(actor, worker.GetSandboxClass(), opName, ateattr.ReasonWorkerPodGone)
 
 	actor.Status = ateapipb.Actor_STATUS_CRASHED
 	actor.AteomPodNamespace = ""
@@ -318,5 +333,10 @@ func (s *WorkerPoolSyncer) releaseActorOnDeadWorker(ctx context.Context, namespa
 	actor.WorkerPoolName = ""
 
 	_, err = s.persistence.UpdateActor(ctx, actor, actor.GetMetadata().GetVersion())
+
+	if err == nil && !wasAlreadyCrashed {
+		recordActorCrash(ctx, crashAttrs)
+	}
 	return err
+
 }