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