diff --git a/cmd/agent/flush.go b/cmd/agent/flush.go new file mode 100644 index 0000000..bb07ae7 --- /dev/null +++ b/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() + } +} diff --git a/cmd/agent/flush_test.go b/cmd/agent/flush_test.go new file mode 100644 index 0000000..023d1d2 --- /dev/null +++ b/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) + } +} diff --git a/cmd/agent/main.go b/cmd/agent/main.go index ff846e0..76df88b 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -718,7 +718,33 @@ 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 { @@ -726,13 +752,7 @@ func run(ctx context.Context, logger *slog.Logger, clientset kubernetes.Interfac 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...) } diff --git a/internal/collector/nodes.go b/internal/collector/nodes.go index 851bce1..4b6dfc5 100644 --- a/internal/collector/nodes.go +++ b/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) }() diff --git a/internal/collector/pods.go b/internal/collector/pods.go index 2c8a822..85a0533 100644 --- a/internal/collector/pods.go +++ b/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) }() diff --git a/internal/collector/shutdown_test.go b/internal/collector/shutdown_test.go new file mode 100644 index 0000000..9cdb46f --- /dev/null +++ b/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) + } +} diff --git a/internal/collector/watchhealth.go b/internal/collector/watchhealth.go index 72beccb..564b37b 100644 --- a/internal/collector/watchhealth.go +++ b/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. //