feat: observation completeness on measured payloads (ADR 0013)

cmd/agent/main.go

@@ -224,22 +224,26 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac
 		pc := cfg.Profiling.Normalized()
 		targetsPublisher = targeting.NewPublisher(cfg.Profiling.EligibleNamespaces, cfg.Profiling.EligibleWorkloads, pc.TopN)
 	}
-	usagePoller := collector.NewUsagePoller(clientset, nodeWatcher.Names, podWatcher,
+	// Declared before construction so the callbacks can read the poller's own
+	// observation state; they run on the poll goroutine, and Observation() is
+	// synchronized regardless.
+	var usagePoller *collector.UsagePoller
+	usagePoller = collector.NewUsagePoller(clientset, nodeWatcher.Names, podWatcher,
 		func(sequence int64, records []*rollup.Record) {
 			logRecords("usage rollup snapshot", sequence, records)
 			if targetsPublisher != nil {
 				targetsPublisher.Publish(records)
 			}
 			if spool != nil {
-				if err := spool.WriteUsageSnapshot(sequence, records); err != nil {
+				if err := spool.WriteUsageSnapshot(sequence, records, usagePoller.Observation()); err != nil {
 					logger.Error("spooling usage snapshot", "error", err)
 				}
 			}
 		},
 		func(records []*rollup.Record) {
 			logRecords("usage rollup closed", 0, records)
 			if spool != nil {
-				if err := spool.WriteClosedWindows(records); err != nil {
+				if err := spool.WriteClosedWindows(records, usagePoller.Observation()); err != nil {
 					logger.Error("spooling closed windows", "error", err)
 				}
 			}

docs/adr/0013-observation-completeness.md

@@ -0,0 +1,98 @@
+# 0013. Observation completeness: measured payloads carry what the agent saw
+
+Date: 2026-08-20
+Status: Accepted
+
+## Context
+
+[ADR 0012](0012-payload-registry-and-provenance.md) fixed the provenance classes
+and recorded the gap it left open: the measured payloads did not yet say
+anything about their own collection. Three consequences of that gap were
+already visible in the code.
+
+**A zero was ambiguous.** `ThrottledPeriods` and `TotalPeriods` were plain
+counters with a zero default. A container that was never throttled and a
+container whose node's `/metrics/cadvisor` scrape failed for the whole window
+produced byte-identical records. The same held for both PSI stall counters. The
+cluster-wide signal set — which existed, but only in a log line — could say "this
+cluster exposes throttling" and still not resolve the question for any single
+record, because a scrape fails per node and per path, not per cluster.
+
+**A partial window looked like a quiet one.** A record's `Samples` counted rate
+observations, but nothing said how much of the window those observations spanned.
+A container that started fifty minutes into an hour and one whose node was
+unreachable for fifty minutes both produced a small number with no way to tell
+which.
+
+**Only the agent can know any of this.** From outside the cluster a failed scrape
+is unrecoverable after the fact — there is no later query that reveals it. This
+is a fact about collection, not a judgment about the data, so it belongs in the
+payload rather than in a log the backend never sees.
+
+Separately, the throttling counters were keyed differently from every other
+kubelet counter, and that was a defect rather than a decision — see below.
+
+## Decision
+
+**1. Per-signal sample counts on the record.** `CPU.ThrottlingSamples`,
+`CPU.PSISamples` and `Memory.PSISamples` count the observations that actually
+carried each signal. Zero means never observed; non-zero makes the accompanying
+zero total mean "observed, and it was quiet". The counts are attributed to the
+window holding the interval's last instant — the same rule rate samples already
+follow — and add under merge, so the merge property is unchanged.
+
+**2. Per-record coverage.** `CPU.CoveredNanoseconds` is the summed duration of
+the counter intervals attributed to a record, split across windows exactly as
+the counters themselves are. Against `WindowSeconds` it answers "how much of this
+window did we see?" without the agent deciding what a sufficient answer is.
+
+**3. Cluster-wide collection state on the payload.** Measured payloads carry an
+`observation` block: the poll cadence, cumulative counts of kubelet requests
+attempted and failed, and the signal set this cluster exposes. Both kubelet
+paths (`/stats/summary` and `/metrics/cadvisor`) count as separate requests
+because they fail independently — and which of them failed is resolved per record
+by the sample counts of (1), not here. The counters are cumulative since agent
+start, like every other counter this agent reports: the backend takes
+differences between deliveries, and no window-scoped bookkeeping is invented for
+them.
+
+**4. Provenance completes.** `usage_snapshot` and `usage_window` declare
+`source: measured`, `oom_kill` declares `journal`, `go_inventory` declares
+`structural`. The registry in ADR 0012 §1 is now fully realized in the bytes; the
+gap that ADR recorded is closed.
+
+**5. Throttling keys on the pod UID.** The cAdvisor exposition labels containers
+by pod *name*, and the poller kept its counter baselines under that name while
+CPU and memory kept theirs under the pod UID. A name is not an identity: a
+StatefulSet recreates a pod under the same name with a new UID, and the
+replacement's counters were then compared against a dead container's baseline —
+dropped as a phantom counter reset, or, when the new counters happened to exceed
+the old baseline, stretched into one continuous run. The exposition's name is now
+resolved to a UID before any state is kept, so all three kubelet signals share
+one identity.
+
+## Consequences
+
+**Easier.** A backend can state what a report rests on and refuse to conclude
+from a record it did not see enough of. "Throttling is not a problem here" is
+now distinguishable from "we could not observe throttling here", which is the
+difference between an answer and a guess. Coverage also carries the
+container-lifetime case for free: a container that ran for ten minutes of the
+window covers ten minutes of it, without the agent modelling lifetimes.
+
+**Harder / given up.** The golden bytes of every usage, OOM and inventory payload
+changed — one migration, taken deliberately and all at once rather than field by
+field later. Records grew four `int64`s. The observation block is cluster-wide,
+so it cannot attribute a failure to a node: it says three scrapes failed, not
+which node lost them. Per-node attribution would mean node-scoped payloads that
+nothing yet needs; the per-record sample counts already answer the question that
+matters for any given record.
+
+**Deliberately not done.** No "expected sample count" is reported. It would
+require the agent to model when a container was supposed to exist, which is a
+judgment; `CoveredNanoseconds` against the window length is the same information
+without the modelling. No reason string for non-coverage: the agent reports the
+counters, and the backend renders the cause.
+
+This ADR records decisions already implemented in the same pull request, per the
+process in [`README.md`](README.md).

docs/adr/README.md

@@ -52,3 +52,4 @@ stays in place.
 - [0010. Node → controller inventory channel: node-initiated HTTP, projected-token auth, local JWKS validation](0010-node-to-controller-inventory-channel.md)
 - [0011. On-node eBPF CPU profiling: embedded profiler, node-side symbolization, config-bounded targeting query](0011-node-ebpf-cpu-profiling.md)
 - [0012. Payload registry, provenance discriminator, and metadata delivery](0012-payload-registry-and-provenance.md)
+- [0013. Observation completeness: measured payloads carry what the agent saw](0013-observation-completeness.md)

docs/backend-requirements.md

@@ -100,10 +100,11 @@ The agent ships, per cluster:
   configuration, carrying no configuration content. Every upload is
   attributable to the configuration that produced it, so stability checks
   can distinguish "the data changed" from "the filters changed"
-- agent self-info: version, install profile, effective RBAC summary, and
-  the active usage signal set (which kubelet signals — e.g. PSI — this
-  cluster exposes and the agent collects), so every report can state what
-  its findings do and do not rest on
+- agent self-info: version, install profile, and effective RBAC summary, so
+  every report can state what its findings do and do not rest on. The active
+  usage signal set (which kubelet signals — e.g. PSI — this cluster exposes)
+  travels in each measured payload's `observation` block rather than here, so
+  it is attached to the data it qualifies (ADR 0013)
 
 All payloads are denominated in **resource units** (core-hours, GiB-hours,
 counts, ratios). The protocol has no monetary fields: the agent knows nothing
@@ -113,16 +114,29 @@ from a versioned pricing snapshot.
 Every payload declares a `kind`, and carries its **provenance** in a `source`
 field: `structural` (read from a spec or status), `measured` (polled from an
 instrument), `journal` (derived from object history), or `sampled`
-(statistical). The field ships today on `workload_metadata`, `node_metadata`
-and `ebpf_profile`; the usage and event payloads gain it with the
-observation-completeness work, and their classes are already fixed by the
-registry, so the backend can map a kind to its provenance either way. The
+(statistical). The
 backend MUST NOT merge payloads of different provenance under one natural key:
 a declared value and a measured one are different kinds of claim, and
 collapsing them yields a confident wrong answer with no failing test to show
 for it. The registry of kinds, their natural keys, and their delivery
 discipline is [ADR 0012](adr/0012-payload-registry-and-provenance.md).
 
+**Measured payloads state what was observed, and the backend MUST use it**
+([ADR 0013](adr/0013-observation-completeness.md)). Each usage payload carries an
+`observation` block — poll cadence, cumulative counts of kubelet requests
+attempted and failed, and the signal set this cluster exposes — and each record
+carries its own completeness:
+
+- `cpu.covered_nanoseconds` — how much of the window this container was actually
+  observed for. Against `window_seconds` it bounds what the record can support.
+- `cpu.throttling_samples`, `cpu.psi_samples`, `memory.psi_samples` — how many
+  observations carried each signal. **A zero total with a zero sample count means
+  "never observed", not "zero".** A backend that reads the first as the second
+  will report a throttling-free workload on the strength of a failed scrape.
+
+The counters are cumulative since agent start; the backend takes differences
+between deliveries, exactly as it does for every other counter here.
+
 ### Delivery semantics
 
 - **At-least-once.** The agent retransmits anything not acknowledged. The

docs/security.md

@@ -379,6 +379,24 @@ identity never appears in either payload. Every payload declares its
 provenance (`source`), so declared values can never be silently merged with
 measured or sampled ones.
 
+### What the agent says about its own collection (ADR 0013)
+
+Usage payloads carry, alongside the numbers, what the agent actually observed
+while producing them. This is metadata about the agent's own polling — it
+describes no workload and identifies nothing in your cluster:
+
+- an `observation` block per payload: the poll cadence, cumulative counts of
+  kubelet requests attempted and failed, and which kubelet signals this cluster
+  exposes (e.g. whether PSI is available);
+- per record: how much of the window that container was observed for, and how
+  many observations carried each signal.
+
+The purpose is honesty about gaps. Without it, a container that was never
+throttled and a container whose node could not be scraped produce identical
+records, and an analysis outside the cluster cannot tell them apart — the
+distinction is unrecoverable once the moment has passed. No pod, node, or
+workload is named in this data; the failure counts are cluster-wide totals.
+
 ### Profile filtering (applies on the node, before egress — ADR 0011)
 
 - Stack frames are filtered by Go module path on the node that captured them.

internal/collector/cadvisor.go

@@ -24,8 +24,10 @@ const (
 	metricStartTime        = "container_start_time_seconds"
 )
 
-// cadvisorKey identifies a container in the cAdvisor exposition, which
-// labels by name, not UID.
+// cadvisorKey identifies a container as the cAdvisor exposition labels it —
+// by pod name, not UID. It is the parse-side key only: before any counter
+// state is kept, the name is resolved to a pod UID so that throttling shares
+// the identity CPU and memory already use (trackerKey).
 type cadvisorKey struct {
 	namespace string
 	pod       string
@@ -160,7 +162,7 @@ func (p *UsagePoller) ingestCadvisor(samples map[cadvisorKey]*cadvisorSample, no
 		if !sample.hasThrottled || !sample.hasPeriods {
 			continue
 		}
-		workload, ok := p.pods.LookupPodByName(key.namespace, key.pod)
+		uid, workload, ok := p.pods.LookupPodByName(key.namespace, key.pod)
 		if !ok {
 			continue
 		}
@@ -176,10 +178,14 @@ func (p *UsagePoller) ingestCadvisor(samples map[cadvisorKey]*cadvisorSample, no
 			WorkloadName: workload.Name,
 			Container:    key.container,
 		}
-		st := p.throttle[key]
+		// Keyed on the pod UID, not the scraped name: a recreated pod reuses
+		// its name, and a name-keyed baseline would let the dead pod's counters
+		// widen the new container's first delta.
+		tk := trackerKey{pod: uid, container: key.container}
+		st := p.throttle[tk]
 		if st == nil {
 			st = &throttleState{}
-			p.throttle[key] = st
+			p.throttle[tk] = st
 		}
 		st.lastSeen = now
 

internal/collector/cadvisor_test.go

@@ -105,6 +105,77 @@ func TestThrottlingDeltaRules(t *testing.T) {
 	}
 }
 
+// The cAdvisor exposition labels by pod name, and a name is not an identity: a
+// StatefulSet recreates its pod under the same name with a new UID. Keying the
+// counter baseline on the name made the two pods share one baseline, so the
+// replacement's counters were compared against a dead container's — either
+// silently dropped as a "counter reset", or, when the new counters happened to
+// exceed the old baseline, attributed as one continuous run. Keyed on the UID,
+// each container is its own series and the replacement is attributed from its
+// own start, exactly as any first observation is.
+func TestThrottlingDoesNotCrossPodRecreation(t *testing.T) {
+	resolver := stubResolver{
+		"uid-old": {namespace: "shop", name: "web-abc", workload: WorkloadRef{Kind: "StatefulSet", Name: "web"}},
+	}
+	p := testPoller(resolver)
+	first := usageTestStart.Add(30 * time.Second)
+	second := first.Add(30 * time.Second)
+	newPodStart := second.Add(-10 * time.Second)
+
+	ingest := func(throttled, periods uint64, at, start, now time.Time) {
+		t.Helper()
+		samples, err := parseCadvisor(strings.NewReader(fixture(throttled, periods, at, start)))
+		if err != nil {
+			t.Fatal(err)
+		}
+		p.ingestCadvisor(samples, now)
+	}
+
+	// The original pod runs and accumulates a large baseline.
+	ingest(400, 1000, first, usageTestStart, first)
+
+	// It is replaced: same namespace and name, new UID, a container that has
+	// been up for ten seconds with counters of its own.
+	delete(resolver, "uid-old")
+	resolver["uid-new"] = podIndexEntry{
+		namespace: "shop", name: "web-abc",
+		workload: WorkloadRef{Kind: "StatefulSet", Name: "web"},
+	}
+	ingest(3, 10, second, newPodStart, second)
+
+	r := onlyRecord(t, p)
+	// 400/1000 from the first container plus the replacement's own 3/10 —
+	// neither dropped as a phantom reset nor stretched into a delta against a
+	// baseline that belonged to a different container.
+	if r.CPU.ThrottledPeriods != 403 || r.CPU.TotalPeriods != 1010 {
+		t.Errorf("throttled/total = %d/%d, want 403/1010 — each container attributed from its own start",
+			r.CPU.ThrottledPeriods, r.CPU.TotalPeriods)
+	}
+	if len(p.throttle) != 2 {
+		t.Errorf("throttle state entries = %d, want 2 (one series per pod UID)", len(p.throttle))
+	}
+}
+
+// A container observed and quiet must be distinguishable from one never
+// observed at all — the whole reason the sample count exists.
+func TestThrottlingSampleCountSeparatesQuietFromUnobserved(t *testing.T) {
+	p := testPoller(webResolver())
+	at := usageTestStart.Add(30 * time.Second)
+	samples, err := parseCadvisor(strings.NewReader(fixture(0, 100, at, usageTestStart)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	p.ingestCadvisor(samples, at)
+
+	r := onlyRecord(t, p)
+	if r.CPU.ThrottledPeriods != 0 {
+		t.Fatalf("throttled periods = %d, want 0", r.CPU.ThrottledPeriods)
+	}
+	if r.CPU.ThrottlingSamples == 0 {
+		t.Error("throttling samples = 0 after a successful observation: a quiet container is now indistinguishable from an unobserved one")
+	}
+}
+
 func TestThrottlingUnknownPodSkipped(t *testing.T) {
 	p := testPoller(stubResolver{})
 	at := usageTestStart.Add(30 * time.Second)

internal/collector/pods.go

@@ -336,17 +336,25 @@ func (w *PodWatcher) LookupPod(uid types.UID) (namespace string, workload Worklo
 	return entry.namespace, entry.workload, ok
 }
 
-// LookupPodByName is LookupPod for sources that identify pods by
-// namespace and name instead of UID.
-func (w *PodWatcher) LookupPodByName(namespace, name string) (workload WorkloadRef, ok bool) {
+// LookupPodByName is LookupPod for sources that identify pods by namespace and
+// name instead of UID — the cAdvisor exposition is one. It returns the UID as
+// well as the workload, because a name is not an identity: a StatefulSet
+// recreates a pod under the same name with a new UID, and a caller keeping
+// per-container counter baselines must key them on something that changes when
+// the container does. Callers that key on the name alone silently attribute the
+// new pod's counters to the dead pod's baseline.
+func (w *PodWatcher) LookupPodByName(namespace, name string) (uid types.UID, workload WorkloadRef, ok bool) {
 	w.indexMu.RLock()
 	defer w.indexMu.RUnlock()
-	uid, ok := w.nameIndex[namespace+"/"+name]
+	uid, ok = w.nameIndex[namespace+"/"+name]
 	if !ok {
-		return WorkloadRef{}, false
+		return "", WorkloadRef{}, false
 	}
 	entry, ok := w.index[uid]
-	return entry.workload, ok
+	if !ok {
+		return "", WorkloadRef{}, false
+	}
+	return uid, entry.workload, true
 }
 
 // LookupContainer resolves a node-role build-info fact — a pod UID and a