fix: a shutdown between cache sync and handler registration is not a failure

cmd/agent/flush.go

@@ -0,0 +1,35 @@
+package main
+
+// The set of payload writers that run on the agent's own cadence, as data.
+//
+// There are two passes over it — the periodic one and the one a shutdown makes
+// — and they must agree about which writers exist and in what order. As two
+// hand-written call sequences they did not: the metadata flush was in the
+// periodic pass and absent from the shutdown pass, correctly, but nothing said
+// so and nothing kept the two lists from drifting further.
+
+// flusher is one payload writer. Its identifier at the call site is its label;
+// the struct carries only what the two passes have to decide between them.
+type flusher struct {
+	run func()
+	// onShutdown is whether this writer also runs on the pass a shutdown makes.
+	//
+	// False for a writer that re-derives its payload from the watchers' live
+	// indexes: by shutdown those have stopped, so the pass would republish the
+	// previous state under a fresh capture instant. The writers that accumulate
+	// — the journals, the inventory — are the opposite case, and are exactly
+	// what the shutdown pass exists for: without it a graceful stop drops up to
+	// one coverage interval of them.
+	onShutdown bool
+}
+
+// runFlushers runs the set in order. On the shutdown pass it skips the writers
+// that declare themselves periodic-only.
+func runFlushers(flushers []flusher, shutdown bool) {
+	for _, f := range flushers {
+		if shutdown && !f.onShutdown {
+			continue
+		}
+		f.run()
+	}
+}

cmd/agent/flush_test.go

@@ -0,0 +1,40 @@
+package main
+
+import (
+	"slices"
+	"testing"
+)
+
+func TestRunFlushersRunsEveryWriterInOrder(t *testing.T) {
+	var ran []string
+	record := func(name string) func() { return func() { ran = append(ran, name) } }
+	flushers := []flusher{
+		{run: record("first"), onShutdown: true},
+		{run: record("second")},
+		{run: record("third"), onShutdown: true},
+	}
+
+	runFlushers(flushers, false)
+
+	want := []string{"first", "second", "third"}
+	if !slices.Equal(ran, want) {
+		t.Fatalf("periodic pass ran %v, want %v", ran, want)
+	}
+}
+
+func TestRunFlushersSkipsPeriodicOnlyWritersOnShutdown(t *testing.T) {
+	var ran []string
+	record := func(name string) func() { return func() { ran = append(ran, name) } }
+	flushers := []flusher{
+		{run: record("first"), onShutdown: true},
+		{run: record("second")},
+		{run: record("third"), onShutdown: true},
+	}
+
+	runFlushers(flushers, true)
+
+	want := []string{"first", "third"}
+	if !slices.Equal(ran, want) {
+		t.Fatalf("shutdown pass ran %v, want %v", ran, want)
+	}
+}

cmd/agent/main.go

@@ -718,21 +718,41 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac
 		}
 	}
 
+	// Everything that writes a payload on the agent's own cadence, in the order
+	// it is written. One list rather than a call sequence per pass: the periodic
+	// pass and the shutdown pass have to agree about this set, and as two
+	// hand-written sequences they had already drifted.
+	flushers := []flusher{
+		{run: logCoverage, onShutdown: true},
+		// Not on shutdown: it re-derives a snapshot from the watchers' live
+		// indexes, and by then the watchers have stopped. It would write the
+		// state of the previous pass under a fresh capture instant, which is
+		// the one way a superseding payload can lie (ADR 0027).
+		{run: flushMetadata},
+		{run: flushInventory, onShutdown: true},
+		{run: flushRestarts, onShutdown: true},
+		{run: flushDisruptions, onShutdown: true},
+		{run: flushJobRuns, onShutdown: true},
+		{run: flushNodeLifecycle, onShutdown: true},
+	}
+
+	// The periodic pass is a lifecycle task like the watchers below, and for a
+	// load-bearing reason: the shutdown pass writes the same payload keys, and
+	// the spool has no lock — two writers of one key share a temp file name and
+	// publish whichever interleaving won. Waiting for this goroutine is what
+	// makes the shutdown pass the only writer it claims to be.
+	var wg sync.WaitGroup
+	wg.Add(1)
 	go func() {
+		defer wg.Done()
 		ticker := time.NewTicker(coverageInterval)
 		defer ticker.Stop()
 		for {
 			select {
 			case <-ctx.Done():
 				return
 			case <-ticker.C:
-				logCoverage()
-				flushMetadata()
-				flushInventory()
-				flushRestarts()
-				flushDisruptions()
-				flushJobRuns()
-				flushNodeLifecycle()
+				runFlushers(flushers, false)
 				// Unconditionally, on the agent's own cadence rather than the
 				// usage poller's: the spool's bounds must hold on a cluster
 				// where no usage record is ever written, which is exactly the
@@ -827,7 +847,6 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac
 		tasks["node-intake"] = intake.Run
 	}
 
-	var wg sync.WaitGroup
 	var mu sync.Mutex
 	var errs []error
 	for name, run := range tasks {
@@ -843,19 +862,13 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac
 		}()
 	}
 	wg.Wait()
-	logCoverage()
 	// A final flush lands whatever arrived since the last periodic write: an
 	// inventory fact joined, a container restarted, a pod preempted. Without it
 	// a graceful shutdown silently drops up to one coverage interval of the
 	// journals, which is exactly the minute a rolling upgrade or a node drain
-	// tends to be interesting in. The coverage goroutine has returned (ctx is
-	// canceled), so this is the only writer; the accumulators serialize
-	// regardless.
-	flushInventory()
-	flushRestarts()
-	flushDisruptions()
-	flushJobRuns()
-	flushNodeLifecycle()
+	// tends to be interesting in. The periodic pass is waited for above, so this
+	// is the only writer; the accumulators serialize regardless.
+	runFlushers(flushers, true)
 	return errors.Join(errs...)
 }
 

