From e2986ef4d2cab703e15f62cc0cae29b1a98ee5c5 Mon Sep 17 00:00:00 2001 From: Tim Bai Date: Tue, 4 Aug 2026 15:36:42 -0400 Subject: [PATCH] ateom-gvisor: measure the sandbox cgroup for GetWorkloadStats Fills in the measurement half of GetWorkloadStats for the gVisor runtime, so it returns real numbers instead of Unimplemented. The micro-VM runtime keeps its stub until the guest-agent read lands. The sample comes from the sandbox's cgroup v2 leaf at /sys/fs/cgroup/pause. gVisor runs every container of a sandbox inside one host process, the sentry, and runsc places that process in the cgroup of the container that created the sandbox -- "pause", the first container RunWorkload and RestoreWorkload create. The actor's own containers get leaves too, but their memory and CPU are the sentry's and are charged there instead, which is why a sample is attributed to the actor rather than to a container. The path follows the "/" + containerName convention runsc.ensureContainerCgroupsPath writes into the OCI spec, resolved against the pod's own cgroup scope that setupCgroupDelegation prepares. The read lives in a new cmd/ateom-gvisor/internal/cgroupstats. It takes the cgroup directory as an argument rather than reaching for an absolute path, so the parsing is testable from a fixture tree without root or a live sandbox, and it carries no build tag so those tests run everywhere. It fails only when memory.current is missing or unparseable -- wrapping fs.ErrNotExist in the first case, so the handler can tell "no cgroup to measure" from "the format is not what we parse". Every other file degrades to zero on its own field, because each has a legitimate reason to be absent on a healthy node: memory.peak predates kernel 5.19, and setupCgroupDelegation enables controllers best-effort, so a cgroup with memory but no cpu is reachable. Reporting no memory numbers because the node could not report CPU would be the wrong trade. Working set is memory.current less memory.stat's inactive_file, saturating at zero rather than wrapping: the two files are read a moment apart and are not a consistent snapshot, so inactive_file can legitimately exceed the memory.current read just before it. The proto's measurement block gains the epoch caveat it was missing. It already said cpu_usage_usec is per-epoch; memory_peak_bytes accumulates the same way and said nothing, so a caller reading it as a lifetime peak would silently under-report -- this source recreates the sandbox cgroup on every restore, and memory.peak restarts with it. The note is phrased per source rather than as one rule, because the two sources do not agree: the guest agent reads counters the guest kernel keeps in its own RAM, and a restored guest brings those back, so an epoch there does not end where it ends here. It also drops the suggestion that watching for a decrease is enough to spot a boundary. An epoch can begin above the value last reported, and then no decrease ever appears. The handler is the first user of the NOT_FOUND / FAILED_PRECONDITION split the previous commit documented, and follows it: available and a UID mismatch are both NOT_FOUND, since each tells the caller the actor is not here and its worker-to-actor mapping wants re-resolving. A missing cgroup under a matching UID is the transient FAILED_PRECONDITION -- usually a poll landing in the boot, since attribution is now retained from the moment the ateom accepts the actor, before runsc has created the leaf. A malformed cgroup is Internal: not a routine race, so it must not be reported as one. The handler takes no lock, which is what the atomic on activeActor bought and what TestGetWorkloadStatsDoesNotTakeLock pins: it holds s.lock across the call, so a handler that reached for it deadlocks. After the read it reloads activeActor and compares pointer identity, so a checkpoint plus a fresh run completing underneath the read is NOT_FOUND rather than misattributed -- the same answer a retry would get, rather than a second code for one state. Deliberately not here: a panic-recovery interceptor. grpc-go serves each RPC on its own goroutine and does not recover handler panics, and the Go runtime kills the process on an unrecovered panic in any goroutine, so a nil dereference in any handler today ends every other RPC the ateom is serving, including an in-flight checkpoint. That gap predates this change and is not made reachable by it: nothing calls GetWorkloadStats yet, and neither the handler (which nil-checks before every dereference) nor the parser (which length-checks before indexing) has a panic path. It wants its own PR, covering every server rather than the two ateoms a telemetry change happens to touch, and it needs to land before the phase that adds a caller polling on a timer. Not covered here: that the sentry lands in /sys/fs/cgroup/pause on a live node is a runsc placement behavior, derived from the spec convention above but not verified from a unit test. Process listings in #161 confirm runsc-sandbox and both gofers sit in the "pause" cgroup, but those runs predate #496, so they establish the leaf name rather than the absolute path. The leaf holds the sentry's own overhead and the gofers alongside the actor's work, which is what the proto means by measuring the SANDBOX; splitting the actor's share out would need the sentry's own accounting. Part of #594 --- .../internal/cgroupstats/cgroupstats.go | 172 ++++++++++++ .../internal/cgroupstats/cgroupstats_test.go | 255 ++++++++++++++++++ cmd/ateom-gvisor/main.go | 23 +- cmd/ateom-gvisor/stats.go | 164 +++++++++++ cmd/ateom-gvisor/stats_test.go | 200 ++++++++++++-- internal/proto/ateompb/ateom.pb.go | 26 +- internal/proto/ateompb/ateom.proto | 22 +- 7 files changed, 818 insertions(+), 44 deletions(-) create mode 100644 cmd/ateom-gvisor/internal/cgroupstats/cgroupstats.go create mode 100644 cmd/ateom-gvisor/internal/cgroupstats/cgroupstats_test.go create mode 100644 cmd/ateom-gvisor/stats.go diff --git a/cmd/ateom-gvisor/internal/cgroupstats/cgroupstats.go b/cmd/ateom-gvisor/internal/cgroupstats/cgroupstats.go new file mode 100644 index 000000000..4d67c0829 --- /dev/null +++ b/cmd/ateom-gvisor/internal/cgroupstats/cgroupstats.go @@ -0,0 +1,172 @@ +// 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 cgroupstats reads resource usage out of a cgroup v2 directory. +// +// The gVisor ateom uses it to answer ateompb.Ateom/GetWorkloadStats: the sentry +// hosts the whole sandbox in one host process, so the sandbox's cgroup leaf is +// where the workload's memory and CPU actually show up. +// +// Every read is scoped to a caller-supplied directory rather than a hardcoded +// /sys/fs/cgroup path, which keeps the parsing testable from a fixture tree +// without root or a live sandbox. The package deliberately does not know what a +// sandbox is; it reads four numbers out of five files. +package cgroupstats + +import ( + "bufio" + "bytes" + "errors" + "fmt" + "io/fs" + "os" + "path/filepath" + "strconv" + "strings" +) + +// Sample is a point-in-time reading of one cgroup v2 directory. +// +// Fields the kernel does not expose read as zero rather than failing the whole +// sample: a partial reading is more useful than none, and the alternative is an +// ateom that reports nothing at all on a kernel missing one file. Read's doc +// comment says which fields can do this and why. +type Sample struct { + // MemoryCurrentBytes is memory.current: bytes currently charged to the + // cgroup, including page cache. + MemoryCurrentBytes uint64 + + // MemoryPeakBytes is memory.peak: the high-water mark of MemoryCurrentBytes + // over the cgroup's lifetime. Zero on kernels below 5.19, which do not have + // the file. + MemoryPeakBytes uint64 + + // MemoryWorkingSetBytes is MemoryCurrentBytes less the reclaimable page + // cache (memory.stat's inactive_file), floored at zero. This is the estimate + // of "memory that would have to be paged in again if reclaimed" that cAdvisor + // and the kubelet report, and it is the field to compare against a memory + // limit; MemoryCurrentBytes drifts upward with cache that the kernel will + // drop for free under pressure. + MemoryWorkingSetBytes uint64 + + // CPUUsageUsec is cpu.stat's usage_usec: cumulative CPU time consumed by the + // cgroup since it was created. Zero if the cpu controller was not delegated + // to this cgroup (see setupCgroupDelegation, which enables controllers + // best-effort and carries on when one cannot be enabled). + CPUUsageUsec uint64 +} + +// Read returns a Sample for the cgroup v2 directory at dir. +// +// It fails only when the cgroup itself cannot be read: a missing directory, or +// a memory.current that is absent or unparseable. That case is reported with an +// error wrapping fs.ErrNotExist when the cause is a missing path, so callers can +// distinguish "this sandbox is gone" from "this file is malformed". +// +// Everything else degrades to zero on that one field, because each has a +// legitimate reason to be missing on a healthy system: memory.peak does not +// exist before kernel 5.19, memory.stat's inactive_file is absent without the +// memory controller's full accounting, and cpu.stat is absent when the cpu +// controller was not delegated. Failing the sample for any of them would mean +// reporting no memory numbers because the node could not report CPU. +func Read(dir string) (Sample, error) { + current, ok, err := readUint(filepath.Join(dir, "memory.current")) + if err != nil { + return Sample{}, err + } + if !ok { + return Sample{}, fmt.Errorf("reading %q: %w", filepath.Join(dir, "memory.current"), fs.ErrNotExist) + } + + // Best-effort from here down: a read error on an optional file is treated the + // same as the file being absent, since both mean "the kernel did not give us + // this number" and neither invalidates the numbers we did get. + peak, _, _ := readUint(filepath.Join(dir, "memory.peak")) + inactiveFile, haveInactiveFile, _ := readKeyedUint(filepath.Join(dir, "memory.stat"), "inactive_file") + cpuUsage, _, _ := readKeyedUint(filepath.Join(dir, "cpu.stat"), "usage_usec") + + // Without inactive_file there is nothing to subtract, so the working set + // collapses to memory.current. That over-reports by however much reclaimable + // cache the cgroup holds, which is the safe direction: it never claims the + // workload is using less than it is. + workingSet := current + if haveInactiveFile { + // Saturating rather than wrapping. The two files are read a moment apart + // and are not a consistent snapshot, so inactive_file can legitimately + // exceed the memory.current read just before it; on uint64 that would + // wrap to an absurd number instead of the near-zero the reading means. + workingSet = 0 + if inactiveFile < current { + workingSet = current - inactiveFile + } + } + + return Sample{ + MemoryCurrentBytes: current, + MemoryPeakBytes: peak, + MemoryWorkingSetBytes: workingSet, + CPUUsageUsec: cpuUsage, + }, nil +} + +// readUint reads a cgroup file holding a single unsigned integer. The bool +// reports whether the file was there; a present but unparseable file is an +// error, since that means the kernel's format is not what we think it is rather +// than the file being unsupported. +func readUint(path string) (uint64, bool, error) { + b, err := os.ReadFile(path) + if errors.Is(err, fs.ErrNotExist) { + return 0, false, nil + } + if err != nil { + return 0, false, fmt.Errorf("reading %q: %w", path, err) + } + v, err := strconv.ParseUint(strings.TrimSpace(string(b)), 10, 64) + if err != nil { + return 0, false, fmt.Errorf("parsing %q: %w", path, err) + } + return v, true, nil +} + +// readKeyedUint reads one key out of a cgroup "flat keyed" file, whose lines are +// " " pairs. The bool reports whether the key was found. +func readKeyedUint(path, key string) (uint64, bool, error) { + b, err := os.ReadFile(path) + if errors.Is(err, fs.ErrNotExist) { + return 0, false, nil + } + if err != nil { + return 0, false, fmt.Errorf("reading %q: %w", path, err) + } + + sc := bufio.NewScanner(bytes.NewReader(b)) + for sc.Scan() { + // Fields, then a length check, rather than indexing straight into the + // split: a blank or single-token line is not worth a panic in an RPC + // handler, and these files are read on a timer for the life of a workload. + f := strings.Fields(sc.Text()) + if len(f) != 2 || f[0] != key { + continue + } + v, err := strconv.ParseUint(f[1], 10, 64) + if err != nil { + return 0, false, fmt.Errorf("parsing %q of %q: %w", key, path, err) + } + return v, true, nil + } + if err := sc.Err(); err != nil { + return 0, false, fmt.Errorf("scanning %q: %w", path, err) + } + return 0, false, nil +} diff --git a/cmd/ateom-gvisor/internal/cgroupstats/cgroupstats_test.go b/cmd/ateom-gvisor/internal/cgroupstats/cgroupstats_test.go new file mode 100644 index 000000000..8c04e0f96 --- /dev/null +++ b/cmd/ateom-gvisor/internal/cgroupstats/cgroupstats_test.go @@ -0,0 +1,255 @@ +// 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 cgroupstats + +import ( + "errors" + "io/fs" + "os" + "path/filepath" + "testing" + + "github.com/google/go-cmp/cmp" +) + +// fullMemoryStat is a trimmed but structurally faithful cgroup v2 memory.stat: +// many keys, inactive_file neither first nor last, and the surrounding keys that +// a naive "first number wins" parser would pick up instead. +const fullMemoryStat = `anon 104857600 +file 52428800 +kernel 8388608 +kernel_stack 262144 +slab 4194304 +sock 0 +shmem 0 +file_mapped 1048576 +file_dirty 0 +file_writeback 0 +inactive_anon 0 +active_anon 104857600 +inactive_file 20971520 +active_file 31457280 +unevictable 0 +` + +const fullCPUStat = `usage_usec 1234567 +user_usec 1000000 +system_usec 234567 +nr_periods 0 +nr_throttled 0 +throttled_usec 0 +` + +// writeCgroup builds a fixture cgroup directory. A nil value omits the file, +// which is how the "kernel does not have this" cases are expressed. +func writeCgroup(t *testing.T, files map[string]string) string { + t.Helper() + dir := t.TempDir() + for name, content := range files { + if err := os.WriteFile(filepath.Join(dir, name), []byte(content), 0o600); err != nil { + t.Fatalf("writing fixture %q: %v", name, err) + } + } + return dir +} + +func TestRead(t *testing.T) { + for _, tc := range []struct { + name string + files map[string]string + want Sample + }{ + { + name: "all files present", + files: map[string]string{ + "memory.current": "157286400\n", + "memory.peak": "209715200\n", + "memory.stat": fullMemoryStat, + "cpu.stat": fullCPUStat, + }, + want: Sample{ + MemoryCurrentBytes: 157286400, + MemoryPeakBytes: 209715200, + // 157286400 - 20971520 + MemoryWorkingSetBytes: 136314880, + CPUUsageUsec: 1234567, + }, + }, + { + // memory.peak arrived in kernel 5.19; older nodes simply have no file. + name: "no memory.peak (pre-5.19 kernel)", + files: map[string]string{ + "memory.current": "157286400\n", + "memory.stat": fullMemoryStat, + "cpu.stat": fullCPUStat, + }, + want: Sample{ + MemoryCurrentBytes: 157286400, + MemoryPeakBytes: 0, + MemoryWorkingSetBytes: 136314880, + CPUUsageUsec: 1234567, + }, + }, + { + // setupCgroupDelegation enables controllers one at a time and carries on + // when one cannot be enabled, so a cgroup with memory but no cpu is a + // state this actually reaches. + name: "no cpu.stat (cpu controller not delegated)", + files: map[string]string{ + "memory.current": "157286400\n", + "memory.peak": "209715200\n", + "memory.stat": fullMemoryStat, + }, + want: Sample{ + MemoryCurrentBytes: 157286400, + MemoryPeakBytes: 209715200, + MemoryWorkingSetBytes: 136314880, + CPUUsageUsec: 0, + }, + }, + { + name: "no memory.stat falls back to current for working set", + files: map[string]string{ + "memory.current": "157286400\n", + "memory.peak": "209715200\n", + "cpu.stat": fullCPUStat, + }, + want: Sample{ + MemoryCurrentBytes: 157286400, + MemoryPeakBytes: 209715200, + MemoryWorkingSetBytes: 157286400, + CPUUsageUsec: 1234567, + }, + }, + { + name: "memory.stat without inactive_file falls back to current", + files: map[string]string{ + "memory.current": "157286400\n", + "memory.stat": "anon 104857600\nfile 52428800\n", + }, + want: Sample{ + MemoryCurrentBytes: 157286400, + MemoryWorkingSetBytes: 157286400, + }, + }, + { + // The two files are read a moment apart, so this ordering is reachable + // on a live cgroup. On uint64 the naive subtraction wraps to ~1.8e19. + name: "inactive_file above memory.current floors the working set at zero", + files: map[string]string{ + "memory.current": "1000\n", + "memory.stat": "inactive_file 4000\n", + }, + want: Sample{ + MemoryCurrentBytes: 1000, + MemoryWorkingSetBytes: 0, + }, + }, + { + name: "zero usage", + files: map[string]string{ + "memory.current": "0\n", + "memory.peak": "0\n", + "memory.stat": "inactive_file 0\n", + "cpu.stat": "usage_usec 0\n", + }, + want: Sample{}, + }, + { + // A short, blank, or over-long line must not panic: this runs in an RPC + // handler on a timer, and grpc-go does not recover handler panics. + name: "malformed lines in keyed files are skipped", + files: map[string]string{ + "memory.current": "157286400\n", + "memory.stat": "\nanon\n\ninactive_file 20971520\nbogus 1 2 3\n \n", + "cpu.stat": "nr_periods\n\nusage_usec 1234567\n", + }, + want: Sample{ + MemoryCurrentBytes: 157286400, + MemoryWorkingSetBytes: 136314880, + CPUUsageUsec: 1234567, + }, + }, + { + // Optional files degrade to zero whether they are missing or garbage; + // neither should cost the caller the memory numbers. + name: "unparseable optional files degrade to zero", + files: map[string]string{ + "memory.current": "157286400\n", + "memory.peak": "not-a-number\n", + "cpu.stat": "usage_usec eleventy\n", + }, + want: Sample{ + MemoryCurrentBytes: 157286400, + MemoryPeakBytes: 0, + MemoryWorkingSetBytes: 157286400, + CPUUsageUsec: 0, + }, + }, + } { + t.Run(tc.name, func(t *testing.T) { + got, err := Read(writeCgroup(t, tc.files)) + if err != nil { + t.Fatalf("Read() error = %v, want nil", err) + } + if diff := cmp.Diff(tc.want, got); diff != "" { + t.Errorf("Read() mismatch (-want +got):\n%s", diff) + } + }) + } +} + +// TestReadMissingCgroup covers the case the RPC handler has to tell apart from a +// bad read: the sandbox's cgroup is not there, because the sandbox is not there. +func TestReadMissingCgroup(t *testing.T) { + for _, tc := range []struct { + name string + dir func(t *testing.T) string + }{ + { + name: "directory does not exist", + dir: func(t *testing.T) string { return filepath.Join(t.TempDir(), "no-such-cgroup") }, + }, + { + name: "directory exists but is empty", + dir: func(t *testing.T) string { return writeCgroup(t, nil) }, + }, + } { + t.Run(tc.name, func(t *testing.T) { + _, err := Read(tc.dir(t)) + if err == nil { + t.Fatal("Read() error = nil, want non-nil") + } + if !errors.Is(err, fs.ErrNotExist) { + t.Errorf("Read() error = %v, want one matching fs.ErrNotExist", err) + } + }) + } +} + +// TestReadMalformedMemoryCurrent separates "the cgroup is gone" from "the cgroup +// is there but its format is not what we parse". Only the former is routine, so +// only the former may match fs.ErrNotExist. +func TestReadMalformedMemoryCurrent(t *testing.T) { + dir := writeCgroup(t, map[string]string{"memory.current": "max\n"}) + + _, err := Read(dir) + if err == nil { + t.Fatal("Read() error = nil, want non-nil") + } + if errors.Is(err, fs.ErrNotExist) { + t.Errorf("Read() error = %v, want one that does not match fs.ErrNotExist", err) + } +} diff --git a/cmd/ateom-gvisor/main.go b/cmd/ateom-gvisor/main.go index e01bda304..d69f2717b 100644 --- a/cmd/ateom-gvisor/main.go +++ b/cmd/ateom-gvisor/main.go @@ -51,9 +51,7 @@ import ( "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc" "golang.org/x/sys/unix" "google.golang.org/grpc" - "google.golang.org/grpc/codes" "google.golang.org/grpc/reflection" - "google.golang.org/grpc/status" ) var ( @@ -279,10 +277,14 @@ type AteomService struct { // exactly what atomic.Pointer is for. // // The type makes a lock-free read possible; it does not make one happen. - // GetWorkloadStats must not take lock at all, including around whatever it - // does with the value. A regression test pins that once there is a handler - // with a body to pin. + // GetWorkloadStats must not take lock at all, including around the cgroup + // read it does with the value. TestGetWorkloadStatsDoesNotTakeLock pins that. activeActor atomic.Pointer[ateomstats.ActorAttribution] + + // cgroupRoot is where the sandbox's cgroup v2 leaves live: the worker pod's + // own cgroup scope, which setupCgroupDelegation prepares. A field rather + // than a constant so tests can point GetWorkloadStats at a fixture tree. + cgroupRoot string } var _ ateompb.AteomServer = (*AteomService)(nil) @@ -298,6 +300,7 @@ func NewService(interiorNetNS netns.NsHandle, actorLogger *actorlog.ActorLogger, workerCredentialBundlePath: workerCredentialBundlePath, podIdentityTrustBundlePath: podIdentityTrustBundlePath, egressGatewayTrustBundlePath: egressGatewayTrustBundlePath, + cgroupRoot: defaultCgroupRoot, } } @@ -516,16 +519,6 @@ func (s *AteomService) CheckpointWorkload(ctx context.Context, req *ateompb.Chec return &ateompb.CheckpointWorkloadResponse{SnapshotFiles: snapshotFiles}, nil } -// GetWorkloadStats implements ateompb.Ateom/GetWorkloadStats. -// -// The attribution half is wired up here; the measurement half is not. Reading the -// sandbox's cgroup (/sys/fs/cgroup/pause) lands in the follow-up to -// https://github.com/agent-substrate/substrate/issues/594, at which point this -// stops returning Unimplemented. -func (s *AteomService) GetWorkloadStats(ctx context.Context, req *ateompb.GetWorkloadStatsRequest) (*ateompb.GetWorkloadStatsResponse, error) { - return nil, status.Error(codes.Unimplemented, "GetWorkloadStats is not implemented yet") -} - // listSnapshotFiles returns the (relative) names of regular files directly under // dir, which atelet ships to object storage as the snapshot. func listSnapshotFiles(dir string) ([]string, error) { diff --git a/cmd/ateom-gvisor/stats.go b/cmd/ateom-gvisor/stats.go new file mode 100644 index 000000000..4514af413 --- /dev/null +++ b/cmd/ateom-gvisor/stats.go @@ -0,0 +1,164 @@ +//go:build linux + +// 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/fs" + "path/filepath" + "time" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/agent-substrate/substrate/cmd/ateom-gvisor/internal/cgroupstats" + "github.com/agent-substrate/substrate/internal/proto/ateompb" +) + +// defaultCgroupRoot is the worker pod's own cgroup scope. The worker runs in a +// private cgroup namespace, so this path is the pod's cgroup rather than the +// host root, and runsc's per-container leaves are its direct children (see +// setupCgroupDelegation). +const defaultCgroupRoot = "/sys/fs/cgroup" + +// sandboxCgroupContainer is the container whose cgroup leaf accounts for the +// sandbox as a whole. +// +// gVisor runs every container of a sandbox as threads inside one host process, +// the sentry. runsc starts that process from the root container's create and +// from inside that container's cgroup — container.createRoot wraps the sandbox +// and gofer spawn in cgroup.RunInCgroup — so the sentry lands in the leaf of +// "pause", the first container RunWorkload and RestoreWorkload create. +// +// The leaf is a direct child of the delegated scope rather than of ateom's own +// cgroup, because runsc resolves cgroupsPath against the parent of the cgroup +// it is running in: cgroup v2 forbids a cgroup from holding processes and +// delegating controllers to children at once, so runsc walks up one level to +// find a directory it is allowed to create in. setupCgroupDelegation moves +// ateom into /sys/fs/cgroup/ateom precisely so that one level up is the +// delegated scope. +// +// The actor's own containers get leaves too, since cmdCreate calls +// ensureContainerCgroupsPath for each of them, but those leaves stay empty by +// design. gVisor's setupCgroupForSubcontainer creates them with empty resources +// and explains why: "Since subcontainers run exclusively inside the sandbox, +// subcontainer cgroups on the host have no effect on them. However, some tools +// (e.g. cAdvisor) uses cgroups paths to discover new containers and report +// stats for them." They are discovery markers, not accounting. +// +// So the actor's memory and CPU are the sentry's and are charged here. That is +// why a sample is attributed to the actor rather than to a container, and why +// this reads one leaf instead of summing them — the others have nothing in them +// to sum. +// +// What the leaf holds besides the actor's own work: the sentry's own overhead +// (its Go heap, page tables, netstack) and the gofers. Process listings taken +// on a live node in #161 put runsc-sandbox and both gofers — the pause +// container's and the actor container's — in the "pause" cgroup. Those runs +// predate #496, so they establish the leaf name and the fact that everything +// lands in one leaf, not the absolute path, which #496's delegation moved under +// the pod scope. +// +// So this measures the sandbox, not the actor's processes in isolation, which +// is what the proto means by "the unit of measurement is the SANDBOX". Sizing +// and chargeback want that number, since the sentry's overhead is a cost the +// node pays for running this actor, but it is not comparable to a container +// figure from a runc-based runtime, and a mostly idle actor reads as a nonzero +// floor. Splitting the actor's share out would need the sentry's own +// accounting, which the proto already says is not reported here. +// +// The name has to agree with the cgroupsPath convention in +// runsc.ensureContainerCgroupsPath, which is "/" + containerName relative to +// the same scope. +const sandboxCgroupContainer = "pause" + +// GetWorkloadStats implements ateompb.Ateom/GetWorkloadStats. +// +// Unlike the three lifecycle RPCs this does not take s.lock, and must not start: +// it is polled on a timer for the whole life of a workload, while lock is held +// across entire boots and checkpoints. Blocking on it would park every poll +// behind a multi-second runsc call, and — worse in the other direction — holding +// it across the cgroup read would put a CheckpointWorkload behind telemetry. +// The attribution is read from an atomic instead, and the cgroup files are read +// with no lock held at all. +func (s *AteomService) GetWorkloadStats(ctx context.Context, req *ateompb.GetWorkloadStatsRequest) (*ateompb.GetWorkloadStatsResponse, error) { + if req.GetActorUid() == "" { + return nil, status.Error(codes.InvalidArgument, "actor_uid is required") + } + + // Both of these are NOT_FOUND rather than FAILED_PRECONDITION: they tell the + // caller the requested actor is not here, which no amount of retrying on the + // same timer will change. Its worker-to-actor mapping wants re-resolving. + active := s.activeActor.Load() + if active == nil { + return nil, status.Errorf(codes.NotFound, "ateom is available; it is not executing actor %q", req.GetActorUid()) + } + if active.UID != req.GetActorUid() { + return nil, status.Errorf(codes.NotFound, "ateom is executing actor %q, not the requested %q", active.UID, req.GetActorUid()) + } + + observedAt := time.Now() + sample, err := cgroupstats.Read(filepath.Join(s.cgroupRoot, sandboxCgroupContainer)) + if err != nil { + // The requested actor is the active one but its cgroup is not there. Most + // often that is a poll landing in the boot: the ateom retains the + // attribution from the moment it accepts the actor, before runsc has + // created the leaf. The other way in is a sandbox that went away between + // the check above and the read, which the next CheckpointWorkload turns + // into the NOT_FOUND above. Either way it is "no numbers right now" and the + // caller should take the next sample, so FAILED_PRECONDITION. Anything else + // is a real read failure. + if errors.Is(err, fs.ErrNotExist) { + return nil, status.Error(codes.FailedPrecondition, "no sandbox cgroup to measure yet") + } + return nil, status.Errorf(codes.Internal, "reading sandbox cgroup: %v", err) + } + + // Re-check that the same workload is still the active one. The read above + // holds no lock, so a checkpoint plus a fresh run can complete underneath it, + // and the numbers would then belong to an actor other than the one being + // reported. Pointer identity is enough: activeActor is stored as a new + // pointer on every Run and Restore and never mutated in place, so an + // unchanged pointer means no transition happened across the read. + // + // NOT_FOUND, like the two checks above and for the same reason: the requested + // actor is no longer the one here, so a retry lands on one of them and gets + // that answer anyway. The same state should not report two different codes + // depending on where in the handler it was noticed. + if s.activeActor.Load() != active { + return nil, status.Errorf(codes.NotFound, "ateom stopped executing actor %q while the sample was being taken", req.GetActorUid()) + } + + return &ateompb.GetWorkloadStatsResponse{ + Atespace: active.Ref.Atespace, + ActorName: active.Ref.Name, + ActorUid: active.UID, + ActorTemplateNamespace: active.TemplateNamespace, + ActorTemplateName: active.TemplateName, + + SandboxClass: ateompb.SandboxClass_SANDBOX_CLASS_GVISOR, + Source: ateompb.StatsSource_STATS_SOURCE_CGROUP, + + MemoryCurrentBytes: sample.MemoryCurrentBytes, + MemoryPeakBytes: sample.MemoryPeakBytes, + MemoryWorkingSetBytes: sample.MemoryWorkingSetBytes, + CpuUsageUsec: sample.CPUUsageUsec, + + ObservedAtUnixNano: observedAt.UnixNano(), + }, nil +} diff --git a/cmd/ateom-gvisor/stats_test.go b/cmd/ateom-gvisor/stats_test.go index 65518c2fe..da57a5782 100644 --- a/cmd/ateom-gvisor/stats_test.go +++ b/cmd/ateom-gvisor/stats_test.go @@ -18,33 +18,191 @@ package main import ( "context" + "os" + "path/filepath" "testing" + "time" + "github.com/google/go-cmp/cmp" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + "google.golang.org/protobuf/testing/protocmp" + "github.com/agent-substrate/substrate/internal/ateomstats" "github.com/agent-substrate/substrate/internal/proto/ateompb" + "github.com/agent-substrate/substrate/internal/resources" ) -// TestGetWorkloadStatsUnimplemented pins the stub's advertised contract. The -// cgroup read replaces this body; until then a caller that gets any other code back -// would be reading numbers that are not there. -// -// The retention this stub will eventually read — s.activeActor, set by -// RunWorkload / RestoreWorkload and cleared by CheckpointWorkload — has no unit -// test, because those three RPCs each reach for netlink, runsc, and the worker -// pod's netns within a few lines of entry and cannot be driven from `go test`. -// Its mapping is covered in internal/ateomstats; the transitions are verified -// end to end once GetWorkloadStats returns real data. -func TestGetWorkloadStatsUnimplemented(t *testing.T) { - s := &AteomService{} - - resp, err := s.GetWorkloadStats(context.Background(), &ateompb.GetWorkloadStatsRequest{ActorUid: "uid-c"}) - if resp != nil { - t.Errorf("GetWorkloadStats() returned response %v, want nil", resp) - } - if got := status.Code(err); got != codes.Unimplemented { - t.Errorf("GetWorkloadStats() error code = %v, want %v (err: %v)", got, codes.Unimplemented, err) +// The lifecycle transitions that maintain s.activeActor — set by RunWorkload and +// RestoreWorkload, cleared by CheckpointWorkload — have no unit test, because +// those three RPCs each reach for netlink, runsc, and the worker pod's netns +// within a few lines of entry and cannot be driven from `go test`. The mapping +// they use is covered in internal/ateomstats; the transitions are verified end +// to end. What is testable here is everything GetWorkloadStats does with the +// result, which is where the polling loop will actually live. + +var testActor = ateomstats.ActorAttribution{ + Ref: resources.ActorRef{Atespace: "space-a", Name: "actor-a"}, + UID: "uid-a", + TemplateNamespace: "ns-a", + TemplateName: "template-a", +} + +var healthyCgroup = map[string]string{ + "memory.current": "157286400\n", + "memory.peak": "209715200\n", + "memory.stat": "anon 104857600\ninactive_file 20971520\nactive_file 31457280\n", + "cpu.stat": "usage_usec 1234567\nuser_usec 1000000\n", +} + +// newStatsService builds a service whose cgroup root is a fixture tree. A nil +// files map leaves the sandbox cgroup directory absent entirely, which is what +// a torn-down sandbox looks like. +func newStatsService(t *testing.T, files map[string]string) *AteomService { + t.Helper() + root := t.TempDir() + if files != nil { + dir := filepath.Join(root, sandboxCgroupContainer) + if err := os.Mkdir(dir, 0o700); err != nil { + t.Fatalf("creating fixture cgroup dir: %v", err) + } + for name, content := range files { + if err := os.WriteFile(filepath.Join(dir, name), []byte(content), 0o600); err != nil { + t.Fatalf("writing fixture %q: %v", name, err) + } + } + } + return &AteomService{cgroupRoot: root} +} + +func TestGetWorkloadStats(t *testing.T) { + s := newStatsService(t, healthyCgroup) + s.activeActor.Store(&testActor) + + before := time.Now().UnixNano() + got, err := s.GetWorkloadStats(context.Background(), &ateompb.GetWorkloadStatsRequest{ActorUid: "uid-a"}) + after := time.Now().UnixNano() + if err != nil { + t.Fatalf("GetWorkloadStats() error = %v, want nil", err) + } + + if got.GetObservedAtUnixNano() < before || got.GetObservedAtUnixNano() > after { + t.Errorf("GetWorkloadStats() observed_at_unix_nano = %d, want within [%d, %d]", got.GetObservedAtUnixNano(), before, after) + } + // Checked above; zeroed so the rest can be compared as a whole. + got.ObservedAtUnixNano = 0 + + want := &ateompb.GetWorkloadStatsResponse{ + Atespace: "space-a", + ActorName: "actor-a", + ActorUid: "uid-a", + ActorTemplateNamespace: "ns-a", + ActorTemplateName: "template-a", + SandboxClass: ateompb.SandboxClass_SANDBOX_CLASS_GVISOR, + Source: ateompb.StatsSource_STATS_SOURCE_CGROUP, + MemoryCurrentBytes: 157286400, + MemoryPeakBytes: 209715200, + MemoryWorkingSetBytes: 136314880, + CpuUsageUsec: 1234567, + } + if diff := cmp.Diff(want, got, protocmp.Transform()); diff != "" { + t.Errorf("GetWorkloadStats() mismatch (-want +got):\n%s", diff) + } +} + +func TestGetWorkloadStatsErrors(t *testing.T) { + for _, tc := range []struct { + name string + // files is the fixture sandbox cgroup; nil means the directory is absent. + files map[string]string + // active is stored into activeActor when non-nil; nil leaves the ateom + // "available". + active *ateomstats.ActorAttribution + actorUID string + want codes.Code + }{ + { + // A required field the caller left off: a client bug, distinct from the + // races below, so it gets a distinct code. + name: "empty actor_uid", + files: healthyCgroup, + active: &testActor, + actorUID: "", + want: codes.InvalidArgument, + }, + { + // Not here at all. NOT_FOUND rather than FAILED_PRECONDITION, because + // what the caller should do about it is re-resolve, not retry. + name: "ateom is available", + files: healthyCgroup, + active: nil, + actorUID: "uid-a", + want: codes.NotFound, + }, + { + // The worker was recycled between the caller's view of the world and + // this call. Reporting anyway would file one actor's numbers under + // another's name, and it is the same "not here" as the case above. + name: "actor_uid does not match the executing workload", + files: healthyCgroup, + active: &testActor, + actorUID: "uid-b", + want: codes.NotFound, + }, + { + // The requested actor is the one here, but there is nothing to read yet: + // a poll landing in the boot, or a sandbox torn down between the + // attribution check and the read. The one transient case, so the one + // FAILED_PRECONDITION. + name: "no sandbox cgroup to measure", + files: nil, + active: &testActor, + actorUID: "uid-a", + want: codes.FailedPrecondition, + }, + { + // The cgroup is there but does not parse: not a routine race, so it + // must not be reported as one. + name: "sandbox cgroup is malformed", + files: map[string]string{"memory.current": "max\n"}, + active: &testActor, + actorUID: "uid-a", + want: codes.Internal, + }, + } { + t.Run(tc.name, func(t *testing.T) { + s := newStatsService(t, tc.files) + if tc.active != nil { + s.activeActor.Store(tc.active) + } + + resp, err := s.GetWorkloadStats(context.Background(), &ateompb.GetWorkloadStatsRequest{ActorUid: tc.actorUID}) + if resp != nil { + t.Errorf("GetWorkloadStats() returned response %v, want nil", resp) + } + if got := status.Code(err); got != tc.want { + t.Errorf("GetWorkloadStats() error code = %v, want %v (err: %v)", got, tc.want, err) + } + }) + } +} + +// TestGetWorkloadStatsDoesNotTakeLock is the regression test for the property +// the design turns on: a stats poll must not queue behind a lifecycle RPC. +// s.lock is held for the duration of the call here, so a handler that reached +// for it would deadlock and fail this test by timing out rather than by +// assertion. +func TestGetWorkloadStatsDoesNotTakeLock(t *testing.T) { + s := newStatsService(t, healthyCgroup) + s.activeActor.Store(&testActor) + + // Stands in for a RunWorkload or CheckpointWorkload in flight, which hold the + // lock across their entire bodies. + s.lock.Lock() + defer s.lock.Unlock() + + if _, err := s.GetWorkloadStats(context.Background(), &ateompb.GetWorkloadStatsRequest{ActorUid: "uid-a"}); err != nil { + t.Errorf("GetWorkloadStats() error = %v, want nil", err) } } @@ -53,7 +211,7 @@ func TestGetWorkloadStatsUnimplemented(t *testing.T) { // is built on this: a non-nil zero value here would make an idle ateom report // an empty actor's usage instead of refusing. func TestAteomServiceStartsAvailable(t *testing.T) { - if s := (&AteomService{}); s.activeActor.Load() != nil { - t.Errorf("new AteomService.activeActor = %v, want nil", s.activeActor.Load()) + if got := (&AteomService{}).activeActor.Load(); got != nil { + t.Errorf("new AteomService.activeActor = %v, want nil", got) } } diff --git a/internal/proto/ateompb/ateom.pb.go b/internal/proto/ateompb/ateom.pb.go index 42020320a..cfe6b3ed5 100644 --- a/internal/proto/ateompb/ateom.pb.go +++ b/internal/proto/ateompb/ateom.pb.go @@ -1099,12 +1099,28 @@ type GetWorkloadStatsResponse struct { Source StatsSource `protobuf:"varint,7,opt,name=source,proto3,enum=ateom.StatsSource" json:"source,omitempty"` // Measurements. All four are zero when source is STATS_SOURCE_UNSPECIFIED, // which means "not measured" rather than "measured as zero". - MemoryCurrentBytes uint64 `protobuf:"varint,8,opt,name=memory_current_bytes,json=memoryCurrentBytes,proto3" json:"memory_current_bytes,omitempty"` - MemoryPeakBytes uint64 `protobuf:"varint,9,opt,name=memory_peak_bytes,json=memoryPeakBytes,proto3" json:"memory_peak_bytes,omitempty"` + // + // Two of them accumulate -- memory_peak_bytes and cpu_usage_usec -- and both + // are scoped to the current EPOCH rather than to the actor's lifetime. An + // epoch begins wherever the accounting behind the sample begins, which is not + // the same event for every source: STATS_SOURCE_CGROUP reads a sandbox cgroup + // that a restore recreates, so both restart at zero there, while + // STATS_SOURCE_GUEST_AGENT reads counters the guest kernel keeps in its own + // RAM, which a restored guest brings back with it. A caller that wants a + // lifetime figure has to accumulate one itself, and must read a decrease as a + // new epoch rather than emit a negative delta -- but not the converse. An + // epoch can also begin at a value above the last one reported, so no + // comparison of consecutive samples detects every boundary. + MemoryCurrentBytes uint64 `protobuf:"varint,8,opt,name=memory_current_bytes,json=memoryCurrentBytes,proto3" json:"memory_current_bytes,omitempty"` + // High-water mark of memory_current_bytes within the current epoch. Also zero + // when the runtime cannot report a peak at all: the cgroup source reads + // memory.peak, which only exists on Linux 5.19 and later. + MemoryPeakBytes uint64 `protobuf:"varint,9,opt,name=memory_peak_bytes,json=memoryPeakBytes,proto3" json:"memory_peak_bytes,omitempty"` + // memory_current_bytes less the reclaimable page cache, floored at zero. This + // is the figure to compare against a memory limit; memory_current_bytes + // drifts upward with cache the kernel would drop for free under pressure. MemoryWorkingSetBytes uint64 `protobuf:"varint,10,opt,name=memory_working_set_bytes,json=memoryWorkingSetBytes,proto3" json:"memory_working_set_bytes,omitempty"` - // Cumulative CPU time within the current epoch. A restore starts a new epoch: - // the sandbox is recreated, so this restarts at zero and callers computing - // deltas must treat a decrease as a reset rather than emit a negative value. + // Cumulative CPU time within the current epoch. CpuUsageUsec uint64 `protobuf:"varint,11,opt,name=cpu_usage_usec,json=cpuUsageUsec,proto3" json:"cpu_usage_usec,omitempty"` ObservedAtUnixNano int64 `protobuf:"varint,12,opt,name=observed_at_unix_nano,json=observedAtUnixNano,proto3" json:"observed_at_unix_nano,omitempty"` unknownFields protoimpl.UnknownFields diff --git a/internal/proto/ateompb/ateom.proto b/internal/proto/ateompb/ateom.proto index be4ce4d2c..2d6a8d8be 100644 --- a/internal/proto/ateompb/ateom.proto +++ b/internal/proto/ateompb/ateom.proto @@ -300,12 +300,28 @@ message GetWorkloadStatsResponse { // Measurements. All four are zero when source is STATS_SOURCE_UNSPECIFIED, // which means "not measured" rather than "measured as zero". + // + // Two of them accumulate -- memory_peak_bytes and cpu_usage_usec -- and both + // are scoped to the current EPOCH rather than to the actor's lifetime. An + // epoch begins wherever the accounting behind the sample begins, which is not + // the same event for every source: STATS_SOURCE_CGROUP reads a sandbox cgroup + // that a restore recreates, so both restart at zero there, while + // STATS_SOURCE_GUEST_AGENT reads counters the guest kernel keeps in its own + // RAM, which a restored guest brings back with it. A caller that wants a + // lifetime figure has to accumulate one itself, and must read a decrease as a + // new epoch rather than emit a negative delta -- but not the converse. An + // epoch can also begin at a value above the last one reported, so no + // comparison of consecutive samples detects every boundary. uint64 memory_current_bytes = 8; + // High-water mark of memory_current_bytes within the current epoch. Also zero + // when the runtime cannot report a peak at all: the cgroup source reads + // memory.peak, which only exists on Linux 5.19 and later. uint64 memory_peak_bytes = 9; + // memory_current_bytes less the reclaimable page cache, floored at zero. This + // is the figure to compare against a memory limit; memory_current_bytes + // drifts upward with cache the kernel would drop for free under pressure. uint64 memory_working_set_bytes = 10; - // Cumulative CPU time within the current epoch. A restore starts a new epoch: - // the sandbox is recreated, so this restarts at zero and callers computing - // deltas must treat a decrease as a reset rather than emit a negative value. + // Cumulative CPU time within the current epoch. uint64 cpu_usage_usec = 11; int64 observed_at_unix_nano = 12;