From 4aa4bd24c6d6776bb158a9d8ab7bcc4048f0e4bc Mon Sep 17 00:00:00 2001 From: Konstantins Feofantovs Date: Wed, 16 Sep 2026 13:28:21 +0300 Subject: [PATCH] fix: a pod that opts out while running is counted (ADR 0054) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The pod informer's AddFunc counted admission decisions and its UpdateFunc did not, so a customer who annotated a running pod removed it from collection without moving any counter. filter.excluded_pod_annotation stayed 0 while the record left the go_inventory payload. ADR 0054 already decided this should not happen. It accepts that a workload leaving the collected set is visible to the backend by subtraction, and rests on the count explaining why — "the alternative is silence, under which the backend concludes the workload was deleted and tells the customer something false". That is what shipped. No new decision here, so no new ADR. Counting on every update would have been worse: a pod emits status updates every few seconds and each carries the same answer. countTransition counts a decision that differs from the one recorded in the index, so a change counts once and a repetition counts nothing. The same rule covers the narrower control: a workload that refuses the profiler mid-flight now moves excluded_profiling_annotation, and removing either annotation counts the pod observed again. This makes the filter counters count decisions rather than pods, so adding pods_observed to the exclusion counters no longer recovers a pod count. backend-requirements.md says so. four of the five new unit tests fail against the prior pods.go; the fifth guards against double counting and passes both ways make inventory-e2e — PASS (428s), coverage reporting excluded_pod_annotation = 1 where it read 0 before Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01KeZECJXTDHL6Nts95eKMJA --- docs/backend-requirements.md | 7 + docs/prose_test.go | 8 +- internal/collector/optout_transition_test.go | 164 +++++++++++++++++++ internal/collector/pods.go | 49 +++++- test/e2e/inventory_test.go | 33 ++++ 5 files changed, 253 insertions(+), 8 deletions(-) create mode 100644 internal/collector/optout_transition_test.go diff --git a/docs/backend-requirements.md b/docs/backend-requirements.md index a23b304..2c5bf35 100644 --- a/docs/backend-requirements.md +++ b/docs/backend-requirements.md @@ -448,6 +448,13 @@ or `failing: true` bounds every finding resting on it — the backend MUST NOT present such a finding as complete. The scan block is a fleet aggregate over the latest pass of each reporting node and MUST NOT be summed across flushes. +The `filter` counters count **decisions, not pods**. A customer who annotates a +running pod changes the answer about it, so a pod already counted in +`pods_observed` then moves an exclusion counter, and removing the annotation +counts it observed again. The backend MUST NOT read `pods_observed` and the +exclusion counters as a partition of the pods seen, and MUST NOT recover a pod +count by adding them. + `filter.excluded_profiling_annotation` counts pods the customer excluded from profiling alone ([ADR 0071](adr/0071-refusing-the-profiler-without-refusing-the-rest.md)): they are still collected and still ship every other kind, so the backend MUST NOT diff --git a/docs/prose_test.go b/docs/prose_test.go index 71f5eac..af07346 100644 --- a/docs/prose_test.go +++ b/docs/prose_test.go @@ -46,10 +46,10 @@ var ceilings = []struct { // the implementer must decode it and stop reading a payload's size off the // request (ADR 0076); a journal window states whether it is finished // (ADR 0078); a new kind arrives whose samples are gauges, where the obvious - // ingest — adding them up — is the wrong one (ADR 0080); and `godebug` now - // carries names where neither value is the safe one by convention, so `1` - // may not be read as the weak setting (ADR 0081). - {"backend-requirements.md", 799, 628}, + // ingest — adding them up — is the wrong one (ADR 0080); `godebug` carries + // names where neither value is safe by convention (ADR 0081); and the filter + // counters count decisions, so adding them no longer recovers a pod count. + {"backend-requirements.md", 806, 635}, } func TestNoCommentRunIsLongerThanAPointerToAnADR(t *testing.T) { diff --git a/internal/collector/optout_transition_test.go b/internal/collector/optout_transition_test.go new file mode 100644 index 0000000..7716751 --- /dev/null +++ b/internal/collector/optout_transition_test.go @@ -0,0 +1,164 @@ +package collector + +import ( + "context" + "testing" + "time" + + "github.com/RebuildStackCo/runtime-agent/internal/model" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/kubernetes/fake" +) + +// A customer annotates a pod that is already running, so the answer changes +// without the pod appearing again. The counters have to move with it: ADR 0054 +// accepts that a workload leaving the collected set is visible by subtraction, +// and rests on the count explaining why — a record that vanishes while every +// number holds still reads to the backend as a workload that was deleted. + +// transitionWatcher runs a watcher over one pod and returns once the filter has +// decided about it, handing back everything needed to change the pod underneath. +func transitionWatcher(t *testing.T, annotations map[string]string) (*PodWatcher, *Filter, kubernetes.Interface, *corev1.Pod) { + t.Helper() + rs := &appsv1.ReplicaSet{ObjectMeta: metav1.ObjectMeta{ + Namespace: "shop", Name: "web-7d9f", UID: "uid-web-7d9f", + OwnerReferences: []metav1.OwnerReference{controllerRef("Deployment", "web")}, + }} + deployment := &appsv1.Deployment{ObjectMeta: metav1.ObjectMeta{ + Namespace: "shop", Name: "web", UID: "uid-web", + }} + p := profilingPod("web-a", "10.0.0.1", annotations) + clientset := fake.NewClientset([]runtime.Object{rs, deployment, p}...) + + watcher := NewPodWatcher(clientset, func(model.PodInfo) {}) + filter := NewFilter(nil, nil) + watcher.SetFilter(filter) + + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + go func() { _ = watcher.Run(ctx) }() + + deadline := time.Now().Add(5 * time.Second) + for { + c := filter.Snapshot() + if c.PodsObserved+c.ExcludedPodAnnotation > 0 { + break + } + if time.Now().After(deadline) { + t.Fatalf("the filter decided nothing about the pod before timeout: %+v", c) + } + time.Sleep(10 * time.Millisecond) + } + return watcher, filter, clientset, p +} + +// setAnnotations replaces the pod's annotations through the API the informer +// watches, which is what a customer editing a live pod does. +func setAnnotations(t *testing.T, cs kubernetes.Interface, p *corev1.Pod, annotations map[string]string) { + t.Helper() + ctx := context.Background() + live, err := cs.CoreV1().Pods(p.Namespace).Get(ctx, p.Name, metav1.GetOptions{}) + if err != nil { + t.Fatalf("reading %s/%s: %v", p.Namespace, p.Name, err) + } + live.Annotations = annotations + if _, err := cs.CoreV1().Pods(p.Namespace).Update(ctx, live, metav1.UpdateOptions{}); err != nil { + t.Fatalf("updating %s/%s: %v", p.Namespace, p.Name, err) + } +} + +func TestOptingOutARunningPodIsCounted(t *testing.T) { + watcher, filter, cs, p := transitionWatcher(t, nil) + if c := filter.Snapshot(); c.PodsObserved != 1 || c.ExcludedPodAnnotation != 0 { + t.Fatalf("coverage before the annotation = %+v, want 1 observed and 0 excluded", c) + } + + setAnnotations(t, cs, p, map[string]string{CollectAnnotation: "false"}) + waitFor(t, 5*time.Second, "the opt-out to be counted", + func() bool { return filter.Snapshot().ExcludedPodAnnotation == 1 }) + + if pods := watcher.Pods(); len(pods) != 0 { + t.Errorf("collected pods = %d, want the opted-out pod dropped", len(pods)) + } +} + +// The pod goes on emitting status updates after the opt-out, each carrying the +// same answer. Counting the state rather than the change would climb forever. +func TestRepeatedUpdatesAfterTheOptOutCountOnce(t *testing.T) { + _, filter, cs, p := transitionWatcher(t, nil) + optOut := map[string]string{CollectAnnotation: "false"} + + setAnnotations(t, cs, p, optOut) + waitFor(t, 5*time.Second, "the opt-out to be counted", + func() bool { return filter.Snapshot().ExcludedPodAnnotation == 1 }) + + for range 5 { + setAnnotations(t, cs, p, optOut) + } + time.Sleep(200 * time.Millisecond) + + if c := filter.Snapshot(); c.ExcludedPodAnnotation != 1 { + t.Errorf("excluded_pod_annotation = %d after five more updates, want 1", c.ExcludedPodAnnotation) + } +} + +// Removing the annotation is a change too, and the pod returns to the collected +// set. A backend that saw it leave has to be able to see it come back. +func TestRemovingTheAnnotationCountsThePodObservedAgain(t *testing.T) { + watcher, filter, cs, p := transitionWatcher(t, nil) + + setAnnotations(t, cs, p, map[string]string{CollectAnnotation: "false"}) + waitFor(t, 5*time.Second, "the opt-out to be counted", + func() bool { return filter.Snapshot().ExcludedPodAnnotation == 1 }) + + setAnnotations(t, cs, p, nil) + waitFor(t, 5*time.Second, "the pod to be counted observed again", + func() bool { return filter.Snapshot().PodsObserved == 2 }) + + if pods := watcher.Pods(); len(pods) != 1 { + t.Errorf("collected pods = %d, want the pod back", len(pods)) + } +} + +// The narrower control moves its own counter on the same terms. The pod stays +// collected throughout, so nothing but the profiling count may change. +func TestRefusingTheProfilerOnARunningPodIsCounted(t *testing.T) { + watcher, filter, cs, p := transitionWatcher(t, nil) + + setAnnotations(t, cs, p, map[string]string{ProfileAnnotation: "false"}) + waitFor(t, 5*time.Second, "the profiling opt-out to be counted", + func() bool { return filter.Snapshot().ExcludedProfilingAnnotation == 1 }) + + if c := filter.Snapshot(); c.ExcludedPodAnnotation != 0 { + t.Errorf("excluded_pod_annotation = %d, want 0: the pod refused the profiler, not collection", + c.ExcludedPodAnnotation) + } + if pods := watcher.Pods(); len(pods) != 1 { + t.Errorf("collected pods = %d, want the pod to stay collected", len(pods)) + } + if cs := watcher.ContainersOnNode("node-1"); len(cs) != 0 { + t.Errorf("targets answer names %+v, want nothing for a pod that refused the profiler", cs) + } +} + +// A pod that was already excluded when it appeared was counted then. The +// updates that follow carry the same answer and must not count it twice. +func TestAPodExcludedOnArrivalIsNotCountedAgain(t *testing.T) { + _, filter, cs, p := transitionWatcher(t, map[string]string{CollectAnnotation: "false"}) + if c := filter.Snapshot(); c.ExcludedPodAnnotation != 1 || c.PodsObserved != 0 { + t.Fatalf("coverage on arrival = %+v, want 1 excluded and 0 observed", c) + } + + for range 5 { + setAnnotations(t, cs, p, map[string]string{CollectAnnotation: "false"}) + } + time.Sleep(200 * time.Millisecond) + + if c := filter.Snapshot(); c.ExcludedPodAnnotation != 1 || c.PodsObserved != 0 { + t.Errorf("coverage = %+v, want the arrival's single count and nothing since", c) + } +} diff --git a/internal/collector/pods.go b/internal/collector/pods.go index a48748e..4eb8f03 100644 --- a/internal/collector/pods.go +++ b/internal/collector/pods.go @@ -465,7 +465,11 @@ func (w *PodWatcher) Run(ctx context.Context) error { if !ok { return } - if d := w.admit(pod, false); d.collect { + d := w.admit(pod, false) + // Read before the index is touched: what is recorded there is the + // decision this one is compared against. + was, known := w.decisionFor(pod.UID) + if d.collect { info := w.describe(pod) w.indexPod(pod, info, d.profile) w.reportPodIfChanged(pod.UID, info) @@ -475,6 +479,7 @@ func (w *PodWatcher) Run(ctx context.Context) error { } else { w.dropPod(pod.UID) } + w.countTransition(was, known, d) }, DeleteFunc: func(obj any) { if unknown, ok := obj.(cache.DeletedFinalStateUnknown); ok { @@ -839,12 +844,14 @@ func (w *PodWatcher) AdmittedPodsOnNode(node string) []string { type admission struct { collect bool profile bool + // reason is why collection was refused, empty when it was not. + reason ExclusionReason } // admit runs the pod through the filter, consulting the namespace's // annotations from the cache (a cache miss reads as no annotations). -// Exclusions are counted only when count is set — once per pod appearance -// on Add, never again on the many status updates that follow. +// Exclusions are counted only when count is set — once per pod appearance on +// Add; the updates that follow go through countTransition instead. // // Both controls are decided from the one owner-chain walk: the profiling // opt-out reads the same three objects, and asking twice would double the @@ -870,7 +877,41 @@ func (w *PodWatcher) admit(pod *corev1.Pod, count bool) admission { w.filter.countUnresolvedWorkload(workload.Unresolved) } } - return admission{collect: allowed, profile: profilable} + return admission{collect: allowed, profile: profilable, reason: reason} +} + +// decisionFor reports the decision currently recorded for a pod. Presence in the +// index is the record: a pod is there because it was admitted, and the entry +// carries whether profiling was admitted with it. +func (w *PodWatcher) decisionFor(uid types.UID) (admission, bool) { + w.indexMu.RLock() + defer w.indexMu.RUnlock() + entry, ok := w.index[uid] + if !ok { + return admission{}, false + } + return admission{collect: true, profile: entry.profilable}, true +} + +// countTransition counts an admission decision that differs from the one already +// recorded for the pod. Annotating a running pod changes the answer without the +// pod appearing again, and ADR 0054 rests on the counter moving when it does: a +// workload leaving the collected set while every count holds still reads to the +// backend as a workload that was deleted. Only the change counts — the status +// updates a pod emits every few seconds carry the previous answer and count +// nothing. +func (w *PodWatcher) countTransition(was admission, known bool, now admission) { + switch { + case known && !now.collect: + w.filter.countExcluded(now.reason) + case !known && now.collect: + w.filter.countObserved() + if !now.profile { + w.filter.countExcludedProfiling() + } + case known && now.collect && was.profile && !now.profile: + w.filter.countExcludedProfiling() + } } // workloadAnnotations reads the annotations of the controller that ultimately diff --git a/test/e2e/inventory_test.go b/test/e2e/inventory_test.go index 53240b6..95748e8 100644 --- a/test/e2e/inventory_test.go +++ b/test/e2e/inventory_test.go @@ -256,6 +256,7 @@ func checkOptOutRemovesTheRecord(ctx context.Context, t *testing.T, config *rest for { snap := findSampleRecord(ctx, t, config, cs, ns, controllerPod) if !snap.Found { + checkTheOptOutIsCounted(ctx, t, config, cs, ns, controllerPod) return } // Every rewrite is logged, so the transcript shows whether the payload @@ -346,6 +347,38 @@ type inventoryCoverage struct { FactsJoined int64 `json:"facts_joined"` } +// checkTheOptOutIsCounted asserts the coverage payload explains the record that +// just left. ADR 0054 accepts that the disappearance is visible by subtraction +// and rests on the count saying why: a workload leaving the collected set while +// every number holds still reads as a workload that was deleted. +func checkTheOptOutIsCounted(ctx context.Context, t *testing.T, config *rest.Config, cs kubernetes.Interface, ns, pod string) { + t.Helper() + // The kind ships on every flush, so one cadence is enough to wait. + deadline := time.Now().Add(3 * time.Minute) + for { + raw, ok := readSpoolFile(ctx, t, config, cs, ns, pod, coverageSpoolPath) + if ok { + var payload struct { + Filter struct { + ExcludedPodAnnotation int64 `json:"excluded_pod_annotation"` + } `json:"filter"` + } + if err := json.Unmarshal([]byte(raw), &payload); err != nil { + t.Fatalf("collection_coverage payload not valid JSON: %v", err) + } + if n := payload.Filter.ExcludedPodAnnotation; n >= 1 { + t.Logf("coverage explains the departure: excluded_pod_annotation = %d", n) + return + } + } + if time.Now().After(deadline) { + t.Fatal("the record left the payload and filter.excluded_pod_annotation stayed 0; " + + "the backend cannot tell an opt-out from a deletion") + } + time.Sleep(5 * time.Second) + } +} + // goInventorySnapshot is one read of the spool file. CapturedAt is what makes a // negative reading meaningful: the kind supersedes in place under one file name, // so a reader that does not date the payload cannot tell a record that left from