internal/collector/nodes.go

@@ -111,6 +111,9 @@ type NodeWatcher struct {
 	taintsDropped     atomic.Int64
 	valuesDropped     atomic.Int64
 
+	// afterSync is the test seam described where the type is declared.
+	afterSync afterSync
+
 	mu    sync.RWMutex
 	nodes map[string]NodeInfo
 
@@ -238,6 +241,9 @@ func (w *NodeWatcher) Run(ctx context.Context) error {
 		}
 		return fmt.Errorf("node informer cache did not sync")
 	}
+	if w.afterSync != nil {
+		w.afterSync(nodesInformer)
+	}
 
 	reg, err := nodesInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj any) {
@@ -265,7 +271,7 @@ func (w *NodeWatcher) Run(ctx context.Context) error {
 		},
 	})
 	if err != nil {
-		return fmt.Errorf("register node handler: %w", err)
+		return registrationFailure(ctx, fatal, "node", err)
 	}
 	defer func() { _ = nodesInformer.RemoveEventHandler(reg) }()
 

internal/collector/pods.go

@@ -178,6 +178,9 @@ type PodWatcher struct {
 	priorityLister     schedulinglisters.PriorityClassLister
 	storageClassLister storagelisters.StorageClassLister
 
+	// afterSync is the test seam described where the type is declared.
+	afterSync afterSync
+
 	mu           sync.Mutex
 	reportedOOMs map[string]struct{}
 	// restartCounts is what the agent remembers about each container's restart
@@ -529,6 +532,9 @@ func (w *PodWatcher) Run(ctx context.Context) error {
 		}
 		return fmt.Errorf("informer caches did not sync")
 	}
+	if w.afterSync != nil {
+		w.afterSync(podsInformer)
+	}
 
 	// Registered after sync: the handler replays every pod already in the
 	// cache as an Add, then receives genuinely new pods as they appear.
@@ -578,7 +584,7 @@ func (w *PodWatcher) Run(ctx context.Context) error {
 		},
 	})
 	if err != nil {
-		return fmt.Errorf("register pod handler: %w", err)
+		return registrationFailure(ctx, fatal, "pod", err)
 	}
 	defer func() { _ = podsInformer.RemoveEventHandler(reg) }()
 
@@ -611,7 +617,7 @@ func (w *PodWatcher) Run(ctx context.Context) error {
 		},
 	})
 	if err != nil {
-		return fmt.Errorf("register job handler: %w", err)
+		return registrationFailure(ctx, fatal, "job", err)
 	}
 	defer func() { _ = jobsInformer.RemoveEventHandler(jobReg) }()
 

internal/collector/shutdown_test.go

