From 1464ca227d6d3ab3d960e880ad84fd43376f2f08 Mon Sep 17 00:00:00 2001 From: Konstantins Feofantovs Date: Mon, 31 Aug 2026 22:37:14 +0300 Subject: [PATCH] feat: the counters are subtracted on the node (ADR 0062) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The rows beside the footprint are a different kind of thing: counters that only rise since the process started. `run_delay` says how long a process has waited for a CPU in its whole life, and a life is not a question anybody asks. Where the subtraction happens is the decision. Doing it on the controller fails twice: a PID is reused, and the controller never sees a start time to tell two processes apart, so it would hold per-PID state for a fleet — growing with the cluster rather than with the workloads. And a process that exits between flushes loses its last interval, because the previous reading is what would have closed it. The node has both halves in one read. So a cache holds one reading per PID between passes, keyed by the process's start time. Three cases produce no delta rather than a wrong one: the first pass that sees a process, a reused PID, and a counter that fell — the last drops the whole delta rather than clamping a field, because a counter that fell means the reading is not of the process the last one described. A delta never travels without `observed_nanos`, the wall-clock time between the two readings. It sums with the deltas, so it is process-time and not a duration, and it is named for that. The payload carries sums and never the division: a shipped ratio cannot be re-divided by a consumer asking a different question. Field two of /proc//stat is the executable's name and may contain spaces and parentheses, so fields are counted from the last ')' — which is also what keeps the name behind the cut rather than filtered after it. go test ./... -race, golangci-lint run ./... (host and GOOS=linux): 0 issues. --- cmd/agent/main.go | 6 + .../adr/0052-the-peak-the-kernel-remembers.md | 2 +- ...-what-the-process-holds-beside-its-peak.md | 2 + ...-the-counters-and-the-window-they-cover.md | 112 +++++++ docs/adr/README.md | 1 + docs/backend-requirements.md | 14 + docs/prose_test.go | 2 +- docs/security.md | 1 + internal/inventory/store.go | 171 ++++++++++- internal/inventory/store_test.go | 107 +++++++ internal/nodescan/proccounters.go | 275 ++++++++++++++++++ internal/nodescan/proccounters_test.go | 215 ++++++++++++++ internal/nodescan/procstatus.go | 18 ++ internal/nodescan/scan.go | 24 +- internal/sink/registry.go | 10 + internal/sink/spool.go | 32 ++ internal/sink/spool_test.go | 44 +++ .../testdata/process-counters.golden.json | 42 +++ test/e2e/inventory_test.go | 78 +++++ 19 files changed, 1149 insertions(+), 7 deletions(-) create mode 100644 docs/adr/0062-the-counters-and-the-window-they-cover.md create mode 100644 internal/nodescan/proccounters.go create mode 100644 internal/nodescan/proccounters_test.go create mode 100644 internal/sink/testdata/process-counters.golden.json diff --git a/cmd/agent/main.go b/cmd/agent/main.go index b39fb9c..a1e72f8 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -420,6 +420,12 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac if err := spool.WriteProcessPeaks(time.Now(), goStore.PeakSnapshot()); err != nil { logger.Error("spooling process peaks", "error", err) } + // What those processes did since each node's previous pass. Differences + // rather than readings, so this kind is the one whose records mean + // nothing without the interval they carry (ADR 0062). + if err := spool.WriteProcessCounters(time.Now(), goStore.CounterSnapshot()); err != nil { + logger.Error("spooling process counters", "error", err) + } // Where those processes accept connections. Structural like the // inventory, and separate from it because it changes without the build // changing (ADR 0056 §3). diff --git a/docs/adr/0052-the-peak-the-kernel-remembers.md b/docs/adr/0052-the-peak-the-kernel-remembers.md index f74cc7a..400cee7 100644 --- a/docs/adr/0052-the-peak-the-kernel-remembers.md +++ b/docs/adr/0052-the-peak-the-kernel-remembers.md @@ -4,7 +4,7 @@ Date: 2026-08-29 Status: Accepted -Amended by: 0061 +Amended by: 0061, 0062 Amends: 0009, 0006 §5 diff --git a/docs/adr/0061-what-the-process-holds-beside-its-peak.md b/docs/adr/0061-what-the-process-holds-beside-its-peak.md index a4be54c..8be5747 100644 --- a/docs/adr/0061-what-the-process-holds-beside-its-peak.md +++ b/docs/adr/0061-what-the-process-holds-beside-its-peak.md @@ -4,6 +4,8 @@ Date: 2026-08-31 Status: Accepted +Amended by: 0062 + Amends: 0052 Adds seven fields to `process_peaks`, read from three files under diff --git a/docs/adr/0062-the-counters-and-the-window-they-cover.md b/docs/adr/0062-the-counters-and-the-window-they-cover.md new file mode 100644 index 0000000..89968ce --- /dev/null +++ b/docs/adr/0062-the-counters-and-the-window-they-cover.md @@ -0,0 +1,112 @@ +# 0062. The counters are subtracted on the node, and never leave without their interval + +Date: 2026-08-31 + +Status: Accepted + +Amends: 0052, 0061 + +Adds the payload kind `process_counters`, and one piece of node-held state: the +previous pass's counter reading per process. Reads three more files under +`/proc//` and two more rows of one already open. No new privilege. + +## Context + +Everything ADR 0052 and ADR 0061 collect is a value: a peak, a footprint, a +count of threads. The rows beside them are a different kind of thing — counters +that only rise, since the process started. `run_delay` says how long a process +has waited for a CPU in its whole life, and a life is not a question anybody +asks. What a finding needs is the change over an interval. + +That subtraction has to happen somewhere, and where it happens is the decision. +Shipping raw counters and subtracting on the controller was the first idea and it +fails on two counts. A PID is reused, so two readings under one number can belong +to two processes, and the controller has no way to tell — it never sees a start +time, and holding per-PID state for a fleet's processes would grow the store with +the cluster rather than with the workloads. And a process that exits between two +flushes has its last interval lost, because the controller's previous reading is +the only thing that would have closed it. + +The node has both halves: it reads the counters and the start time from the same +file at the same moment, and it already keeps a cache across passes for the +binary marker (ADR 0056). + +## Decision + +**1. The node subtracts, and the start time keys the subtraction.** A cache holds +one reading per PID between passes — the counters, the instant they were read, +and the process's start time. A pass compares against the previous reading only +when the start time matches; otherwise the PID belongs to a different process and +there is nothing to subtract from. + +Three cases produce no delta rather than a wrong one: the first pass that sees a +process, a PID that has been reused, and a counter that fell. The last should be +impossible when the start time matches, and it drops the whole delta rather than +clamping one field, because a counter that fell means the reading is not of the +process the last one described, whatever the start time says. + +Field two of `/proc//stat` is the executable's own name, and it may contain +spaces and parentheses. Fields are counted from the *last* `)` rather than by +splitting the line — which is also why the name never reaches a variable here: +it is behind the cut, not filtered after it (the discipline ADR 0052 set for +`status`). + +**2. A delta never travels without the interval it covers.** Each carries +`observed_nanos`, the wall-clock time between the two readings. A delta alone is +a number no consumer can turn into a rate, and the interval is not a constant: +a pass can be late, a report can be lost, and a node restarts. + +**3. The payload carries sums, never ratios.** `on_cpu_nanos` beside +`run_delay_nanos` is CPU starvation stated as the two numbers it is made of. The +agent could divide them, and does not: a payload that ships the division cannot +be re-divided by a consumer asking a different question, which is the shape +ADR 0004 settled for money and holds for every derived figure. + +`observed_nanos` sums across processes, so it is process-time and not a duration. +Two processes observed for a minute each contribute two minutes. That makes it +the denominator of a rate and not a window, and it is named for what it is. + +**4. A node's next report replaces its interval, it does not extend it.** The +same wholesale replacement ADR 0052 §3 uses for peaks, for a different reason: +adding successive intervals would turn a rate into a total over a window nobody +chose, drifting further from any wall-clock boundary with every pass. The payload +is therefore what the fleet's processes did over the most recent interval each +node measured — ragged across nodes by construction, and honest about it, because +each record carries the process-time it rests on. + +**5. Two rows come from a file already open.** The context-switch counters are in +`/proc//status`, which the scanner reads for the footprint. They are +monotonic like the rest, so they join the reading rather than the footprint. + +## Consequences + +**Easier.** CPU starvation becomes measurable without depending on CFS +throttling being visible: `run_delay` counts the wait even when the quota was +never hit, and the non-voluntary switch count is the same fact from the +scheduler's side. The user/system split says whether a service burns its own +code or the kernel acting for it. Major faults separate a working set that does +not fit from one that is merely large. Disk bytes per workload exist at all, +which they do not in any kubelet reading. + +**Harder / given up.** The node now holds state between passes that it did not +before: one reading per live process. It is bounded by the process table and +dropped for every PID a pass does not see, and losing it costs one interval and +never a total (ADR 0003). But it is state, and a node that restarts every minute +would produce no counters at all — which is visible, because the record's +`processes` count would stay at zero. + +The interval between a process's last observed pass and its exit is lost. There +is no way to close it without the node observing the exit, which would mean +watching every process rather than sampling the table. + +`processes` in this record is lower than in `process_peaks` for one interval +after a pod starts, because a process seen once has a footprint and no delta. +The two counts answer different questions and are not the same number. + +**Not changed.** No new capability — `/proc//io` sits behind the same ptrace +check the node already passes for `/proc//exe` (ADR 0037). No identity: the +executable name in `stat` is behind the field cut, and what a byte was read from +is not in any of these files. + +This ADR records decisions implemented in the same pull request, per the process +in [`README.md`](README.md). diff --git a/docs/adr/README.md b/docs/adr/README.md index 43cac6d..5dd7225 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -119,3 +119,4 @@ line to paste. - [0059. The node knows whose code it profiles, so the allow-list stops being a prerequisite](0059-the-node-knows-whose-code-it-profiles.md) - [0060. The node says what its profiler did, on the channel that fires when it did nothing](0060-the-node-says-what-its-profiler-did.md) - [0061. What the process holds, beside the peak it once reached](0061-what-the-process-holds-beside-its-peak.md) +- [0062. The counters are subtracted on the node, and never leave without their interval](0062-the-counters-and-the-window-they-cover.md) diff --git a/docs/backend-requirements.md b/docs/backend-requirements.md index d9c71e0..7554f7c 100644 --- a/docs/backend-requirements.md +++ b/docs/backend-requirements.md @@ -413,6 +413,20 @@ a record is absent, never zero, when none were read. The record is keyed by image digest as well as container: a peak belongs to the build that reached it, and MUST NOT be carried across a rollout. +**`process_counters` records are differences, and `observed_nanos` is what they +rest on** ([ADR 0062](adr/0062-the-counters-and-the-window-they-cover.md)). Each +is the change since the reporting node's previous pass, summed over processes and +nodes, so it MUST NOT be summed across flushes: a node's next report replaces its +interval rather than extending it. `observed_nanos` sums with the deltas, so it +is process-time and not a duration — two processes watched for a minute each +contribute two minutes — and it is the denominator every rate MUST be formed +against, never the flush interval or the wall clock. `on_cpu_nanos` beside +`run_delay_nanos` is CPU starvation; the backend forms the ratio, because the +agent ships numerators and denominators and never the division. A `processes` +count lower than the same container's in `process_peaks` is expected for one +interval after a pod starts: a process seen once has a footprint and no +difference yet, and MUST NOT be read as processes having exited. + **The footprint fields beside it are samples, and the field names say so** ([ADR 0061](adr/0061-what-the-process-holds-beside-its-peak.md)). `peak_rss_bytes` is a mark the kernel maintains between readings; every `*_max` beside it is the diff --git a/docs/prose_test.go b/docs/prose_test.go index 06bee55..7cee733 100644 --- a/docs/prose_test.go +++ b/docs/prose_test.go @@ -35,7 +35,7 @@ var ceilings = []struct { section int }{ {"security.md", 1036, 360}, - {"backend-requirements.md", 610, 467}, + {"backend-requirements.md", 624, 481}, } func TestNoCommentRunIsLongerThanAPointerToAnADR(t *testing.T) { diff --git a/docs/security.md b/docs/security.md index 5af44a9..be47b08 100644 --- a/docs/security.md +++ b/docs/security.md @@ -524,6 +524,7 @@ test fails if a kind ships without appearing here (ADR 0022). | `go_inventory` | Per (namespace, workload, container): Go version, main module, image digest, PGO flag, plus a fleet-coverage block | no | | `process_peaks` | Per collected workload container and build: the largest `VmHWM` among its Go processes, how many processes that was taken over, and the range of permitted CPU counts. A **floor** under the container's peak, never the figure the OOM killer compares against — the cgroup also holds page cache and every other process in it ([ADR 0052](adr/0052-the-peak-the-kernel-remembers.md)). Beside it, the largest sample taken of each process's footprint: resident memory split into the program's own pages and file-backed ones, the memory it shares with its replicas and the memory nothing else maps, its OS threads, and how many file descriptors it held against the ceiling in force ([ADR 0061](adr/0061-what-the-process-holds-beside-its-peak.md)). **What each descriptor points at is never read** — they are counted, not followed | no | | `go_build` | Per image digest, written once: Go version, main module, each dependency module's **path and version**, allow-listed build settings, two allow-listed GODEBUG defaults, and whether `net/http/pprof` is linked into the build. This is a bill of materials for the build — see [§10.4](#104-the-build-inventory-is-a-bill-of-materials) | no | +| `process_counters` | Per collected workload container and build: what its processes did since each node's previous scan pass — time on CPU and time waiting for one, that CPU time split between your code and the kernel acting for it, page faults that reached storage, context switches, and bytes that moved to and from block storage. Each record carries the process-time the numbers cover, because a difference without its interval is not a rate ([ADR 0062](adr/0062-the-counters-and-the-window-they-cover.md)). Sums only — the agent never ships the division | no | | `listening_ports` | Per collected workload container and build: the TCP ports its Go processes accept connections on, each with whether it is bound to loopback. Read from the processes' own sockets, so a port nothing declared is here and a declared port nothing binds is not ([ADR 0056](adr/0056-a-pprof-endpoint-is-proved-not-probed.md)). No address, and no connection the workload has open | no | | `ebpf_profile` | One capture: allow-list-filtered symbolized pprof bytes, keyed by workload, image digest and capture window | no | | `pprof_profile` | One ten-second CPU profile fetched from the workload's own `/debug/pprof` endpoint, allow-list-filtered, under the same key ([ADR 0058](adr/0058-the-pull-starts-a-profiler-and-the-binary-says-whose-code-it-is.md)). The mappings — the executable's path on disk and its build ID — the sample labels, and the full source paths are removed with the redacted frames, so what remains is function names your allow-list admits, base file names and line numbers. Beside it travel the counts of what was redacted, never what | no | diff --git a/internal/inventory/store.go b/internal/inventory/store.go index 9e44f1b..e902e29 100644 --- a/internal/inventory/store.go +++ b/internal/inventory/store.go @@ -102,6 +102,56 @@ type nodePeaks struct { filesLimitMin int64 } +// CounterKey extends Key with the build, for the reason PeakKey and PortKey do: +// what a process spent belongs to the code that spent it, and a rollout must not +// inherit the previous build's interval. +type CounterKey struct { + Key + ImageDigest string `json:"image_digest,omitempty"` +} + +// CounterRecord is what the processes of one workload container, of one build, +// did over the most recent interval each reporting node measured — summed over +// every process and every node. +// +// It is not a wall-clock window. ObservedNanos is the sum of the intervals the +// deltas cover, so it is process-time and grows with the number of processes: +// it is the denominator for a rate, never a duration (ADR 0062 §3). +type CounterRecord struct { + CounterKey + // Processes is how many processes contributed a delta. A process seen for + // the first time contributes none, so this is lower than the peaks record's + // count for one interval after a pod starts. + Processes int `json:"processes"` + // ObservedNanos is the summed wall-clock time those deltas cover. + ObservedNanos int64 `json:"observed_nanos"` + // OnCPUNanos against RunDelayNanos is CPU starvation stated as the two + // numbers it is made of, rather than as a ratio computed here: a payload + // that ships the division cannot be re-divided by a consumer asking a + // different question (ADR 0004's shape). + OnCPUNanos int64 `json:"on_cpu_nanos"` + RunDelayNanos int64 `json:"run_delay_nanos"` + // The same running time split between the program and the kernel acting + // for it. + CPUUserNanos int64 `json:"cpu_user_nanos"` + CPUSystemNanos int64 `json:"cpu_system_nanos"` + // MajorFaults reached storage; the two switch counts are the process + // yielding and the scheduler taking the CPU away. + MajorFaults int64 `json:"major_faults"` + VoluntarySwitches int64 `json:"voluntary_switches"` + NonvoluntarySwitches int64 `json:"nonvoluntary_switches"` + // Bytes that actually moved to and from block storage. + ReadBytes int64 `json:"read_bytes"` + WriteBytes int64 `json:"write_bytes"` +} + +// nodeCounters is one node's contribution to a CounterRecord: the sum of the +// deltas its latest report carried for that container. +type nodeCounters struct { + processes int + delta nodescan.CounterDelta +} + // PortKey extends Key with the build, for the reason PeakKey does: a rollout // that moves an endpoint must not inherit the port the previous build served. type PortKey struct { @@ -199,6 +249,13 @@ type Store struct { // survives only while some node still serves it (ADR 0056 §3). ports map[PortKey]map[string][]nodescan.ListeningPort + // counters holds, per (record key, build) and per reporting node, the + // deltas that node's latest report carried. Node-keyed and replaced + // wholesale for the reason peaks are: a node's next report describes a new + // interval, and adding it to the last would turn a rate into a total over + // a window nobody chose (ADR 0062 §3). + counters map[CounterKey]map[string]nodeCounters + // scans holds the latest scan counters each reporting node sent. Latest // rather than summed: they describe one pass over a node's process table, // and adding successive passes would count the same long-lived process once @@ -242,6 +299,7 @@ func NewStore(since time.Time) *Store { written: make(map[string]struct{}), peaks: make(map[PeakKey]map[string]nodePeaks), ports: make(map[PortKey]map[string][]nodescan.ListeningPort), + counters: make(map[CounterKey]map[string]nodeCounters), nodesReported: make(map[string]struct{}), scans: make(map[string]nodescan.Counters), profiling: make(map[string]nodescan.ProfilingCoverage), @@ -268,6 +326,7 @@ func (s *Store) Ingest(report nodescan.Report, resolver ContainerResolver) { } seenPeaks := make(map[PeakKey]bool, len(report.Binaries)) seenPorts := make(map[PortKey]bool, len(report.Binaries)) + seenCounters := make(map[CounterKey]bool, len(report.Binaries)) for _, b := range report.Binaries { s.factsReceived++ res, ok := resolver.LookupContainer(types.UID(b.PodUID), b.ContainerID) @@ -292,9 +351,11 @@ func (s *Store) Ingest(report nodescan.Report, resolver ContainerResolver) { s.ingestBuild(b, res.ImageDigest) s.ingestPeaks(b, key, res.ImageDigest, report.Node, seenPeaks) s.ingestPorts(b, key, res.ImageDigest, report.Node, seenPorts) + s.ingestCounters(b, key, res.ImageDigest, report.Node, seenCounters) } s.forgetUnreported(report.Node, seenPeaks) s.forgetUnreportedPorts(report.Node, seenPorts) + s.forgetUnreportedCounters(report.Node, seenCounters) } // ingestPeaks folds one binary's measured facts into this node's contribution @@ -389,6 +450,59 @@ func (s *Store) ingestPorts(b nodescan.BinaryInfo, key Key, digest, node string, byNode[node] = nodescan.MergeListeningPorts(acc, b.ListeningPorts) } +// ingestCounters sums one binary's delta into this node's contribution for its +// key, as ingestPeaks folds its measured facts. A binary carrying no delta — +// the first pass that saw it, or a PID the node could not compare against — +// contributes nothing, not a zero, and is not counted among the processes. +// +// Callers hold s.mu. +func (s *Store) ingestCounters(b nodescan.BinaryInfo, key Key, digest, node string, seen map[CounterKey]bool) { + if b.Counters == nil { + return + } + ck := CounterKey{Key: key, ImageDigest: digest} + byNode, ok := s.counters[ck] + if !ok { + byNode = make(map[string]nodeCounters, 1) + s.counters[ck] = byNode + } + var n nodeCounters + if seen[ck] { + n = byNode[node] + } + seen[ck] = true + n.processes++ + d := *b.Counters + n.delta.ObservedNanos += d.ObservedNanos + n.delta.OnCPUNanos += d.OnCPUNanos + n.delta.RunDelayNanos += d.RunDelayNanos + n.delta.CPUUserNanos += d.CPUUserNanos + n.delta.CPUSystemNanos += d.CPUSystemNanos + n.delta.MajorFaults += d.MajorFaults + n.delta.VoluntarySwitches += d.VoluntarySwitches + n.delta.NonvoluntarySwitches += d.NonvoluntarySwitches + n.delta.ReadBytes += d.ReadBytes + n.delta.WriteBytes += d.WriteBytes + byNode[node] = n +} + +// forgetUnreportedCounters drops this node's contribution to every counter key +// its latest report did not mention, and the key itself once no node +// contributes. Without it a rate would outlive the pod that produced it. +// +// Callers hold s.mu. +func (s *Store) forgetUnreportedCounters(node string, seen map[CounterKey]bool) { + for ck, byNode := range s.counters { + if seen[ck] { + continue + } + delete(byNode, node) + if len(byNode) == 0 { + delete(s.counters, ck) + } + } +} + // forgetUnreportedPorts drops this node's contribution to every port key its // latest report did not mention, and the key itself once no node contributes. // @@ -486,6 +600,8 @@ type Evicted struct { Peaks int // Ports is port records whose workload container is gone. Ports int + // Counters is counter records whose workload container is gone. + Counters int } // Retain drops every record whose key is absent from live, and then the build @@ -523,6 +639,12 @@ func (s *Store) Retain(live []Key) Evicted { ev.Ports++ } } + for ck := range s.counters { + if _, ok := set[ck.Key]; !ok { + delete(s.counters, ck) + ev.Counters++ + } + } referenced := make(map[string]struct{}, len(s.records)) for _, r := range s.records { @@ -563,8 +685,8 @@ func (s *Store) RetainNodes(live []string) { delete(s.profiling, n) } } - // A departed node's peaks and ports go with it. Its pods are gone, so what - // it contributed stands behind nothing (ADR 0052 §3). + // A departed node's peaks, ports and counters go with it. Its pods are + // gone, so what it contributed stands behind nothing (ADR 0052 §3). for pk, byNode := range s.peaks { for node := range byNode { if _, ok := set[node]; !ok { @@ -585,6 +707,16 @@ func (s *Store) RetainNodes(live []string) { delete(s.ports, pk) } } + for ck, byNode := range s.counters { + for node := range byNode { + if _, ok := set[node]; !ok { + delete(byNode, node) + } + } + if len(byNode) == 0 { + delete(s.counters, ck) + } + } } // MarkBuildWritten records that the facts for digest reached the sink, so they @@ -652,6 +784,41 @@ func (s *Store) PeakSnapshot() []PeakRecord { return out } +// CounterSnapshot returns one record per (workload container, build) that some +// node still reports a delta for, sorted deterministically. +func (s *Store) CounterSnapshot() []CounterRecord { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]CounterRecord, 0, len(s.counters)) + for ck, byNode := range s.counters { + rec := CounterRecord{CounterKey: ck} + for _, n := range byNode { + rec.Processes += n.processes + rec.ObservedNanos += n.delta.ObservedNanos + rec.OnCPUNanos += n.delta.OnCPUNanos + rec.RunDelayNanos += n.delta.RunDelayNanos + rec.CPUUserNanos += n.delta.CPUUserNanos + rec.CPUSystemNanos += n.delta.CPUSystemNanos + rec.MajorFaults += n.delta.MajorFaults + rec.VoluntarySwitches += n.delta.VoluntarySwitches + rec.NonvoluntarySwitches += n.delta.NonvoluntarySwitches + rec.ReadBytes += n.delta.ReadBytes + rec.WriteBytes += n.delta.WriteBytes + } + if rec.Processes == 0 { + continue + } + out = append(out, rec) + } + sort.Slice(out, func(i, j int) bool { + if out[i].Key != out[j].Key { + return lessKey(out[i].Key, out[j].Key) + } + return out[i].ImageDigest < out[j].ImageDigest + }) + return out +} + // PortSnapshot returns one record per (workload container, build) that some node // still reports a listening port for, sorted deterministically. func (s *Store) PortSnapshot() []PortRecord { diff --git a/internal/inventory/store_test.go b/internal/inventory/store_test.go index bb90104..e74947a 100644 --- a/internal/inventory/store_test.go +++ b/internal/inventory/store_test.go @@ -892,3 +892,110 @@ func TestAnUnreadFootprintContributesNothing(t *testing.T) { t.Errorf("processes = %d, want 1: the process that measured nothing stands behind nothing", p.Processes) } } + +// withCounters attaches one process's delta to a binary. +func withCounters(b nodescan.BinaryInfo, d nodescan.CounterDelta) nodescan.BinaryInfo { + b.Counters = &d + return b +} + +// Deltas sum across processes and nodes, and the interval sums with them: what +// leaves is a numerator and its denominator, never the division (ADR 0062 §3). +func TestCountersSumWithTheIntervalTheyCover(t *testing.T) { + s := NewStore(testSince) + s.Ingest(nodescan.Report{Node: "node-1", Binaries: []nodescan.BinaryInfo{ + withCounters(binary("pod-web-1", "cid-web", "go1.26.1", "github.com/acme/web", false), nodescan.CounterDelta{ + ObservedNanos: 60e9, OnCPUNanos: 9e9, RunDelayNanos: 4e9, + CPUUserNanos: 7e9, CPUSystemNanos: 2e9, NonvoluntarySwitches: 3000, ReadBytes: 4096, + }), + }}, webResolver()) + s.Ingest(nodescan.Report{Node: "node-2", Binaries: []nodescan.BinaryInfo{ + withCounters(binary("pod-web-2", "cid-web", "go1.26.1", "github.com/acme/web", false), nodescan.CounterDelta{ + ObservedNanos: 30e9, OnCPUNanos: 1e9, RunDelayNanos: 500e6, + CPUUserNanos: 800e6, CPUSystemNanos: 200e6, NonvoluntarySwitches: 122, ReadBytes: 0, + }), + }}, webResolver()) + + recs := s.CounterSnapshot() + if len(recs) != 1 { + t.Fatalf("counters = %+v, want one record", recs) + } + r := recs[0] + if r.Processes != 2 || r.ObservedNanos != 90e9 { + t.Errorf("processes/observed = %d/%d, want 2 over 90s of process time", r.Processes, r.ObservedNanos) + } + if r.OnCPUNanos != 10e9 || r.RunDelayNanos != 4.5e9 { + t.Errorf("on cpu/run delay = %d/%d, want the sums", r.OnCPUNanos, r.RunDelayNanos) + } + if r.NonvoluntarySwitches != 3122 || r.ReadBytes != 4096 { + t.Errorf("switches/read = %d/%d", r.NonvoluntarySwitches, r.ReadBytes) + } + if r.ImageDigest != "sha256:web" { + t.Errorf("image digest = %q; an interval belongs to the build that spent it", r.ImageDigest) + } +} + +// A binary with no delta — the first pass that saw its process — contributes +// nothing and is not counted, so `processes` is the number the sums rest on +// rather than the number that were running. +func TestAProcessWithNoDeltaIsNotCounted(t *testing.T) { + s := NewStore(testSince) + s.Ingest(nodescan.Report{Node: "node-1", Binaries: []nodescan.BinaryInfo{ + withCounters(binary("pod-web-1", "cid-web", "go1.26.1", "github.com/acme/web", false), nodescan.CounterDelta{ + ObservedNanos: 60e9, OnCPUNanos: 9e9, + }), + binary("pod-web-2", "cid-web", "go1.26.1", "github.com/acme/web", false), + }}, webResolver()) + + r := s.CounterSnapshot()[0] + if r.Processes != 1 || r.ObservedNanos != 60e9 { + t.Errorf("processes/observed = %d/%d, want only the process that had a delta", + r.Processes, r.ObservedNanos) + } +} + +// A node's next report describes a new interval. Adding it to the last would +// turn a rate into a total over a window nobody chose, so the contribution is +// replaced wholesale — the discipline the peaks beside it use. +func TestANodesNextReportReplacesItsInterval(t *testing.T) { + s := NewStore(testSince) + for range 3 { + s.Ingest(nodescan.Report{Node: "node-1", Binaries: []nodescan.BinaryInfo{ + withCounters(binary("pod-web-1", "cid-web", "go1.26.1", "github.com/acme/web", false), nodescan.CounterDelta{ + ObservedNanos: 60e9, OnCPUNanos: 9e9, + }), + }}, webResolver()) + } + r := s.CounterSnapshot()[0] + if r.ObservedNanos != 60e9 || r.OnCPUNanos != 9e9 { + t.Errorf("after three reports: observed = %d, on cpu = %d; the intervals accumulated", + r.ObservedNanos, r.OnCPUNanos) + } +} + +// A departed node takes its interval with it, and so does an evicted workload. +func TestCountersLeaveWithTheirNodeAndTheirWorkload(t *testing.T) { + s := NewStore(testSince) + s.Ingest(nodescan.Report{Node: "node-1", Binaries: []nodescan.BinaryInfo{ + withCounters(binary("pod-web-1", "cid-web", "go1.26.1", "github.com/acme/web", false), nodescan.CounterDelta{ + ObservedNanos: 60e9, OnCPUNanos: 9e9, + }), + }}, webResolver()) + + s.RetainNodes([]string{"node-2"}) + if recs := s.CounterSnapshot(); len(recs) != 0 { + t.Errorf("counters = %+v after the node left, want none", recs) + } + + s.Ingest(nodescan.Report{Node: "node-1", Binaries: []nodescan.BinaryInfo{ + withCounters(binary("pod-web-1", "cid-web", "go1.26.1", "github.com/acme/web", false), nodescan.CounterDelta{ + ObservedNanos: 60e9, OnCPUNanos: 9e9, + }), + }}, webResolver()) + if ev := s.Retain(nil); ev.Counters != 1 { + t.Errorf("evicted counters = %d, want 1", ev.Counters) + } + if recs := s.CounterSnapshot(); len(recs) != 0 { + t.Errorf("counters = %+v after the workload left, want none", recs) + } +} diff --git a/internal/nodescan/proccounters.go b/internal/nodescan/proccounters.go new file mode 100644 index 0000000..33ec3f8 --- /dev/null +++ b/internal/nodescan/proccounters.go @@ -0,0 +1,275 @@ +package nodescan + +import ( + "bufio" + "io" + "os" + "path/filepath" + "strconv" + "strings" + "time" +) + +// The counter reader (ADR 0062). Everything here is monotonic since the process +// started, so a reading alone says nothing about now: what a finding needs is +// the change over an interval, and this file is where that subtraction happens. +// +// It happens on the node because the node is the only place that can do it +// safely. A PID is reused, so two readings under one number can belong to two +// processes; the start time distinguishes them, and it is read alongside the +// counters rather than fetched afterwards (ADR 0043's discipline). + +// userHZ is the unit of the CPU-time fields in /proc//stat. It is 100 on +// every Linux architecture regardless of the kernel's own tick rate: USER_HZ is +// part of the kernel's ABI, not of its configuration. +const userHZ = 100 + +// nanosPerTick converts a clock tick to nanoseconds. +const nanosPerTick = int64(time.Second) / userHZ + +// maxCounterFileBytes bounds each read. The three files are a few hundred bytes. +const maxCounterFileBytes = 16 << 10 + +// ProcessCounters is one reading of the monotonic counters of one process. Every +// field is cumulative since that process started, and none is meaningful on its +// own — CounterDelta below is what leaves the node. +type ProcessCounters struct { + // OnCPUNanos is time spent running and RunDelayNanos time spent runnable + // but waiting for a CPU. Their ratio is CPU starvation measured directly, + // independent of whether a CFS quota was ever hit. + OnCPUNanos int64 + RunDelayNanos int64 + // CPUUserNanos and CPUSystemNanos split that running time between the + // program and the kernel acting for it. + CPUUserNanos int64 + CPUSystemNanos int64 + // MajorFaults are page faults that had to reach storage: the working set + // does not fit, rather than merely growing. + MajorFaults int64 + // VoluntarySwitches is the process yielding — normal for a server waiting + // on I/O. NonvoluntarySwitches is the scheduler taking the CPU away, which + // is the same starvation RunDelayNanos measures, seen from the other side. + VoluntarySwitches int64 + NonvoluntarySwitches int64 + // ReadBytes and WriteBytes are bytes that actually reached block storage, + // not bytes the process asked for: the page cache absorbs the difference. + ReadBytes int64 + WriteBytes int64 +} + +// CounterDelta is what changed between two readings of one process, and how much +// wall-clock time separates them. Both travel together: a delta without its +// interval is a number no consumer can turn into a rate (ADR 0062 §2). +type CounterDelta struct { + ObservedNanos int64 `json:"observed_nanos"` + OnCPUNanos int64 `json:"on_cpu_nanos,omitempty"` + RunDelayNanos int64 `json:"run_delay_nanos,omitempty"` + CPUUserNanos int64 `json:"cpu_user_nanos,omitempty"` + CPUSystemNanos int64 `json:"cpu_system_nanos,omitempty"` + MajorFaults int64 `json:"major_faults,omitempty"` + VoluntarySwitches int64 `json:"voluntary_switches,omitempty"` + NonvoluntarySwitches int64 `json:"nonvoluntary_switches,omitempty"` + ReadBytes int64 `json:"read_bytes,omitempty"` + WriteBytes int64 `json:"write_bytes,omitempty"` +} + +// reading is what the cache holds between passes: one process's counters, when +// they were read, and the start time that says which process they belong to. +type reading struct { + at time.Time + startTicks int64 + counters ProcessCounters +} + +// counterCache holds the previous pass's reading per PID. It is the only state +// the node keeps between passes besides the marker cache, and it is +// loss-harmless: a restart costs one interval of counters, never a total, and +// the next pass re-baselines (ADR 0003). +type counterCache struct { + last map[int]reading + seen map[int]struct{} +} + +func newCounterCache() *counterCache { + return &counterCache{last: map[int]reading{}, seen: map[int]struct{}{}} +} + +// delta reads one process's counters and returns the change since this cache's +// previous reading of the same process, or nil when there is nothing to compare +// against: the first pass that saw it, a PID that now belongs to a different +// process, or a counter that moved backwards. +func (c *counterCache) delta(procRoot string, pid int, now time.Time, status ProcessStatus) *CounterDelta { + cur, startTicks, ok := readProcessCounters(procRoot, pid, status) + if !ok { + return nil + } + c.seen[pid] = struct{}{} + prev, had := c.last[pid] + c.last[pid] = reading{at: now, startTicks: startTicks, counters: cur} + if !had || prev.startTicks != startTicks { + // A process seen for the first time, or a PID that has been reused. + // Neither has a previous reading of *this* process to subtract from. + return nil + } + elapsed := now.Sub(prev.at) + if elapsed <= 0 { + return nil + } + d := CounterDelta{ + ObservedNanos: elapsed.Nanoseconds(), + OnCPUNanos: cur.OnCPUNanos - prev.counters.OnCPUNanos, + RunDelayNanos: cur.RunDelayNanos - prev.counters.RunDelayNanos, + CPUUserNanos: cur.CPUUserNanos - prev.counters.CPUUserNanos, + CPUSystemNanos: cur.CPUSystemNanos - prev.counters.CPUSystemNanos, + MajorFaults: cur.MajorFaults - prev.counters.MajorFaults, + VoluntarySwitches: cur.VoluntarySwitches - prev.counters.VoluntarySwitches, + NonvoluntarySwitches: cur.NonvoluntarySwitches - prev.counters.NonvoluntarySwitches, + ReadBytes: cur.ReadBytes - prev.counters.ReadBytes, + WriteBytes: cur.WriteBytes - prev.counters.WriteBytes, + } + // A monotonic counter cannot fall. One that did means the reading is not of + // the process the previous one described, whatever the start time says, so + // the whole delta is dropped rather than one field being clamped. + if negative(d) { + return nil + } + return &d +} + +func negative(d CounterDelta) bool { + return d.OnCPUNanos < 0 || d.RunDelayNanos < 0 || d.CPUUserNanos < 0 || + d.CPUSystemNanos < 0 || d.MajorFaults < 0 || d.VoluntarySwitches < 0 || + d.NonvoluntarySwitches < 0 || d.ReadBytes < 0 || d.WriteBytes < 0 +} + +// startPass begins a pass; endPass drops every PID the pass did not see, so the +// cache holds one entry per live process rather than growing with every process +// the node has ever run. +func (c *counterCache) startPass() { clear(c.seen) } + +func (c *counterCache) endPass() { + for pid := range c.last { + if _, ok := c.seen[pid]; !ok { + delete(c.last, pid) + } + } +} + +// readProcessCounters assembles one reading from three files plus the two +// switch counts the status read already produced. It reports the process's +// start time separately: that value keys the reading rather than travelling +// with it, and it never leaves the node. +func readProcessCounters(procRoot string, pid int, status ProcessStatus) (ProcessCounters, int64, bool) { + utime, stime, majflt, startTicks, ok := readProcStat(procRoot, pid) + if !ok { + return ProcessCounters{}, 0, false + } + onCPU, runDelay := readSchedStat(procRoot, pid) + read, written := readProcIO(procRoot, pid) + return ProcessCounters{ + OnCPUNanos: onCPU, + RunDelayNanos: runDelay, + CPUUserNanos: utime * nanosPerTick, + CPUSystemNanos: stime * nanosPerTick, + MajorFaults: majflt, + VoluntarySwitches: status.VoluntarySwitches, + NonvoluntarySwitches: status.NonvoluntarySwitches, + ReadBytes: read, + WriteBytes: written, + }, startTicks, true +} + +// readProcStat takes four fields of /proc//stat by position. +// +// Field two is the executable's own name in parentheses, and it may itself +// contain spaces and parentheses — so the fields are counted from the last ')' +// rather than by splitting the line. That is also why the name never reaches a +// variable here: it is behind the cut, not filtered after it. +func readProcStat(procRoot string, pid int) (utime, stime, majflt, startTicks int64, ok bool) { + body, err := readSmallFile(filepath.Join(procRoot, strconv.Itoa(pid), "stat")) + if err != nil { + return 0, 0, 0, 0, false + } + end := strings.LastIndexByte(body, ')') + if end < 0 { + return 0, 0, 0, 0, false + } + // After the name, field 3 is the first: index i holds field i+3. + fields := strings.Fields(body[end+1:]) + at := func(field int) (int64, bool) { + i := field - 3 + if i < 0 || i >= len(fields) { + return 0, false + } + n, err := strconv.ParseInt(fields[i], 10, 64) + return n, err == nil && n >= 0 + } + majflt, okMaj := at(12) + utime, okU := at(14) + stime, okS := at(15) + startTicks, okStart := at(22) + if !okMaj || !okU || !okS || !okStart { + return 0, 0, 0, 0, false + } + return utime, stime, majflt, startTicks, true +} + +// readSchedStat takes the first two of the three numbers in +// /proc//schedstat: nanoseconds on CPU, and nanoseconds runnable but +// waiting for one. A kernel built without CONFIG_SCHEDSTATS has no such file, +// and both are then zero. +func readSchedStat(procRoot string, pid int) (onCPU, runDelay int64) { + body, err := readSmallFile(filepath.Join(procRoot, strconv.Itoa(pid), "schedstat")) + if err != nil { + return 0, 0 + } + fields := strings.Fields(body) + if len(fields) < 2 { + return 0, 0 + } + a, errA := strconv.ParseInt(fields[0], 10, 64) + b, errB := strconv.ParseInt(fields[1], 10, 64) + if errA != nil || errB != nil || a < 0 || b < 0 { + return 0, 0 + } + return a, b +} + +// readProcIO takes the two block-storage counters. The file also holds rchar +// and wchar, which count bytes the process asked for rather than bytes that +// moved, and neither is read: a finding about storage rests on what reached it. +func readProcIO(procRoot string, pid int) (read, written int64) { + f, err := os.Open(filepath.Join(procRoot, strconv.Itoa(pid), "io")) // #nosec G304 -- procRoot is an operator-set flag; pid is a live process + if err != nil { + return 0, 0 + } + defer func() { _ = f.Close() }() + scanner := bufio.NewScanner(io.LimitReader(f, maxCounterFileBytes)) + for scanner.Scan() { + key, value, ok := strings.Cut(scanner.Text(), ":") + if !ok { + continue + } + n, err := strconv.ParseInt(strings.TrimSpace(value), 10, 64) + if err != nil || n < 0 { + continue + } + switch key { + case "read_bytes": + read = n + case "write_bytes": + written = n + } + } + return read, written +} + +func readSmallFile(path string) (string, error) { + f, err := os.Open(path) // #nosec G304 -- path is built from an operator-set proc root and a live pid + if err != nil { + return "", err + } + defer func() { _ = f.Close() }() + body, err := io.ReadAll(io.LimitReader(f, maxCounterFileBytes)) + return string(body), err +} diff --git a/internal/nodescan/proccounters_test.go b/internal/nodescan/proccounters_test.go new file mode 100644 index 0000000..31314bf --- /dev/null +++ b/internal/nodescan/proccounters_test.go @@ -0,0 +1,215 @@ +package nodescan + +import ( + "os" + "path/filepath" + "strconv" + "testing" + "time" +) + +// A real stat line. Field two is the executable's name in parentheses, and this +// one contains both a space and a closing paren — the case that breaks every +// parser that splits the line on spaces. +const realWorldStat = `1 (checkout api (v2)) S 0 1 1 0 -1 4194560 91823 0 1841 0 4210 1337 0 0 20 0 17 0 8641234 2652856320 100534 18446744073709551615 1 1 0 0 0 0 0 0 0 0 0 0 17 3 0 0 0 0 0 0 0 0 0 0 0 0 0` + +const realWorldSchedStat = "48210934812 13440928374 91823\n" + +const realWorldIO = `rchar: 90128374 +wchar: 1209384 +syscr: 88123 +syscw: 4410 +read_bytes: 268435456 +write_bytes: 134217728 +cancelled_write_bytes: 0 +` + +func writeCounterFiles(t *testing.T, root string, pid int, stat, sched, procIO string) { + t.Helper() + dir := filepath.Join(root, strconv.Itoa(pid)) + if err := os.MkdirAll(dir, 0o750); err != nil { + t.Fatal(err) + } + for name, body := range map[string]string{"stat": stat, "schedstat": sched, "io": procIO} { + if body == "" { + continue + } + if err := os.WriteFile(filepath.Join(dir, name), []byte(body), 0o600); err != nil { + t.Fatal(err) + } + } +} + +// The name in field two is the identity this parser must not pick up, and +// counting fields from the last ')' is what keeps it behind the cut rather than +// filtered after it (ADR 0062 §1). +func TestStatFieldsAreCountedFromTheNameNotThroughIt(t *testing.T) { + root := t.TempDir() + writeCounterFiles(t, root, 1, realWorldStat, "", "") + + utime, stime, majflt, start, ok := readProcStat(root, 1) + if !ok { + t.Fatal("a stat line with a space and a paren in the name was not parsed") + } + if utime != 4210 || stime != 1337 { + t.Errorf("utime/stime = %d/%d ticks, want 4210/1337", utime, stime) + } + if majflt != 1841 { + t.Errorf("major faults = %d, want 1841", majflt) + } + if start != 8641234 { + t.Errorf("start ticks = %d, want 8641234", start) + } +} + +func TestSchedStatAndIOTakeOnlyWhatTheyNeed(t *testing.T) { + root := t.TempDir() + writeCounterFiles(t, root, 1, realWorldStat, realWorldSchedStat, realWorldIO) + + onCPU, runDelay := readSchedStat(root, 1) + if onCPU != 48210934812 || runDelay != 13440928374 { + t.Errorf("on cpu/run delay = %d/%d", onCPU, runDelay) + } + read, written := readProcIO(root, 1) + if read != 268435456 || written != 134217728 { + t.Errorf("read/write = %d/%d, want the block-storage counters", read, written) + } + // rchar and wchar count bytes asked for rather than bytes that moved, and + // they sit directly above the two that are read. + if read == 90128374 || written == 1209384 { + t.Error("rchar/wchar were read as the block-storage counters") + } +} + +// A kernel without CONFIG_SCHEDSTATS has no such file, and a container may deny +// the io file. Both are zero rather than a failed reading: the rest of the +// counters are still worth having. +func TestMissingCounterFilesAreZeroNotFailure(t *testing.T) { + root := t.TempDir() + writeCounterFiles(t, root, 1, realWorldStat, "", "") + + c, _, ok := readProcessCounters(root, 1, ProcessStatus{}) + if !ok { + t.Fatal("a process with only stat produced no reading") + } + if c.OnCPUNanos != 0 || c.ReadBytes != 0 { + t.Errorf("absent files produced %+v, want zeros", c) + } + if c.CPUUserNanos != 4210*nanosPerTick { + t.Errorf("cpu user = %d, want the stat reading converted from ticks", c.CPUUserNanos) + } +} + +// statWith rewrites the counter fields of the fixture: utime, stime and the +// start time, which is what says whether two readings are of one process. +func statWith(utime, stime, startTicks int64) string { + return "1 (checkout api (v2)) S 0 1 1 0 -1 4194560 91823 0 1841 0 " + + strconv.FormatInt(utime, 10) + " " + strconv.FormatInt(stime, 10) + + " 0 0 20 0 17 0 " + strconv.FormatInt(startTicks, 10) + " 2652856320 100534" +} + +// The first pass has nothing to subtract from, and says so by producing no +// delta at all rather than a delta equal to the whole life of the process. +func TestTheFirstReadingOfAProcessIsNotADelta(t *testing.T) { + root := t.TempDir() + writeCounterFiles(t, root, 1, statWith(100, 50, 8641234), realWorldSchedStat, realWorldIO) + + c := newCounterCache() + now := time.Unix(1000, 0) + if d := c.delta(root, 1, now, ProcessStatus{}); d != nil { + t.Errorf("first reading produced a delta %+v", d) + } + writeCounterFiles(t, root, 1, statWith(160, 70, 8641234), realWorldSchedStat, realWorldIO) + d := c.delta(root, 1, now.Add(60*time.Second), ProcessStatus{}) + if d == nil { + t.Fatal("the second reading of the same process produced no delta") + } + if d.CPUUserNanos != 60*nanosPerTick || d.CPUSystemNanos != 20*nanosPerTick { + t.Errorf("cpu delta = %d/%d, want the difference of the two readings", + d.CPUUserNanos, d.CPUSystemNanos) + } + if d.ObservedNanos != int64(60*time.Second) { + t.Errorf("observed = %d, want the wall-clock time between the readings", d.ObservedNanos) + } +} + +// A PID is reused. Two readings under one number then belong to two processes, +// and subtracting them yields a number about neither — the failure this cache's +// start-time key exists to prevent. +func TestAReusedPIDIsNotSubtractedFromItsPredecessor(t *testing.T) { + root := t.TempDir() + writeCounterFiles(t, root, 1, statWith(5000, 900, 8641234), realWorldSchedStat, realWorldIO) + + c := newCounterCache() + now := time.Unix(1000, 0) + c.delta(root, 1, now, ProcessStatus{}) + + // The same PID, a lower counter, and a different start time: a new process. + writeCounterFiles(t, root, 1, statWith(12, 3, 9000000), realWorldSchedStat, realWorldIO) + if d := c.delta(root, 1, now.Add(60*time.Second), ProcessStatus{}); d != nil { + t.Errorf("a reused PID produced a delta %+v against its predecessor", d) + } + // And the pass after that compares against the new process, not the old. + writeCounterFiles(t, root, 1, statWith(20, 5, 9000000), realWorldSchedStat, realWorldIO) + d := c.delta(root, 1, now.Add(120*time.Second), ProcessStatus{}) + if d == nil || d.CPUUserNanos != 8*nanosPerTick { + t.Errorf("delta after re-baselining = %+v, want 8 ticks of user time", d) + } +} + +// A monotonic counter cannot fall. One that did means the reading is not of the +// process the last one described, whatever the start time claims, so the whole +// delta goes rather than one field being clamped to zero. +func TestACounterThatFellDropsTheWholeDelta(t *testing.T) { + root := t.TempDir() + writeCounterFiles(t, root, 1, statWith(5000, 900, 8641234), realWorldSchedStat, realWorldIO) + + c := newCounterCache() + now := time.Unix(1000, 0) + c.delta(root, 1, now, ProcessStatus{}) + + writeCounterFiles(t, root, 1, statWith(4000, 900, 8641234), realWorldSchedStat, realWorldIO) + if d := c.delta(root, 1, now.Add(60*time.Second), ProcessStatus{}); d != nil { + t.Errorf("a counter that fell produced a delta %+v", d) + } +} + +// The cache holds one entry per live process rather than growing with every +// process the node has ever run. +func TestTheCacheForgetsProcessesThatAreGone(t *testing.T) { + root := t.TempDir() + writeCounterFiles(t, root, 1, statWith(100, 50, 8641234), realWorldSchedStat, realWorldIO) + writeCounterFiles(t, root, 2, statWith(100, 50, 8641234), realWorldSchedStat, realWorldIO) + + c := newCounterCache() + now := time.Unix(1000, 0) + c.startPass() + c.delta(root, 1, now, ProcessStatus{}) + c.delta(root, 2, now, ProcessStatus{}) + c.endPass() + if len(c.last) != 2 { + t.Fatalf("cache holds %d entries after a pass that saw two", len(c.last)) + } + + c.startPass() + c.delta(root, 1, now.Add(time.Minute), ProcessStatus{}) + c.endPass() + if len(c.last) != 1 { + t.Errorf("cache holds %d entries after a pass that saw one", len(c.last)) + } +} + +// The two switch counts come from the status file the scanner already read, so +// they arrive through the reading rather than through a fourth open. +func TestSwitchCountsRideTheStatusRead(t *testing.T) { + root := t.TempDir() + writeCounterFiles(t, root, 1, statWith(100, 50, 8641234), realWorldSchedStat, realWorldIO) + + c := newCounterCache() + now := time.Unix(1000, 0) + c.delta(root, 1, now, ProcessStatus{VoluntarySwitches: 1000, NonvoluntarySwitches: 10}) + d := c.delta(root, 1, now.Add(time.Minute), ProcessStatus{VoluntarySwitches: 1600, NonvoluntarySwitches: 42}) + if d == nil || d.VoluntarySwitches != 600 || d.NonvoluntarySwitches != 32 { + t.Errorf("switch deltas = %+v, want 600 and 32", d) + } +} diff --git a/internal/nodescan/procstatus.go b/internal/nodescan/procstatus.go index 3582358..69ce803 100644 --- a/internal/nodescan/procstatus.go +++ b/internal/nodescan/procstatus.go @@ -47,6 +47,11 @@ type ProcessStatus struct { // blocking syscall it cannot park, so a number far above GOMAXPROCS is the // signature of blocking or cgo work. Threads int + // VoluntarySwitches and NonvoluntarySwitches are monotonic since the + // process started, unlike everything above them. They are read here because + // this file is already open; what happens to them is in ADR 0062. + VoluntarySwitches int64 + NonvoluntarySwitches int64 } // ReadProcessStatus reads the allow-listed fields of one process's status file. @@ -84,11 +89,24 @@ func parseProcessStatus(r io.Reader) ProcessStatus { if n, err := strconv.Atoi(value); err == nil && n > 0 { out.Threads = n } + case "voluntary_ctxt_switches": + out.VoluntarySwitches = countOf(value) + case "nonvoluntary_ctxt_switches": + out.NonvoluntarySwitches = countOf(value) } } return out } +// countOf parses a plain count, dropping anything that is not one. +func countOf(value string) int64 { + n, err := strconv.ParseInt(value, 10, 64) + if err != nil || n < 0 { + return 0 + } + return n +} + // peakBytes converts the kernel's "12345 kB" to bytes. A value in any other // unit is not what this parser assumes and is dropped whole rather than // misread by a factor of 1024. diff --git a/internal/nodescan/scan.go b/internal/nodescan/scan.go index 82e6f4a..dc679d4 100644 --- a/internal/nodescan/scan.go +++ b/internal/nodescan/scan.go @@ -15,6 +15,7 @@ import ( "path/filepath" "strconv" "strings" + "time" ) // BinaryInfo is what the scanner extracts about one kept Go process. It is the @@ -61,6 +62,10 @@ type BinaryInfo struct { CPUsAllowed int `json:"cpus_allowed,omitempty"` // Footprint is what the process holds now, beside the mark above (ADR 0061). Footprint Footprint `json:"footprint,omitzero"` + // Counters is what changed since this node's previous pass saw this same + // process, with the interval it covers. Absent on the first pass that sees + // a process, and whenever there is nothing to subtract from (ADR 0062). + Counters *CounterDelta `json:"counters,omitempty"` // PodUID and ContainerID come from the process cgroup; either may be empty // for a host process outside the kubepods hierarchy. PodUID string `json:"pod_uid,omitempty"` @@ -145,12 +150,18 @@ type Scanner struct { procRoot string filter *ModuleFilter marks *markCache + counters *counterCache } // NewScanner builds a scanner rooted at procRoot (normally "/proc"; a fixture // tree in tests) using filter to classify module paths. func NewScanner(procRoot string, filter *ModuleFilter) *Scanner { - return &Scanner{procRoot: procRoot, filter: filter, marks: newMarkCache()} + return &Scanner{ + procRoot: procRoot, + filter: filter, + marks: newMarkCache(), + counters: newCounterCache(), + } } // Scan performs one full pass over the process tree, restricted to the pods @@ -167,15 +178,20 @@ func (s *Scanner) Scan(scope Scope) (Result, error) { } var res Result s.marks.startPass() + s.counters.startPass() + // One instant for the whole pass, so every delta in it is measured against + // the same clock rather than against when its process happened to be read. + now := time.Now() for _, entry := range entries { pid, ok := pidFromName(entry.Name()) if !ok { continue } res.Counters.ProcessesScanned++ - s.scanPID(pid, scope, &res) + s.scanPID(pid, scope, now, &res) } s.marks.endPass() + s.counters.endPass() return res, nil } @@ -186,7 +202,7 @@ func (s *Scanner) Scan(scope Scope) (Result, error) { // executable left unopened and its module path never extracted: nothing about it // is collected, rather than collected and dropped (CLAUDE.md invariant 4). Only // the aggregate count moves. -func (s *Scanner) scanPID(pid int, scope Scope, res *Result) { +func (s *Scanner) scanPID(pid int, scope Scope, now time.Time, res *Result) { binding := s.binding(pid) if !scope.Admits(binding.PodUID) { res.Counters.FilteredScope++ @@ -222,6 +238,7 @@ func (s *Scanner) scanPID(pid int, scope Scope, res *Result) { status := ReadProcessStatus(s.procRoot, pid) memory := ReadProcessMemory(s.procRoot, pid) files := ReadProcessFiles(s.procRoot, pid) + counters := s.counters.delta(s.procRoot, pid, now, status) hasPprof := s.marks.lookup(exePath) ports := ReadListeningPorts(s.procRoot, pid) @@ -247,6 +264,7 @@ func (s *Scanner) scanPID(pid int, scope Scope, res *Result) { OpenFiles: files.Open, OpenFilesLimit: files.Limit, }, + Counters: counters, PodUID: binding.PodUID, ContainerID: binding.ContainerID, }) diff --git a/internal/sink/registry.go b/internal/sink/registry.go index 1ecd02c..acdf077 100644 --- a/internal/sink/registry.go +++ b/internal/sink/registry.go @@ -149,6 +149,16 @@ var registry = []PayloadKind{ Delivery: DeliverySupersedes, ADR: "0052", }, + { + // Measured, and the one kind whose records are differences rather than + // readings: each is what changed since the node's previous pass, with + // the interval it covers (ADR 0062). + Kind: "process_counters", + Source: SourceMeasured, + NaturalKey: "the kind itself (one per cluster)", + Delivery: DeliverySupersedes, + ADR: "0062", + }, { // Structural, not measured: a bound port is a fact about how the // workload is configured, and reading it from the process rather than diff --git a/internal/sink/spool.go b/internal/sink/spool.go index 5191000..77d627f 100644 --- a/internal/sink/spool.go +++ b/internal/sink/spool.go @@ -249,6 +249,21 @@ type listeningPortsPayload struct { Records []inventory.PortRecord `json:"records"` } +// processCountersPayload is what each collected workload container's processes +// did over the most recent interval each node measured: a superseding batch +// under a fixed key, like the peaks and ports beside it (ADR 0062). +// +// CapturedAt dates the assembly, and it is not the window. Each record carries +// its own `observed_nanos`, which is the process-time the deltas cover — the +// denominator of a rate, and the only thing that makes the sums comparable +// between two flushes. +type processCountersPayload struct { + Kind string `json:"kind"` + Source string `json:"source"` + CapturedAt time.Time `json:"captured_at"` + Records []inventory.CounterRecord `json:"records"` +} + // ProfileDrops is how much of a pulled profile the allow-list removed. It is the // count and never the identity: which functions were redacted is the thing the // filter exists to keep inside the cluster (CLAUDE.md invariant 6). @@ -735,6 +750,23 @@ func (s *Spool) WriteProcessPeaks(capturedAt time.Time, records []inventory.Peak return s.write("process-peaks.json", payload) } +// WriteProcessCounters writes the current counter records as one superseding +// batch. Records must arrive sorted (Store.CounterSnapshot sorts them) so the +// payload bytes are stable — the golden contract. +// +// A cluster where no node has produced a second reading yet writes an empty +// record list rather than nothing: "every process was seen for the first time" +// is an answer, and it is the one that explains an empty first flush. +func (s *Spool) WriteProcessCounters(capturedAt time.Time, records []inventory.CounterRecord) error { + payload := processCountersPayload{ + Kind: "process_counters", + Source: SourceMeasured, + CapturedAt: capturedAt.UTC(), + Records: records, + } + return s.write("process-counters.json", payload) +} + // WriteListeningPorts writes where every collected workload container accepts // connections, as of capturedAt. One file, superseding, for the reason // process-peaks is one file: the batch is the answer, and a record per container diff --git a/internal/sink/spool_test.go b/internal/sink/spool_test.go index bd4ed3c..4456867 100644 --- a/internal/sink/spool_test.go +++ b/internal/sink/spool_test.go @@ -638,6 +638,50 @@ func fixedPeaks() []inventory.PeakRecord { } } +// The counters fixture is the finding in one payload: `web/app` spends nearly a +// third of its runnable time waiting for a CPU while barely using the CPU it +// gets, and `db` is doing real disk work. Both are stated as the numbers they +// are made of, never as a ratio (ADR 0062 §3). +func fixedCounters() []inventory.CounterRecord { + return []inventory.CounterRecord{ + { + CounterKey: inventory.CounterKey{ + Key: inventory.Key{ + Namespace: "shop", WorkloadKind: "Deployment", + WorkloadName: "web", Container: "app", + }, + ImageDigest: "sha256:2222222222222222222222222222222222222222222222222222222222222222", + }, + Processes: 2, ObservedNanos: 120 * int64(time.Second), + OnCPUNanos: 9 * int64(time.Second), RunDelayNanos: 4 * int64(time.Second), + CPUUserNanos: 7 * int64(time.Second), CPUSystemNanos: 2 * int64(time.Second), + MajorFaults: 0, VoluntarySwitches: 48210, NonvoluntarySwitches: 3122, + ReadBytes: 0, WriteBytes: 4096, + }, + { + CounterKey: inventory.CounterKey{ + Key: inventory.Key{ + Namespace: "shop", WorkloadKind: "StatefulSet", + WorkloadName: "db", Container: "db", + }, + }, + Processes: 1, ObservedNanos: 60 * int64(time.Second), + OnCPUNanos: 21 * int64(time.Second), RunDelayNanos: 90 * int64(time.Millisecond), + CPUUserNanos: 12 * int64(time.Second), CPUSystemNanos: 9 * int64(time.Second), + MajorFaults: 1841, VoluntarySwitches: 9004, NonvoluntarySwitches: 61, + ReadBytes: 268435456, WriteBytes: 134217728, + }, + } +} + +func TestGoldenProcessCountersPayload(t *testing.T) { + s, dir := newTestSpool(t) + if err := s.WriteProcessCounters(capturedAt, fixedCounters()); err != nil { + t.Fatal(err) + } + checkGolden(t, filepath.Join(dir, "process-counters.json"), "process-counters.golden.json") +} + func TestGoldenProcessPeaksPayload(t *testing.T) { s, dir := newTestSpool(t) if err := s.WriteProcessPeaks(capturedAt, fixedPeaks()); err != nil { diff --git a/internal/sink/testdata/process-counters.golden.json b/internal/sink/testdata/process-counters.golden.json new file mode 100644 index 0000000..83fc6ef --- /dev/null +++ b/internal/sink/testdata/process-counters.golden.json @@ -0,0 +1,42 @@ +{ + "kind": "process_counters", + "source": "measured", + "captured_at": "2026-08-06T10:12:00Z", + "records": [ + { + "namespace": "shop", + "workload_kind": "Deployment", + "workload_name": "web", + "container": "app", + "image_digest": "sha256:2222222222222222222222222222222222222222222222222222222222222222", + "processes": 2, + "observed_nanos": 120000000000, + "on_cpu_nanos": 9000000000, + "run_delay_nanos": 4000000000, + "cpu_user_nanos": 7000000000, + "cpu_system_nanos": 2000000000, + "major_faults": 0, + "voluntary_switches": 48210, + "nonvoluntary_switches": 3122, + "read_bytes": 0, + "write_bytes": 4096 + }, + { + "namespace": "shop", + "workload_kind": "StatefulSet", + "workload_name": "db", + "container": "db", + "processes": 1, + "observed_nanos": 60000000000, + "on_cpu_nanos": 21000000000, + "run_delay_nanos": 90000000, + "cpu_user_nanos": 12000000000, + "cpu_system_nanos": 9000000000, + "major_faults": 1841, + "voluntary_switches": 9004, + "nonvoluntary_switches": 61, + "read_bytes": 268435456, + "write_bytes": 134217728 + } + ] +} diff --git a/test/e2e/inventory_test.go b/test/e2e/inventory_test.go index 15f84d3..f16470a 100644 --- a/test/e2e/inventory_test.go +++ b/test/e2e/inventory_test.go @@ -44,6 +44,8 @@ const nodeMetadataSpoolPath = spoolDir + "/node-metadata.json" const processPeaksSpoolPath = spoolDir + "/process-peaks.json" +const processCountersSpoolPath = spoolDir + "/process-counters.json" + const listeningPortsSpoolPath = spoolDir + "/listening-ports.json" const coverageSpoolPath = spoolDir + "/collection-coverage.json" @@ -165,6 +167,7 @@ func TestGoInventoryEndToEnd(t *testing.T) { } checkSampleBuild(ctx, t, config, clientset, ns, controllerPod, rec.ImageDigest) checkSamplePeak(ctx, t, config, clientset, ns, controllerPod, rec.ImageDigest) + checkSampleCounters(ctx, t, config, clientset, ns, controllerPod, rec.ImageDigest) checkSamplePorts(ctx, t, config, clientset, ns, controllerPod, rec.ImageDigest) checkEndpointConfirmed(ctx, t, config, clientset, ns, controllerPod) checkProfilePulled(ctx, t, config, clientset, ns, controllerPod) @@ -481,6 +484,81 @@ func checkSamplePeak(ctx context.Context, t *testing.T, config *rest.Config, cs } } +// checkSampleCounters asserts that the subtraction really happened on a node: +// the first pass produces no delta at all, so a record here proves a second pass +// compared against the first for the same process (ADR 0062). +// +// It polls for longer than its siblings for that reason — one scan interval must +// elapse before the payload can carry anything. +func checkSampleCounters(ctx context.Context, t *testing.T, config *rest.Config, cs kubernetes.Interface, ns, pod, digest string) { + t.Helper() + deadline := time.Now().Add(4 * time.Minute) + for { + raw, ok := readSpoolFile(ctx, t, config, cs, ns, pod, processCountersSpoolPath) + if ok { + var payload struct { + Kind string `json:"kind"` + Source string `json:"source"` + Records []struct { + WorkloadName string `json:"workload_name"` + ImageDigest string `json:"image_digest"` + Processes int `json:"processes"` + ObservedNanos int64 `json:"observed_nanos"` + OnCPUNanos int64 `json:"on_cpu_nanos"` + RunDelayNanos int64 `json:"run_delay_nanos"` + CPUUserNanos int64 `json:"cpu_user_nanos"` + CPUSysNanos int64 `json:"cpu_system_nanos"` + VolSwitches int64 `json:"voluntary_switches"` + } `json:"records"` + } + if err := json.Unmarshal([]byte(raw), &payload); err != nil { + t.Fatalf("process counters payload not valid JSON: %v", err) + } + for _, r := range payload.Records { + if r.WorkloadName != "goworkload" { + continue + } + if payload.Kind != "process_counters" || payload.Source != "measured" { + t.Errorf("kind/source = %q/%q, want process_counters/measured", payload.Kind, payload.Source) + } + // The interval is what makes the rest a rate, so it must be + // real and it must be at least one scan interval. + if r.ObservedNanos < int64(20*time.Second) { + t.Errorf("observed_nanos = %d, want at least one scan interval", r.ObservedNanos) + } + if r.Processes < 1 { + t.Errorf("processes = %d, want the one that had a delta", r.Processes) + } + // The sample burns CPU, so it used some; and no process can be + // on CPU for longer than it was observed times its processes. + if r.CPUUserNanos+r.CPUSysNanos <= 0 { + t.Errorf("cpu delta = %d, want a workload that ran", r.CPUUserNanos+r.CPUSysNanos) + } + if r.OnCPUNanos > r.ObservedNanos*int64(r.Processes) { + t.Errorf("on_cpu %d exceeds the process-time observed %d; the interval is not this delta's", + r.OnCPUNanos, r.ObservedNanos*int64(r.Processes)) + } + if r.RunDelayNanos < 0 || r.VolSwitches < 0 { + t.Errorf("a counter delta is negative: run_delay %d, switches %d", r.RunDelayNanos, r.VolSwitches) + } + if r.ImageDigest != digest { + t.Errorf("image_digest = %q, want %q — an interval belongs to the build that spent it", r.ImageDigest, digest) + } + t.Logf("counters for %s: %d process(es) over %v, %v on cpu (%v user, %v system), %v waiting", + r.WorkloadName, r.Processes, time.Duration(r.ObservedNanos), + time.Duration(r.OnCPUNanos), time.Duration(r.CPUUserNanos), + time.Duration(r.CPUSysNanos), time.Duration(r.RunDelayNanos)) + return + } + } + if time.Now().After(deadline) { + t.Errorf("process_counters payload never carried the sample workload") + return + } + time.Sleep(10 * time.Second) + } +} + // checkSamplePorts asserts that the ports the sample really binds arrived, with // the reachability of each. The sample listens on 0.0.0.0:6060 and on // 127.0.0.1:9090 and declares neither in its spec, so this also proves the read