fix: a pod that opts out while running is counted (ADR 0054)

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

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) {

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)
+	}
+}

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

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