From 38977fde5c78853ae97b95ef16c0d110208623f5 Mon Sep 17 00:00:00 2001 From: Tim Bai Date: Fri, 14 Aug 2026 10:45:31 -0400 Subject: [PATCH] atelet: poll ateom workload stats into template-level metrics The reader half of #896: a poller that discovers the node's ateoms from the filesystem and turns their GetActiveWorkloadStats samples into the TSDB half of #174's split -- per-ActorTemplate gauges with the bounded label set (template, sandbox class, stats source), actor and atespace identity never reaching a metric label. Discovery holds no state and never asks the control plane: every ateom registers itself on disk by creating its socket directory at boot, so a sweep is one readdir plus one probe per entry, and an atelet restart loses nothing. One tolerance rule covers the scan's noise -- any dial or call failure means "not a target this tick" -- which uniformly handles stale directories, half-born ateoms, and teardowns mid-sweep. The no-sample reasons are skips by the RPC's own contract. Attribution comes solely from the echoed identity, per the same contract. The interval is a flag (--actor-stats-poll-interval, default 1m, 0 disables) clamped to the worst-case micro-VM sweep so a low setting cannot pile overlapping polls onto one guest agent; within a sweep, distinct ateoms are probed concurrently (bounded), which stacks nothing on any one guest and keeps a node of stuck sockets from serializing into minutes. Three gauges: sampled actors, memory current, memory working set -- observable rather than synchronous, so a template whose actors leave the node disappears from the export instead of freezing at its last value. CPU is a counter, not a gauge: the raw cpu_usage_usec is cumulative per-epoch per actor, so the poller tracks each actor's last seen value and adds only the per-sweep INCREASE (a decrease is an epoch reset, charged from zero), which keeps rate() meaningful across actors joining, leaving, and resetting. Undercounts across atelet restarts and misses the tail before a checkpoint; the events channel carries per-actor precision. Samples are enriched with the owning WorkerPool (ate.workerpool.namespace/name) by resolving the node's worker pods -- one field- and label-selected list per sweep, joined on the pod UID the ateom directory is named for; an unresolved pod groups without pool labels rather than vanishing. Part of #896, toward #550. --- cmd/atelet/main.go | 16 + cmd/atelet/statspoller.go | 534 ++++++++++++++++++++++++++++++ cmd/atelet/statspoller_test.go | 359 ++++++++++++++++++++ internal/ateattr/ateattr.go | 12 + internal/ateompath/ateompath.go | 13 +- manifests/ate-install/atelet.yaml | 4 + 6 files changed, 933 insertions(+), 5 deletions(-) create mode 100644 cmd/atelet/statspoller.go create mode 100644 cmd/atelet/statspoller_test.go diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 46de6e0d4e..2a82f61b74 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -101,6 +101,8 @@ var ( otlpRelaySocket = pflag.String("otlp-relay-socket", ateompath.AteletOTLPSocketPath(), "Unix socket to serve the OTLP relay on, which forwards the node's ateom 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)) + 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.") ) @@ -254,6 +256,20 @@ func main() { serverboot.Fatal(ctx, "Failed to create Kubernetes clients", err) } + if interval := clampActorStatsPollInterval(ctx, *actorStatsPollInterval); interval > 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 + // (conflicting registration), which the poller's own tests catch + // in CI -- and the poller has an official disabled state, so a + // broken one degrades to that state, loudly, instead of + // 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) + } + } + // TODO: Revisit scalability implications of using a shared informer. This lister // is unlikely to be used with frequency. ateFactory := externalversions.NewSharedInformerFactory(ateClient, 0) diff --git a/cmd/atelet/statspoller.go b/cmd/atelet/statspoller.go new file mode 100644 index 0000000000..dcc36f037e --- /dev/null +++ b/cmd/atelet/statspoller.go @@ -0,0 +1,534 @@ +// 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" + "io" + "log/slog" + "os" + "sync" + "sync/atomic" + "time" + + "golang.org/x/sync/errgroup" + + "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes" + + "github.com/agent-substrate/substrate/internal/ateattr" + "github.com/agent-substrate/substrate/internal/ateompath" + "github.com/agent-substrate/substrate/internal/proto/ateompb" +) + +// workerPoolLabel is the label the pool controller stamps on every worker pod +// it creates; the value is the WorkerPool's name and the pod's namespace is +// the pool's. Must agree with cmd/atecontroller/internal/controllers/ +// workerpool_apply.go. +const workerPoolLabel = "ate.dev/worker-pool" + +// minActorStatsPollInterval is the floor a configured poll interval is clamped +// to. It is the worst-case duration of one ateom's sweep on the micro-VM +// runtime -- maxActorContainers containers at statsCallTimeout each, constants +// that live with that runtime -- so a shorter interval could start a new poll +// into a guest agent still serving the previous one. +const minActorStatsPollInterval = 50 * time.Second + +// statsRPCTimeout bounds one ateom's GetActiveWorkloadStats call. It has to +// cover the ateom's own worst-case sweep (see minActorStatsPollInterval); +// anything still unanswered past that is a stuck socket, not a slow guest. +const statsRPCTimeout = 55 * time.Second + +// statsSweepConcurrency bounds how many ateoms one sweep probes at once. The +// interval floor protects a single guest from overlapping polls; probing +// DISTINCT ateoms concurrently puts one probe on each guest, so the only +// stacking the cap prevents is on atelet itself -- without it, a node of +// stuck-but-accepting sockets would hold one hung call per ateom for the full +// statsRPCTimeout. With it, such a node degrades the sweep to +// ceil(n/statsSweepConcurrency) timeouts instead of n. +const statsSweepConcurrency = 8 + +// workerPoolListTimeout bounds the per-sweep pod list that resolves worker +// pools. Pool labels are enrichment: better one unlabeled tick than a sweep +// blocked on the apiserver. +const workerPoolListTimeout = 10 * time.Second + +// clampActorStatsPollInterval enforces the floor on a nonzero configured +// interval, warning rather than obeying: an interval below the worst-case +// sweep would pile overlapping polls onto the same guest agent. +func clampActorStatsPollInterval(ctx context.Context, configured time.Duration) time.Duration { + if configured > 0 && configured < minActorStatsPollInterval { + slog.WarnContext(ctx, "actor-stats-poll-interval below the worst-case sweep; clamping", + slog.Duration("configured", configured), slog.Duration("clamped_to", minActorStatsPollInterval)) + return minActorStatsPollInterval + } + return configured +} + +// activeStatsClient is the one RPC the poller makes, as a narrow interface so +// tests can fake an ateom without a socket. ateompb.AteomClient satisfies it. +type activeStatsClient interface { + GetActiveWorkloadStats(ctx context.Context, req *ateompb.GetActiveWorkloadStatsRequest, opts ...grpc.CallOption) (*ateompb.GetActiveWorkloadStatsResponse, error) +} + +// statsPoller discovers the node's ateoms from the filesystem and turns their +// workload samples into template-level metrics. +// +// It holds no worker-to-actor mapping and never asks the control plane: every +// ateom registers itself on disk by creating its socket directory at boot (the +// same sockets the lifecycle RPCs dial), so one readdir plus one probe per +// socket is complete discovery, and an atelet restart loses nothing because +// nothing was held. Attribution comes solely from the identity echoed inside +// each sample, per the RPC's contract. +type statsPoller struct { + // interval between the end of one sweep and the start of the next. Sweeps + // never overlap: a slow sweep delays the next tick rather than stacking a + // second poll onto the same guests. + interval time.Duration + + // ateomsDir is the directory whose entries are worker pod UIDs + // (ateompath.AteomsDir on a real node; a fixture in tests). + ateomsDir string + + // dial returns a stats client for one ateom plus the closer that releases + // its connection; the probe closes it before returning, so a connection + // lives exactly one probe. Deliberately NOT the lifecycle RPCs' cached + // AteomDialer: at one probe per ateom per minute over a local unix socket + // a cache saves nothing, and sweeping the node's stale sockets through a + // shared cache would let telemetry evict connections the lifecycle RPCs + // are using. + dial func(ctx context.Context, podUID string) (activeStatsClient, io.Closer, error) + + // workerPools resolves this node's worker pod UIDs to the pool that owns + // them, called once per sweep. Nil (or a nil map, or a missing entry) + // degrades to samples grouped without pool labels rather than dropped: the + // pool is enrichment, the sample is the point. The real resolver lists the + // node's pods by the ate.dev/worker-pool label the pool controller stamps + // on every worker (see workerpool_apply.go); the ateom directory name IS + // the worker pod UID, which is the join key. + workerPools func(ctx context.Context) map[string]workerPoolRef + + inst *statsInstruments + + // lastCPU is the previous sweep's cpu_usage_usec per actor uid, the + // baseline the next sweep's deltas are computed against. Only the sweep + // loop touches it (under collect's mutex), and entries for actors a sweep + // did not see are dropped at its end -- an actor that leaves the node + // stops occupying memory here, and one that comes BACK later simply + // re-baselines. Empty after an atelet restart, so the first sweep + // contributes zero deltas: an undercount, never an overcount. + lastCPU map[string]uint64 +} + +// templateAggregate is one tick's sums for one templateKey group: the bounded +// label set #174 permits on a TSDB series. Actor and +// atespace identity deliberately never reach a metric label; per-actor detail +// is the events channel's job, not this one's. +// +// The memory fields are point-in-time sums the gauges observe. cpuDeltaUsec is +// different: cpu_usage_usec is a cumulative per-epoch counter per actor, so +// summing the raw values across a churning actor set would be meaningless to +// rate() -- instead the poller tracks each actor's last seen value and this +// carries the sweep's INCREASE, which tick adds onto a monotonic counter. +// Counter semantics survive actors joining, leaving, and resetting epochs by +// construction. +type templateAggregate struct { + sampledActors int64 + memoryCurrentBytes int64 + memoryWorkingSetBytes int64 + cpuDeltaUsec int64 +} + +// workerPoolRef names one WorkerPool: the pod's namespace and the +// ate.dev/worker-pool label value. +type workerPoolRef struct { + namespace string + name string +} + +// templateKey groups samples for aggregation. +type templateKey struct { + templateNamespace string + templateName string + sandboxClass string + source string + // workerPool is zero-valued when the pod could not be resolved to a pool + // (resolver disabled, list failure, pod already gone): those samples group + // together without pool labels rather than vanish. + workerPool workerPoolRef +} + +// attrs is the bounded label set for one aggregation group. The pool keys are +// omitted while unresolved rather than emitted as empty-string series, +// following the snapshotOp precedent. +func (k templateKey) attrs() metric.MeasurementOption { + attrs := make([]attribute.KeyValue, 0, 6) + attrs = append(attrs, + ateattr.TemplateNamespaceKey.String(k.templateNamespace), + ateattr.TemplateNameKey.String(k.templateName), + ateattr.SandboxClassKey.String(k.sandboxClass), + ateattr.StatsSourceKey.String(k.source), + ) + if k.workerPool != (workerPoolRef{}) { + attrs = append(attrs, + ateattr.WorkerPoolNamespaceKey.String(k.workerPool.namespace), + ateattr.WorkerPoolNameKey.String(k.workerPool.name), + ) + } + return metric.WithAttributes(attrs...) +} + +// run polls until ctx is cancelled. The caller has already validated and +// clamped interval. +func (p *statsPoller) run(ctx context.Context) { + slog.InfoContext(ctx, "Actor stats poller starting", slog.Duration("interval", p.interval)) + for { + p.tick(ctx) + select { + case <-ctx.Done(): + return + case <-time.After(p.interval): + } + } +} + +// tick sweeps every ateom on the node once, adds the sweep's CPU increases +// onto the counters, and publishes the aggregates for the next metric +// collection to observe. +func (p *statsPoller) tick(ctx context.Context) { + aggs := p.collect(ctx) + p.inst.addCPU(ctx, aggs) + p.inst.publish(aggs) +} + +// collect probes every ateom directory, statsSweepConcurrency at a time, and +// aggregates the samples it gets. +// +// One tolerance rule covers all the noise a scan meets: any failure to dial or +// call an entry means "not a target this tick", never an error worth more than +// a debug line. That uniformly handles stale directories left by deleted +// worker pods (nothing garbage-collects them eagerly), ateoms that have made +// their directory but not yet listened, and workers torn down mid-sweep. The +// no-sample reasons are equally routine: NO_WORKLOAD is an idle worker, +// NOT_MEASURABLE_YET is a boot or restore in progress -- both are skips by the +// RPC's own contract. +func (p *statsPoller) collect(ctx context.Context) map[templateKey]*templateAggregate { + entries, err := os.ReadDir(p.ateomsDir) + if err != nil { + // A node with no ateoms directory yet has no workers to measure; the + // first RunWorkload dispatch creates it. + slog.DebugContext(ctx, "Actor stats sweep: no ateoms directory", slog.Any("err", err)) + return nil + } + + var pools map[string]workerPoolRef + if p.workerPools != nil { + pools = p.workerPools(ctx) + } + + var ( + mu sync.Mutex + aggs = make(map[templateKey]*templateAggregate) + seenCPU = make(map[string]uint64) + g errgroup.Group + ) + g.SetLimit(statsSweepConcurrency) + for _, e := range entries { + if !e.IsDir() { + continue + } + podUID := e.Name() + + g.Go(func() error { + // One deadline over the whole probe, dial included. The real dial + // is lazy (grpc.NewClient touches no socket), but the seam does not + // promise that: a blocking dial implementation must not be able to + // park a sweep slot past the probe budget. + callCtx, cancel := context.WithTimeout(ctx, statsRPCTimeout) + defer cancel() + + client, closer, err := p.dial(callCtx, podUID) + if err != nil { + slog.DebugContext(ctx, "Actor stats sweep: skipping ateom", slog.String("pod_uid", podUID), slog.Any("err", err)) + return nil + } + defer closer.Close() + + resp, err := client.GetActiveWorkloadStats(callCtx, &ateompb.GetActiveWorkloadStatsRequest{}) + if err != nil { + slog.DebugContext(ctx, "Actor stats sweep: skipping ateom", slog.String("pod_uid", podUID), slog.Any("err", err)) + return nil + } + + sample := resp.GetSample() + if sample == nil { + // NO_WORKLOAD or NOT_MEASURABLE_YET: normal answers, nothing to + // add. + return nil + } + + key := templateKey{ + templateNamespace: sample.GetActorTemplateNamespace(), + templateName: sample.GetActorTemplateName(), + sandboxClass: sandboxClassLabel(sample.GetSandboxClass()), + source: statsSourceLabel(sample.GetSource()), + workerPool: pools[podUID], + } + mu.Lock() + defer mu.Unlock() + agg := aggs[key] + if agg == nil { + agg = &templateAggregate{} + aggs[key] = agg + } + agg.sampledActors++ + agg.memoryCurrentBytes += int64(sample.GetMemoryCurrentBytes()) + agg.memoryWorkingSetBytes += int64(sample.GetMemoryWorkingSetBytes()) + + // The counter increase this sample represents. A decrease means the + // epoch reset underneath us (the cgroup source restarts at zero on + // restore), so the new value IS the usage since the reset. A sample + // with NO baseline charges nothing and only records one: atelet + // cannot tell a new actor from its own restart, and charging the + // whole epoch-so-far would re-count hours of usage the previous + // atelet already counted, as one artificial spike. The bounded + // price is that every actor's boot-to-first-poll usage goes + // uncounted -- the events channel carries per-actor precision. + cpu := sample.GetCpuUsageUsec() + seenCPU[sample.GetActorUid()] = cpu + if last, ok := p.lastCPU[sample.GetActorUid()]; ok { + if last <= cpu { + agg.cpuDeltaUsec += int64(cpu - last) + } else { + agg.cpuDeltaUsec += int64(cpu) + } + } + return nil + }) + } + // The tasks only ever return nil: a probe that fails is "not a target this + // tick", never a failed sweep. + _ = g.Wait() + // Replacing (not merging) the baselines drops actors this sweep did not + // see, so lastCPU cannot grow with actor churn. + p.lastCPU = seenCPU + return aggs +} + +// sandboxClassLabel maps the wire enum to the ate.sandbox.class label values +// the rest of the system uses. +func sandboxClassLabel(c ateompb.SandboxClass) string { + switch c { + case ateompb.SandboxClass_SANDBOX_CLASS_GVISOR: + return "gvisor" + case ateompb.SandboxClass_SANDBOX_CLASS_MICROVM: + return "microvm" + default: + return ateattr.SandboxClassUnknown + } +} + +// statsSourceLabel maps the wire enum to the ate.stats.source label values. +func statsSourceLabel(s ateompb.StatsSource) string { + switch s { + case ateompb.StatsSource_STATS_SOURCE_CGROUP: + return ateattr.StatsSourceCgroup + case ateompb.StatsSource_STATS_SOURCE_GUEST_AGENT: + return ateattr.StatsSourceGuestAgent + default: + return ateattr.StatsSourceUnspecified + } +} + +// nodeWorkerPools returns a resolver that lists nodeName's worker pods once +// per sweep and maps pod UID to the pool that owns it. One field-selected, +// label-selected LIST per interval per node is deliberately chosen over a +// standing informer: at the poll cadence the apiserver cost is negligible, +// there is no cache to sync before the first sweep, and a failed list +// degrades to unlabeled samples for one tick instead of blocking anything. +func nodeWorkerPools(client kubernetes.Interface, nodeName string) func(ctx context.Context) map[string]workerPoolRef { + return func(ctx context.Context) map[string]workerPoolRef { + // Bounded so a hung apiserver connection cannot stall the sweep it is + // merely enriching: past the deadline, this tick's samples group + // without pool labels, which is the same answer as any other failed + // list. + listCtx, cancel := context.WithTimeout(ctx, workerPoolListTimeout) + defer cancel() + pods, err := client.CoreV1().Pods(metav1.NamespaceAll).List(listCtx, metav1.ListOptions{ + FieldSelector: "spec.nodeName=" + nodeName, + LabelSelector: workerPoolLabel, + }) + if err != nil { + slog.DebugContext(ctx, "Actor stats sweep: worker pool resolution failed; samples group without pool labels", slog.Any("err", err)) + return nil + } + pools := make(map[string]workerPoolRef, len(pods.Items)) + for _, pod := range pods.Items { + pools[string(pod.UID)] = workerPoolRef{ + namespace: pod.Namespace, + name: pod.Labels[workerPoolLabel], + } + } + return pools + } +} + +const ( + sampledActorsMetric = "ate.actor.stats.sampled_actors" + memoryCurrentMetric = "ate.actor.stats.memory_current_bytes" + workingSetMetric = "ate.actor.stats.memory_working_set_bytes" + cpuUsageMetric = "ate.actor.stats.cpu_usage" +) + +// statsInstruments exposes the latest sweep's aggregates as observable +// gauges: each metric collection observes exactly the groups the last sweep +// found, so a group that vanishes (last actor of a template leaves the node) +// genuinely disappears from the export. Synchronous gauges would not do that +// -- the SDK re-exports a sync instrument's last recorded value on every +// collection until process exit, which would keep reporting memory for actors +// long gone. +type statsInstruments struct { + // latest is the snapshot the callback reads: written whole by publish, + // never mutated in place. + latest atomic.Pointer[map[templateKey]*templateAggregate] + + // cpuUsage is a plain synchronous counter, unlike the gauges: the sweep's + // per-actor increases are ADDED here, and cumulative-counter semantics -- + // including a vanished template's series holding its final value rather + // than disappearing -- are exactly what rate() consumers expect. + cpuUsage metric.Float64Counter +} + +func newStatsInstruments(meter metric.Meter) (*statsInstruments, error) { + i := &statsInstruments{} + + cpuUsage, err := meter.Float64Counter( + cpuUsageMetric, + metric.WithUnit("s"), + metric.WithDescription("Cumulative CPU time consumed by running actors, in seconds."), + ) + if err != nil { + return nil, fmt.Errorf("create %s counter: %w", cpuUsageMetric, err) + } + i.cpuUsage = cpuUsage + + sampledActors, err := meter.Int64ObservableGauge( + sampledActorsMetric, + metric.WithUnit("{actor}"), + metric.WithDescription("Number of running actors with a current resource usage measurement."), + ) + if err != nil { + return nil, fmt.Errorf("create %s gauge: %w", sampledActorsMetric, err) + } + memoryCurrent, err := meter.Int64ObservableGauge( + memoryCurrentMetric, + metric.WithUnit("By"), + metric.WithDescription("Aggregated current memory usage of running actors, in bytes, including reclaimable page cache."), + ) + if err != nil { + return nil, fmt.Errorf("create %s gauge: %w", memoryCurrentMetric, err) + } + workingSet, err := meter.Int64ObservableGauge( + workingSetMetric, + metric.WithUnit("By"), + metric.WithDescription("Aggregated current memory working set of running actors, in bytes."), + ) + if err != nil { + return nil, fmt.Errorf("create %s gauge: %w", workingSetMetric, err) + } + + _, err = meter.RegisterCallback(func(_ context.Context, o metric.Observer) error { + snapshot := i.latest.Load() + if snapshot == nil { + return nil + } + for key, agg := range *snapshot { + opt := key.attrs() + o.ObserveInt64(sampledActors, agg.sampledActors, opt) + o.ObserveInt64(memoryCurrent, agg.memoryCurrentBytes, opt) + o.ObserveInt64(workingSet, agg.memoryWorkingSetBytes, opt) + } + return nil + }, sampledActors, memoryCurrent, workingSet) + if err != nil { + return nil, fmt.Errorf("register actor stats callback: %w", err) + } + + return i, nil +} + +// publish makes aggs the snapshot the next collection observes. A nil +// receiver is a valid no-op, like Instruments. +func (i *statsInstruments) publish(aggs map[templateKey]*templateAggregate) { + if i == nil { + return + } + i.latest.Store(&aggs) +} + +// addCPU adds one sweep's CPU increases onto the counters. A nil receiver is +// a valid no-op, like Instruments. +func (i *statsInstruments) addCPU(ctx context.Context, aggs map[templateKey]*templateAggregate) { + if i == nil { + return + } + for key, agg := range aggs { + // The wire carries microseconds; the metric is seconds, the base unit + // CPU time is exported in everywhere else (cAdvisor's + // container_cpu_usage_seconds_total, OTel's *.cpu.time), so the + // existing rate() idioms read directly as cores. + i.cpuUsage.Add(ctx, float64(agg.cpuDeltaUsec)/1e6, key.attrs()) + } +} + +// startStatsPoller assembles the poller and starts it. Split from main's boot +// sequence so the sampling subsystem has one obvious entry point. +// +// The poller dials its own per-probe connections (see statsPoller.dial) and +// takes no AteomDialer: 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) { + poller := &statsPoller{ + interval: interval, + ateomsDir: ateompath.AteomsDir(), + dial: func(_ context.Context, podUID string) (activeStatsClient, io.Closer, error) { + conn, err := grpc.NewClient( + "unix://"+ateompath.AteomSocketPath(podUID), + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithStatsHandler(otelgrpc.NewClientHandler()), + ) + if err != nil { + return nil, nil, err + } + return ateompb.NewAteomClient(conn), conn, nil + }, + inst: inst, + } + // NODE_NAME comes from the Downward API; without it the samples still + // flow, just grouped without pool labels. + if nodeName := os.Getenv("NODE_NAME"); nodeName != "" { + poller.workerPools = nodeWorkerPools(k8sClient, nodeName) + } else { + slog.WarnContext(ctx, "NODE_NAME not set; actor stats will carry no worker pool labels") + } + go poller.run(ctx) +} diff --git a/cmd/atelet/statspoller_test.go b/cmd/atelet/statspoller_test.go new file mode 100644 index 0000000000..f1d8d45d36 --- /dev/null +++ b/cmd/atelet/statspoller_test.go @@ -0,0 +1,359 @@ +// 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" + "io" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/google/go-cmp/cmp" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" + "google.golang.org/grpc" + + "github.com/agent-substrate/substrate/internal/proto/ateompb" +) + +// fakeStatsAteom answers GetActiveWorkloadStats with a canned response or +// error, standing in for one ateom socket. +type fakeStatsAteom struct { + resp *ateompb.GetActiveWorkloadStatsResponse + err error + + // mu guards the recordings below: the sweep probes ateoms concurrently. + mu sync.Mutex + // calls counts probes, so tests can tell "skipped" from "never found". + calls int + // sawDeadline records whether the probe's context carried one, pinning the + // per-call timeout. + sawDeadline bool +} + +func (f *fakeStatsAteom) GetActiveWorkloadStats(ctx context.Context, req *ateompb.GetActiveWorkloadStatsRequest, opts ...grpc.CallOption) (*ateompb.GetActiveWorkloadStatsResponse, error) { + f.mu.Lock() + f.calls++ + _, f.sawDeadline = ctx.Deadline() + f.mu.Unlock() + return f.resp, f.err +} + +// executingResponse builds the sample an executing ateom would echo. +func executingResponse(templateNS, templateName string, class ateompb.SandboxClass, source ateompb.StatsSource, current, workingSet uint64) *ateompb.GetActiveWorkloadStatsResponse { + return &ateompb.GetActiveWorkloadStatsResponse{ + Result: &ateompb.GetActiveWorkloadStatsResponse_Sample{Sample: &ateompb.WorkloadStatsSample{ + ActorTemplateNamespace: templateNS, + ActorTemplateName: templateName, + SandboxClass: class, + Source: source, + MemoryCurrentBytes: current, + MemoryWorkingSetBytes: workingSet, + }}, + } +} + +func noSampleResponse(reason ateompb.NoSampleReason) *ateompb.GetActiveWorkloadStatsResponse { + return &ateompb.GetActiveWorkloadStatsResponse{ + Result: &ateompb.GetActiveWorkloadStatsResponse_NoSampleReason{NoSampleReason: reason}, + } +} + +// closeRecorder counts Close calls, standing in for a probe's connection. +type closeRecorder struct { + mu sync.Mutex + closes int +} + +func (c *closeRecorder) Close() error { + c.mu.Lock() + c.closes++ + c.mu.Unlock() + return nil +} + +// newPollerFixture builds a poller over a fixture ateoms directory with one +// subdirectory (and one fake) per entry in fakes. Dialing a UID without a fake +// fails, which is the shape of a stale directory whose socket is gone. Every +// successful dial hands out a recorded closer; assertClosed checks the +// connections-live-exactly-one-probe contract. +func newPollerFixture(t *testing.T, fakes map[string]*fakeStatsAteom) (*statsPoller, map[string]*closeRecorder) { + t.Helper() + dir := t.TempDir() + closers := make(map[string]*closeRecorder) + for uid := range fakes { + if err := os.Mkdir(filepath.Join(dir, uid), 0o700); err != nil { + t.Fatalf("creating fixture ateom dir %q: %v", uid, err) + } + closers[uid] = &closeRecorder{} + } + return &statsPoller{ + ateomsDir: dir, + dial: func(_ context.Context, podUID string) (activeStatsClient, io.Closer, error) { + f, ok := fakes[podUID] + if !ok || f == nil { + return nil, nil, errors.New("no such socket") + } + return f, closers[podUID], nil + }, + }, closers +} + +// assertClosed checks that every successfully dialed probe closed its +// connection exactly once per sweep -- the RPC failing must not leak it. +func assertClosed(t *testing.T, fakes map[string]*fakeStatsAteom, closers map[string]*closeRecorder, sweeps int) { + t.Helper() + for uid, f := range fakes { + if f == nil { + continue // dial fails: no connection to close + } + if got := closers[uid].closes; got != sweeps { + t.Errorf("ateom %s connection closed %d times over %d sweeps, want %d", uid, got, sweeps, sweeps) + } + } +} + +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 + // nothing. + fakes := map[string]*fakeStatsAteom{ + "uid-1": {resp: executingResponse("ns-a", "tmpl-a", ateompb.SandboxClass_SANDBOX_CLASS_GVISOR, ateompb.StatsSource_STATS_SOURCE_CGROUP, 1000, 700)}, + "uid-2": {resp: executingResponse("ns-a", "tmpl-a", ateompb.SandboxClass_SANDBOX_CLASS_GVISOR, ateompb.StatsSource_STATS_SOURCE_CGROUP, 500, 300)}, + "uid-3": {resp: executingResponse("ns-b", "tmpl-b", ateompb.SandboxClass_SANDBOX_CLASS_MICROVM, ateompb.StatsSource_STATS_SOURCE_GUEST_AGENT, 42, 40)}, + "uid-4": {resp: noSampleResponse(ateompb.NoSampleReason_NO_SAMPLE_REASON_NO_WORKLOAD)}, + "uid-5": {resp: noSampleResponse(ateompb.NoSampleReason_NO_SAMPLE_REASON_NOT_MEASURABLE_YET)}, + } + p, closers := newPollerFixture(t, fakes) + + got := p.collect(context.Background()) + + want := map[templateKey]*templateAggregate{ + {templateNamespace: "ns-a", templateName: "tmpl-a", sandboxClass: "gvisor", source: "cgroup"}: { + sampledActors: 2, memoryCurrentBytes: 1500, memoryWorkingSetBytes: 1000, + }, + {templateNamespace: "ns-b", templateName: "tmpl-b", sandboxClass: "microvm", source: "guest-agent"}: { + sampledActors: 1, memoryCurrentBytes: 42, memoryWorkingSetBytes: 40, + }, + } + if diff := cmp.Diff(want, got, cmp.AllowUnexported(templateAggregate{}, templateKey{}, workerPoolRef{})); diff != "" { + t.Errorf("collect() mismatch (-want +got):\n%s", diff) + } + + for uid, f := range fakes { + if f.calls != 1 { + t.Errorf("ateom %s probed %d times, want 1", uid, f.calls) + } + if !f.sawDeadline { + t.Errorf("ateom %s probed without a deadline; every probe must carry the per-call timeout", uid) + } + } + assertClosed(t, fakes, closers, 1) +} + +// TestStatsPollerCollectSkipsFailures pins the scan's one tolerance rule: a +// dial or call failure means "not a target this tick", never a failed sweep. +// The healthy ateom's sample must still be aggregated. +func TestStatsPollerCollectSkipsFailures(t *testing.T) { + fakes := map[string]*fakeStatsAteom{ + "uid-healthy": {resp: executingResponse("ns-a", "tmpl-a", ateompb.SandboxClass_SANDBOX_CLASS_GVISOR, ateompb.StatsSource_STATS_SOURCE_CGROUP, 100, 80)}, + "uid-stale": nil, // directory with no reachable socket: dial fails + "uid-broken": {err: errors.New("rpc error: connection refused")}, + } + p, closers := newPollerFixture(t, fakes) + + got := p.collect(context.Background()) + + want := map[templateKey]*templateAggregate{ + {templateNamespace: "ns-a", templateName: "tmpl-a", sandboxClass: "gvisor", source: "cgroup"}: { + sampledActors: 1, memoryCurrentBytes: 100, memoryWorkingSetBytes: 80, + }, + } + if diff := cmp.Diff(want, got, cmp.AllowUnexported(templateAggregate{}, templateKey{}, workerPoolRef{})); diff != "" { + t.Errorf("collect() mismatch (-want +got):\n%s", diff) + } + // The broken ateom's RPC failed, but its connection was dialed -- it must + // be closed all the same. + assertClosed(t, fakes, closers, 1) +} + +// TestStatsPollerCollectNoAteomsDir: a node whose first workload has not +// arrived has no ateoms directory, which is empty coverage, not an error. +func TestStatsPollerCollectNoAteomsDir(t *testing.T) { + p := &statsPoller{ateomsDir: filepath.Join(t.TempDir(), "does-not-exist")} + if got := p.collect(context.Background()); len(got) != 0 { + t.Errorf("collect() with no ateoms dir = %v, want empty", got) + } +} + +func TestClampActorStatsPollInterval(t *testing.T) { + tests := []struct { + name string + in time.Duration + want time.Duration + }{ + {name: "zero stays disabled", in: 0, want: 0}, + {name: "below floor clamps", in: time.Second, want: minActorStatsPollInterval}, + {name: "at floor passes", in: minActorStatsPollInterval, want: minActorStatsPollInterval}, + {name: "above floor passes", in: 5 * time.Minute, want: 5 * time.Minute}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := clampActorStatsPollInterval(context.Background(), tc.in); got != tc.want { + t.Errorf("clampActorStatsPollInterval(%v) = %v, want %v", tc.in, got, tc.want) + } + }) + } +} + +// TestStatsInstrumentsObserveLatestSnapshotOnly pins the reason the gauges are +// observable rather than synchronous: each collection reports exactly the +// groups the latest sweep found. A synchronous gauge would re-export its last +// recorded value on every collection until process exit, so a template whose +// actors left the node would keep reporting their memory forever. +func TestStatsInstrumentsObserveLatestSnapshotOnly(t *testing.T) { + reader := sdkmetric.NewManualReader() + mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + defer mp.Shutdown(context.Background()) + + inst, err := newStatsInstruments(mp.Meter("test")) + if err != nil { + t.Fatalf("newStatsInstruments() error = %v", err) + } + + key := templateKey{templateNamespace: "ns-a", templateName: "tmpl-a", sandboxClass: "gvisor", source: "cgroup"} + inst.publish(map[templateKey]*templateAggregate{ + key: {sampledActors: 2, memoryCurrentBytes: 1500, memoryWorkingSetBytes: 1000}, + }) + + if got := gaugePointCount(t, reader, workingSetMetric); got != 1 { + t.Fatalf("after publish: %s has %d datapoints, want 1", workingSetMetric, got) + } + + // The template's actors leave the node: an empty sweep must make the + // series disappear, not freeze at its last value. + inst.publish(map[templateKey]*templateAggregate{}) + if got := gaugePointCount(t, reader, workingSetMetric); got != 0 { + t.Errorf("after empty sweep: %s has %d datapoints, want 0", workingSetMetric, got) + } +} + +// gaugePointCount collects once and returns how many datapoints name has. +func gaugePointCount(t *testing.T, reader *sdkmetric.ManualReader, name string) int { + t.Helper() + var rm metricdata.ResourceMetrics + if err := reader.Collect(context.Background(), &rm); err != nil { + t.Fatalf("Collect() error = %v", err) + } + for _, sm := range rm.ScopeMetrics { + for _, m := range sm.Metrics { + if m.Name != name { + continue + } + g, ok := m.Data.(metricdata.Gauge[int64]) + if !ok { + t.Fatalf("metric %s has data type %T, want Gauge[int64]", name, m.Data) + } + return len(g.DataPoints) + } + } + return 0 +} + +// cpuResponse is executingResponse with only the CPU counter set, for the +// delta tests. +func cpuResponse(actorUID string, cpuUsec uint64) *ateompb.GetActiveWorkloadStatsResponse { + return &ateompb.GetActiveWorkloadStatsResponse{ + Result: &ateompb.GetActiveWorkloadStatsResponse_Sample{Sample: &ateompb.WorkloadStatsSample{ + ActorUid: actorUID, + ActorTemplateNamespace: "ns-a", + ActorTemplateName: "tmpl-a", + SandboxClass: ateompb.SandboxClass_SANDBOX_CLASS_GVISOR, + Source: ateompb.StatsSource_STATS_SOURCE_CGROUP, + CpuUsageUsec: cpuUsec, + }}, + } +} + +// TestStatsPollerCPUDeltas pins the increase computation across sweeps: the +// first sight of an actor establishes a baseline and charges nothing (atelet +// cannot tell a new actor from its own restart, and re-charging an epoch the +// previous atelet counted would spike the counter), a later sweep charges +// only the increase, a decrease is an epoch reset whose new value is the +// usage since the reset, and an actor that disappears stops contributing and +// is dropped from the baselines. +func TestStatsPollerCPUDeltas(t *testing.T) { + key := templateKey{templateNamespace: "ns-a", templateName: "tmpl-a", sandboxClass: "gvisor", source: "cgroup"} + fake := &fakeStatsAteom{resp: cpuResponse("uid-a", 1000)} + p, _ := newPollerFixture(t, map[string]*fakeStatsAteom{"uid-1": fake}) + + if got := p.collect(context.Background())[key].cpuDeltaUsec; got != 0 { + t.Errorf("first sweep delta = %d, want 0 (baseline only on first sight)", got) + } + + fake.resp = cpuResponse("uid-a", 1600) + if got := p.collect(context.Background())[key].cpuDeltaUsec; got != 600 { + t.Errorf("second sweep delta = %d, want 600 (the increase)", got) + } + + // Epoch reset: the counter went backwards, so the new value is the usage + // since the reset. + fake.resp = cpuResponse("uid-a", 250) + if got := p.collect(context.Background())[key].cpuDeltaUsec; got != 250 { + t.Errorf("post-reset sweep delta = %d, want 250", got) + } + + // The actor leaves: nothing to contribute, and its baseline must be + // dropped so a later return re-baselines instead of comparing against a + // dead value. + fake.resp = noSampleResponse(ateompb.NoSampleReason_NO_SAMPLE_REASON_NO_WORKLOAD) + if got := p.collect(context.Background()); len(got) != 0 { + t.Errorf("empty sweep aggregates = %v, want none", got) + } + if len(p.lastCPU) != 0 { + t.Errorf("baselines after empty sweep = %v, want pruned empty", p.lastCPU) + } +} + +// TestStatsPollerWorkerPoolLabels pins the pool enrichment: a resolved pod +// groups under its pool, an unresolved one groups without pool labels rather +// than vanishing, and the two never merge. +func TestStatsPollerWorkerPoolLabels(t *testing.T) { + fakes := map[string]*fakeStatsAteom{ + "uid-pooled": {resp: executingResponse("ns-a", "tmpl-a", ateompb.SandboxClass_SANDBOX_CLASS_GVISOR, ateompb.StatsSource_STATS_SOURCE_CGROUP, 100, 80)}, + "uid-unpooled": {resp: executingResponse("ns-a", "tmpl-a", ateompb.SandboxClass_SANDBOX_CLASS_GVISOR, ateompb.StatsSource_STATS_SOURCE_CGROUP, 10, 8)}, + } + p, _ := newPollerFixture(t, fakes) + p.workerPools = func(context.Context) map[string]workerPoolRef { + return map[string]workerPoolRef{"uid-pooled": {namespace: "pool-ns", name: "pool-a"}} + } + + got := p.collect(context.Background()) + + base := templateKey{templateNamespace: "ns-a", templateName: "tmpl-a", sandboxClass: "gvisor", source: "cgroup"} + pooled := base + pooled.workerPool = workerPoolRef{namespace: "pool-ns", name: "pool-a"} + want := map[templateKey]*templateAggregate{ + pooled: {sampledActors: 1, memoryCurrentBytes: 100, memoryWorkingSetBytes: 80}, + base: {sampledActors: 1, memoryCurrentBytes: 10, memoryWorkingSetBytes: 8}, + } + if diff := cmp.Diff(want, got, cmp.AllowUnexported(templateAggregate{}, templateKey{}, workerPoolRef{})); diff != "" { + t.Errorf("collect() mismatch (-want +got):\n%s", diff) + } +} diff --git a/internal/ateattr/ateattr.go b/internal/ateattr/ateattr.go index 7a938d0ca4..2ed261e318 100644 --- a/internal/ateattr/ateattr.go +++ b/internal/ateattr/ateattr.go @@ -80,6 +80,18 @@ const ( RouterResumeKey = attribute.Key("ate.router.resume") RouterOutcomeKey = attribute.Key("ate.router.outcome") FailureReasonKey = attribute.Key("ate.failure.reason") + StatsSourceKey = attribute.Key("ate.stats.source") +) + +// Values for StatsSourceKey, mirroring ateompb.StatsSource. The two sources do +// not measure the same thing (the cgroup source charges the sandbox runtime's +// overhead along with the workload, the guest-agent source sees only the +// workload's containers), so rollups must group by this key rather than sum +// across it. +const ( + StatsSourceUnspecified = "unspecified" + StatsSourceCgroup = "cgroup" + StatsSourceGuestAgent = "guest-agent" ) // Values for SchedulingConstraintKey. diff --git a/internal/ateompath/ateompath.go b/internal/ateompath/ateompath.go index bd715695c0..0f719fb160 100644 --- a/internal/ateompath/ateompath.go +++ b/internal/ateompath/ateompath.go @@ -80,12 +80,15 @@ func AteletOTLPSocketPath() string { ) } +// AteomsDir is the parent of every per-ateom directory. Each ateom creates +// AteomPath(podUID) under it when it boots, so listing this directory is how a +// scraper with no prior knowledge discovers the node's ateoms. +func AteomsDir() string { + return filepath.Join(BasePath, "ateoms") +} + func AteomPath(podUID string) string { - return filepath.Join( - BasePath, - "ateoms", - podUID, - ) + return filepath.Join(AteomsDir(), podUID) } func AteomSocketPath(podUID string) string { diff --git a/manifests/ate-install/atelet.yaml b/manifests/ate-install/atelet.yaml index 8f28579a35..a484498ad1 100644 --- a/manifests/ate-install/atelet.yaml +++ b/manifests/ate-install/atelet.yaml @@ -136,6 +136,10 @@ spec: drop: - ALL env: + - name: NODE_NAME + valueFrom: + fieldRef: + fieldPath: spec.nodeName - name: POD_NAME valueFrom: fieldRef: