diff --git a/cmd/atelet/actorguestmetrics.go b/cmd/atelet/actorguestmetrics.go new file mode 100644 index 0000000000..8c30467bbd --- /dev/null +++ b/cmd/atelet/actorguestmetrics.go @@ -0,0 +1,144 @@ +// Copyright 2026 The Agent Substrate Authors +// +// 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 ( + "errors" + "fmt" + "sync/atomic" + "time" + + "github.com/prometheus/client_golang/prometheus" + "go.ate.dev/substrate/internal/proto/ateworkerpb" + "go.ate.dev/substrate/internal/resources" + "k8s.io/apimachinery/pkg/util/validation/field" +) + +const maxDebugActorUIDs = 16 + +var ( + guestMetricLabels = []string{"actor_uid", "actor_name", "actor_atespace"} + guestCPUCores = prometheus.NewDesc("ate_debug_actor_guest_cpu_usage_cores", "Mean guest CPU cores used between two samples in the same activation.", guestMetricLabels, nil) + guestMemoryUsage = prometheus.NewDesc("ate_debug_actor_guest_memory_usage_bytes", "Guest memory usage including reclaimable page cache.", guestMetricLabels, nil) + guestMemoryWorkingSet = prometheus.NewDesc("ate_debug_actor_guest_memory_working_set_bytes", "Guest memory working set.", guestMetricLabels, nil) +) + +type actorGuestReading struct { + sample *ateworkerpb.WorkloadStatsSample + cores float64 + hasCPU bool +} + +// actorGuestMetrics exposes samples for the operator's allowlist. +// The stats poller publishes a complete snapshot after each sweep. +type actorGuestMetrics struct { + allowed map[string]struct{} + interval time.Duration + latest atomic.Pointer[map[string]actorGuestReading] +} + +func newActorGuestMetrics(uids []string, interval time.Duration) (*actorGuestMetrics, error) { + if len(uids) == 0 { + return nil, nil + } + if interval <= 0 { + return nil, errors.New("actor stats polling must be enabled") + } + if len(uids) > maxDebugActorUIDs { + return nil, fmt.Errorf("at most %d actor UIDs are allowed", maxDebugActorUIDs) + } + m := &actorGuestMetrics{allowed: make(map[string]struct{}, len(uids)), interval: interval} + for _, uid := range uids { + if errs := resources.ValidateUUID(uid, field.NewPath("uid")); len(errs) > 0 { + return nil, errs.ToAggregate() + } + m.allowed[uid] = struct{}{} + } + return m, nil +} + +func (m *actorGuestMetrics) selectSample(samples map[string]*ateworkerpb.WorkloadStatsSample, sample *ateworkerpb.WorkloadStatsSample) { + uid := sample.GetActorUid() + if _, ok := m.allowed[uid]; !ok || sample.GetEpochUnixNano() <= 0 { + return + } + switch sample.GetSource() { + case ateworkerpb.StatsSource_STATS_SOURCE_UNSPECIFIED, ateworkerpb.StatsSource_STATS_SOURCE_CGROUP: + // A restore can supersede the guest reading before the new activation + // is measured, or move the actor to gVisor. + case ateworkerpb.StatsSource_STATS_SOURCE_GUEST_AGENT: + if sample.GetObservedAtUnixNano() <= 0 { + return + } + default: + return + } + old := samples[uid] + if sample.GetEpochUnixNano() > old.GetEpochUnixNano() || + (sample.GetEpochUnixNano() == old.GetEpochUnixNano() && sample.GetObservedAtUnixNano() > old.GetObservedAtUnixNano()) { + samples[uid] = sample + } +} + +func (m *actorGuestMetrics) publish(samples map[string]*ateworkerpb.WorkloadStatsSample) { + previous := m.latest.Load() + next := make(map[string]actorGuestReading, len(samples)) + for uid, sample := range samples { + if sample.GetSource() != ateworkerpb.StatsSource_STATS_SOURCE_GUEST_AGENT { + continue + } + reading := actorGuestReading{sample: sample} + if previous != nil { + last := (*previous)[uid] + if last.sample != nil && last.sample.GetEpochUnixNano() == sample.GetEpochUnixNano() { + elapsed := sample.GetObservedAtUnixNano() - last.sample.GetObservedAtUnixNano() + if elapsed == 0 { + reading = last + } else if elapsed > 0 && sample.GetCpuUsageUsec() >= last.sample.GetCpuUsageUsec() { + reading.cores = 1000 * float64(sample.GetCpuUsageUsec()-last.sample.GetCpuUsageUsec()) / float64(elapsed) + reading.hasCPU = true + } + } + } + next[uid] = reading + } + m.latest.Store(&next) +} + +func (m *actorGuestMetrics) Describe(ch chan<- *prometheus.Desc) { + ch <- guestCPUCores + ch <- guestMemoryUsage + ch <- guestMemoryWorkingSet +} + +func (m *actorGuestMetrics) Collect(ch chan<- prometheus.Metric) { + snapshot := m.latest.Load() + if snapshot == nil { + return + } + cutoff := time.Now().Add(-m.interval).Add(-m.interval).UnixNano() + for uid, reading := range *snapshot { + sample := reading.sample + if sample.GetObservedAtUnixNano() < cutoff { + continue + } + labels := []string{uid, sample.GetActorName(), sample.GetAtespace()} + ch <- prometheus.MustNewConstMetric(guestMemoryUsage, prometheus.GaugeValue, float64(sample.GetMemoryCurrentBytes()), labels...) + ch <- prometheus.MustNewConstMetric(guestMemoryWorkingSet, prometheus.GaugeValue, float64(sample.GetMemoryWorkingSetBytes()), labels...) + if reading.hasCPU { + ch <- prometheus.MustNewConstMetric(guestCPUCores, prometheus.GaugeValue, reading.cores, labels...) + } + } +} diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 42d1e4d75b..367f9e139d 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -31,6 +31,7 @@ import ( "time" "github.com/google/go-containerregistry/pkg/authn" + "github.com/prometheus/client_golang/prometheus" "github.com/spf13/pflag" "github.com/spiffe/go-spiffe/v2/spiffeid" "github.com/spiffe/go-spiffe/v2/svid/x509svid" @@ -128,6 +129,7 @@ var ( otlpRelaySocket = pflag.String("otlp-relay-socket", nodepath.AteletOTLPSocketPath(), "Unix socket to serve the OTLP relay on, which forwards the node's worker telemetry to OTEL_EXPORTER_OTLP_ENDPOINT so worker pods need no network path to the collector. Empty disables the relay.") actorStatsPollInterval = pflag.Duration("actor-stats-poll-interval", time.Minute, fmt.Sprintf("Actor resource utilization sampling frequency. 0 disables the sampling entirely; minimum accepted value is %v.", minActorStatsPollInterval)) + actorGuestMetricsUIDs = pflag.StringSlice("debug-actor-guest-metrics-uids", nil, "Actor UIDs whose guest metrics to expose on /metrics. Empty disables per-actor metrics.") drainDelay = pflag.Duration("drain-delay", 0, "How long to keep accepting new RPCs after SIGTERM before starting the gRPC drain.") drainTimeout = pflag.Duration("drain-timeout", 5*time.Minute, "Deadline for the graceful gRPC drain on shutdown. In-flight RPCs still running past it are forcefully cancelled.") @@ -212,6 +214,16 @@ func main() { readiness := &serverboot.Readiness{} serving := make(chan struct{}) readiness.AddCheck("serving", servingCheck(serving)) + statsInterval := clampActorStatsPollInterval(ctx, *actorStatsPollInterval) + guestMetrics, err := newActorGuestMetrics(*actorGuestMetricsUIDs, statsInterval) + if err != nil { + serverboot.Fatal(ctx, "Invalid actor guest debug metrics configuration", err) + } + if guestMetrics != nil { + if err := prometheus.Register(guestMetrics); err != nil { + serverboot.Fatal(ctx, "Failed to register actor guest debug metrics", err) + } + } go serverboot.StartMetricsServer(ctx, *metricsListenAddr) go serverboot.StartHealthServer(ctx, *healthListenAddr, readiness) @@ -329,7 +341,7 @@ func main() { serverboot.Fatal(ctx, "Failed to create Kubernetes clients", err) } - if interval := clampActorStatsPollInterval(ctx, *actorStatsPollInterval); interval > 0 { + if statsInterval > 0 { if statsInst, err := newStatsInstruments(otel.Meter("atelet")); err != nil { // Telemetry must not take the node's lifecycle daemon down with // it. Instrument creation only fails on programmer error @@ -339,7 +351,7 @@ func main() { // crash-looping every actor operation on the node. slog.ErrorContext(ctx, "Actor stats sampling disabled: failed to create instruments", slog.Any("err", err)) } else { - startStatsPoller(ctx, interval, statsInst, k8sClient) + startStatsPoller(ctx, statsInterval, statsInst, k8sClient, guestMetrics) } } diff --git a/cmd/atelet/statspoller.go b/cmd/atelet/statspoller.go index 305cfaf04b..b1e54f0e6a 100644 --- a/cmd/atelet/statspoller.go +++ b/cmd/atelet/statspoller.go @@ -123,7 +123,8 @@ type statsPoller struct { // which is the join key. fetchWorkerPools func(ctx context.Context) map[string]workerPoolRef - inst *statsInstruments + inst *statsInstruments + guestMetrics *actorGuestMetrics // cachedPools carries pool resolutions across sweeps, so one failed pod // list cannot re-home a tick's samples -- and the CPU counter's @@ -259,6 +260,9 @@ func (p *statsPoller) collect(ctx context.Context) map[templateKey]*templateAggr // A node with no workers directory yet has no workers to measure; the // first RunWorkload dispatch creates it. slog.DebugContext(ctx, "Actor stats sweep: no workers directory", slog.Any("err", err)) + if p.guestMetrics != nil { + p.guestMetrics.publish(nil) + } return nil } @@ -271,6 +275,10 @@ func (p *statsPoller) collect(ctx context.Context) map[templateKey]*templateAggr } } pools := p.resolveWorkerPools(ctx, podUIDs) + var guestSamples map[string]*ateworkerpb.WorkloadStatsSample + if p.guestMetrics != nil { + guestSamples = make(map[string]*ateworkerpb.WorkloadStatsSample) + } var ( mu sync.Mutex @@ -311,6 +319,11 @@ func (p *statsPoller) collect(ctx context.Context) map[templateKey]*templateAggr // One entry per workload the worker is hosting; empty when it is // available. for _, sample := range resp.GetSamples() { + if p.guestMetrics != nil { + mu.Lock() + p.guestMetrics.selectSample(guestSamples, sample) + mu.Unlock() + } cpuAt := cpuKey{actorUID: sample.GetActorUid(), epoch: sample.GetEpochUnixNano()} if sample.GetSource() == ateworkerpb.StatsSource_STATS_SOURCE_UNSPECIFIED { // Pending: hosted but not measured, so it adds nothing @@ -363,6 +376,9 @@ func (p *statsPoller) collect(ctx context.Context) map[templateKey]*templateAggr // The tasks only ever return nil: a probe that fails is "not a target this // tick", never a failed sweep. _ = g.Wait() + if p.guestMetrics != nil { + p.guestMetrics.publish(guestSamples) + } // A worker that did not answer keeps its activations' baselines: dropping // one would charge the activation in full again once the worker answers. // Everything else this sweep did not see is dropped, so lastCPU does not @@ -582,7 +598,7 @@ func (i *statsInstruments) addCPU(ctx context.Context, aggs map[templateKey]*tem // The poller dials its own short-lived connection per probe (see // dialWorkerStats) and takes no WorkerDialer: the isolation from the lifecycle // RPCs' connection cache is structural, not just behavioral. -func startStatsPoller(ctx context.Context, interval time.Duration, inst *statsInstruments, k8sClient kubernetes.Interface) { +func startStatsPoller(ctx context.Context, interval time.Duration, inst *statsInstruments, k8sClient kubernetes.Interface, guestMetrics *actorGuestMetrics) { poller := &statsPoller{ interval: interval, workersDir: nodepath.WorkersDir(), @@ -593,8 +609,9 @@ func startStatsPoller(ctx context.Context, interval time.Duration, inst *statsIn } return ateworkerpb.NewWorkerClient(conn), closer, nil }, - inst: inst, - startedAt: time.Now().UnixNano(), + inst: inst, + guestMetrics: guestMetrics, + startedAt: time.Now().UnixNano(), } // NODE_NAME comes from the Downward API; without it the samples still // flow, just grouped without pool labels. diff --git a/cmd/atelet/statspoller_test.go b/cmd/atelet/statspoller_test.go index f2dc0c727a..4c9adc494b 100644 --- a/cmd/atelet/statspoller_test.go +++ b/cmd/atelet/statspoller_test.go @@ -19,13 +19,18 @@ import ( "errors" "io" "math" + "net/http" + "net/http/httptest" "os" "path/filepath" + "strings" "sync" "testing" "time" "github.com/google/go-cmp/cmp" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promhttp" sdkmetric "go.opentelemetry.io/otel/sdk/metric" "go.opentelemetry.io/otel/sdk/metric/metricdata" "google.golang.org/grpc" @@ -188,6 +193,102 @@ func assertClosed(t *testing.T, fakes map[string]*fakeStatsWorker, closers map[s } } +func TestActorGuestDebugMetrics(t *testing.T) { + const uid = "c7f3fb50-ef2d-4075-b59a-4a8e192e48a0" + const other = "c7f3fb50-ef2d-4075-b59a-4a8e192e48a1" + const unselected = "c7f3fb50-ef2d-4075-b59a-4a8e192e48a2" + const labels = `{actor_atespace="team",actor_name="my-agent",actor_uid="` + uid + `"}` + const cpu = "ate_debug_actor_guest_cpu_usage_cores" + labels + " 1" + const memory = "ate_debug_actor_guest_memory_usage_bytes" + labels + " 4096" + const workingSet = "ate_debug_actor_guest_memory_working_set_bytes" + labels + " 2048" + now := time.Now() + start := now.Add(-10 * time.Second) + guest := func(actorUID string, epoch, observed time.Time, cpuUsec uint64) *ateworkerpb.WorkloadStatsSample { + sample := atEpoch(measuredSample(actorUID, "ns", "template", 4096, 2048, cpuUsec), epoch.UnixNano()) + sample.Source = ateworkerpb.StatsSource_STATS_SOURCE_GUEST_AGENT + sample.SandboxClass = "microvm" + sample.Atespace = "team" + sample.ActorName = "my-agent" + sample.ObservedAtUnixNano = observed.UnixNano() + return sample + } + + metrics, err := newActorGuestMetrics([]string{uid, other}, time.Minute) + if err != nil { + t.Fatal(err) + } + worker := &fakeStatsWorker{} + poller, _ := newPollerFixture(t, map[string]*fakeStatsWorker{"worker": worker}) + poller.guestMetrics = metrics + if err := prometheus.Register(metrics); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { prometheus.Unregister(metrics) }) + handler := promhttp.Handler() + scrape := func(samples ...*ateworkerpb.WorkloadStatsSample) string { + t.Helper() + worker.resp = &ateworkerpb.GetActiveWorkloadStatsResponse{Samples: samples} + poller.collect(t.Context()) + res := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/metrics", nil) + handler.ServeHTTP(res, req) + if res.Code != http.StatusOK { + t.Fatalf("scrape returned %d: %s", res.Code, res.Body) + } + return res.Body.String() + } + + first := guest(uid, start, start, 100_000) + if got := scrape(first, guest(unselected, start, start, 1), + measuredSample(other, "ns", "template", 4096, 2048, 1)); !strings.Contains(got, memory) || + !strings.Contains(got, workingSet) || strings.Contains(got, "cpu_usage_cores") || + strings.Contains(got, other) || strings.Contains(got, unselected) { + t.Fatalf("first scrape should contain only the selected guest memory: %s", got) + } + second := guest(uid, start, start.Add(time.Second), 1_100_000) + for _, sample := range []*ateworkerpb.WorkloadStatsSample{second, second} { + if got := scrape(sample, first); !strings.Contains(got, cpu+"\n") { + t.Fatalf("new or cached reading should report one CPU core: %s", got) + } + } + if got := scrape(second, guest(uid, now, now, 2_100_000)); !strings.Contains(got, memory) || + strings.Contains(got, "cpu_usage_cores") { + t.Fatalf("a new activation should start with a fresh CPU baseline: %s", got) + } + if got := scrape(guest(other, now, now, 2_200_000)); !strings.Contains(got, other) || + strings.Contains(got, uid) || strings.Contains(got, "cpu_usage_cores") { + t.Fatalf("a recreated actor should have its own series and CPU baseline: %s", got) + } + scrape(second) + pending := atEpoch(pendingSample(uid, "ns", "template"), now.UnixNano()) + if got := scrape(second, pending); strings.Contains(got, uid) { + t.Fatalf("pending restore exposed the previous activation's series: %s", got) + } + expired := now.Add(-3 * time.Minute) + if got := scrape(guest(uid, expired, expired, 1)); strings.Contains(got, uid) { + t.Fatalf("expired guest sample still had a series: %s", got) + } +} + +func TestActorGuestDebugMetricsConfig(t *testing.T) { + const uid = "c7f3fb50-ef2d-4075-b59a-4a8e192e48a0" + if metrics, err := newActorGuestMetrics(nil, 0); metrics != nil || err != nil { + t.Fatalf("empty allowlist should disable per-actor metrics, got %v, %v", metrics, err) + } + for _, tc := range []struct { + uids []string + interval time.Duration + }{ + {[]string{uid}, 0}, + {[]string{"*"}, time.Minute}, + {make([]string, maxDebugActorUIDs+1), time.Minute}, + } { + if _, err := newActorGuestMetrics(tc.uids, tc.interval); err == nil { + t.Errorf("accepted invalid config with UIDs %q and interval %v", tc.uids, tc.interval) + } + } +} + func TestStatsPollerCollectAggregates(t *testing.T) { // Two actors of the same template on this node, one of another, one idle // worker, one mid-boot: the same-template pair sums, the others contribute diff --git a/docs/guides/observability.md b/docs/guides/observability.md index 8969b174fe..53b549d423 100644 --- a/docs/guides/observability.md +++ b/docs/guides/observability.md @@ -268,6 +268,33 @@ jsonPayload.msg="Actor usage sampled" AND jsonPayload."ate.actor.uid"="8f2a…" **Do not put actor identity on a log-based metric.** Aggregating these events in the log store is what they are for, but a log-based metric built over them must label only by the bounded set (template, sandbox class, source, pool) — promoting `ate.actor.uid` or `ate.actor.name` into a metric label reintroduces exactly the per-actor cardinality this split keeps out of the TSDB. +### Guest Metrics for Selected Actors + +For microVM troubleshooting, set `--debug-actor-guest-metrics-uids=[,...]` +on atelet to expose guest resource usage for up to 16 explicit actor UIDs on its +existing Prometheus `/metrics` endpoint. An empty allowlist disables these +series. Actor stats polling must be enabled. + +The series carry `actor_uid`, `actor_name` and `actor_atespace`. Use the UID +to distinguish actors recreated with the same name. The default +`ate-system-observability` NetworkPolicy allows traffic to the metrics port from +any source. Before enabling these metrics, restrict access to the scrape +endpoint, pod port-forwarding, and the metrics backend to readers authorized +to see those identities. + +* `ate_debug_actor_guest_cpu_usage_cores{actor_uid="…"}` is the mean CPU cores + used between two distinct samples in the same activation. It is absent until + two samples are available. +* `ate_debug_actor_guest_memory_usage_bytes{actor_uid="…"}` includes reclaimable + page cache. `ate_debug_actor_guest_memory_working_set_bytes{actor_uid="…"}` + reports the working set. + +The debug metrics reuse cached samples from the stats sweep. A series disappears +when the sweep no longer finds a usable reading or the sample exceeds twice the +atelet polling interval. To inspect an actor after it moves, query the atelet on +the new node, with that UID also allowlisted there. These series are available +through Prometheus scraping only. + ### Secrets in Component Logs Control plane and node component logs mask proto fields marked `debug_redact`, such as actor JWTs and env var values: a request or response body in a log line shows `[REDACTED]` in their place. Not every process masks them; the developer guide below lists the ones that do not. An actor's own container output is not touched. @@ -354,7 +381,7 @@ identity to metric labels. **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. -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 runtime on the actor's worker assignment or snapshot (each worker pool has exactly one class), so it adds no extra series next to the pool label; it exists so dashboards can aggregate by class without enumerating pool names. High-cardinality actor identity (name/uid/atespace) stays off metrics entirely and lives on logs and traces instead — for resource usage, on the [per-actor usage events](#per-actor-usage-events). +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 runtime on the actor's worker assignment or snapshot (each worker pool has exactly one class), so it adds no extra series next to the pool label; it exists so dashboards can aggregate by class without enumerating pool names. Actor name, UID and atespace stay off the standard metrics pipeline. They appear on logs and traces, including [per-actor usage events](#per-actor-usage-events), and on the opt-in [guest debug metrics](#guest-metrics-for-selected-actors). ### The metric registry diff --git a/docs/metrics/substrate.yaml b/docs/metrics/substrate.yaml index 03cf473c4e..cf7fa94a6c 100644 --- a/docs/metrics/substrate.yaml +++ b/docs/metrics/substrate.yaml @@ -132,9 +132,13 @@ bridged_metric_families: cardinality_rules: - id: no-actor-identity brief: > - The name, the UID and the atespace of an actor are never metric labels. + The name, the UID and the atespace of an actor are never labels on + metrics in the standard telemetry pipeline. They have too many values at the target of 1 billion actors. Put them on spans and logs. The trace_id field connects a log to a metric. + The opt-in ate_debug_actor_guest_* Prometheus series may use actor_uid, + actor_name and actor_atespace for an explicit operator allowlist of at + most 16 UIDs per atelet. These series are excluded from OTLP export. enforced: false could_be_enforced_by: > A Weaver Rego policy can refuse a forbidden attribute in this registry. It @@ -189,7 +193,7 @@ cardinality_rules: lint_exceptions: - attribute: outcome metric: atenet.router.parking.wait.duration - rule: A metric label of Substrate must start with ate. + rule: A Substrate OTLP metric label must start with ate. brief: > This label is older than the naming convention of the router. A change to ate.router.park.outcome needs a change to the dashboards at the same time.