@@ -0,0 +1,88 @@
+package collector
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"k8s.io/client-go/kubernetes/fake"
+	"k8s.io/client-go/tools/cache"
+)
+
+// Handlers are registered after the caches sync, and the informers stop the
+// moment the run context is canceled — so a shutdown landing between the two
+// has its registration refused. Reporting that as an error takes the whole
+// controller down with exit 1 on an ordinary SIGTERM, and it is what made
+// TestCollectsContainersAndImages flake: that test cancels as soon as it has
+// seen its pods, which is exactly this window.
+
+func TestPodWatcherShutdownBeforeRegistrationIsNotAFailure(t *testing.T) {
+	watcher := NewPodWatcher(fake.NewClientset(pod("checkout-1", nil)), func(PodInfo) {})
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	watcher.afterSync = func(informer cache.SharedIndexInformer) {
+		cancel()
+		waitUntilStopped(t, informer)
+	}
+
+	if err := watcher.Run(ctx); err != nil {
+		t.Fatalf("Run = %v, want nil: a canceled context is a shutdown, not a watch failure", err)
+	}
+}
+
+func TestNodeWatcherShutdownBeforeRegistrationIsNotAFailure(t *testing.T) {
+	watcher := NewNodeWatcher(fake.NewClientset(node("node-a", nil)), func(NodeInfo) {})
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	watcher.afterSync = func(informer cache.SharedIndexInformer) {
+		cancel()
+		waitUntilStopped(t, informer)
+	}
+
+	if err := watcher.Run(ctx); err != nil {
+		t.Fatalf("Run = %v, want nil: a canceled context is a shutdown, not a watch failure", err)
+	}
+}
+
+// A watch failure still stops the agent, canceled context or not: the verdict
+// is the reason the run context was canceled in the first place, and losing it
+// here would turn a blind agent into a clean exit.
+func TestRegistrationFailureKeepsTheWatchdogVerdict(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	fatal := make(chan error, 1)
+	verdict := context.DeadlineExceeded
+	fatal <- verdict
+
+	if err := registrationFailure(ctx, fatal, "pod", context.Canceled); !errors.Is(err, verdict) {
+		t.Fatalf("registrationFailure = %v, want the watchdog verdict %v", err, verdict)
+	}
+}
+
+// With neither a verdict nor a cancellation, a refused registration is what it
+// looks like: the agent cannot watch, and must say so.
+func TestRegistrationFailureReportsAGenuineError(t *testing.T) {
+	err := registrationFailure(context.Background(), make(chan error, 1), "pod", context.Canceled)
+	if err == nil {
+		t.Fatal("registrationFailure = nil on a live context, want an error")
+	}
+	if want := "register pod handler"; !strings.Contains(err.Error(), want) {
+		t.Fatalf("registrationFailure = %q, want it to name %q", err, want)
+	}
+}
+
+// waitUntilStopped blocks until the informer refuses new handlers. IsStopped
+// reads the flag AddEventHandler checks, under the same lock, so waiting on it
+// is what makes the registration below deterministic rather than a race.
+func waitUntilStopped(t *testing.T, informer cache.SharedIndexInformer) {
+	t.Helper()
+	deadline := time.Now().Add(5 * time.Second)
+	for !informer.IsStopped() {
+		if time.Now().After(deadline) {
+			t.Fatal("informer still running 5s after its run context was canceled")
+		}
+		time.Sleep(time.Millisecond)
+	}
+}

internal/collector/watchhealth.go

@@ -178,6 +178,14 @@ func watchdog(ctx context.Context, now func() time.Time, limits watchLimits, gat
 	}
 }
 
+// afterSync runs between the cache-sync wait and handler registration when set,
+// receiving the informer the registration is about to be made on.
+//
+// Only tests set it, and only to stop that informer at that instant: the window
+// where a registration is refused is otherwise reached by a race, and a race is
+// not something a regression test can stand on.
+type afterSync func(cache.SharedIndexInformer)
+
 // useWatchLimits re-times an assembled watcher, for tests that cannot wait five
 // minutes to watch a five-minute rule fire. It must be called before Run: the
 // health records are already registered with their informers, so only their
@@ -204,6 +212,24 @@ func takeWatchFailure(fatal <-chan error) error {
 	}
 }
 
+// registrationFailure classifies an AddEventHandler error, which an informer
+// returns once it has stopped.
+//
+// Informers stop when the run context is canceled, so a shutdown landing
+// between the sync wait and registration has its handler refused: the agent was
+// asked to stop, not prevented from watching, and the wait above already draws
+// that distinction. The watchdog's verdict is read first — it is the other
+// reason the context is canceled, and the one the caller must return.
+func registrationFailure(ctx context.Context, fatal <-chan error, resource string, err error) error {
+	if verdict := takeWatchFailure(fatal); verdict != nil {
+		return verdict
+	}
+	if ctx.Err() != nil {
+		return nil
+	}
+	return fmt.Errorf("register %s handler: %w", resource, err)
+}
+
 // recordWatchFailures returns the handler an informer calls when ListAndWatch
 // drops, bound to one cache's health record.
 //