diff --git a/cmd/ateapi/internal/controlapi/metrics.go b/cmd/ateapi/internal/controlapi/metrics.go index 790cdfd00..3df184ed1 100644 --- a/cmd/ateapi/internal/controlapi/metrics.go +++ b/cmd/ateapi/internal/controlapi/metrics.go @@ -175,10 +175,12 @@ func (i *Instruments) recordLifecycleOp(ctx context.Context, op string, start ti } // 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 { +// state. Nil-safe, and omits the pool, snapshot-kind and snapshot-scope 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; snapshotScope applies to all three and is what separates a restore +// combined with the template's golden state from a plain one of the same kind. +func lifecycleOpAttrs(actor *ateapipb.Actor, template *atev1alpha1.ActorTemplate, snapshotKind, snapshotScope string) []attribute.KeyValue { attrs := []attribute.KeyValue{ ateattr.TemplateNameKey.String(actor.GetActorTemplateName()), ateattr.TemplateNamespaceKey.String(actor.GetActorTemplateNamespace()), @@ -192,6 +194,9 @@ func lifecycleOpAttrs(actor *ateapipb.Actor, template *atev1alpha1.ActorTemplate if snapshotKind != "" { attrs = append(attrs, ateattr.SnapshotKindKey.String(snapshotKind)) } + if snapshotScope != "" { + attrs = append(attrs, ateattr.SnapshotScopeKey.String(snapshotScope)) + } return attrs } diff --git a/cmd/ateapi/internal/controlapi/metrics_test.go b/cmd/ateapi/internal/controlapi/metrics_test.go index ed0779c9e..9ae491528 100644 --- a/cmd/ateapi/internal/controlapi/metrics_test.go +++ b/cmd/ateapi/internal/controlapi/metrics_test.go @@ -216,7 +216,7 @@ func TestLifecycleOpDurationShape(t *testing.T) { Spec: atev1alpha1.ActorTemplateSpec{SandboxClass: atev1alpha1.SandboxClassGvisor}, } inst.recordLifecycleOp(context.Background(), ateattr.OperationResume, time.Now(), nil, - lifecycleOpAttrs(actor, template, ateattr.SnapshotKindLatest)...) + lifecycleOpAttrs(actor, template, ateattr.SnapshotKindLatest, ateattr.SnapshotScopeDataOnGolden)...) dp := singleHistogramDP(t, reader, lifecycleOpDurationMetric) assertAttrKeys(t, dp, @@ -226,10 +226,31 @@ func TestLifecycleOpDurationShape(t *testing.T) { ateattr.WorkerPoolNameKey, ateattr.SandboxClassKey, ateattr.SnapshotKindKey, + ateattr.SnapshotScopeKey, ) if op, _ := attrString(dp, ateattr.ActorOperationNameKey); op != ateattr.OperationResume { t.Errorf("operation = %q, want %q", op, ateattr.OperationResume) } + // Kind and scope are independent: a data_on_golden restore of the actor's + // own latest snapshot must stay distinguishable from one of a local snapshot. + if scope, _ := attrString(dp, ateattr.SnapshotScopeKey); scope != ateattr.SnapshotScopeDataOnGolden { + t.Errorf("snapshot scope = %q, want %q", scope, ateattr.SnapshotScopeDataOnGolden) + } + if kind, _ := attrString(dp, ateattr.SnapshotKindKey); kind != ateattr.SnapshotKindLatest { + t.Errorf("snapshot kind = %q, want %q", kind, ateattr.SnapshotKindLatest) + } +} + +// TestLifecycleOpAttrsOmitsUnknownScope guards the failure path: a resume that +// dies before the restore request is built has no scope, and an empty-string +// series would be indistinguishable from a real one. +func TestLifecycleOpAttrsOmitsUnknownScope(t *testing.T) { + actor := &ateapipb.Actor{ActorTemplateName: "support-agent", ActorTemplateNamespace: "ate-agents"} + for _, kv := range lifecycleOpAttrs(actor, nil, "", "") { + if kv.Key == ateattr.SnapshotScopeKey || kv.Key == ateattr.SnapshotKindKey { + t.Errorf("attribute %s must be omitted while unknown, got %q", kv.Key, kv.Value.AsString()) + } + } } // TestRecordLifecycleOp_OutcomeClassification asserts success omits error.type and diff --git a/cmd/ateapi/internal/controlapi/workflow.go b/cmd/ateapi/internal/controlapi/workflow.go index afb4b62ad..be546e98e 100644 --- a/cmd/ateapi/internal/controlapi/workflow.go +++ b/cmd/ateapi/internal/controlapi/workflow.go @@ -187,7 +187,7 @@ func (w *ActorWorkflow) ResumeActor(ctx context.Context, actorRef resources.Acto return } w.instruments.recordLifecycleOp(ctx, ateattr.OperationResume, start, err, - lifecycleOpAttrs(state.Actor, state.ActorTemplate, state.SnapshotKind)...) + lifecycleOpAttrs(state.Actor, state.ActorTemplate, state.SnapshotKind, state.WireSnapshotScope)...) }() lockCtx, lock, err := w.acquireActorLock(ctx, actorRef) @@ -222,7 +222,7 @@ func (w *ActorWorkflow) SuspendActor(ctx context.Context, actorRef resources.Act defer func() { w.instruments.recordLifecycleOp(ctx, ateattr.OperationSuspend, start, err, - lifecycleOpAttrs(state.Actor, state.ActorTemplate, "")...) + lifecycleOpAttrs(state.Actor, state.ActorTemplate, "", state.WireSnapshotScope)...) }() lockCtx, lock, err := w.acquireActorLock(ctx, actorRef) @@ -256,7 +256,7 @@ func (w *ActorWorkflow) PauseActor(ctx context.Context, actorRef resources.Actor defer func() { w.instruments.recordLifecycleOp(ctx, ateattr.OperationPause, start, err, - lifecycleOpAttrs(state.Actor, state.ActorTemplate, "")...) + lifecycleOpAttrs(state.Actor, state.ActorTemplate, "", state.WireSnapshotScope)...) }() lockCtx, lock, err := w.acquireActorLock(ctx, actorRef) diff --git a/cmd/ateapi/internal/controlapi/workflow_pause.go b/cmd/ateapi/internal/controlapi/workflow_pause.go index a29e65cae..f084b1cd5 100644 --- a/cmd/ateapi/internal/controlapi/workflow_pause.go +++ b/cmd/ateapi/internal/controlapi/workflow_pause.go @@ -41,8 +41,9 @@ type PauseInput struct { // PauseState holds the mutable state loaded and modified during execution. type PauseState struct { - Actor *ateapipb.Actor - ActorTemplate *atev1alpha1.ActorTemplate + Actor *ateapipb.Actor + ActorTemplate *atev1alpha1.ActorTemplate + WireSnapshotScope string } type LoadActorForPauseStep struct { @@ -168,6 +169,7 @@ func (s *CallAteletPauseStep) Execute(ctx context.Context, input *PauseInput, st Scope: toAteletSnapshotScope(state.ActorTemplate.Spec.SnapshotsConfig.OnPause), ActorUid: state.Actor.GetMetadata().Uid, } + state.WireSnapshotScope = ateattr.SnapshotScopeValue(req.Scope) _, err = client.Checkpoint(ctx, req) return maybeCrashActor(ctx, s.store, input.ActorRef, err, "while checkpointing workload", ateattr.OperationPause) diff --git a/cmd/ateapi/internal/controlapi/workflow_resume.go b/cmd/ateapi/internal/controlapi/workflow_resume.go index c3ec2b14b..d96fabe1a 100644 --- a/cmd/ateapi/internal/controlapi/workflow_resume.go +++ b/cmd/ateapi/internal/controlapi/workflow_resume.go @@ -54,6 +54,9 @@ type ResumeState struct { SnapshotLocation string SnapshotScope ateapipb.SnapshotContentScope SnapshotKind string + // WireSnapshotScope labels the restore requested, not the stored snapshot's + // SnapshotScope: a data snapshot restored on golden goes out as data_on_golden. + WireSnapshotScope string // GoldenSnapshotLocation is the storage location of the ActorTemplate's // golden snapshot. Populated only when the template's onResume // configuration selects the golden snapshot as the boot source for the @@ -581,6 +584,7 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput, req.Scope = ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN req.GoldenSnapshotUriPrefix = state.GoldenSnapshotLocation } + state.WireSnapshotScope = ateattr.SnapshotScopeValue(req.Scope) _, err = client.Restore(ctx, req) return maybeCrashActor(ctx, s.store, input.ActorRef, err, "while restoring workload", ateattr.OperationResume) @@ -600,6 +604,7 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput, if state.GoldenSnapshotLocation != "" { scope = ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN } + state.WireSnapshotScope = ateattr.SnapshotScopeValue(scope) req := &ateletpb.RestoreRequest{ TargetAteomUid: assignment.GetWorkerPodUid(), Atespace: state.Actor.GetMetadata().GetAtespace(), diff --git a/cmd/ateapi/internal/controlapi/workflow_suspend.go b/cmd/ateapi/internal/controlapi/workflow_suspend.go index cf63b15c7..d926ba76c 100644 --- a/cmd/ateapi/internal/controlapi/workflow_suspend.go +++ b/cmd/ateapi/internal/controlapi/workflow_suspend.go @@ -42,9 +42,10 @@ type SuspendInput struct { // SuspendState holds the mutable state loaded and modified during execution. type SuspendState struct { - Actor *ateapipb.Actor - ActorTemplate *atev1alpha1.ActorTemplate - SourceVersion int64 + Actor *ateapipb.Actor + ActorTemplate *atev1alpha1.ActorTemplate + SourceVersion int64 + WireSnapshotScope string } type LoadActorForSuspendStep struct { @@ -186,6 +187,7 @@ func (s *CallAteletSuspendStep) Execute(ctx context.Context, input *SuspendInput Scope: toAteletSnapshotScope(commitSnapshotScope(state.Actor.GetMetadata().GetAtespace(), state.ActorTemplate)), ActorUid: state.Actor.GetMetadata().Uid, } + state.WireSnapshotScope = ateattr.SnapshotScopeValue(req.Scope) _, err = client.Checkpoint(ctx, req) return maybeCrashActor(ctx, s.store, input.ActorRef, err, "while checkpointing workload", ateattr.OperationSuspend) diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 6b371d5ca..1d661f574 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -34,6 +34,7 @@ import ( "cloud.google.com/go/storage" "github.com/agent-substrate/substrate/cmd/atelet/internal/ategcs" + "github.com/agent-substrate/substrate/internal/ateattr" "github.com/agent-substrate/substrate/internal/ateerrors" "github.com/agent-substrate/substrate/internal/ateinterceptors" "github.com/agent-substrate/substrate/internal/ateompath" @@ -51,8 +52,8 @@ import ( "github.com/spf13/pflag" "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc" "go.opentelemetry.io/otel" - "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/metric" + semconv "go.opentelemetry.io/otel/semconv/v1.40.0" "golang.org/x/sync/errgroup" "google.golang.org/api/option" "google.golang.org/grpc" @@ -121,6 +122,11 @@ func main() { serverboot.Fatal(ctx, "Failed to create snapshot size metric", err) } + instruments, err := NewInstruments(otel.Meter("atelet")) + if err != nil { + serverboot.Fatal(ctx, "Failed to create atelet metrics", err) + } + // readiness flips to not-ready on SIGTERM so /readyz reports 503 while the // pod drains, while /healthz stays 200 for liveness. readiness := &serverboot.Readiness{} @@ -198,6 +204,7 @@ func main() { wrappedAnonGCS, wrappedGCS, imageCache, + instruments, ) lis, err := net.Listen("tcp", ":"+strconv.Itoa(*port)) @@ -268,6 +275,7 @@ type AteomHerder struct { imageCache *imagecache.Store anonGCSClient ategcs.ObjectStorage gcsClient ategcs.ObjectStorage + instruments *Instruments } var _ ateletpb.AteomHerderServer = (*AteomHerder)(nil) @@ -279,12 +287,14 @@ func NewService( anonGCSClient ategcs.ObjectStorage, gcsClient ategcs.ObjectStorage, imageCache *imagecache.Store, + instruments *Instruments, ) *AteomHerder { wms := &AteomHerder{ ateomDialer: ateomDialer, imageCache: imageCache, anonGCSClient: anonGCSClient, gcsClient: gcsClient, + instruments: instruments, } return wms } @@ -368,7 +378,10 @@ func initSnapshotSizeMetric() error { return err } -func recordSnapshotSize(ctx context.Context, kind, path, atNamespace, atName string) { +// recordSnapshotSize labels each image with the registry's file.name. That +// label used to be spelled "kind", which means the snapshot's provenance +// everywhere else in the ate.* namespace, not one of its files. +func recordSnapshotSize(ctx context.Context, file, path, atNamespace, atName string) { if snapshotSizeBytes == nil { return } @@ -378,17 +391,17 @@ func recordSnapshotSize(ctx context.Context, kind, path, atNamespace, atName str } if err != nil { slog.WarnContext(ctx, "Failed to stat snapshot image for size metric", - slog.String("kind", kind), slog.String("path", path), slog.Any("err", err)) + slog.String("file", file), slog.String("path", path), slog.Any("err", err)) return } snapshotSizeBytes.Record(ctx, fi.Size(), metric.WithAttributes( - attribute.String("kind", kind), - attribute.String("actor_template_namespace", atNamespace), - attribute.String("actor_template_name", atName), + semconv.FileNameKey.String(file), + ateattr.TemplateNamespaceKey.String(atNamespace), + ateattr.TemplateNameKey.String(atName), )) } -func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRequest) (*ateletpb.CheckpointResponse, error) { +func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRequest) (_ *ateletpb.CheckpointResponse, err error) { if err := validateCheckpointRequest(req); err != nil { return nil, status.Error(codes.InvalidArgument, err.Error()) } @@ -396,6 +409,24 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe actorUID := req.GetActorUid() actorRef := resources.ActorRef{Atespace: req.GetAtespace(), Name: req.GetActorName()} + // Per-phase timing, recorded on the way out so a failed checkpoint still + // reports the phases it completed. Phases left at zero never ran. + tStart := time.Now() + var dAssets, dAteom, dPersist time.Duration + op := snapshotOp{ + templateNamespace: req.GetActorTemplateNamespace(), + templateName: req.GetActorTemplateName(), + kind: checkpointSnapshotKind(req), + scope: ateattr.SnapshotScopeValue(req.GetScope()), + } + defer func() { + s.instruments.recordCheckpoint(ctx, op, err, + phase{ateattr.SnapshotPhaseSandboxAssets, dAssets}, + phase{ateattr.SnapshotPhaseAteomCheckpoint, dAteom}, + phase{ateattr.SnapshotPhasePersist, dPersist}, + phase{ateattr.SnapshotPhaseTotal, time.Since(tStart)}) + }() + // Checkpoint requests no longer carry the sandbox config; recover the // version this actor was started with from the on-node record and re-fetch // it (a cache hit) so ateom can drive runsc, and so we can pin it into the @@ -404,8 +435,13 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe if err != nil { return nil, ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonInvalidSandboxAsset, ateerrors.ReasonTerminalFileSystemError) } + op.sandboxClass = sandboxRec.SandboxClass + + tAssets := time.Now() assetPaths, err := s.ensureSandboxAssets(ctx, sandboxRec) + dAssets = time.Since(tAssets) if err != nil { + op.failedPhase = ateattr.SnapshotPhaseSandboxAssets return nil, ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonInvalidSandboxAsset, ateerrors.ReasonTerminalFileSystemError, ateerrors.ReasonFailedGetExternalObject, ateerrors.ReasonInvalidObjectURL) } @@ -419,6 +455,7 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe // Tell ateom to take the checkpoint and delete containers. ateom reports the // exact files it wrote so we ship precisely that set (gVisor's image files, // cloud-hypervisor's snapshot set, ...) rather than a hardcoded list. + tAteom := time.Now() resp, err := client.CheckpointWorkload(ctx, &ateompb.CheckpointWorkloadRequest{ Atespace: actorRef.Atespace, ActorName: actorRef.Name, @@ -430,9 +467,11 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe Scope: toAteomSnapshotScope(req.GetScope()), ActorUid: actorUID, }) + dAteom = time.Since(tAteom) if err != nil { // TODO: Ateom should classify checkpoint failures, and set "should-crash" // in the metadata if the error is not retriable. + op.failedPhase = ateattr.SnapshotPhaseAteomCheckpoint return nil, fmt.Errorf("while calling ateom.CheckpointWorkload: %w", err) } @@ -452,19 +491,28 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe // checkpoint either overwrites (pause) or clears (suspend). pruneLocalCheckpoints(ctx, actorUID) + // Pruning stays outside the persist window: it collects superseded + // snapshots on both paths, so timing it as part of an external upload would + // mix local disk deletion into the object-storage measurement. + tPersist := time.Now() switch req.GetType() { case ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL: // TODO(#362): Because we do not cache the external snapshot files when upload fails, we have to mark the Actor as CRASHED. if err := s.uploadExternalCheckpoint(ctx, req, checkpointDir, sandboxRec); err != nil { + dPersist = time.Since(tPersist) + op.failedPhase = ateattr.SnapshotPhasePersist return nil, ateerrors.NewGRPCError(ctx, codes.DataLoss, ateerrors.ReasonFaileSaveSnapshot, ateerrors.ActorCrashedMetadata(), fmt.Errorf("%w: while uploading external snapshot: %w", ateerrors.ReasonFaileSaveSnapshot, err)) } case ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL: if err := s.moveLocalCheckpoint(ctx, req, checkpointDir, sandboxRec); err != nil { + dPersist = time.Since(tPersist) + op.failedPhase = ateattr.SnapshotPhasePersist return nil, ateerrors.NewGRPCError(ctx, codes.DataLoss, ateerrors.ReasonFaileSaveSnapshot, ateerrors.ActorCrashedMetadata(), fmt.Errorf("%w: while moving to local snapshot: %w", ateerrors.ReasonFaileSaveSnapshot, err)) } default: return nil, fmt.Errorf("unexpected checkpoint type: %v", req.GetType()) } + dPersist = time.Since(tPersist) if err := s.unmountExternalVolumes(ctx, actorUID, req.GetSpec().GetVolumes()); err != nil { return nil, ateerrors.NewGRPCError(ctx, codes.DataLoss, ateerrors.ReasonTerminalFileSystemError, ateerrors.ActorCrashedMetadata(), fmt.Errorf("while unmounting external volumes: %w", err)) @@ -500,7 +548,7 @@ func (s *AteomHerder) moveLocalCheckpoint(ctx context.Context, req *ateletpb.Che for _, fileName := range rec.SnapshotFiles { src := filepath.Join(checkpointDir, fileName) dst := filepath.Join(localCheckpointPath, fileName) - recordSnapshotSize(ctx, strings.TrimSuffix(fileName, ".img"), src, req.GetActorTemplateNamespace(), req.GetActorTemplateName()) + recordSnapshotSize(ctx, fileName, src, req.GetActorTemplateNamespace(), req.GetActorTemplateName()) if err := os.Rename(src, dst); err != nil { return fmt.Errorf("failed to move %s to %s: %w", src, dst, err) @@ -527,7 +575,7 @@ func (s *AteomHerder) uploadExternalCheckpoint(ctx context.Context, req *ateletp for _, fileName := range rec.SnapshotFiles { fileName := fileName local := filepath.Join(checkpointDir, fileName) - recordSnapshotSize(ctx, strings.TrimSuffix(fileName, ".img"), local, req.GetActorTemplateNamespace(), req.GetActorTemplateName()) + recordSnapshotSize(ctx, fileName, local, req.GetActorTemplateNamespace(), req.GetActorTemplateName()) g.Go(func() error { if err := ategcs.SendLocalFileToGCSWithZstd(gCtx, s.gcsClient, prefix+"/"+fileName+".zstd", local); err != nil { return fmt.Errorf("while uploading %s to GCS: %w", fileName, err) @@ -558,27 +606,56 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest) actorUID := req.GetActorUid() actorRef := resources.ActorRef{Atespace: req.GetAtespace(), Name: req.GetActorName()} + // Per-step timing so we can attribute resume latency between the rustfs + // download/decompress, the OCI image unpack, and ateom's own work. Logged at + // the end, and recorded per phase on the way out so a failed restore still + // reports the phases it completed. Phases left at zero never ran. + tStart := time.Now() + var dMount, dManifest, dAssets, dDownload, dBundles, dAteom time.Duration + op := snapshotOp{ + templateNamespace: req.GetActorTemplateNamespace(), + templateName: req.GetActorTemplateName(), + scope: ateattr.SnapshotScopeValue(req.GetScope()), + } + defer func() { + s.instruments.recordRestore(ctx, op, err, + phase{ateattr.SnapshotPhaseVolumeMount, dMount}, + phase{ateattr.SnapshotPhaseManifestFetch, dManifest}, + phase{ateattr.SnapshotPhaseSandboxAssets, dAssets}, + phase{ateattr.SnapshotPhaseDownload, dDownload}, + phase{ateattr.SnapshotPhaseOCIUnpack, dBundles}, + phase{ateattr.SnapshotPhaseAteomRestore, dAteom}, + phase{ateattr.SnapshotPhaseTotal, time.Since(tStart)}) + }() + // Not crashing the actor, because terminal errors here indicate problems with atelet, // node or the disk itself. if err := resetActorDirs(actorUID); err != nil { return nil, fmt.Errorf("while resetting actor dirs: %w", err) } - if err := s.mountExternalVolumes(ctx, actorUID, req.GetSpec().GetVolumes()); err != nil { - return nil, err + tMount := time.Now() + mountErr := s.mountExternalVolumes(ctx, actorUID, req.GetSpec().GetVolumes()) + dMount = time.Since(tMount) + if mountErr != nil { + op.failedPhase = ateattr.SnapshotPhaseVolumeMount + return nil, mountErr } checkpointDir := ateompath.RestoreStateDir(actorUID) - // Per-step timing so we can attribute resume latency between the rustfs - // download/decompress, the OCI image unpack, and ateom's own work. Logged at the end. - tStart := time.Now() - var dDownload, dBundles, dAteom time.Duration - // The snapshot is self-describing: recover the sandbox binaries that created // it from the manifest stored beside the checkpoint images (the Restore // request no longer carries the sandbox config). Fetch the (small) manifest // first — both the checkpoint download and the OCI/asset prep below need it. + tManifest := time.Now() + manifestDone := false + defer func() { + if !manifestDone { + dManifest = time.Since(tManifest) + op.failedPhase = ateattr.SnapshotPhaseManifestFetch + } + }() var sandboxRec *sandboxAssetsRecord switch req.GetType() { case ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL: @@ -627,6 +704,13 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest) return nil, status.Errorf(codes.FailedPrecondition, "golden snapshot sandbox class %q does not match actor snapshot sandbox class %q", goldenRec.SandboxClass, sandboxRec.SandboxClass) } } + dManifest = time.Since(tManifest) + manifestDone = true + + // The manifest is what tells a golden restore from a latest one, so the + // metric dimensions only become knowable here. + op.kind = restoreSnapshotKind(req, sandboxRec) + op.sandboxClass = sandboxRec.SandboxClass // The record whose pinned binaries run the restored workload: the golden's // for a DATA_ON_GOLDEN restore, the snapshot's own otherwise. The golden's @@ -647,9 +731,16 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest) // TODO(dberkov): the old pause checkpoint files are not deleted after they are // copied to checkpointDir for the LOCAL case. var assetPaths map[string]string + // One per leg: a single field written from both goroutines would race. + var downloadErr, prepErr error + var prepFailedPhase string g, gctx := errgroup.WithContext(ctx) - g.Go(func() error { + g.Go(func() (err error) { t := time.Now() + defer func() { + dDownload = time.Since(t) + downloadErr = err + }() switch req.GetType() { case ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL: if req.GetScope() == ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN { @@ -689,22 +780,34 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest) return err } } - dDownload = time.Since(t) return nil }) - g.Go(func() error { - var err error - if assetPaths, err = s.ensureSandboxAssets(gctx, runtimeRec); err != nil { + g.Go(func() (err error) { + defer func() { prepErr = err }() + tAssets := time.Now() + assetPaths, err = s.ensureSandboxAssets(gctx, runtimeRec) + dAssets = time.Since(tAssets) + if err != nil { + prepFailedPhase = ateattr.SnapshotPhaseSandboxAssets return ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonFailedGetExternalObject, ateerrors.ReasonInvalidObjectURL, ateerrors.ReasonTerminalFileSystemError, ateerrors.ReasonInvalidSandboxAsset) } t := time.Now() - if err := s.prepareOCIBundles(gctx, actorUID, actorRef.Name, req.GetSpec(), req.GetTargetAteomUid()); err != nil { + err = s.prepareOCIBundles(gctx, actorUID, actorRef.Name, req.GetSpec(), req.GetTargetAteomUid()) + dBundles = time.Since(t) + if err != nil { + prepFailedPhase = ateattr.SnapshotPhaseOCIUnpack return ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonTerminalFileSystemError, ateerrors.ReasonInvalidContainerConfig) } - dBundles = time.Since(t) return nil }) if err := g.Wait(); err != nil { + op.failedPhase = groupFailedPhase(err, downloadErr, prepErr, prepFailedPhase) + if isCollateral(err, downloadErr) { + dDownload = 0 + } + if isCollateral(err, prepErr) { + dAssets, dBundles = 0, 0 + } return nil, err } @@ -716,7 +819,7 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest) // Tell ateom to do runsc create + runsc restore for pause container and // all application containers. tAteom := time.Now() - if _, err := client.RestoreWorkload(ctx, &ateompb.RestoreWorkloadRequest{ + _, err = client.RestoreWorkload(ctx, &ateompb.RestoreWorkloadRequest{ Atespace: actorRef.Atespace, ActorName: actorRef.Name, ActorTemplateNamespace: req.GetActorTemplateNamespace(), @@ -730,11 +833,13 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest) // already staged into the restore dir by the combined download above; // ateom restores from the shared dir and never fetches this URI. GoldenSnapshotUriPrefix: req.GetGoldenSnapshotUriPrefix(), - }); err != nil { + }) + dAteom = time.Since(tAteom) + if err != nil { // TODO: classify the errors returned by Ateom and crash the actor if needed. + op.failedPhase = ateattr.SnapshotPhaseAteomRestore return nil, fmt.Errorf("while calling ateom.RestoreWorkload: %w", err) } - dAteom = time.Since(tAteom) // Record the (manifest-pinned) sandbox binaries on-node so a subsequent // Checkpoint of this restored actor can re-pin the same version. For a diff --git a/cmd/atelet/metrics.go b/cmd/atelet/metrics.go new file mode 100644 index 000000000..c4adb05d5 --- /dev/null +++ b/cmd/atelet/metrics.go @@ -0,0 +1,202 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "context" + "fmt" + "time" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" + + "github.com/agent-substrate/substrate/internal/ateattr" + "github.com/agent-substrate/substrate/internal/proto/ateletpb" + "github.com/agent-substrate/substrate/internal/resources" +) + +const ( + restoreDurationMetric = "ate.actor.restore.duration" + checkpointDurationMetric = "ate.actor.checkpoint.duration" +) + +// snapshotPhaseBuckets have to cover both ends of a phase breakdown: a warm OCI +// unpack or a local rename lands in single-digit milliseconds, while a cold node +// fetching a multi-GiB snapshot runs for tens of seconds. +var snapshotPhaseBuckets = []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 20, 30, 60} + +// Instruments holds atelet's cold-start histograms. A nil *Instruments is a +// valid no-op, so call sites need no guard. +type Instruments struct { + restoreDuration metric.Float64Histogram + checkpointDuration metric.Float64Histogram +} + +func NewInstruments(meter metric.Meter) (*Instruments, error) { + restoreDuration, err := meter.Float64Histogram( + restoreDurationMetric, + metric.WithUnit("s"), + metric.WithDescription("Duration of one phase of an actor restore on atelet. Phases overlap, so they are independent observations rather than a partition of the total phase."), + metric.WithExplicitBucketBoundaries(snapshotPhaseBuckets...), + ) + if err != nil { + return nil, fmt.Errorf("create %s histogram: %w", restoreDurationMetric, err) + } + + checkpointDuration, err := meter.Float64Histogram( + checkpointDurationMetric, + metric.WithUnit("s"), + metric.WithDescription("Duration of one phase of an actor checkpoint on atelet. Phases overlap, so they are independent observations rather than a partition of the total phase."), + metric.WithExplicitBucketBoundaries(snapshotPhaseBuckets...), + ) + if err != nil { + return nil, fmt.Errorf("create %s histogram: %w", checkpointDurationMetric, err) + } + + return &Instruments{ + restoreDuration: restoreDuration, + checkpointDuration: checkpointDuration, + }, nil +} + +// snapshotOp is the dimension set shared by every phase of one restore or +// checkpoint. +type snapshotOp struct { + templateNamespace string + templateName string + kind string + scope string + sandboxClass string + // failedPhase is the step the operation died in, so error.type lands there + // and on the total rather than on the phases that had already succeeded. + failedPhase string +} + +// attrs omits kind and sandbox class while they are unknown (a restore that +// failed before reading the snapshot manifest) rather than emitting an +// empty-string series. +func (o snapshotOp) attrs() []attribute.KeyValue { + attrs := make([]attribute.KeyValue, 0, 5) + attrs = append(attrs, + ateattr.TemplateNamespaceKey.String(o.templateNamespace), + ateattr.TemplateNameKey.String(o.templateName), + ateattr.SnapshotScopeKey.String(o.scope), + ) + if o.kind != "" { + attrs = append(attrs, ateattr.SnapshotKindKey.String(o.kind)) + } + if o.sandboxClass != "" { + attrs = append(attrs, ateattr.SandboxClassKey.String(ateattr.NormalizeSandboxClass(o.sandboxClass))) + } + return attrs +} + +// phase is one timed step of a snapshot operation. +type phase struct { + name string + d time.Duration +} + +func (i *Instruments) recordRestore(ctx context.Context, op snapshotOp, err error, phases ...phase) { + if i == nil || i.restoreDuration == nil { + return + } + recordPhases(ctx, i.restoreDuration, op, err, phases) +} + +func (i *Instruments) recordCheckpoint(ctx context.Context, op snapshotOp, err error, phases ...phase) { + if i == nil || i.checkpointDuration == nil { + return + } + recordPhases(ctx, i.checkpointDuration, op, err, phases) +} + +// recordPhases skips zero-valued phases: those never started, because the +// operation died before reaching them, and reporting them as instantaneous +// would drag every percentile down. +// +// ate.failure.reason marks only the phase that failed and the total. It carries +// substrate's taxonomy rather than a gRPC code, which would read Unknown for +// almost every failure here: the interceptor maps these wrapped domain errors +// to a status only after the handler returns. +func recordPhases(ctx context.Context, h metric.Float64Histogram, op snapshotOp, err error, phases []phase) { + base := op.attrs() + for _, p := range phases { + if p.d == 0 { + continue + } + attrs := make([]attribute.KeyValue, 0, len(base)+2) + attrs = append(attrs, base...) + attrs = append(attrs, ateattr.SnapshotPhaseKey.String(p.name)) + if err != nil && (p.name == ateattr.SnapshotPhaseTotal || p.name == op.failedPhase) { + attrs = append(attrs, ateattr.FailureReasonKey.String(ateattr.FailureReason(err))) + } + h.Record(ctx, p.d.Seconds(), metric.WithAttributes(attrs...)) + } +} + +// groupFailedPhase attributes a failed restore errgroup to the leg that +// produced err. errgroup cancels the shared context on the first failure, so +// the other leg aborts as collateral and would otherwise claim the phase; Wait +// returns that first error verbatim, so identity separates the two. +func groupFailedPhase(err, downloadErr, prepErr error, prepPhase string) string { + switch err { + case downloadErr: + return ateattr.SnapshotPhaseDownload + case prepErr: + return prepPhase + } + return "" +} + +// isCollateral reports whether legErr is only fallout from the other leg +// canceling the shared context. That leg stopped part way, so its duration +// would read as an unusually fast success. +func isCollateral(groupErr, legErr error) bool { + return legErr != nil && groupErr != legErr +} + +// restoreSnapshotKind classifies which snapshot a restore reads. A local +// restore is evident from the wire; golden and latest both arrive as an external +// URI prefix, so they are told apart by the identity the manifest records for +// the actor that wrote the snapshot. An empty result means the manifest has not +// been read yet, so the kind is not knowable. +func restoreSnapshotKind(req *ateletpb.RestoreRequest, rec *sandboxAssetsRecord) string { + if req.GetType() == ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL { + return ateattr.SnapshotKindLocal + } + if rec == nil { + return "" + } + // Manifests written before the identity fields existed carry no atespace and + // fall through to latest, which is the common case for them anyway. + if rec.Atespace == resources.GoldenActorAtespace { + return ateattr.SnapshotKindGolden + } + return ateattr.SnapshotKindLatest +} + +// checkpointSnapshotKind classifies which snapshot a checkpoint writes: a pause +// writes the node-local one, a suspend the actor's durable latest, and a commit +// by an actor in the golden atespace the template's golden image. +func checkpointSnapshotKind(req *ateletpb.CheckpointRequest) string { + if req.GetType() == ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL { + return ateattr.SnapshotKindLocal + } + if req.GetAtespace() == resources.GoldenActorAtespace { + return ateattr.SnapshotKindGolden + } + return ateattr.SnapshotKindLatest +} diff --git a/cmd/atelet/metrics_test.go b/cmd/atelet/metrics_test.go new file mode 100644 index 000000000..1e5e13ea3 --- /dev/null +++ b/cmd/atelet/metrics_test.go @@ -0,0 +1,450 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "context" + "errors" + "fmt" + "testing" + "time" + + "go.opentelemetry.io/otel/attribute" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" + + "github.com/agent-substrate/substrate/internal/ateattr" + "github.com/agent-substrate/substrate/internal/ateerrors" + "github.com/agent-substrate/substrate/internal/proto/ateletpb" + "github.com/agent-substrate/substrate/internal/resources" +) + +const ( + testTemplateNamespace = "ate-agents" + testTemplateName = "support-agent" +) + +// newTestInstruments builds the histograms against a local ManualReader 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("atelet")) + if err != nil { + t.Fatalf("NewInstruments: %v", err) + } + return inst, reader +} + +func collectHistogram(t *testing.T, reader *sdkmetric.ManualReader, name string) metricdata.Metrics { + 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 == name { + return m + } + } + } + t.Fatalf("metric %q not collected", name) + return metricdata.Metrics{} +} + +// phaseValues maps each recorded phase to the attribute set it carries. +func phaseValues(t *testing.T, m metricdata.Metrics) map[string]attribute.Set { + t.Helper() + hist, ok := m.Data.(metricdata.Histogram[float64]) + if !ok { + t.Fatalf("%s is %T, want a float64 histogram", m.Name, m.Data) + } + byPhase := make(map[string]attribute.Set, len(hist.DataPoints)) + for _, dp := range hist.DataPoints { + v, ok := dp.Attributes.Value(ateattr.SnapshotPhaseKey) + if !ok { + t.Errorf("datapoint without a phase attribute: %v", dp.Attributes.ToSlice()) + continue + } + byPhase[v.AsString()] = dp.Attributes + } + return byPhase +} + +func attrString(t *testing.T, set attribute.Set, k attribute.Key) string { + t.Helper() + v, ok := set.Value(k) + if !ok { + t.Errorf("missing attribute %s in %v", k, set.ToSlice()) + return "" + } + return v.AsString() +} + +func TestRestoreDurationShape(t *testing.T) { + inst, reader := newTestInstruments(t) + + op := snapshotOp{ + templateNamespace: testTemplateNamespace, + templateName: testTemplateName, + kind: ateattr.SnapshotKindLatest, + scope: ateattr.SnapshotScopeDataOnGolden, + sandboxClass: "gvisor", + } + inst.recordRestore(context.Background(), op, nil, + phase{ateattr.SnapshotPhaseDownload, 2 * time.Second}, + phase{ateattr.SnapshotPhaseTotal, 3 * time.Second}) + + m := collectHistogram(t, reader, restoreDurationMetric) + if m.Unit != "s" { + t.Errorf("unit = %q, want %q", m.Unit, "s") + } + if m.Description == "" { + t.Error("description is empty") + } + + byPhase := phaseValues(t, m) + if len(byPhase) != 2 { + t.Fatalf("recorded %d phases, want download and total", len(byPhase)) + } + got := byPhase[ateattr.SnapshotPhaseDownload] + for _, tc := range []struct { + key attribute.Key + want string + }{ + {ateattr.TemplateNamespaceKey, testTemplateNamespace}, + {ateattr.TemplateNameKey, testTemplateName}, + {ateattr.SnapshotKindKey, ateattr.SnapshotKindLatest}, + {ateattr.SnapshotScopeKey, ateattr.SnapshotScopeDataOnGolden}, + {ateattr.SandboxClassKey, "gvisor"}, + } { + if v := attrString(t, got, tc.key); v != tc.want { + t.Errorf("%s = %q, want %q", tc.key, v, tc.want) + } + } + if _, ok := got.Value(ateattr.ErrorTypeKey); ok { + t.Error("error.type present on a successful restore") + } + if _, ok := got.Value(ateattr.ActorNameKey); ok { + t.Error("actor identity must never reach a metric datapoint") + } +} + +func TestCheckpointDurationShape(t *testing.T) { + inst, reader := newTestInstruments(t) + + inst.recordCheckpoint(context.Background(), snapshotOp{ + templateNamespace: testTemplateNamespace, + templateName: testTemplateName, + kind: ateattr.SnapshotKindLocal, + scope: ateattr.SnapshotScopeFull, + sandboxClass: "microvm", + }, nil, phase{ateattr.SnapshotPhasePersist, time.Second}) + + m := collectHistogram(t, reader, checkpointDurationMetric) + if m.Unit != "s" { + t.Errorf("unit = %q, want %q", m.Unit, "s") + } + set := phaseValues(t, m)[ateattr.SnapshotPhasePersist] + if v := attrString(t, set, ateattr.SnapshotKindKey); v != ateattr.SnapshotKindLocal { + t.Errorf("snapshot kind = %q, want %q", v, ateattr.SnapshotKindLocal) + } + if v := attrString(t, set, ateattr.SandboxClassKey); v != "microvm" { + t.Errorf("sandbox class = %q, want microvm", v) + } +} + +// TestRecordPhasesFailurePath is the failure-path contract: a restore that dies +// in the download marks ate.failure.reason on that phase and on the total, +// leaves the phases that already succeeded unlabeled so their latency stays +// queryable, and does not report phases that never started as instantaneous. +func TestRecordPhasesFailurePath(t *testing.T) { + inst, reader := newTestInstruments(t) + + downloadErr := fmt.Errorf("%w: while downloading snapshot", ateerrors.ReasonFailedGetExternalObject) + inst.recordRestore(context.Background(), + snapshotOp{scope: ateattr.SnapshotScopeFull, failedPhase: ateattr.SnapshotPhaseDownload}, + downloadErr, + phase{ateattr.SnapshotPhaseManifestFetch, 50 * time.Millisecond}, + phase{ateattr.SnapshotPhaseDownload, 2 * time.Second}, + phase{ateattr.SnapshotPhaseAteomRestore, 0}, + phase{ateattr.SnapshotPhaseTotal, 2 * time.Second}) + + byPhase := phaseValues(t, collectHistogram(t, reader, restoreDurationMetric)) + if _, ok := byPhase[ateattr.SnapshotPhaseAteomRestore]; ok { + t.Error("a phase that never started was recorded as a zero observation") + } + + wantReason := string(ateerrors.ReasonFailedGetExternalObject) + tests := []struct { + phase string + wantReason string // empty means ate.failure.reason must be absent + }{ + {phase: ateattr.SnapshotPhaseDownload, wantReason: wantReason}, + {phase: ateattr.SnapshotPhaseTotal, wantReason: wantReason}, + {phase: ateattr.SnapshotPhaseManifestFetch, wantReason: ""}, + } + for _, tt := range tests { + t.Run(tt.phase, func(t *testing.T) { + set, ok := byPhase[tt.phase] + if !ok { + t.Fatalf("phase %q missing", tt.phase) + } + got, present := set.Value(ateattr.FailureReasonKey) + if tt.wantReason == "" { + if present { + t.Errorf("ate.failure.reason = %q on a phase that succeeded, want absent", got.AsString()) + } + return + } + if !present || got.AsString() != tt.wantReason { + t.Errorf("ate.failure.reason = %q (present=%v), want %q", got.AsString(), present, tt.wantReason) + } + }) + } +} + +// TestRecordPhasesUnclassifiedFailure covers the infrastructure failures that +// carry no ateerrors.Reason (a dead object-storage endpoint, say): they must +// collapse onto UNKNOWN rather than leaking an error message into the label. +func TestRecordPhasesUnclassifiedFailure(t *testing.T) { + inst, reader := newTestInstruments(t) + + inst.recordRestore(context.Background(), + snapshotOp{scope: ateattr.SnapshotScopeFull, failedPhase: ateattr.SnapshotPhaseManifestFetch}, + fmt.Errorf("dial tcp 10.96.192.187:9000: connect: connection refused"), + phase{ateattr.SnapshotPhaseManifestFetch, 30 * time.Millisecond}, + phase{ateattr.SnapshotPhaseTotal, 30 * time.Millisecond}) + + set := phaseValues(t, collectHistogram(t, reader, restoreDurationMetric))[ateattr.SnapshotPhaseManifestFetch] + if v := attrString(t, set, ateattr.FailureReasonKey); v != ateattr.ReasonUnknown { + t.Errorf("ate.failure.reason = %q, want %q", v, ateattr.ReasonUnknown) + } +} + +// TestSnapshotOpAttrsOmitsUnknownDimensions covers a restore that fails before +// the manifest resolves: kind and sandbox class are unknowable there, and an +// empty-string series would be indistinguishable from a real one. +func TestSnapshotOpAttrsOmitsUnknownDimensions(t *testing.T) { + attrs := snapshotOp{ + templateNamespace: testTemplateNamespace, + templateName: testTemplateName, + scope: ateattr.SnapshotScopeFull, + }.attrs() + for _, kv := range attrs { + if kv.Key == ateattr.SnapshotKindKey || kv.Key == ateattr.SandboxClassKey { + t.Errorf("attribute %s must be omitted while unknown, got %q", kv.Key, kv.Value.AsString()) + } + } +} + +// TestSnapshotOpAttrsNormalizesSandboxClass keeps an unvalidated manifest value +// from becoming an unbounded label. +func TestSnapshotOpAttrsNormalizesSandboxClass(t *testing.T) { + attrs := snapshotOp{sandboxClass: "definitely-not-a-runtime"}.attrs() + for _, kv := range attrs { + if kv.Key == ateattr.SandboxClassKey && kv.Value.AsString() != ateattr.SandboxClassUnknown { + t.Errorf("sandbox class = %q, want %q", kv.Value.AsString(), ateattr.SandboxClassUnknown) + } + } +} + +// TestGroupFailedPhase covers the concurrent leg of a restore: whichever +// goroutine fails first cancels the shared context, so the other one also +// returns an error, and only the one whose error errgroup actually surfaced may +// claim the phase. +func TestGroupFailedPhase(t *testing.T) { + download := errors.New("download: connection reset") + prep := errors.New("prepare bundles: no entrypoint") + cancelled := errors.New("context canceled") + + tests := []struct { + name string + err error + downloadErr error + prepErr error + prepPhase string + want string + }{ + { + name: "download failed alone", + err: download, + downloadErr: download, + want: ateattr.SnapshotPhaseDownload, + }, + { + name: "prep failed alone during the asset fetch", + err: prep, + prepErr: prep, + prepPhase: ateattr.SnapshotPhaseSandboxAssets, + want: ateattr.SnapshotPhaseSandboxAssets, + }, + { + name: "prep failed first and the in-flight download was collateral", + err: prep, + downloadErr: cancelled, + prepErr: prep, + prepPhase: ateattr.SnapshotPhaseOCIUnpack, + want: ateattr.SnapshotPhaseOCIUnpack, + }, + { + name: "download failed first and prep was collateral", + err: download, + downloadErr: download, + prepErr: cancelled, + prepPhase: ateattr.SnapshotPhaseSandboxAssets, + want: ateattr.SnapshotPhaseDownload, + }, + { + name: "error from neither leg claims no phase", + err: errors.New("something else entirely"), + want: "", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := groupFailedPhase(tt.err, tt.downloadErr, tt.prepErr, tt.prepPhase); got != tt.want { + t.Errorf("groupFailedPhase() = %q, want %q", got, tt.want) + } + }) + } +} + +// TestIsCollateral guards the durations the same way TestGroupFailedPhase +// guards the label: a leg cancelled by the other leg's failure recorded an +// unlabeled partial duration, which would land in the healthy-path percentiles +// as a fast success and drag them down on every failed restore. +func TestIsCollateral(t *testing.T) { + owner := errors.New("prepare bundles: no entrypoint") + cancelled := errors.New("context canceled") + + tests := []struct { + name string + groupErr error + legErr error + want bool + }{ + {name: "leg that owns the error keeps its duration", groupErr: owner, legErr: owner, want: false}, + {name: "leg cancelled as collateral drops its duration", groupErr: owner, legErr: cancelled, want: true}, + {name: "leg that succeeded keeps its duration", groupErr: owner, legErr: nil, want: false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := isCollateral(tt.groupErr, tt.legErr); got != tt.want { + t.Errorf("isCollateral() = %v, want %v", got, tt.want) + } + }) + } +} + +func TestRestoreSnapshotKind(t *testing.T) { + tests := []struct { + name string + req *ateletpb.RestoreRequest + rec *sandboxAssetsRecord + want string + }{ + { + name: "local pause snapshot", + req: &ateletpb.RestoreRequest{Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL}, + rec: &sandboxAssetsRecord{Atespace: "team-a"}, + want: ateattr.SnapshotKindLocal, + }, + { + name: "local restore is classifiable before the manifest is read", + req: &ateletpb.RestoreRequest{Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL}, + rec: nil, + want: ateattr.SnapshotKindLocal, + }, + { + name: "external snapshot written by a golden actor", + req: &ateletpb.RestoreRequest{Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL}, + rec: &sandboxAssetsRecord{Atespace: resources.GoldenActorAtespace}, + want: ateattr.SnapshotKindGolden, + }, + { + name: "external snapshot written by a tenant actor", + req: &ateletpb.RestoreRequest{Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL}, + rec: &sandboxAssetsRecord{Atespace: "team-a"}, + want: ateattr.SnapshotKindLatest, + }, + { + name: "manifest predating the identity fields", + req: &ateletpb.RestoreRequest{Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL}, + rec: &sandboxAssetsRecord{}, + want: ateattr.SnapshotKindLatest, + }, + { + name: "external kind is unknowable until the manifest is read", + req: &ateletpb.RestoreRequest{Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL}, + rec: nil, + want: "", + }, + { + name: "data on golden keeps the actor snapshot's own kind", + req: &ateletpb.RestoreRequest{ + Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL, + Scope: ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN, + }, + rec: &sandboxAssetsRecord{Atespace: "team-a"}, + want: ateattr.SnapshotKindLocal, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := restoreSnapshotKind(tt.req, tt.rec); got != tt.want { + t.Errorf("restoreSnapshotKind() = %q, want %q", got, tt.want) + } + }) + } +} + +func TestCheckpointSnapshotKind(t *testing.T) { + tests := []struct { + name string + req *ateletpb.CheckpointRequest + want string + }{ + { + name: "pause writes the node-local snapshot", + req: &ateletpb.CheckpointRequest{Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL, Atespace: "team-a"}, + want: ateattr.SnapshotKindLocal, + }, + { + name: "suspend writes the actor's durable snapshot", + req: &ateletpb.CheckpointRequest{Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL, Atespace: "team-a"}, + want: ateattr.SnapshotKindLatest, + }, + { + name: "a golden actor's commit writes the template's golden", + req: &ateletpb.CheckpointRequest{Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL, Atespace: resources.GoldenActorAtespace}, + want: ateattr.SnapshotKindGolden, + }, + { + name: "a local checkpoint in the golden atespace is still local", + req: &ateletpb.CheckpointRequest{Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL, Atespace: resources.GoldenActorAtespace}, + want: ateattr.SnapshotKindLocal, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := checkpointSnapshotKind(tt.req); got != tt.want { + t.Errorf("checkpointSnapshotKind() = %q, want %q", got, tt.want) + } + }) + } +} diff --git a/docs/observability.md b/docs/observability.md index feefef2b0..231c4c4d0 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -113,10 +113,12 @@ Agent Substrate emits foundational OpenTelemetry system and server metrics to mo | `ate.actor.crashes` | ateapi | counter | Number of times actors transitioned to `STATUS_CRASHED` with failure reasons (labels `ate.actor.operation.name`, `ate.failure.reason`, `ate.template.namespace`, `ate.template.name`, `ate.workerpool.name`, `ate.sandbox.class`) | | `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 given the constraint filters (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`) | +| `atelet.snapshot.size` | atelet | histogram | uncompressed size in bytes of each gVisor snapshot image written during checkpoint (labels `file.name`, `ate.template.namespace`, `ate.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; already-running resume no-ops are not recorded so the histogram tracks actual activations, not router traffic | +| `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 and scope on resume; already-running resume no-ops are not recorded so the histogram tracks actual activations, not router traffic | | `ate.scheduler.assignment.duration` | ateapi | histogram | time it takes for an actor to be assigned to a worker, per attempt (version-conflict retries record only the final attempt), with the outcome (`assigned` / `no_free_worker` / `error`) and sandbox class to catch scheduling latency and capacity starvation problems | +| `ate.actor.restore.duration` | atelet | histogram | how long each phase of a restore takes on the worker node, which is where cold-start latency actually goes once ateapi hands off (labels `ate.snapshot.phase`, `ate.snapshot.kind`, `ate.snapshot.scope`, `ate.template.namespace`, `ate.template.name`, `ate.sandbox.class`, plus `ate.failure.reason` on failure) | +| `ate.actor.checkpoint.duration` | atelet | histogram | the same phase breakdown for writing a snapshot, so a slow suspend can be attributed to ateom or to the upload (same labels as the restore histogram) | The table lists the OpenTelemetry instrument names. How a name appears in a query depends on the backend (Cloud Monitoring (GMP) / Kind collector). @@ -127,7 +129,16 @@ For `atenet.router.route.duration`: 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). -The `ate.*` control-plane metric labels are either fixed value sets (operation, outcome, state, class, kind) or scoped to the deployment catalog (template and pool names are operator-created, never derived from request payloads), and the label set varies per operation: resume carries the most dimensions, delete only the operation and error type. `ate.sandbox.class` is derived from the template (each template has exactly one class), so it adds no extra series next to the template labels; it exists so dashboards can aggregate by class without enumerating template names. High-cardinality actor identity (name/uid/atespace) stays off metrics entirely and lives on logs and traces instead. +The three snapshot labels are orthogonal and mean the same thing on every histogram that carries them: +* `ate.snapshot.kind`: which snapshot the operation reads or writes. `local` (node-local, written by a pause), `latest` (the actor's own durable snapshot), `golden` (the template's image), or `boot` (from scratch, so it never appears on the atelet histograms). +* `ate.snapshot.scope`: what content it covers. `full`, `data`, or `data_on_golden` (restore-only: the actor's data combined with the golden guest state). +* `ate.snapshot.phase`: which step was timed. `volume_mount`, `manifest_fetch`, `sandbox_assets`, `download`, `oci_unpack`, `ateom_restore` on restore; `sandbox_assets`, `ateom_checkpoint`, `persist` on checkpoint; `total` on both. + +**Phases overlap and do not sum to `total`.** The download runs concurrently with the asset fetch and OCI unpack, so each is an independent observation; use `total` as the denominator. A phase that never started is absent rather than zero. + +On a failure, `ate.failure.reason` marks the phase that died and the `total`, and nothing else, so `ate.actor.restore.duration{ate.snapshot.phase="download", ate.failure.reason!=""}` says how often the download is what breaks and why, while the phases that succeeded stay queryable as successes. The atelet histograms classify with substrate's own reason taxonomy (the same one `ate.actor.crashes` uses) rather than `error.type`, because these handlers return wrapped domain errors and the gRPC status is only assigned after the handler returns, so a status code would read `Unknown` for nearly every real failure. Infrastructure failures that carry no reason report `UNKNOWN`. + +The `ate.*` control-plane metric labels are either fixed value sets (operation, outcome, state, class, kind, scope, phase) or scoped to the deployment catalog (template and pool names are operator-created, never derived from request payloads), and the label set varies per operation: resume carries the most dimensions, delete only the operation and error type. `ate.sandbox.class` is derived from the template (each template has exactly one class), so it adds no extra series next to the template labels; it exists so dashboards can aggregate by class without enumerating template names. High-cardinality actor identity (name/uid/atespace) stays off metrics entirely and lives on logs and traces instead. ### Bridged controller-runtime metrics (atecontroller) diff --git a/internal/ateattr/ateattr.go b/internal/ateattr/ateattr.go index 9ce5d6edf..94b225fc6 100644 --- a/internal/ateattr/ateattr.go +++ b/internal/ateattr/ateattr.go @@ -24,8 +24,10 @@ import ( "go.opentelemetry.io/otel/attribute" "github.com/agent-substrate/substrate/internal/ateerrors" + "github.com/agent-substrate/substrate/internal/proto/ateletpb" "github.com/agent-substrate/substrate/internal/resources" + atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" ) @@ -56,6 +58,10 @@ const ( // rather than nesting under the pool so it can grow siblings. // WorkerPoolNamespaceKey pairs with WorkerPoolNameKey: a WorkerPool is // namespaced, so the name alone does not identify one. +// The snapshot keys are orthogonal: kind is which snapshot, scope is what +// content it covers, and phase is which step of the operation an observation +// timed. Naming one image within a snapshot is the registry's file.name, not an +// ate.* key of its own. const ( ActorOperationNameKey = attribute.Key("ate.actor.operation.name") WorkerPoolNamespaceKey = attribute.Key("ate.workerpool.namespace") @@ -63,6 +69,8 @@ const ( WorkerStateKey = attribute.Key("ate.worker.state") SandboxClassKey = attribute.Key("ate.sandbox.class") SnapshotKindKey = attribute.Key("ate.snapshot.kind") + SnapshotScopeKey = attribute.Key("ate.snapshot.scope") + SnapshotPhaseKey = attribute.Key("ate.snapshot.phase") SchedulerOutcomeKey = attribute.Key("ate.scheduler.outcome") SchedulingConstraintKey = attribute.Key("ate.scheduling.constraint") RouterResumeKey = attribute.Key("ate.router.resume") @@ -149,6 +157,8 @@ const ( // the label is bounded at the producer: Local restores an in-node snapshot, // Latest pulls the actor's durable snapshot from object storage, Golden pulls the // template's golden image, Boot is a from-scratch start (not a restore). +// atelet derives the same values for its own histograms, where the kind is the +// snapshot a restore reads or a checkpoint writes; Boot never appears there. const ( SnapshotKindGolden = "golden" SnapshotKindLatest = "latest" @@ -156,6 +166,76 @@ const ( SnapshotKindBoot = "boot" ) +// Values for SnapshotScopeKey, mirroring ateletpb.SnapshotScope. Checkpoints +// only ever capture Full or Data; DataOnGolden is restore-only. +const ( + SnapshotScopeFull = "full" + SnapshotScopeData = "data" + SnapshotScopeDataOnGolden = "data_on_golden" + SnapshotScopeUnknown = "unknown" +) + +// SnapshotScopeValue maps the wire enum onto its label value, shared so ateapi +// (which sets the scope) and atelet (which receives it) cannot drift. An +// unrecognized scope reports as unknown rather than stringified, so no wire +// value can widen the label set. +func SnapshotScopeValue(scope ateletpb.SnapshotScope) string { + switch scope { + case ateletpb.SnapshotScope_SNAPSHOT_SCOPE_FULL: + return SnapshotScopeFull + case ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA: + return SnapshotScopeData + case ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN: + return SnapshotScopeDataOnGolden + default: + return SnapshotScopeUnknown + } +} + +// Values for SnapshotPhaseKey. Phases overlap (the download runs concurrently +// with the asset fetch and OCI unpack), so they are independent observations, +// not a partition of Total: summing across them is meaningless. +const ( + SnapshotPhaseVolumeMount = "volume_mount" + SnapshotPhaseManifestFetch = "manifest_fetch" + SnapshotPhaseSandboxAssets = "sandbox_assets" + SnapshotPhaseDownload = "download" + SnapshotPhaseOCIUnpack = "oci_unpack" + SnapshotPhaseAteomRestore = "ateom_restore" + SnapshotPhaseAteomCheckpoint = "ateom_checkpoint" + // Persist is one step with two destinations (upload for external, rename + // for local); SnapshotKindKey already says which. + SnapshotPhasePersist = "persist" + SnapshotPhaseTotal = "total" +) + +// FailureReason classifies err onto the bounded ateerrors taxonomy, reading the +// wrapped Reason or the AIP-193 ErrorInfo detail. An error carrying neither +// reports ReasonUnknown rather than anything derived from its message, which is +// what keeps the label bounded. +func FailureReason(err error) string { + if r := ateerrors.ExtractReason(err); r != "" { + return r + } + return ReasonUnknown +} + +// SandboxClassUnknown is the NormalizeSandboxClass fallback. +const SandboxClassUnknown = "unknown" + +// NormalizeSandboxClass bounds the label: atelet reads the class from a +// snapshot manifest in object storage that nothing validates on the way in. +// Empty reports as unknown rather than the gvisor default, so a manifest +// problem stays visible. +func NormalizeSandboxClass(class string) string { + switch atev1alpha1.SandboxClass(class) { + case atev1alpha1.SandboxClassGvisor, atev1alpha1.SandboxClassMicroVM: + return class + default: + return SandboxClassUnknown + } +} + // 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 eb37ad8a1..4792a8db0 100644 --- a/internal/ateattr/ateattr_test.go +++ b/internal/ateattr/ateattr_test.go @@ -15,11 +15,20 @@ package ateattr import ( + "context" + "errors" + "fmt" "testing" "go.opentelemetry.io/otel/attribute" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "github.com/agent-substrate/substrate/internal/ateerrors" + "github.com/agent-substrate/substrate/internal/proto/ateletpb" "github.com/agent-substrate/substrate/internal/resources" + + atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" ) @@ -154,6 +163,8 @@ func TestKeySpellings(t *testing.T) { {WorkerStateKey, "ate.worker.state"}, {SandboxClassKey, "ate.sandbox.class"}, {SnapshotKindKey, "ate.snapshot.kind"}, + {SnapshotScopeKey, "ate.snapshot.scope"}, + {SnapshotPhaseKey, "ate.snapshot.phase"}, {SchedulerOutcomeKey, "ate.scheduler.outcome"}, {ErrorTypeKey, "error.type"}, {FailureReasonKey, "ate.failure.reason"}, @@ -194,6 +205,23 @@ func TestMetricLabelValues(t *testing.T) { {SnapshotKindLatest, "latest"}, {SnapshotKindLocal, "local"}, {SnapshotKindBoot, "boot"}, + + {SnapshotScopeFull, "full"}, + {SnapshotScopeData, "data"}, + {SnapshotScopeDataOnGolden, "data_on_golden"}, + {SnapshotScopeUnknown, "unknown"}, + + {SnapshotPhaseVolumeMount, "volume_mount"}, + {SnapshotPhaseManifestFetch, "manifest_fetch"}, + {SnapshotPhaseSandboxAssets, "sandbox_assets"}, + {SnapshotPhaseDownload, "download"}, + {SnapshotPhaseOCIUnpack, "oci_unpack"}, + {SnapshotPhaseAteomRestore, "ateom_restore"}, + {SnapshotPhaseAteomCheckpoint, "ateom_checkpoint"}, + {SnapshotPhasePersist, "persist"}, + {SnapshotPhaseTotal, "total"}, + + {SandboxClassUnknown, "unknown"}, } for _, tt := range tests { t.Run(tt.want, func(t *testing.T) { @@ -256,6 +284,108 @@ func TestActorMetricAttributes(t *testing.T) { }) } +// TestSnapshotScopeValue pins the enum-to-label mapping ateapi and atelet share. +// An unmapped enum value must report unknown rather than its stringified form, +// which would let a wire value widen the label set. +func TestSnapshotScopeValue(t *testing.T) { + tests := []struct { + name string + scope ateletpb.SnapshotScope + want string + }{ + {name: "full", scope: ateletpb.SnapshotScope_SNAPSHOT_SCOPE_FULL, want: SnapshotScopeFull}, + {name: "data", scope: ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA, want: SnapshotScopeData}, + {name: "data on golden", scope: ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA_ON_GOLDEN, want: SnapshotScopeDataOnGolden}, + {name: "unspecified", scope: ateletpb.SnapshotScope_SNAPSHOT_SCOPE_UNSPECIFIED, want: SnapshotScopeUnknown}, + {name: "value outside the enum", scope: ateletpb.SnapshotScope(9999), want: SnapshotScopeUnknown}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := SnapshotScopeValue(tt.scope); got != tt.want { + t.Errorf("SnapshotScopeValue(%v) = %q, want %q", tt.scope, got, tt.want) + } + }) + } +} + +// TestNormalizeSandboxClass covers the cardinality guard: atelet reads the class +// out of a snapshot manifest nothing validates, so anything unrecognized has to +// collapse onto a single value. +func TestNormalizeSandboxClass(t *testing.T) { + tests := []struct { + name string + class string + want string + }{ + {name: "gvisor", class: string(atev1alpha1.SandboxClassGvisor), want: string(atev1alpha1.SandboxClassGvisor)}, + {name: "microvm", class: string(atev1alpha1.SandboxClassMicroVM), want: string(atev1alpha1.SandboxClassMicroVM)}, + {name: "empty", class: "", want: SandboxClassUnknown}, + {name: "unknown runtime", class: "kvm", want: SandboxClassUnknown}, + {name: "casing is not normalized away", class: "GVISOR", want: SandboxClassUnknown}, + {name: "attacker-controlled manifest value", class: "gvisor\";evil=\"1", want: SandboxClassUnknown}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := NormalizeSandboxClass(tt.class); got != tt.want { + t.Errorf("NormalizeSandboxClass(%q) = %q, want %q", tt.class, got, tt.want) + } + }) + } +} + +// TestFailureReason pins the error-to-label mapping: only the registered +// ateerrors taxonomy may reach the label, so anything unclassified collapses +// onto UNKNOWN instead of leaking an unbounded error message. +func TestFailureReason(t *testing.T) { + tests := []struct { + name string + err error + want string + }{ + { + name: "wrapped reason", + err: fmt.Errorf("%w: while uploading external snapshot", ateerrors.ReasonFaileSaveSnapshot), + want: string(ateerrors.ReasonFaileSaveSnapshot), + }, + { + name: "reason nested several wraps deep", + err: fmt.Errorf("restore: %w", fmt.Errorf("%w: bad manifest", ateerrors.ReasonInvalidSandboxAsset)), + want: string(ateerrors.ReasonInvalidSandboxAsset), + }, + { + name: "gRPC status carrying the reason as an ErrorInfo detail", + err: ateerrors.NewGRPCError(context.Background(), codes.DataLoss, ateerrors.ReasonTerminalFileSystemError, nil, errors.New("no space left on device")), + want: string(ateerrors.ReasonTerminalFileSystemError), + }, + { + name: "infrastructure error with no reason attached", + err: errors.New("dial tcp 10.96.192.187:9000: connect: connection refused"), + want: ReasonUnknown, + }, + { + name: "plain gRPC status with no ErrorInfo", + err: status.Error(codes.Unavailable, "unavailable"), + want: ReasonUnknown, + }, + { + name: "nil error", + err: nil, + want: ReasonUnknown, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := FailureReason(tt.err) + if got != tt.want { + t.Errorf("FailureReason() = %q, want %q", got, tt.want) + } + if !ateerrors.IsValidReason(got) { + t.Errorf("FailureReason() = %q, which is not a registered reason", got) + } + }) + } +} + func TestNormalizeOperationName(t *testing.T) { tests := []struct { op string diff --git a/internal/e2e/collector_metrics.go b/internal/e2e/collector_metrics.go index 9f25259c0..cae679bc3 100644 --- a/internal/e2e/collector_metrics.go +++ b/internal/e2e/collector_metrics.go @@ -43,6 +43,8 @@ var PlatformMetricPrefixes = []string{ "ate_actor_crashes", "ate_actor_lifecycle_operation_duration", "ate_scheduler_assignment_duration", + "ate_actor_restore_duration", + "ate_actor_checkpoint_duration", "atenet_router_route_duration", "ate_scheduler_eligible_workers", } diff --git a/internal/e2e/suites/metrics/metrics_test.go b/internal/e2e/suites/metrics/metrics_test.go index bf7af18fa..344c44ce5 100644 --- a/internal/e2e/suites/metrics/metrics_test.go +++ b/internal/e2e/suites/metrics/metrics_test.go @@ -89,7 +89,15 @@ func TestPlatformMetricsEmitted(t *testing.T) { } _ = resp.Body.Close() - // Trigger an actor crash to verify ate_actor_crashes counter emission. + // The first resume restored the template's golden snapshot; the actor has + // none of its own yet. Suspend writes one, and the second resume reads it + // back, so the checkpoint histogram gets a datapoint and the restore + // histogram covers both the golden and the latest kind. + suspend(t, ctx, clients, actorID) + resume(t, ctx, clients, actorID) + + // Trigger an actor crash to verify ate_actor_crashes counter emission. Last, + // because it deletes the worker pod. triggerActorCrash(t, ctx, clients, actorID) deadline := time.Now().Add(2 * time.Minute) @@ -217,6 +225,10 @@ func TestPlatformMetricsEmitted(t *testing.T) { errs = append(errs, "ate_actor_crashes metric line not found in collector scrape output") } + if err := validateSnapshotPhaseLabels(scrape); err != nil { + errs = append(errs, err.Error()) + } + if len(errs) == 0 { return } @@ -274,6 +286,46 @@ func resume(t *testing.T, ctx context.Context, clients *e2e.Clients, actorID str waitForStatus(t, ctx, clients, actorID, ateapipb.Actor_STATUS_RUNNING) } +// validateSnapshotPhaseLabels guards atelet's cold-start histograms against a +// silent regression the prefix check cannot see: kind and sandbox class are +// derived from the snapshot manifest and omitted when they cannot be resolved, +// so a broken derivation would keep emitting the metric with the labels that +// make it useful missing. +func validateSnapshotPhaseLabels(scrape string) error { + for _, m := range []string{"ate_actor_restore_duration_seconds_count", "ate_actor_checkpoint_duration_seconds_count"} { + var labelled bool + for _, line := range strings.Split(scrape, "\n") { + if !strings.HasPrefix(line, m) { + continue + } + phase := extractLabelValue(line, "ate_snapshot_phase") + kind := extractLabelValue(line, "ate_snapshot_kind") + scope := extractLabelValue(line, "ate_snapshot_scope") + class := extractLabelValue(line, "ate_sandbox_class") + if phase == "" { + return fmt.Errorf("%s line is missing ate_snapshot_phase: %q", m, line) + } + if kind != "" && scope != "" && class != "" { + labelled = true + } + } + if !labelled { + return fmt.Errorf("no %s line carried all of ate_snapshot_kind, ate_snapshot_scope and ate_sandbox_class", m) + } + } + return nil +} + +func suspend(t *testing.T, ctx context.Context, clients *e2e.Clients, actorID string) { + t.Helper() + if _, err := clients.SubstrateAPI.SuspendActor(ctx, &ateapipb.SuspendActorRequest{ + Actor: &ateapipb.ObjectRef{Atespace: metricsAtespace, Name: actorID}, + }); err != nil { + t.Fatalf("SuspendActor: %v", err) + } + waitForStatus(t, ctx, clients, actorID, ateapipb.Actor_STATUS_SUSPENDED) +} + func waitForStatus(t *testing.T, ctx context.Context, clients *e2e.Clients, actorID string, want ateapipb.Actor_Status) { t.Helper() deadline := time.Now().Add(2 * time.Minute) diff --git a/tools/setup-gcp/dashboards/ate-snapshot-dashboard.json b/tools/setup-gcp/dashboards/ate-snapshot-dashboard.json index 94ea189ff..aaaf881b9 100644 --- a/tools/setup-gcp/dashboards/ate-snapshot-dashboard.json +++ b/tools/setup-gcp/dashboards/ate-snapshot-dashboard.json @@ -14,7 +14,7 @@ "dataSets": [ { "timeSeriesQuery": { - "prometheusQuery": "histogram_quantile(0.99, sum by (le, actor_template_name) (rate({\"atelet.snapshot.size_bucket\", top_level_controller_name=\"atelet\", kind=\"pages\"}[1h])))", + "prometheusQuery": "histogram_quantile(0.99, sum by (le, ate_template_name) (rate({\"atelet.snapshot.size_bucket\", top_level_controller_name=\"atelet\", file_name=\"pages.img\"}[1h])))", "unitOverride": "By" }, "plotType": "LINE", @@ -83,7 +83,7 @@ "dataSets": [ { "timeSeriesQuery": { - "prometheusQuery": "sum by (actor_template_name) (rate({\"atelet.snapshot.size_count\", top_level_controller_name=\"atelet\", kind=\"pages\"}[1h]))", + "prometheusQuery": "sum by (ate_template_name) (rate({\"atelet.snapshot.size_count\", top_level_controller_name=\"atelet\", file_name=\"pages.img\"}[1h]))", "unitOverride": "1/s" }, "plotType": "LINE", @@ -109,7 +109,7 @@ { "timeSeriesQuery": { "timeSeriesFilter": { - "filter": "metric.type=\"prometheus.googleapis.com/atelet.snapshot.size/histogram\" resource.type=\"prometheus_target\" metric.labels.kind=\"pages\"", + "filter": "metric.type=\"prometheus.googleapis.com/atelet.snapshot.size/histogram\" resource.type=\"prometheus_target\" metric.labels.file_name=\"pages.img\"", "aggregation": { "alignmentPeriod": "60s", "perSeriesAligner": "ALIGN_DELTA",