diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index adbe3f993..ea3f33214 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -172,6 +172,7 @@ func main() { imagecache.WithLocalhostRegistryReplacement(*localhostRegistryReplacement), imagecache.WithActorsDir(ateompath.ActorsDir), imagecache.WithMinAge(*imageCacheMinAge), + imagecache.WithMeter(otel.Meter("atelet")), ) if err != nil { serverboot.Fatal(ctx, "Failed to open image cache", err) diff --git a/docs/observability.md b/docs/observability.md index 5e1c93d9f..61d92f6f6 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -123,6 +123,7 @@ Agent Substrate emits foundational OpenTelemetry system and server metrics to mo | `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) | +| `ate.imagecache.requests` | atelet | counter | image lookups in the node-local image cache, by outcome (`ate.imagecache.outcome`), with `error.type` on the `error` outcome. A miss pays for the pull and the unpack, so the hit ratio per node is a leading indicator of resume latency | The table lists the OpenTelemetry instrument names. How a name appears in a query depends on the backend (Cloud Monitoring (GMP) / Kind collector). @@ -137,6 +138,10 @@ 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). +For `ate.imagecache.requests`: +* `ate.imagecache.outcome` is `hit` when the node holds a complete image record — every layer directory the record names is present — and `miss` when the lookup must pull. A failed lookup is neither: `error` is a failed lookup whatever the cause, and `cancelled` or `timeout` is the caller giving up, as on `ate.router.outcome`. So the hit ratio is `hit / (hit + miss)`, with failures and abandoned lookups out of the denominator. +* `error.type` is present only on the `error` outcome, and carries the registry's own HTTP status for its rejection, from a fixed set: `401`, `403`, `404`, `429`, `500`, `502`, `503`, `504`. The set is an allow-list because the registry client reports whatever the remote returned. Each other status, and each failure that carries no status, reports `_OTHER`. + 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). diff --git a/internal/ateattr/ateattr.go b/internal/ateattr/ateattr.go index 94b225fc6..7a938d0ca 100644 --- a/internal/ateattr/ateattr.go +++ b/internal/ateattr/ateattr.go @@ -62,6 +62,9 @@ const ( // 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. +// ImageCacheOutcomeKey is rooted at the subsystem, not under actor: the layer +// pool is node state every actor shares. For the same reason it is the only +// ate.* label on its counter. const ( ActorOperationNameKey = attribute.Key("ate.actor.operation.name") WorkerPoolNamespaceKey = attribute.Key("ate.workerpool.namespace") @@ -71,6 +74,7 @@ const ( SnapshotKindKey = attribute.Key("ate.snapshot.kind") SnapshotScopeKey = attribute.Key("ate.snapshot.scope") SnapshotPhaseKey = attribute.Key("ate.snapshot.phase") + ImageCacheOutcomeKey = attribute.Key("ate.imagecache.outcome") SchedulerOutcomeKey = attribute.Key("ate.scheduler.outcome") SchedulingConstraintKey = attribute.Key("ate.scheduling.constraint") RouterResumeKey = attribute.Key("ate.router.resume") @@ -103,6 +107,17 @@ const ( RouterResumeJoined = "joined" ) +// Values for ImageCacheOutcomeKey. A hit is a complete image record; a miss +// must pull. A failed lookup is neither: Error is the only one that carries an +// error.type, Cancelled and Timeout mean the caller gave up. +const ( + ImageCacheOutcomeHit = "hit" + ImageCacheOutcomeMiss = "miss" + ImageCacheOutcomeError = "error" + ImageCacheOutcomeCancelled = "cancelled" + ImageCacheOutcomeTimeout = "timeout" +) + // ErrorTypeKey is the OTel registry attribute, reused verbatim (not aliased into // ate.*): failures are reported on the same instrument via this key, its absence // meaning success, never as a parallel _failures counter. diff --git a/internal/ateattr/ateattr_test.go b/internal/ateattr/ateattr_test.go index 4792a8db0..0813e3260 100644 --- a/internal/ateattr/ateattr_test.go +++ b/internal/ateattr/ateattr_test.go @@ -165,6 +165,7 @@ func TestKeySpellings(t *testing.T) { {SnapshotKindKey, "ate.snapshot.kind"}, {SnapshotScopeKey, "ate.snapshot.scope"}, {SnapshotPhaseKey, "ate.snapshot.phase"}, + {ImageCacheOutcomeKey, "ate.imagecache.outcome"}, {SchedulerOutcomeKey, "ate.scheduler.outcome"}, {ErrorTypeKey, "error.type"}, {FailureReasonKey, "ate.failure.reason"}, @@ -197,6 +198,12 @@ func TestMetricLabelValues(t *testing.T) { {OperationDelete, "delete"}, {OperationUnknown, "unknown"}, + {ImageCacheOutcomeHit, "hit"}, + {ImageCacheOutcomeMiss, "miss"}, + {ImageCacheOutcomeError, "error"}, + {ImageCacheOutcomeCancelled, "cancelled"}, + {ImageCacheOutcomeTimeout, "timeout"}, + {SchedulerOutcomeAssigned, "assigned"}, {SchedulerOutcomeNoFreeWorker, "no_free_worker"}, {SchedulerOutcomeError, "error"}, diff --git a/internal/imagecache/imagecache.go b/internal/imagecache/imagecache.go index da6fb04e4..98bdc68cf 100644 --- a/internal/imagecache/imagecache.go +++ b/internal/imagecache/imagecache.go @@ -65,8 +65,11 @@ import ( "github.com/google/go-containerregistry/pkg/name" v1 "github.com/google/go-containerregistry/pkg/v1" "github.com/google/go-containerregistry/pkg/v1/remote" + "go.opentelemetry.io/otel/metric" "golang.org/x/sync/errgroup" "golang.org/x/sync/singleflight" + + "github.com/agent-substrate/substrate/internal/ateattr" ) const ( @@ -119,6 +122,13 @@ type Store struct { // spec write / ateom mount that roots it. minAge time.Duration + // meter, when set, is the meter the store reports on. See WithMeter. + meter metric.Meter + + // requests counts EnsureImage lookups by outcome. Nil without a meter, + // which recordRequest treats as a no-op. + requests metric.Int64Counter + imageSF singleflight.Group layerSF singleflight.Group @@ -168,6 +178,13 @@ func WithMinAge(d time.Duration) Option { return func(s *Store) { s.minAge = d } } +// WithMeter attaches the meter the store reports ate.imagecache.requests on. +// Without it the store records nothing, so a caller with no metrics pipeline +// needs no meter provider. +func WithMeter(m metric.Meter) Option { + return func(s *Store) { s.meter = m } +} + // Image describes one cached, ready-to-compose image. type Image struct { // Digest is the manifest digest the caller's ref resolved to (for a @@ -198,6 +215,14 @@ func New(root string, opts ...Option) (*Store, error) { o(s) } + if s.meter != nil { + requests, err := newRequestsCounter(s.meter) + if err != nil { + return nil, err + } + s.requests = requests + } + for _, d := range []string{s.layersDir(), s.manifestsDir()} { if err := os.MkdirAll(d, 0o700); err != nil { return nil, fmt.Errorf("while creating image cache dir %q: %w", d, err) @@ -302,7 +327,12 @@ func (s *Store) sweepTempDirs() error { // I/O; tag refs cost one HEAD request to resolve the tag to a manifest // digest (so tag refs are cacheable, and a moved tag is picked up on the // next call). -func (s *Store) EnsureImage(ctx context.Context, ref string) (*Image, error) { +func (s *Store) EnsureImage(ctx context.Context, ref string) (_ *Image, err error) { + // A miss until a complete record proves otherwise; recordRequest + // reclassifies a failure onto its own outcome. + outcome := ateattr.ImageCacheOutcomeMiss + defer func() { s.recordRequest(ctx, outcome, err) }() + parsedRef, err := s.parseRef(ref) if err != nil { return nil, fmt.Errorf("while parsing reference: %w", err) @@ -317,9 +347,10 @@ func (s *Store) EnsureImage(ctx context.Context, ref string) (*Image, error) { } else { // Tag ref: one small HEAD request pins it to an immutable manifest // digest, which is the only safe cache key for mutable tags. - desc, err := remote.Head(parsedRef, s.remoteOpts(ctx, parsedRef)...) - if err != nil { - return nil, fmt.Errorf("while resolving tag %q to a digest: %w", ref, err) + desc, headErr := remote.Head(parsedRef, s.remoteOpts(ctx, parsedRef)...) + if headErr != nil { + err = fmt.Errorf("while resolving tag %q to a digest: %w", ref, headErr) + return nil, err } digest = desc.Digest } @@ -329,6 +360,7 @@ func (s *Store) EnsureImage(ctx context.Context, ref string) (*Image, error) { return nil, err } if img != nil { + outcome = ateattr.ImageCacheOutcomeHit slog.InfoContext(ctx, "Image cache hit", slog.String("ref", ref), slog.String("digest", digest.String())) return img, nil } diff --git a/internal/imagecache/metrics.go b/internal/imagecache/metrics.go new file mode 100644 index 000000000..6e014b4f7 --- /dev/null +++ b/internal/imagecache/metrics.go @@ -0,0 +1,100 @@ +// 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 imagecache + +import ( + "context" + "errors" + "fmt" + "strconv" + + "github.com/google/go-containerregistry/pkg/v1/remote/transport" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" + + "github.com/agent-substrate/substrate/internal/ateattr" +) + +const requestsMetric = "ate.imagecache.requests" + +// errTypeOther is the OTel registry's fallback, used for a failure that +// carries no status of its own. It is spelled _OTHER, not the "unknown" of the +// ate.* labels, because error.type is that registry's attribute reused +// verbatim. +const errTypeOther = "_OTHER" + +// reportedStatuses bounds the label: transport.Error carries whatever the +// remote returned, so a registry or a proxy could otherwise mint a new series +// per status. Each listed one has its own action: credentials (401, 403), the +// ref (404), throttling (429), a registry-side fault (5xx). +var reportedStatuses = map[int]bool{ + 401: true, 403: true, 404: true, 429: true, + 500: true, 502: true, 503: true, 504: true, +} + +// newRequestsCounter creates the ate.imagecache.requests instrument. +func newRequestsCounter(meter metric.Meter) (metric.Int64Counter, error) { + counter, err := meter.Int64Counter( + requestsMetric, + metric.WithUnit("{request}"), + metric.WithDescription("Number of image lookups in the node-local image cache, by outcome."), + ) + if err != nil { + return nil, fmt.Errorf("create %s counter: %w", requestsMetric, err) + } + return counter, nil +} + +// recordRequest counts one EnsureImage lookup. A store with no meter records +// nothing. A failure replaces the hit-or-miss outcome the caller passed, and +// only the Error outcome carries an error.type. +func (s *Store) recordRequest(ctx context.Context, outcome string, err error) { + if s.requests == nil { + return + } + if err != nil { + outcome = failureOutcome(err) + } + attrs := []attribute.KeyValue{ateattr.ImageCacheOutcomeKey.String(outcome)} + if outcome == ateattr.ImageCacheOutcomeError { + attrs = append(attrs, ateattr.ErrorTypeKey.String(errorType(err))) + } + // A cancelled lookup still reports: its pull was started and paid for. + s.requests.Add(context.WithoutCancel(ctx), 1, metric.WithAttributes(attrs...)) +} + +// failureOutcome separates a failed lookup from a caller that gave up. +// Cancellation is read first: an abandoned request also carries a transport +// error, which would otherwise read as a registry outage. +func failureOutcome(err error) string { + switch { + case errors.Is(err, context.Canceled): + return ateattr.ImageCacheOutcomeCancelled + case errors.Is(err, context.DeadlineExceeded): + return ateattr.ImageCacheOutcomeTimeout + } + return ateattr.ImageCacheOutcomeError +} + +// errorType reports the registry's own status for its rejection, the only +// domain status this path has. Each other failure, and each status outside the +// reported set, carries none. +func errorType(err error) string { + var terr *transport.Error + if errors.As(err, &terr) && reportedStatuses[terr.StatusCode] { + return strconv.Itoa(terr.StatusCode) + } + return errTypeOther +} diff --git a/internal/imagecache/metrics_test.go b/internal/imagecache/metrics_test.go new file mode 100644 index 000000000..2280d5d58 --- /dev/null +++ b/internal/imagecache/metrics_test.go @@ -0,0 +1,247 @@ +// 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 imagecache + +import ( + "archive/tar" + "context" + "errors" + "fmt" + "os" + "syscall" + "testing" + + v1 "github.com/google/go-containerregistry/pkg/v1" + "github.com/google/go-containerregistry/pkg/v1/remote/transport" + "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" +) + +// newMeteredStore opens a store against a local ManualReader-backed provider, +// so the tests never touch the global meter provider. +func newMeteredStore(t *testing.T, opts ...Option) (*Store, *sdkmetric.ManualReader) { + t.Helper() + reader := sdkmetric.NewManualReader() + mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + return newTestStore(t, append(opts, WithMeter(mp.Meter("atelet")))...), reader +} + +// requestSeries is one ate.imagecache.requests datapoint, flattened to its +// label values. +type requestSeries struct { + outcome string + errorType string +} + +func collectRequests(t *testing.T, reader *sdkmetric.ManualReader) map[requestSeries]int64 { + t.Helper() + var rm metricdata.ResourceMetrics + if err := reader.Collect(context.Background(), &rm); err != nil { + t.Fatalf("collect: %v", err) + } + got := make(map[requestSeries]int64) + for _, sm := range rm.ScopeMetrics { + for _, m := range sm.Metrics { + if m.Name != requestsMetric { + continue + } + if m.Unit != "{request}" { + t.Errorf("unit = %q, want {request}", m.Unit) + } + sum, ok := m.Data.(metricdata.Sum[int64]) + if !ok { + t.Fatalf("data type = %T, want Sum[int64]", m.Data) + } + if !sum.IsMonotonic { + t.Error("IsMonotonic = false, want true (a counter, not an updowncounter)") + } + for _, dp := range sum.DataPoints { + got[seriesOf(t, dp.Attributes)] += dp.Value + } + } + } + return got +} + +func seriesOf(t *testing.T, set attribute.Set) requestSeries { + t.Helper() + s := requestSeries{} + if v, ok := set.Value(ateattr.ImageCacheOutcomeKey); ok { + s.outcome = v.AsString() + } + // An absent error.type means success; the zero value stands for it. + if v, ok := set.Value(ateattr.ErrorTypeKey); ok { + s.errorType = v.AsString() + } + return s +} + +func pushOneLayerImage(t *testing.T, ref string) { + t.Helper() + layer := layerFromEntries(t, []tarEntry{ + {name: "app/", typeflag: tar.TypeDir}, + {name: "app/main", typeflag: tar.TypeReg, mode: 0o755, body: "main"}, + }) + pushImage(t, ref, v1.Config{}, layer) +} + +// TestRequestsCountsMissThenHit: the first lookup pays for the pull, every +// later one is free. +func TestRequestsCountsMissThenHit(t *testing.T) { + _, host := newTestRegistry(t) + ref := host + "/test/metrics:latest" + pushOneLayerImage(t, ref) + + store, reader := newMeteredStore(t) + ctx := context.Background() + + for range 3 { + if _, err := store.EnsureImage(ctx, ref); err != nil { + t.Fatalf("EnsureImage: %v", err) + } + } + + got := collectRequests(t, reader) + want := map[requestSeries]int64{ + {outcome: ateattr.ImageCacheOutcomeMiss}: 1, + {outcome: ateattr.ImageCacheOutcomeHit}: 2, + } + for series, count := range want { + if got[series] != count { + t.Errorf("series %+v = %d, want %d (all series: %+v)", series, got[series], count, got) + } + } + if len(got) != len(want) { + t.Errorf("collected %d series, want %d: %+v", len(got), len(want), got) + } +} + +// TestRequestsRecordsFailures pins the outcome of each failure kind, and that +// only the error outcome carries an error.type. +func TestRequestsRecordsFailures(t *testing.T) { + _, host := newTestRegistry(t) + ref := host + "/test/fails:latest" + pushOneLayerImage(t, ref) + + tests := []struct { + name string + ref string + ctx func() context.Context + want requestSeries + }{ + { + // The ref never reached a registry, so there is no status. + name: "unparseable ref", + ref: "NOT A REFERENCE", + want: requestSeries{outcome: ateattr.ImageCacheOutcomeError, errorType: errTypeOther}, + }, + { + // The registry's own identifier, reported verbatim. + name: "no such tag", + ref: host + "/test/fails:absent", + want: requestSeries{outcome: ateattr.ImageCacheOutcomeError, errorType: "404"}, + }, + { + // The cache is healthy; the caller went away. No error.type. + name: "caller gave up", + ref: ref, + ctx: func() context.Context { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + return ctx + }, + want: requestSeries{outcome: ateattr.ImageCacheOutcomeCancelled}, + }, + { + name: "caller ran out of time", + ref: ref, + ctx: func() context.Context { + ctx, cancel := context.WithTimeout(context.Background(), 0) + t.Cleanup(cancel) + return ctx + }, + want: requestSeries{outcome: ateattr.ImageCacheOutcomeTimeout}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + store, reader := newMeteredStore(t) + ctx := context.Background() + if tt.ctx != nil { + ctx = tt.ctx() + } + + if _, err := store.EnsureImage(ctx, tt.ref); err == nil { + t.Fatal("EnsureImage succeeded, want a failure") + } + + got := collectRequests(t, reader) + if got[tt.want] != 1 { + t.Errorf("series %+v = %d, want 1 (all series: %+v)", tt.want, got[tt.want], got) + } + }) + } +} + +// TestClassifyFailure pins the bounded label set. Each value comes from a +// sentinel or a typed error, never from a message. +func TestClassifyFailure(t *testing.T) { + tests := []struct { + name string + err error + wantOutcome string + wantErrorType string + }{ + {"caller cancelled", fmt.Errorf("while resolving tag: %w", context.Canceled), ateattr.ImageCacheOutcomeCancelled, ""}, + {"caller timed out", fmt.Errorf("while resolving tag: %w", context.DeadlineExceeded), ateattr.ImageCacheOutcomeTimeout, ""}, + {"registry rejection", &transport.Error{StatusCode: 404}, ateattr.ImageCacheOutcomeError, "404"}, + {"credentials expired", fmt.Errorf("in remote.Image: %w", &transport.Error{StatusCode: 401}), ateattr.ImageCacheOutcomeError, "401"}, + {"throttled", &transport.Error{StatusCode: 429}, ateattr.ImageCacheOutcomeError, "429"}, + // A remote can return any status; only the reported set is a label. + {"unlisted status", &transport.Error{StatusCode: 418}, ateattr.ImageCacheOutcomeError, errTypeOther}, + {"disk full", &os.PathError{Op: "write", Path: "/var/lib/x", Err: syscall.ENOSPC}, ateattr.ImageCacheOutcomeError, errTypeOther}, + {"unclassified", errors.New("layer dir vanished during pull"), ateattr.ImageCacheOutcomeError, errTypeOther}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := failureOutcome(tt.err); got != tt.wantOutcome { + t.Errorf("failureOutcome(%v) = %q, want %q", tt.err, got, tt.wantOutcome) + } + if tt.wantErrorType == "" { + return + } + if got := errorType(tt.err); got != tt.wantErrorType { + t.Errorf("errorType(%v) = %q, want %q", tt.err, got, tt.wantErrorType) + } + }) + } +} + +// TestRequestsWithoutMeterIsNoOp: the validation tool and most tests open the +// cache with no metrics pipeline. +func TestRequestsWithoutMeterIsNoOp(t *testing.T) { + _, host := newTestRegistry(t) + ref := host + "/test/nometer:latest" + pushOneLayerImage(t, ref) + + store := newTestStore(t) + if _, err := store.EnsureImage(context.Background(), ref); err != nil { + t.Fatalf("EnsureImage: %v", err) + } +}