From 6ca1d112913fcc775ed15a88f4895b55d395b500 Mon Sep 17 00:00:00 2001 From: davidby-influx Date: Mon, 31 Aug 2026 17:52:17 -0700 Subject: [PATCH 1/6] feat: hold /health and /ready open after a failed startup A failed start names the failing subsystem on both endpoints, and the process exits before a scraper can read it. The body lives for microseconds, so the log line was the one durable copy. --startup-error-linger keeps both endpoints answering for a fixed duration after a failed start, capped at 30 minutes: the window holds the HTTP port on a process a supervisor is waiting to restart. The default of 0 preserves the current behavior. Teardown splits around the wait. Every subsystem except the HTTP listener and the PID file closes ahead of it, so the bolt flock, the sqlite file and the engine directory belong to the next run. The listener serves the window; the PID file keeps a second influxd from starting against a port this process holds. Shutdown releases both, and consumes each closer as it runs, so no phase can close a store a second time. The check set freezes ahead of that teardown. A closing store reports its own shutdown as a fresh failure, and failures sort ahead of passes, so it would outrank the startup error the window exists to publish. Check.Freeze bounds each probe on its own, so one wedged subsystem cannot spend the budget the checks after it need. Supporting changes: Response gains Snapshotter, for a coherent read of a FreshnessResponse; Shutdown aggregates closer failures with errors.Join and runs on the startup-failure path, which leaves no orphaned PID file. HEALTH_READY.md documents the flag, the cap, and the reduced check detail an operator sees during the window under --health-auth-enabled. --- HEALTH_READY.md | 168 ++++++++- cmd/influxd/launcher/cmd.go | 110 +++++- cmd/influxd/launcher/cmd_test.go | 54 +++ cmd/influxd/launcher/export_test.go | 34 ++ cmd/influxd/launcher/launcher.go | 251 ++++++++++++- cmd/influxd/launcher/shutdown_test.go | 369 +++++++++++++++++++ cmd/influxd/launcher/startup_failure_test.go | 133 +++++++ http/check_handler.go | 46 +++ http/check_handler_freeze_test.go | 172 +++++++++ kit/check/check.go | 121 +++++- kit/check/check_race_test.go | 123 +++++++ kit/check/checktest/checktest.go | 84 +++++ kit/check/freeze_test.go | 261 +++++++++++++ kit/check/freshness.go | 53 +-- kit/check/freshness_test.go | 114 ++++++ kit/check/helpers.go | 20 +- kit/check/response.go | 48 +++ 17 files changed, 2103 insertions(+), 58 deletions(-) create mode 100644 cmd/influxd/launcher/export_test.go create mode 100644 cmd/influxd/launcher/shutdown_test.go create mode 100644 http/check_handler_freeze_test.go create mode 100644 kit/check/checktest/checktest.go create mode 100644 kit/check/freeze_test.go diff --git a/HEALTH_READY.md b/HEALTH_READY.md index 44ee0fb52f6..00cf7a62412 100644 --- a/HEALTH_READY.md +++ b/HEALTH_READY.md @@ -302,6 +302,42 @@ flood runs. `influxd` logs a throttled warning while the budget is exhausted, which is the only way to tell a starved probe from a rejected credential — they produce the same body. +### Check detail during the window + +`--startup-error-linger` and `--health-auth-enabled` interact badly, and +the interaction cannot be designed away. During the +[linger window](#keeping-the-endpoints-alive-after-a-failed-startup) the +store credentials are resolved against has been closed — releasing it is +the point of closing everything but the listener before the wait — so no +caller can be identified for the whole window. By the rule above, being +unable to *ask* releases shape and never content: **every caller, +including one presenting a valid operator token, sees check names and +statuses with the messages stripped.** + +What an operator still gets is more than nothing. The `503` is correct, +every check is named with its status, and exactly one `/health` entry is +failing — the subsystems that came up read `pass`, and `shards` reads +`pass` on an engine failure — so *which* phase failed is unambiguous. +Only the reason string is withheld, and the log already carries it at +`ERROR` with a `subsystem` field. + +There are three honest mitigations and no fourth: + +- Read the log for the reason; the endpoint tells you which subsystem. +- The answer is at least consistent: the auth dependency checker is + retired at the moment of the freeze, so the body has the same shape + from the first request of the window to the last. +- Run with health auth off if the message must be readable over HTTP — + accepting that it is then readable by anyone who can reach the port. + +Keeping the store open across the window would restore the detail, but it +would re-hold the flock and the PID file that the split teardown exists +to release, blocking the restart the operator is presumably attempting. + +Note that this is the existing policy applied consistently, not a new +hole: a startup failure *before* the authorization store opens already +reached the same reduced body, window or no window. + ### Cost A caller presenting no credential costs nothing extra: the scheme probe @@ -493,6 +529,13 @@ The phases that can appear, beyond the subsystem names already listed: can be served to a probe: their failures reach the log only. They are listed for completeness. +Every other phase in the table registers its check on a listener that is +already bound — but by default the process exits immediately afterwards, +so nothing has time to scrape it. +[`--startup-error-linger`](#keeping-the-endpoints-alive-after-a-failed-startup) +is what makes these entries reachable by a monitoring system rather than +only by an in-process test. + `meta-store` rather than `bolt` names the KV migrations and the unknown-store-type case because those run for every `--store` value: a migration failure under `--store=memory` is not a bolt problem, and @@ -775,6 +818,122 @@ With health auth enabled, this early phase is also the and the authorization store opens, no credential can be checked, and both endpoints report check names and statuses without their messages. +**They stop serving when the process exits**, which on a failed startup +is immediately — the listener is torn down microseconds after the +failure is recorded. See +[Keeping the endpoints alive after a failed startup](#keeping-the-endpoints-alive-after-a-failed-startup) +for the flag that changes this. + +### Keeping the endpoints alive after a failed startup + +A startup failure is recorded on both endpoints (see [Startup failure +checks](#startup-failure-checks)), and then the process exits and the +listener goes with it. In practice a monitoring system never sees it: +the body exists for microseconds. The log line is the reliable copy. + +`--startup-error-linger` keeps both endpoints answering for a fixed +duration after a failed startup, so a scraper can retrieve which +subsystem failed and why before the process goes away: + +``` +influxd --startup-error-linger=30s +``` + +| | default (`0`) | `--startup-error-linger=30s` | +|---|---|---| +| `/health` after a failed start | connection refused | `503`, frozen, naming the failing subsystem, for 30s | +| `/ready` after a failed start | connection refused | `503` `"starting"`, frozen, per-gate reasons, for 30s | +| bolt flock, sqlite, engine | released by process death | released **before** the window opens | +| PID file | released by process death | **held** for the window, released on exit | +| exit code and stderr | `1`, the startup error | unchanged | +| `SIGINT` during the window | — | cuts the window short, then exits | + +The equivalent environment variable is +`INFLUXD_STARTUP_ERROR_LINGER`, and the config file key is +`startup-error-linger`. Any Go duration string works (`45s`, `1m`). + +**The value is capped at 30 minutes.** The window holds the HTTP port on +a process that has already failed, and every supervisor that would +restart it is waiting on that process to exit, so an unbounded value +turns a failed start into an indefinite outage — a worse failure than the +one the window exists to report. A larger value is accepted, capped, and +logged at `WARN` naming the flag, the value you asked for and the one you +got; `print-config` still reports what you configured. + +**Everything except the listener and the PID file is released before the +wait begins.** The bolt flock, the sqlite file and the engine directory +belong to the next run rather than to one that already failed, so a +supervisor with `Restart=on-failure` is not blocked for the length of the +window. + +The PID file is deliberately *not* released with them. It is the +interlock that stops a second `influxd` starting against this data +directory, and for the length of the window this process is still running +and still holding its port — so releasing it would let a concurrent start +past the check that exists to catch exactly this, only to fail it later +on `listen tcp: address already in use`, which names the wrong cause. A +PID file describes a live process for as long as the process is alive. +It is removed by the final teardown, after the listener closes. + +One exception, which predates this flag: an engine that failed partway +through `Open` registers no closer, so it is not closed here either. +Nothing holds a lock on it in that state. + +**The report is frozen at the moment of failure.** Tearing a subsystem +down makes its own check start failing — a closed sqlite handle fails its +ping, and the `bolt` prober's last result ages into `"stale: last probe +…"` — and because failing checks sort first and `/health`'s top-level +`message` is the first of them, a closed `bolt` would otherwise outrank +and mask the `engine` failure the window exists to publish. So the whole +check set is snapshotted before any teardown runs and served unchanged +for the rest of the process's life. + +What is frozen is the *check set*, not the whole envelope. Two `/health` +scrapes 30 seconds apart return byte-identical documents: every field it +carries — `status`, `message`, `checks`, `version`, `commit` — comes from +the frozen set or from build info. `/ready` additionally reports +`started` and `up`, and `up` is recomputed per request as the elapsed +time since the handler was built, so it advances across the window like +it does at any other time. A scraper diffing `/ready` bodies to decide +whether the report has changed must ignore `up`; the `checks` array is +the part that is pinned. + +`/ready` reports `"starting"` throughout, exactly as it does during a +normal boot. `/health` is the endpoint that distinguishes the two: it +passes for the whole of a normal startup and fails only once a phase has +failed. + +> [!IMPORTANT] +> `/health` returns `503` from the **start** of the window. A liveness +> probe whose `periodSeconds × failureThreshold` is shorter than the +> linger will kill the container before anyone scrapes the reason, and +> the feature will appear to work in manual testing and silently not in +> production. Nothing in this repository configures a probe — those live +> in Helm charts and operator manifests — so check yours before choosing +> a value. A startup probe with a generous `failureThreshold`, or a +> liveness probe that does not start until the startup probe succeeds, is +> the usual arrangement. + +> [!WARNING] +> `SIGTERM` is not trapped. `influxd` registers only `os.Interrupt`, so a +> `systemctl stop` or a pod deletion during the window kills the process +> where it stands and the final teardown never runs. The split above is +> what limits the damage: the file locks are already gone, so the next +> start is not blocked on them. What is left behind is a stale PID file — +> the same thing an uncatchable signal leaves behind at any other point in +> the process's life, and what `--overwrite-pid-file` is for. `SIGINT` +> (Ctrl-C) does cut the window short. + +With health auth enabled the window is less useful than it looks; see +[Check detail during the window](#check-detail-during-the-window). + +**One behavior change at the default.** `Shutdown` now runs on the +startup-failure path even at `--startup-error-linger=0`. It did not +before: a failed startup returned without reaching it, so a `--pid-file` +was left behind and the next start met +`PID file exists (possible unclean shutdown or another instance already +running)`. That is now cleaned up. + ### Picking the right endpoint Use `/ready`: @@ -1036,9 +1195,12 @@ with a `subsystem` field naming the phase. This state is terminal: the process exits rather than retrying, so an orchestrator restarting the container will hit the same failure until the underlying cause is fixed. -**Note on timing:** `influxd` currently exits as soon as `run` returns, -so a scraper may or may not catch the body before the listener closes. -The log line is the reliable copy. +**Note on timing:** by default `influxd` exits as soon as `run` returns, +so a scraper will almost certainly not catch the body before the listener +closes — the log line is the reliable copy. Set +[`--startup-error-linger`](#keeping-the-endpoints-alive-after-a-failed-startup) +to hold both endpoints open, with this body frozen, long enough to be +scraped. ### `/health` 503 — sqlite not open diff --git a/cmd/influxd/launcher/cmd.go b/cmd/influxd/launcher/cmd.go index d791fab6ebe..01596bf6cf5 100644 --- a/cmd/influxd/launcher/cmd.go +++ b/cmd/influxd/launcher/cmd.go @@ -2,6 +2,7 @@ package launcher import ( "context" + "errors" "fmt" "os" "path/filepath" @@ -25,6 +26,52 @@ import ( "go.uber.org/zap/zapcore" ) +// startupErrorLingerFlag names the --startup-error-linger option. Shared by the +// option's definition and the log line holdForStartupError writes when the wait +// begins, so an operator reading the log knows which flag produced it. +const startupErrorLingerFlag = "startup-error-linger" + +// maxStartupErrorLinger caps --startup-error-linger. The window holds the HTTP +// port open on a process that has already failed, and every supervisor that +// would restart it -- systemd, a container runtime, a shell loop -- is waiting +// on that process to exit, so an unbounded value turns a failed start into an +// indefinite outage. That is a far worse failure than the one the window exists +// to report. Thirty minutes is well past any scrape interval a monitoring +// system uses and well short of the point at which nobody notices. +// +// Enforced in Launcher.holdForStartupError rather than on the option, so that +// every caller is covered and so print-config keeps reporting what the operator +// configured rather than silently rewriting it -- print-config output is +// routinely redirected into a config file. +const maxStartupErrorLinger = 30 * time.Minute + +// shutdownTimeout bounds teardown, giving in-progress requests a few seconds to +// finish. It applies to the normal exit path and to both phases of the +// startup-failure path. The check freeze that precedes them has a budget of its +// own; see freezeTimeout. +const shutdownTimeout = 2 * time.Second + +// freezeTimeout is the backstop on the whole check freeze that precedes a +// startup-failure teardown. It is emphatically not the probe budget: +// kit/check.Check.Freeze bounds every probe individually at +// check.DefaultProbeTimeout, and that per-probe bound is what keeps one slow +// subsystem from spending the time the checks after it need. This caps only the +// sum, so a check set that has grown, or a run in which many subsystems are +// slow at once, cannot hold open a process that has already failed. +// +// It is sized so that a healthy freeze never reaches it. Once it expires the +// remaining probes run on a dead context, and a cancelled probe snapshots as a +// failure that can outrank the attribution the freeze exists to preserve -- +// precisely the drift the per-probe bound was introduced to stop. The launcher +// registers on the order of fifteen checks, so the worst case at +// DefaultProbeTimeout apiece is around 7.5s; this leaves that room and then +// some for the set to grow. +// +// Separate from shutdownTimeout because the two bound unrelated work -- probing +// subsystems that are still up, versus draining in-flight HTTP requests -- and +// resizing one must not silently resize the other. +const freezeTimeout = 15 * time.Second + func errInvalidFlags(flags []string, configFile string) error { return fmt.Errorf( "error: found flags from an InfluxDB 1.x configuration in config file at %s - see https://docs.influxdata.com/influxdb/latest/reference/config-options/ for flags supported on this version of InfluxDB: %s", @@ -128,17 +175,44 @@ func cmdRunE(ctx context.Context, o *InfluxdOpts) func() error { } l.log = logger - // Start the launcher and wait for it to exit on SIGINT or SIGTERM. - if err := l.run(signals.WithStandardSignals(ctx), o); err != nil { - return err + // Start the launcher and wait for it to exit on SIGINT. SIGTERM is not + // trapped — kit/signals registers os.Interrupt and os.Kill, and SIGKILL + // cannot be caught — so a SIGTERM kills the process where it stands. + runErr := l.run(signals.WithStandardSignals(ctx), o) + if runErr != nil { + // Startup failed. Release everything a restart needs and, if the + // operator asked for it, keep /health and /ready answering long + // enough for a scraper to read which subsystem failed and why. + l.holdForStartupError(ctx, o.StartupErrorLinger) + } else { + <-l.Done() } - <-l.Done() - // Tear down the launcher, allowing it a few seconds to finish any - // in-progress requests. - shutdownCtx, cancel := context.WithTimeout(ctx, 2*time.Second) + // Tear down whatever is left, allowing it a few seconds to finish any + // in-progress requests. Derived from the outer ctx rather than the + // signal-wrapped one so a signal cannot truncate teardown. + // + // This runs on the startup-failure path too, which it did not before: + // that path used to return above and leave a --pid-file orphaned for + // the next start to trip over. + shutdownCtx, cancel := context.WithTimeout(ctx, shutdownTimeout) defer cancel() - return l.Shutdown(shutdownCtx) + serr := l.Shutdown(shutdownCtx) + + // Join rather than pick a winner. runErr leads, so the exit code and + // the first line influxd prints are what they have always been, while a + // teardown failure stays reachable through errors.Is and errors.As + // instead of living only in the log -- which is what a caller inspecting + // the error, a test among them, has to work with. errors.Join returns + // nil when both are nil and the surviving error's own message when only + // one is, so neither single-error case changes at all; only a startup + // failure whose teardown ALSO failed gains a second line. + // + // No aggregate log line here: runClosers already logs every closer + // failure at Error with the subsystem that produced it, which is the + // same reasoning holdForStartupError documents for discarding the error + // from its own phase. + return errors.Join(runErr, serr) } } @@ -155,6 +229,15 @@ type InfluxdOpts struct { PIDFile string OverwritePIDFile bool + // StartupErrorLinger is how long a failed startup keeps /health and /ready + // serving before the process exits, so a monitoring system can retrieve the + // subsystem attribution that would otherwise die with the listener. Zero, + // the default, exits immediately as before, and anything above + // maxStartupErrorLinger is capped to it. Everything except the listener and + // the PID file is released before the wait begins; see + // Launcher.holdForStartupError. + StartupErrorLinger time.Duration + AssetsPath string BoltPath string SqLitePath string @@ -233,8 +316,9 @@ func NewOpts(viper *viper.Viper) *InfluxdOpts { FluxLogEnabled: false, ReportingDisabled: false, - PIDFile: "", - OverwritePIDFile: false, + PIDFile: "", + OverwritePIDFile: false, + StartupErrorLinger: 0, BoltPath: filepath.Join(dir, bolt.DefaultFilename), SqLitePath: filepath.Join(dir, sqlite.DefaultFilename), @@ -388,6 +472,12 @@ func (o *InfluxdOpts) BindCliOpts() []cli.Opt { Default: o.OverwritePIDFile, Desc: "overwrite PID file if it already exists instead of exiting", }, + { + DestP: &o.StartupErrorLinger, + Flag: startupErrorLingerFlag, + Default: o.StartupErrorLinger, + Desc: fmt.Sprintf("how long to keep /health and /ready serving after a failed startup, so the error can be retrieved, before exiting. Set to 0 to exit immediately; capped at %s", maxStartupErrorLinger), + }, { DestP: &o.SessionLength, Flag: "session-length", diff --git a/cmd/influxd/launcher/cmd_test.go b/cmd/influxd/launcher/cmd_test.go index 9ff9e0df554..4f1db5cf096 100644 --- a/cmd/influxd/launcher/cmd_test.go +++ b/cmd/influxd/launcher/cmd_test.go @@ -4,6 +4,7 @@ import ( "context" "strings" "testing" + "time" "github.com/spf13/viper" "github.com/stretchr/testify/assert" @@ -263,3 +264,56 @@ func TestNewInfluxdCommand_HealthAuthModeInvalidEnvIgnored(t *testing.T) { assert.Equal(t, HealthAuthAuto, o.HealthAuthMode) assert.True(t, o.healthAuthRequired()) } + +// TestNewInfluxdCommand_StartupErrorLinger covers every way the option can be +// supplied. It is a duration, which is the part worth pinning: viper's +// cast.ToDurationE reads "30s" from a config file, and the derived env var name +// is generated rather than declared, so nothing else in the tree would catch a +// rename. +func TestNewInfluxdCommand_StartupErrorLinger(t *testing.T) { + t.Parallel() + + t.Run("absent", func(t *testing.T) { + t.Parallel() + + o := resolveOpts(t, viper.New()) + assert.Zero(t, o.StartupErrorLinger, "the default must exit immediately, as before") + }) + + t.Run("command line", func(t *testing.T) { + t.Parallel() + + o := resolveOpts(t, viper.New(), "--startup-error-linger=30s") + assert.Equal(t, 30*time.Second, o.StartupErrorLinger) + }) + + t.Run("config file", func(t *testing.T) { + t.Parallel() + + v := viper.New() + v.SetConfigType("yaml") + require.NoError(t, v.ReadConfig(strings.NewReader("startup-error-linger: 1m\n"))) + + o := resolveOpts(t, v) + assert.Equal(t, time.Minute, o.StartupErrorLinger) + }) +} + +// TestNewInfluxdCommand_StartupErrorLingerFromEnv pins the derived env var. +// INFLUXD_* is how a containerized influxd is configured, which is also where a +// failed startup is hardest to observe. Not parallel: t.Setenv forbids it. +func TestNewInfluxdCommand_StartupErrorLingerFromEnv(t *testing.T) { + t.Setenv("INFLUXD_STARTUP_ERROR_LINGER", "45s") + + o := resolveOpts(t, viper.New()) + assert.Equal(t, 45*time.Second, o.StartupErrorLinger) +} + +// TestPrintConfig_ReportsStartupErrorLinger keeps the option discoverable: an +// operator finds it by reading what print-config emits. +func TestPrintConfig_ReportsStartupErrorLinger(t *testing.T) { + t.Parallel() + + got := printConfig(t) + assert.Contains(t, got, "startup-error-linger: 0s") +} diff --git a/cmd/influxd/launcher/export_test.go b/cmd/influxd/launcher/export_test.go new file mode 100644 index 00000000000..3d8c77d0b58 --- /dev/null +++ b/cmd/influxd/launcher/export_test.go @@ -0,0 +1,34 @@ +package launcher + +import ( + "context" + "testing" + "time" +) + +// Test-only exports for the external launcher_test package. Both packages +// compile into the same test binary, so an identifier defined here is visible +// there while staying out of the production build. Nothing in this file may be +// referenced by non-test code. + +// HoldForStartupError exposes holdForStartupError, the wait a failed startup +// takes before the process exits. It blocks for up to d; a test that needs it +// to end sooner calls CancelRun, which is what a SIGINT does in production. +func (tl *TestLauncher) HoldForStartupError(ctx context.Context, d time.Duration) { + tl.Launcher.holdForStartupError(ctx, d) +} + +// CancelRun cancels the context run was given, closing Done and so ending any +// hold in progress. Safe only after Run has been called: run is what installs +// the cancel func. +func (tl *TestLauncher) CancelRun() { + tl.Launcher.cancel() +} + +// RequireReturnsWithin exposes requireReturnsWithin, so the internal and +// external test packages share one bounded-wait helper rather than a copy +// each. See there for why a blocking call must never be made directly. +func RequireReturnsWithin(t *testing.T, d time.Duration, fn func()) { + t.Helper() + requireReturnsWithin(t, d, fn) +} diff --git a/cmd/influxd/launcher/launcher.go b/cmd/influxd/launcher/launcher.go index b8de2f84f6a..c71cb6ea037 100644 --- a/cmd/influxd/launcher/launcher.go +++ b/cmd/influxd/launcher/launcher.go @@ -10,8 +10,8 @@ import ( nethttp "net/http" "os" "path/filepath" + "slices" "strconv" - "strings" "sync" "time" @@ -174,6 +174,21 @@ type Launcher struct { // a startup failure to, used only to notice a failure that reached no // attribution at all. Written and read from run's goroutine. failedSubsystem string + + // shutdownMu guards the closer list, the accumulated teardown state, and + // httpServing. Teardown runs in two phases on the startup-failure path -- + // see shutdownSubsystems -- so a closer is consumed from m.closers as it + // runs rather than the whole teardown being gated by a single sync.Once. + shutdownMu sync.Mutex + + // httpServing reports whether runHTTP bound a listener. Consulted by + // holdForStartupError: with no listener there is nothing to scrape, and + // waiting only delays the error. It is guarded rather than plain because + // run writes it from its own goroutine while holdForStartupError, which is + // exported to tests and is the only reader, is reachable from another. + httpServing bool + shutdownErrs []error + shutdownDone bool } type stoppingScheduler interface { @@ -206,20 +221,29 @@ func (m *Launcher) ReadyCheckNames() []string { return m.checkHandler.ReadyCheckNames() } -// Shutdown shuts down the HTTP server and waits for all services to clean up. +// Shutdown closes whatever is left of the launcher and waits for all services +// to clean up. It is the final teardown phase: after it returns, nothing the +// launcher registered is still running. +// +// Every registered closer runs at most once across all calls, because each is +// consumed as it runs. A caller that cannot tell whether the launcher was +// already torn down — in whole, or in part via shutdownSubsystems — can call +// Shutdown unconditionally without double-closing a store or reporting a +// spurious error for already-released state. The returned error accumulates +// every phase's closer failures, so a single call site reports the whole +// teardown. func (m *Launcher) Shutdown(ctx context.Context) error { - var errs []string - - // Shut down subsystems in the reverse order of their registration. - for i := len(m.closers); i > 0; i-- { - lc := m.closers[i-1] - m.log.Info("Stopping subsystem", zap.String("subsystem", lc.label)) - if err := lc.closer(ctx); err != nil { - m.log.Error("Failed to stop subsystem", zap.String("subsystem", lc.label), zap.Error(err)) - errs = append(errs, err.Error()) - } + m.shutdownMu.Lock() + defer m.shutdownMu.Unlock() + if m.shutdownDone { + return m.shutdownError() } + m.runClosers(ctx) + + // Safe only here, and not in shutdownSubsystems: the HTTP serve goroutine + // is tracked in m.wg and returns only once the server closes, which the + // closer above has now done. m.wg.Wait() // N.B. We ignore any errors here because Sync is known to fail with EINVAL @@ -229,10 +253,204 @@ func (m *Launcher) Shutdown(ctx context.Context) error { // See: https://github.com/uber-go/zap/issues/328 _ = m.log.Sync() - if len(errs) > 0 { - return fmt.Errorf("failed to shut down server: [%s]", strings.Join(errs, ",")) + m.shutdownDone = true + return m.shutdownError() +} + +// shutdownSubsystems runs every registered closer except the two that describe +// the process itself — the HTTP server's and the PID file's — releasing the +// bolt flock, the sqlite file and the engine directory while leaving the +// listener, and so /health and /ready, serving. It is the first of the two +// teardown phases used by holdForStartupError; Shutdown is the second and +// releases both of the ones kept here. +// +// The PID file is retained deliberately, and not merely as a companion to the +// listener. It is the interlock that stops a second influxd starting against +// the same data directory, and this process is still running and still holding +// its port: releasing it early would let a concurrent start past the check that +// exists to catch exactly this, only to fail it on "address already in use" — +// a worse error, naming the wrong cause. A PID file must describe a live +// process for as long as the process is alive. +// +// It deliberately does not wait on m.wg. The HTTP serve goroutine is tracked +// there and returns only once the server closes, so waiting here would block +// for exactly as long as the listener is retained. +func (m *Launcher) shutdownSubsystems(ctx context.Context) error { + m.shutdownMu.Lock() + defer m.shutdownMu.Unlock() + if m.shutdownDone { + return m.shutdownError() + } + + m.runClosers(ctx, SubsystemHTTPServer, SubsystemPIDFile) + return m.shutdownError() +} + +// runClosers runs the registered closers in reverse registration order, +// skipping any whose label is in keep, and records each failure. The closers +// it is about to run are removed from m.closers before any of them run, so no +// closer can run twice even if one panics. The kept closers stay in +// registration order, so a later phase still tears down in reverse. +// +// Caller must hold shutdownMu. +func (m *Launcher) runClosers(ctx context.Context, keep ...string) { + kept := make([]labeledCloser, 0, len(keep)) + pending := make([]labeledCloser, 0, len(m.closers)) + for _, lc := range m.closers { + if slices.Contains(keep, lc.label) { + kept = append(kept, lc) + continue + } + pending = append(pending, lc) } - return nil + m.closers = kept + + // Shut down subsystems in the reverse order of their registration. + for i := len(pending); i > 0; i-- { + lc := pending[i-1] + m.log.Info("Stopping subsystem", zap.String("subsystem", lc.label)) + if err := lc.closer(ctx); err != nil { + m.log.Error("Failed to stop subsystem", zap.String("subsystem", lc.label), zap.Error(err)) + m.shutdownErrs = append(m.shutdownErrs, fmt.Errorf("%s: %w", lc.label, err)) + } + } +} + +// shutdownError renders the closer failures accumulated across every teardown +// phase run so far. errors.Join rather than a flattened message: every failure +// stays reachable through errors.Is and errors.As, and each one already names +// the subsystem it came from. Caller must hold shutdownMu. +func (m *Launcher) shutdownError() error { + if len(m.shutdownErrs) == 0 { + return nil + } + return fmt.Errorf("failed to shut down server: %w", errors.Join(m.shutdownErrs...)) +} + +// freezeChecks pins /health and /ready to the report they serve right now, so +// the teardown that follows cannot rewrite it. This is the whole reason the +// freeze exists rather than an optimization: sqlite.SqlStore.Check pings a +// closed handle, and bolt.KVStore.Check ages into "stale: last probe ..." once +// its prober stops. check.Responses sorts failures first and then by name, and +// /health's top-level message is the first of them, so a closed bolt would +// outrank and mask the engine failure the hold exists to publish. +// +// The freeze is bounded at two levels, and the distinction between them is +// load-bearing. check.Check.Freeze gives every probe a context of its own, +// bounded at check.DefaultProbeTimeout, so a wedged subsystem holds the process +// open for the length of its own probe and no longer -- that is what stops an +// early slow checker leaving the rest to snapshot as failed probes, which, +// since failures sort ahead of passes by name, could outrank and mask the very +// attribution this freeze is taken to preserve. freezeTimeout then caps the sum +// as a backstop, sized so a healthy freeze never reaches it. +func (m *Launcher) freezeChecks(ctx context.Context) { + // m.httpServing implies m.checkHandler is non-nil: the handler is built + // before runHTTP is called, so a bound listener means both exist. No guard + // here would ever fire. + ctx, cancel := context.WithTimeout(ctx, freezeTimeout) + defer cancel() + m.checkHandler.FreezeChecks(ctx) +} + +// setHTTPServing records that runHTTP bound a listener, so a failure from here +// on has somewhere to be read from. +func (m *Launcher) setHTTPServing() { + m.shutdownMu.Lock() + defer m.shutdownMu.Unlock() + m.httpServing = true +} + +// httpIsServing reports whether runHTTP bound a listener. It takes and releases +// shutdownMu, so a caller must not already hold it -- holdForStartupError reads +// this before the phased teardown that acquires it. +func (m *Launcher) httpIsServing() bool { + m.shutdownMu.Lock() + defer m.shutdownMu.Unlock() + return m.httpServing +} + +// cappedLinger bounds d at maxStartupErrorLinger, warning when it has to. +// +// The cap is enforced here rather than on the option so that it cannot be +// bypassed -- every path into the window goes through holdForStartupError -- +// and so print-config keeps reporting the configured value rather than a +// rewritten one. The warning is the operator's only notice that the duration +// they chose is not the duration they will get, and it names the flag so the +// line is actionable on its own. +func (m *Launcher) cappedLinger(d time.Duration) time.Duration { + if d <= maxStartupErrorLinger { + return d + } + m.log.Warn("Startup error linger exceeds the maximum; capping it", + zap.String("flag", startupErrorLingerFlag), + zap.Duration("requested", d), + zap.Duration("maximum", maxStartupErrorLinger)) + return maxStartupErrorLinger +} + +// holdForStartupError releases everything the failed process no longer needs +// and then keeps /health and /ready scrapeable for d, so the startup error +// latched by failSubsystem can be retrieved before the process exits. +// +// The check set is frozen first — see freezeChecks — and then teardown is +// split around the wait. Every subsystem except the HTTP listener and the PID +// file is closed before the process parks, so the bolt flock, the sqlite file +// and the engine directory belong to the next run rather than to one that +// already failed. The listener is what the wait needs; the PID file is what +// keeps the next run from starting on top of this one while it still holds the +// port (see shutdownSubsystems). The Shutdown that follows releases both. +// +// Non-check requests are unaffected: the delegate handler is installed as the +// last statement of a successful run, so on this path there is none and they +// still get the 503 "starting" body. +// +// It returns immediately, tearing nothing down and freezing nothing, when d is +// non-positive or no listener was ever established: there is then nothing to +// scrape, and the caller's Shutdown does the whole teardown in one phase +// exactly as before. At the other end d is capped; see cappedLinger. +// +// The wait also ends when the launcher's context is done, which covers a +// SIGINT and a serve goroutine that already gave up and cancelled. It is NOT +// cut short by SIGTERM: influxd traps only os.Interrupt (see +// kit/signals.WithStandardSignals), so a SIGTERM during the window kills the +// process where it stands. Splitting the teardown is what limits the damage: +// the file locks are already gone, and what is left behind is a stale PID file +// — the same thing any uncatchable signal leaves behind at any other point in +// the process's life, and what --overwrite-pid-file is for. +// +// ctx bounds the teardown at shutdownTimeout and the freeze at freezeTimeout, +// not the wait. Pass the process context rather than the signal-wrapped one, so +// a signal racing either neither truncates the teardown nor poisons the freeze. +func (m *Launcher) holdForStartupError(ctx context.Context, d time.Duration) { + if d <= 0 || !m.httpIsServing() { + return + } + d = m.cappedLinger(d) + m.freezeChecks(ctx) + + subsysCtx, cancel := context.WithTimeout(ctx, shutdownTimeout) + // Failures are logged per subsystem by runClosers and accumulate into the + // error the caller's Shutdown returns, so there is nothing to report here. + _ = m.shutdownSubsystems(subsysCtx) + cancel() + + m.log.Warn("Startup failed; serving /health and /ready before exiting", + zap.Duration(startupErrorLingerFlag, d), zap.Int("port", m.httpPort)) + + timer := time.NewTimer(d) + defer timer.Stop() + select { + case <-timer.C: + case <-m.Done(): + } + + // Cancel the launcher context so the final Shutdown's m.wg.Wait() has the + // same precondition it has on the normal exit path, where cmdRunE reaches + // Shutdown only after <-l.Done(). Not load-bearing today — the serve + // goroutine returns when its closer shuts the server down, not on this + // cancel, and runReporter is the only other wg member and never starts on + // a failure path — so it is tidiness, not a fix. + m.cancel() } func (m *Launcher) Done() <-chan struct{} { @@ -471,6 +689,9 @@ func (m *Launcher) run(ctx context.Context, opts *InfluxdOpts) (err error) { if err != nil { return m.failSubsystem(SubsystemHTTPServer, "Failed starting HTTP server", err) } + // A listener is bound, so a failure from here on has somewhere to be read + // from; see holdForStartupError. + m.setHTTPServing() m.reg = prom.NewRegistry(m.log.With(zap.String("service", "prom_registry"))) m.reg.MustRegister(collectors.NewGoCollector()) diff --git a/cmd/influxd/launcher/shutdown_test.go b/cmd/influxd/launcher/shutdown_test.go new file mode 100644 index 00000000000..1e07174e621 --- /dev/null +++ b/cmd/influxd/launcher/shutdown_test.go @@ -0,0 +1,369 @@ +package launcher + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/influxdata/influxdb/v2/http" + "github.com/influxdata/influxdb/v2/kit/check" + "github.com/stretchr/testify/require" + "go.uber.org/zap" + "go.uber.org/zap/zaptest" + "go.uber.org/zap/zaptest/observer" +) + +// requireReturnsWithin fails the test unless fn returns within d. +// +// On timeout the goroutine running fn is abandoned, still holding whatever fn +// holds: there is no way to interrupt it from here. So fn must have an +// independent way out — cancelling the launcher context ends a hold — and the +// caller must arrange one, typically as a t.Cleanup, rather than leaving a +// blocked goroutine to outlive the test binary's other cases. +// +// It exists because the alternative is calling a blocking method directly: +// a regression that never returns then hangs the whole package instead of +// failing one subtest. +func requireReturnsWithin(t *testing.T, d time.Duration, fn func()) { + t.Helper() + + done := make(chan struct{}) + go func() { + defer close(done) + fn() + }() + + select { + case <-done: + case <-time.After(d): + require.FailNowf(t, "call did not return", "still blocked after %s", d) + } +} + +// newShutdownLauncher returns a Launcher with just enough wired up to tear +// down: a logger, a check handler, and the run context holdForStartupError +// waits on. No subsystem is started; closers are added by the test. +func newShutdownLauncher(t *testing.T) *Launcher { + t.Helper() + + m := NewLauncher() + m.log = zaptest.NewLogger(t) + m.checkHandler = http.NewHealthReadyHandler(m.log) + + ctx, cancel := context.WithCancel(context.Background()) + m.cancel = cancel + m.doneChan = ctx.Done() + t.Cleanup(cancel) + + return m +} + +// recordCloser registers a closer under label that appends its label to *order +// when it runs and then returns err. Closers run serially under shutdownMu, so +// the slice needs no lock of its own. +func recordCloser(m *Launcher, order *[]string, label string, err error) { + m.closers = append(m.closers, labeledCloser{ + label: label, + closer: func(context.Context) error { + *order = append(*order, label) + return err + }, + }) +} + +// TestLauncher_ShutdownPhases pins the split: phase 1 releases everything but +// the listener and the PID file, phase 2 releases those two, and no closer runs +// twice across the two. The PID file goes last of all, which is reverse +// registration order -- it is written before anything else is opened. +func TestLauncher_ShutdownPhases(t *testing.T) { + ctx := context.Background() + m := newShutdownLauncher(t) + + var order []string + recordCloser(m, &order, SubsystemPIDFile, nil) + recordCloser(m, &order, SubsystemKV, nil) + recordCloser(m, &order, SubsystemHTTPServer, nil) + recordCloser(m, &order, SubsystemEngine, nil) + + require.NoError(t, m.shutdownSubsystems(ctx)) + require.Equal(t, []string{SubsystemEngine, SubsystemKV}, order, + "phase 1 must run every closer but the listener's and the PID file's, "+ + "once, in reverse registration order") + + require.NoError(t, m.Shutdown(ctx)) + require.Equal(t, + []string{SubsystemEngine, SubsystemKV, SubsystemHTTPServer, SubsystemPIDFile}, + order, "phase 2 must run the two kept closers and nothing else again") +} + +// TestLauncher_ShutdownWithoutPhaseOne is the unsplit path, which is what every +// existing caller takes and what --startup-error-linger=0 still takes. +func TestLauncher_ShutdownWithoutPhaseOne(t *testing.T) { + ctx := context.Background() + m := newShutdownLauncher(t) + + var order []string + recordCloser(m, &order, SubsystemKV, nil) + recordCloser(m, &order, SubsystemHTTPServer, nil) + + require.NoError(t, m.Shutdown(ctx)) + require.Equal(t, []string{SubsystemHTTPServer, SubsystemKV}, order) + + require.NoError(t, m.Shutdown(ctx), "a second Shutdown must be a no-op") + require.Len(t, order, 2) +} + +// TestLauncher_ShutdownAccumulatesErrors pins that a single Shutdown call site +// reports failures from both phases, which is what lets cmdRunE call Shutdown +// unconditionally and still report the whole teardown. Each failure stays +// individually matchable: the accumulated error joins them rather than +// flattening them into one message. +func TestLauncher_ShutdownAccumulatesErrors(t *testing.T) { + ctx := context.Background() + m := newShutdownLauncher(t) + + errKV := errors.New("kv close failed") + errListener := errors.New("listener close failed") + + var order []string + recordCloser(m, &order, SubsystemKV, errKV) + recordCloser(m, &order, SubsystemHTTPServer, errListener) + + err := m.shutdownSubsystems(ctx) + require.ErrorIs(t, err, errKV) + require.NotErrorIs(t, err, errListener, "phase 1 has not touched the listener") + require.ErrorContains(t, err, SubsystemKV, "a closer failure must name its subsystem") + + err = m.Shutdown(ctx) + require.ErrorIs(t, err, errKV) + require.ErrorIs(t, err, errListener) + require.ErrorContains(t, err, SubsystemHTTPServer) + + again := m.Shutdown(ctx) + require.Equal(t, err.Error(), again.Error(), + "a repeat call must report the same accumulated error") + require.Len(t, order, 2, "a repeat call must run nothing") +} + +// TestLauncher_HoldForStartupError_NoWait covers every case in which the hold +// declines to do anything at all: there is nothing to scrape, or the operator +// asked for no window. Nothing may be torn down and nothing may be frozen — +// the caller's Shutdown still owns the whole teardown, exactly as before this +// flag existed. +func TestLauncher_HoldForStartupError_NoWait(t *testing.T) { + ctx := context.Background() + + for _, tc := range []struct { + name string + linger time.Duration + httpServing bool + }{ + {name: "zero linger", linger: 0, httpServing: true}, + {name: "negative linger", linger: -time.Second, httpServing: true}, + {name: "no listener", linger: time.Hour, httpServing: false}, + } { + t.Run(tc.name, func(t *testing.T) { + m := newShutdownLauncher(t) + if tc.httpServing { + m.setHTTPServing() + } + + var order []string + recordCloser(m, &order, SubsystemKV, nil) + recordCloser(m, &order, SubsystemHTTPServer, nil) + + requireReturnsWithin(t, time.Second, func() { + m.holdForStartupError(ctx, tc.linger) + }) + require.Empty(t, order, "the hold tore something down") + + // Not frozen: a check registered afterwards still appears. + m.checkHandler.AddNamedHealthCheck(check.Named("late", check.ErrCheck(func() error { return nil }))) + require.Contains(t, healthCheckNames(t, m), "late") + }) + } +} + +// TestLauncher_HoldForStartupError_ReleasesAndWaits is the shape of the real +// window: everything but the listener and the PID file is released before the +// wait, those two survive it, and the wait ends when the launcher context is +// cancelled — which is what a SIGINT does. +func TestLauncher_HoldForStartupError_ReleasesAndWaits(t *testing.T) { + ctx := context.Background() + m := newShutdownLauncher(t) + m.setHTTPServing() + + var order []string + recordCloser(m, &order, SubsystemPIDFile, nil) + recordCloser(m, &order, SubsystemKV, nil) + recordCloser(m, &order, SubsystemHTTPServer, nil) + + // Long enough that the timer cannot be what ends this: only the cancel can. + const linger = time.Hour + released := make(chan struct{}) + go func() { + // The hold releases the subsystems before it parks, so this fires well + // before the cancel below. Only kv is released: the listener and the + // PID file are both held for the window. + for { + m.shutdownMu.Lock() + done := len(order) == 1 + m.shutdownMu.Unlock() + if done { + close(released) + return + } + time.Sleep(time.Millisecond) + } + }() + + go func() { + <-released + m.cancel() + }() + + requireReturnsWithin(t, 30*time.Second, func() { + m.holdForStartupError(ctx, linger) + }) + + require.Equal(t, []string{SubsystemKV}, order, + "the listener's and the PID file's closers must survive the window") + require.NoError(t, m.Shutdown(ctx)) + require.Equal(t, []string{SubsystemKV, SubsystemHTTPServer, SubsystemPIDFile}, order) +} + +// TestLauncher_HoldForStartupError_AlreadyDone covers a launcher whose context +// was cancelled before the hold began — a serve goroutine that gave up, or a +// signal that arrived during the failure itself. The teardown still runs; only +// the wait is skipped. +func TestLauncher_HoldForStartupError_AlreadyDone(t *testing.T) { + ctx := context.Background() + m := newShutdownLauncher(t) + m.setHTTPServing() + m.cancel() + + var order []string + recordCloser(m, &order, SubsystemKV, nil) + recordCloser(m, &order, SubsystemHTTPServer, nil) + + requireReturnsWithin(t, 5*time.Second, func() { + m.holdForStartupError(ctx, time.Hour) + }) + require.Equal(t, []string{SubsystemKV}, order) +} + +// TestLauncher_HoldForStartupError_ElapsesTimer pins that the timer alone ends +// the wait, with no cancel involved. Short enough to keep the test quick, long +// enough that a hold returning instantly would be visible as an ordering +// failure rather than passing by luck. +func TestLauncher_HoldForStartupError_ElapsesTimer(t *testing.T) { + ctx := context.Background() + m := newShutdownLauncher(t) + m.setHTTPServing() + + const linger = 50 * time.Millisecond + start := time.Now() + requireReturnsWithin(t, 30*time.Second, func() { + m.holdForStartupError(ctx, linger) + }) + require.GreaterOrEqual(t, time.Since(start), linger) +} + +// TestLauncher_FreezeChecks_PinsStartupAttribution runs the freeze over the +// state a real startup failure leaves behind: one subsystem latched with its +// reason, and every gate downstream of it reporting that it was never reached. +// Both envelopes must survive the freeze unchanged, and the set must be closed +// to anything registered afterwards — a closer that registers a check while +// tearing down cannot rewrite the report. +func TestLauncher_FreezeChecks_PinsStartupAttribution(t *testing.T) { + ctx := context.Background() + m := newCheckLauncher(t) + + require.Error(t, m.failSubsystem(SubsystemEngine, "Failed to open engine", + errors.New("not a directory"))) + m.failUnreachedGates(ctx) + + healthBefore, healthStatusBefore := serveCheck(t, m, "/health") + readyBefore, readyStatusBefore := serveCheck(t, m, "/ready") + + m.freezeChecks(ctx) + + m.checkHandler.AddNamedHealthCheck(check.Named("late", check.ErrCheck(func() error { + return errors.New("registered while tearing down") + }))) + m.checkHandler.AddNamedReadyCheck(check.Named("late", check.ErrCheck(func() error { + return errors.New("registered while tearing down") + }))) + + healthAfter, healthStatusAfter := serveCheck(t, m, "/health") + readyAfter, readyStatusAfter := serveCheck(t, m, "/ready") + + require.Equal(t, healthBefore, healthAfter) + require.Equal(t, healthStatusBefore, healthStatusAfter) + require.Equal(t, readyBefore, readyAfter) + require.Equal(t, readyStatusBefore, readyStatusAfter) + + require.NotContains(t, checkNamesOf(healthAfter.Checks), "late") + require.NotContains(t, checkNamesOf(readyAfter.Checks), "late") + + // The attribution itself, so this test fails loudly if the freeze ever + // starts pinning an empty report. + require.Contains(t, healthAfter.Message, "Failed to open engine") + require.Contains(t, checkNamesOf(healthAfter.Checks), SubsystemEngine) +} + +func checkNamesOf(rs []check.BasicResponse) []string { + out := make([]string, len(rs)) + for i, r := range rs { + out[i] = r.Name() + } + return out +} + +// healthCheckNames returns the names on the launcher's /health envelope. +func healthCheckNames(t *testing.T, m *Launcher) []string { + t.Helper() + body, _ := serveCheck(t, m, "/health") + return checkNamesOf(body.Checks) +} + +// TestLauncher_CappedLinger pins the upper bound on --startup-error-linger. +// The window holds the HTTP port on a process that has already failed, and the +// supervisor that would restart it is waiting on that process to exit, so an +// operator typo — an hour meant as a minute, a duration string read as +// something else — must not be able to turn a failed start into an indefinite +// outage. The warning is pinned alongside the value because it is the only +// notice an operator gets that the duration they chose is not the one they +// will get. +func TestLauncher_CappedLinger(t *testing.T) { + for _, tc := range []struct { + name string + in time.Duration + want time.Duration + wantWarn bool + }{ + {name: "under the cap", in: 30 * time.Second, want: 30 * time.Second}, + {name: "at the cap", in: maxStartupErrorLinger, want: maxStartupErrorLinger}, + {name: "over the cap", in: 24 * time.Hour, want: maxStartupErrorLinger, wantWarn: true}, + } { + t.Run(tc.name, func(t *testing.T) { + m := newShutdownLauncher(t) + core, logs := observer.New(zap.WarnLevel) + m.log = zap.New(core) + + require.Equal(t, tc.want, m.cappedLinger(tc.in)) + + if !tc.wantWarn { + require.Zero(t, logs.Len(), "capping nothing must not warn") + return + } + require.Equal(t, 1, logs.Len(), + "an operator whose value was capped must be told exactly once") + entry := logs.All()[0] + require.Contains(t, entry.Message, "capping") + require.Equal(t, startupErrorLingerFlag, entry.ContextMap()["flag"], + "the warning must name the flag to be actionable on its own") + }) + } +} diff --git a/cmd/influxd/launcher/startup_failure_test.go b/cmd/influxd/launcher/startup_failure_test.go index aebc62041b2..dab879aa262 100644 --- a/cmd/influxd/launcher/startup_failure_test.go +++ b/cmd/influxd/launcher/startup_failure_test.go @@ -5,10 +5,14 @@ import ( "os" "path/filepath" "testing" + "time" + "github.com/influxdata/influxdb/v2/bolt" "github.com/influxdata/influxdb/v2/cmd/influxd/launcher" "github.com/influxdata/influxdb/v2/kit/check" + "github.com/influxdata/influxdb/v2/kit/check/checktest" "github.com/stretchr/testify/require" + bbolt "go.etcd.io/bbolt" ) // readyBody mirrors the JSON shape served by /ready. Its checks are populated @@ -121,6 +125,135 @@ func TestLauncher_StartupFailure_EngineOpen(t *testing.T) { assertEngineFailure(t, l) } +// failEngineOpen puts a regular file where the engine expects its directory, so +// engine.Open fails with ENOTDIR well after runHTTP has a listener bound. It is +// the same failure the two tests above force, hoisted because the linger tests +// need it too. +func failEngineOpen(t *testing.T, l *launcher.TestLauncher) { + t.Helper() + require.NoError(t, os.WriteFile(filepath.Join(l.Path, "engine"), + []byte("not a directory"), 0600)) +} + +// fetchCheckDocuments returns the /health and /ready bodies with the values that +// move on their own replaced by sentinels, so two captures taken a moment apart +// can be compared for equality. See checktest.Normalize for which those are. +// +// The presence assertions stay here rather than moving into Normalize, which +// masks only what it finds: /health carries neither field, so a Normalize that +// required them could not serve both endpoints. That /ready reports them at all +// is this endpoint's contract, and worth failing on. +func fetchCheckDocuments(t *testing.T, l *launcher.TestLauncher) (health, ready map[string]any) { + t.Helper() + + health = make(map[string]any) + status := httpGetJSON(t, l.URL().String()+"/health", "", &health) + require.Equal(t, nethttp.StatusServiceUnavailable, status) + + ready = make(map[string]any) + status = httpGetJSON(t, l.URL().String()+"/ready", "", &ready) + require.Equal(t, nethttp.StatusServiceUnavailable, status) + require.Contains(t, ready, checktest.FieldStarted) + require.Contains(t, ready, checktest.FieldUp) + + return checktest.Normalize(t, health), checktest.Normalize(t, ready) +} + +// TestLauncher_StartupFailure_LingerServesFrozenAttribution is the end-to-end +// shape of --startup-error-linger: a failed startup releases the PID file and +// the store locks a restart needs, and goes on serving the report it had at the +// moment of failure until the window closes. +func TestLauncher_StartupFailure_LingerServesFrozenAttribution(t *testing.T) { + l := launcher.NewTestLauncherServer() + + // Outside l.Path: TestLauncher.Shutdown removes that tree wholesale, which + // would make "the PID file is gone" pass without the teardown doing + // anything at all. + pidFile := filepath.Join(t.TempDir(), "influxd.pid") + failEngineOpen(t, l) + + defer func() { require.NoError(t, l.Shutdown(ctx)) }() + + err := l.Run(t, ctx, func(o *launcher.InfluxdOpts) { o.PIDFile = pidFile }) + require.Error(t, err, "engine.Open must fail with a file in place of its directory") + require.FileExists(t, pidFile, "run must have written the PID file before failing") + + assertEngineFailure(t, l) + healthBefore, readyBefore := fetchCheckDocuments(t, l) + + // A window long enough that nothing but the cancel below can end it. + held := make(chan struct{}) + go func() { + defer close(held) + l.HoldForStartupError(ctx, time.Hour) + }() + + // Phase 1 has run once the bolt flock is released: it belongs to the next + // run, not to this one. The endpoints are still answering at that point, + // which is the whole property under test -- the state a restart needs is + // released while the report stays readable. A live handle makes Open block + // for the timeout and fail, so this polls rather than asserting once. + boltPath := filepath.Join(l.Path, bolt.DefaultFilename) + require.Eventually(t, func() bool { + db, err := bbolt.Open(boltPath, 0600, &bbolt.Options{Timeout: 10 * time.Millisecond}) + if err != nil { + return false + } + return db.Close() == nil + }, 30*time.Second, 10*time.Millisecond, + "the bolt flock was not released while the endpoints were still up") + + // The PID file is not released with it, and that asymmetry is the point. + // This process is still running and still holding its port, so the + // interlock that keeps a second influxd off this directory has to outlive + // the window: released here, a concurrent start would get past the check + // that exists to catch exactly this and fail on "address already in use" + // instead -- a worse error, naming the wrong cause. + require.FileExists(t, pidFile, + "the PID file was released while the process still held the port") + + healthAfter, readyAfter := fetchCheckDocuments(t, l) + require.Equal(t, healthBefore, healthAfter, + "the frozen /health document changed after the stores were torn down") + require.Equal(t, readyBefore, readyAfter, + "the frozen /ready document changed after the stores were torn down") + + l.CancelRun() + launcher.RequireReturnsWithin(t, 30*time.Second, func() { <-held }) + + // Phase 2 is what releases the PID file, and it runs on this path now. + require.NoError(t, l.Shutdown(ctx)) + require.NoFileExists(t, pidFile) +} + +// TestLauncher_StartupFailure_NoLingerTearsNothingDown pins the default. At +// zero the hold does nothing at all — no freeze, no early teardown — and the +// whole teardown belongs to Shutdown, exactly as before the flag existed. +func TestLauncher_StartupFailure_NoLingerTearsNothingDown(t *testing.T) { + l := launcher.NewTestLauncherServer() + + pidFile := filepath.Join(t.TempDir(), "influxd.pid") + failEngineOpen(t, l) + + // Registered before the first assertion so a failing require cannot leak a + // running launcher. Shutdown is idempotent, so the explicit call below is + // still the one that proves the PID file is released. + defer func() { require.NoError(t, l.Shutdown(ctx)) }() + + err := l.Run(t, ctx, func(o *launcher.InfluxdOpts) { o.PIDFile = pidFile }) + require.Error(t, err, "engine.Open must fail with a file in place of its directory") + + launcher.RequireReturnsWithin(t, 5*time.Second, func() { l.HoldForStartupError(ctx, 0) }) + require.FileExists(t, pidFile, "a zero linger must tear nothing down") + assertEngineFailure(t, l) + + // Shutdown is what releases it, and cmdRunE now reaches Shutdown on this + // path — which it did not before, leaving the PID file for the next start + // to trip over. + require.NoError(t, l.Shutdown(ctx)) + require.NoFileExists(t, pidFile) +} + // TestLauncher_StartupFailure_PriorVersion covers the prior-version check, // which called os.Exit(1) directly. That skipped every deferred function in // run — the HTTP server's closer and the PID file's among them — and made the diff --git a/http/check_handler.go b/http/check_handler.go index 0aa002ebd93..c3854b31b19 100644 --- a/http/check_handler.go +++ b/http/check_handler.go @@ -1,6 +1,7 @@ package http import ( + "context" "encoding/json" "io" "net/http" @@ -432,6 +433,51 @@ func (h *HealthReadyHandler) AddNamedHealthCheck(nc check.NamedChecker) { // in registration order. func (h *HealthReadyHandler) ReadyCheckNames() []string { return h.check.ReadyCheckNames() } +// msgFrozenAuthDep is the message carried by the auth dependency checker +// FreezeChecks installs. Nothing renders it -- detail reads only the status -- +// but a checker that fails without saying why is a trap for the next reader. +const msgFrozenAuthDep = "check set frozen: the store credentials resolve against is being closed" + +// frozenAuthDep is the auth dependency checker FreezeChecks installs over any +// existing one. A package-level value rather than a closure per call: it is +// immutable, and Check is invoked once per credentialed request for the rest +// of the process's life. +var frozenAuthDep = check.CheckerFunc(func(context.Context) check.Response { + return check.Fail(msgFrozenAuthDep) +}) + +// FreezeChecks makes /health and /ready serve a static snapshot of what they +// report right now, for the rest of the process's life. See check.Check.Freeze: +// the registered set and its order are unchanged, only the values are pinned. +// No wire format changes -- /health still reports pass/fail, /ready still +// reports "ready"/"starting", same status codes. +// +// It also retires the auth dependency checker, if one was installed. That +// checker stands in for the store credential resolution reads (see +// SetAuthDependencyChecker), and a caller freezing the checks is on its way to +// closing that store, after which no caller can be identified at all. Pinning +// it to fail is both true and the case detail already has a rule for -- "could +// not ask who the caller is" yields detailNames -- and it makes the answer the +// same for the whole frozen period. Left live it would instead answer detailNone +// until the store's own probe aged out and only then detailNames, changing the +// body's shape mid-window and starting at the least useful level. +// +// When no auth dependency checker was installed, none is installed here: +// nothing was gated on one, the store may still be resolvable, and a failing +// one would withhold detail an operator can legitimately have. That branch is +// not a production path -- the launcher installs a checker for every real +// (bolt) KV store and only skips it for the in-memory store, which is +// test-only -- so on a real server this always retires a live checker. +// +// The load-then-store races nothing: the launcher installs at most one +// checker, before it begins serving, and never removes one. +func (h *HealthReadyHandler) FreezeChecks(ctx context.Context) { + h.check.Freeze(ctx) + if h.authDep.Load() != nil { + h.authDep.Store(&checkerHolder{frozenAuthDep}) + } +} + // SetHandler installs the delegate handler used for any request that is not // /health or /ready. A nil next is ignored to prevent a nil delegate from // being published. Note: Go's typed-nil-through-interface gotcha means a diff --git a/http/check_handler_freeze_test.go b/http/check_handler_freeze_test.go new file mode 100644 index 00000000000..19f7a43ee8c --- /dev/null +++ b/http/check_handler_freeze_test.go @@ -0,0 +1,172 @@ +package http + +import ( + "context" + "encoding/json" + "io" + "net/http" + "testing" + + platform "github.com/influxdata/influxdb/v2" + "github.com/influxdata/influxdb/v2/kit/check" + "github.com/influxdata/influxdb/v2/kit/check/checktest" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/zap/zaptest" +) + +// captureCheckDocuments fetches /health and /ready and returns their decoded +// bodies with the values that move on their own replaced by sentinels, so two +// captures taken either side of the freeze can be compared for equality. See +// checktest.Normalize for which values those are and why each is masked. +func captureCheckDocuments(t *testing.T, h http.Handler) (health, ready map[string]any) { + t.Helper() + + healthRes := doRequest(t, h, http.MethodGet, "/health") + defer closeBody(t, healthRes) + health = normalizedBody(t, healthRes) + + readyRes := doRequest(t, h, http.MethodGet, "/ready") + defer closeBody(t, readyRes) + ready = normalizedBody(t, readyRes) + + return health, ready +} + +// normalizedBody reads a rendered check document off res and normalizes it. +func normalizedBody(t *testing.T, res *http.Response) map[string]any { + t.Helper() + body, err := io.ReadAll(res.Body) + require.NoError(t, err) + return checktest.NormalizeJSON(t, body) +} + +// TestHealthReadyHandler_FreezeChecks_BodiesUnchanged is the wire-compatibility +// pin for the freeze: a frozen handler serves the same document it served a +// moment earlier, with the same status codes. Freezing is meant to be invisible +// to a scraper except that the answer stops moving. +func TestHealthReadyHandler_FreezeChecks_BodiesUnchanged(t *testing.T) { + h := NewHealthReadyHandler(zaptest.NewLogger(t)) + h.AddNamedHealthCheck(failingChecker{name: "engine", message: "failed to open engine: not a directory"}) + h.AddNamedHealthCheck(check.Named("bolt", check.CheckerFunc(func(context.Context) check.Response { + return check.NamedPass("bolt") + }))) + h.AddNamedReadyCheck(check.NewReadyGate("bolt")) + h.AddNamedReadyCheck(failingChecker{name: "engine", message: "failed to open engine: not a directory"}) + + beforeHealth, beforeReady := captureCheckDocuments(t, h) + + h.FreezeChecks(context.Background()) + + afterHealth, afterReady := captureCheckDocuments(t, h) + require.Equal(t, beforeHealth, afterHealth) + require.Equal(t, beforeReady, afterReady) + + // And the codes, which the comparison above does not cover. + healthRes := doRequest(t, h, http.MethodGet, "/health") + defer closeBody(t, healthRes) + require.Equal(t, http.StatusServiceUnavailable, healthRes.StatusCode) + readyRes := doRequest(t, h, http.MethodGet, "/ready") + defer closeBody(t, readyRes) + require.Equal(t, http.StatusServiceUnavailable, readyRes.StatusCode) +} + +// TestHealthReadyHandler_FreezeChecks_PinsTornDownSubsystem is the reason the +// freeze exists. After the freeze the stores are closed, and their checks start +// reporting that teardown as a failure of their own; because failures sort +// first and /health's top-level message is the first of them, an alphabetically +// earlier subsystem would otherwise mask the one that actually failed. +func TestHealthReadyHandler_FreezeChecks_PinsTornDownSubsystem(t *testing.T) { + h := NewHealthReadyHandler(zaptest.NewLogger(t)) + + // bolt sorts ahead of engine, so once it starts failing it owns the + // top-level message. + var boltClosed bool + h.AddNamedHealthCheck(check.Named("bolt", check.CheckerFunc(func(context.Context) check.Response { + if boltClosed { + return check.NamedFail("bolt", "stale: last probe 6s ago (threshold 5s)") + } + return check.NamedPass("bolt") + }))) + h.AddNamedHealthCheck(failingChecker{name: "engine", message: "failed to open engine: not a directory"}) + + h.FreezeChecks(context.Background()) + boltClosed = true + + res := doRequest(t, h, http.MethodGet, "/health") + defer closeBody(t, res) + require.Equal(t, http.StatusServiceUnavailable, res.StatusCode) + + var got testHealthBody + require.NoError(t, json.NewDecoder(res.Body).Decode(&got)) + require.Equal(t, "failed to open engine: not a directory", got.Message, + "the teardown of bolt masked the failure the window exists to publish") + require.Len(t, got.Checks, 2) + for _, c := range got.Checks { + if c.Name() == "bolt" { + require.Equal(t, check.StatusPass, c.Status()) + } + } +} + +// TestHealthReadyHandler_FreezeChecks_RetiresAuthDependency covers decision 6. +// The store credentials resolve against is closed by the teardown that follows +// the freeze, so the auth dependency checker is pinned to fail and the window +// answers detailNames from its first request rather than degrading from +// detailNone as the store's own probe ages out. +func TestHealthReadyHandler_FreezeChecks_RetiresAuthDependency(t *testing.T) { + t.Run("with an auth dependency installed", func(t *testing.T) { + h, resolver := authHandler(t, platform.OperPermissions()) + h.SetAuthDependencyChecker(staticChecker{name: "bolt", resp: check.NamedPass("bolt")}) + h.AddNamedHealthCheck(failingChecker{name: "engine", message: "failed to open engine: not a directory"}) + + // Before the freeze an operator reads everything. + res := doAuthRequest(t, h, http.MethodGet, "/health") + got := decodeBody(t, res) + closeBody(t, res) + require.Equal(t, "failed to open engine: not a directory", got["message"]) + require.Contains(t, got, "version", "an operator should have had detailFull") + require.Positive(t, resolver.called.Load()) + + h.FreezeChecks(context.Background()) + + // After it the same operator reads names and statuses only, and the + // credential is not resolved at all -- there is nothing left to resolve + // it against. + calledBefore := resolver.called.Load() + res = doAuthRequest(t, h, http.MethodGet, "/health") + got = decodeBody(t, res) + closeBody(t, res) + require.NotContains(t, got, "version") + require.NotContains(t, got, "message", + "detailNames withholds messages, which is where startup error text lives") + checks, ok := got["checks"].([]any) + require.True(t, ok, "names and statuses must survive: %v", got) + require.Len(t, checks, 1) + entry, ok := checks[0].(map[string]any) + require.True(t, ok) + assert.Equal(t, "engine", entry["name"]) + assert.Equal(t, "fail", entry["status"]) + assert.NotContains(t, entry, "message") + require.Equal(t, calledBefore, resolver.called.Load(), + "the credential was resolved against a store that is being closed") + }) + + t.Run("with no auth dependency installed", func(t *testing.T) { + // Nothing was gated on a store's liveness, so nothing is retired and an + // operator keeps the detail they are entitled to. No production + // configuration reaches this: the launcher skips + // SetAuthDependencyChecker only for the in-memory KV store, which is + // test-only. Pinned anyway, because the branch exists. + h, _ := authHandler(t, platform.OperPermissions()) + h.AddNamedHealthCheck(failingChecker{name: "engine", message: "failed to open engine: not a directory"}) + + h.FreezeChecks(context.Background()) + + res := doAuthRequest(t, h, http.MethodGet, "/health") + defer closeBody(t, res) + got := decodeBody(t, res) + require.Equal(t, "failed to open engine: not a directory", got["message"]) + require.Contains(t, got, "version") + }) +} diff --git a/kit/check/check.go b/kit/check/check.go index 8536fc97061..0fbc477b0e2 100644 --- a/kit/check/check.go +++ b/kit/check/check.go @@ -34,6 +34,11 @@ type Check struct { healthChecks []Checker readyChecks []Checker readyNames []string + + // frozen reports that Freeze has installed a static snapshot. Once set, + // the check sets never change again: later registrations are dropped and + // a second Freeze is a no-op. + frozen bool } // Checker indicates a service whose health can be checked. @@ -51,6 +56,7 @@ func NewCheck() *Check { // so the name is recorded; otherwise the check is stored as-is and its // recorded name is empty. Prefer AddNamedHealthCheck when the caller // already knows the name. +// A registration after Freeze is ignored; see there. func (c *Check) AddHealthCheck(check Checker) { if nc, ok := check.(NamedChecker); ok { c.AddNamedHealthCheck(nc) @@ -58,23 +64,34 @@ func (c *Check) AddHealthCheck(check Checker) { } c.mu.Lock() defer c.mu.Unlock() + if c.frozen { + return + } c.healthChecks = append(c.healthChecks, check) } // AddNamedHealthCheck registers nc as a health check. The name is taken // from nc.CheckName(); nc.Check is responsible for stamping Response.Name // (see NamedChecker), so no additional wrapping happens here. +// +// A registration after Freeze is ignored; see there. func (c *Check) AddNamedHealthCheck(nc NamedChecker) { c.mu.Lock() defer c.mu.Unlock() + if c.frozen { + return + } c.healthChecks = append(c.healthChecks, nc) } // AddNamedReadyCheck registers nc as a ready check. See AddNamedHealthCheck -// for naming semantics. +// for naming semantics and for what a registration after Freeze does. func (c *Check) AddNamedReadyCheck(nc NamedChecker) { c.mu.Lock() defer c.mu.Unlock() + if c.frozen { + return + } c.readyChecks = append(c.readyChecks, nc) c.readyNames = append(c.readyNames, nc.CheckName()) } @@ -135,3 +152,105 @@ func (c *Check) evaluate(ctx context.Context, name string, snap func() []Checker sort.Sort(results) return NewBasicResponse(name, overall, "", results) } + +// frozenChecker answers with a fixed Response. It implements NamedChecker so a +// frozen set can rebuild readyNames and so evaluate needs no special case: to +// everything downstream a frozen check is an ordinary registered check that +// happens never to change its mind. +type frozenChecker struct{ resp BasicResponse } + +func (f frozenChecker) CheckName() string { return f.resp.Name() } +func (f frozenChecker) Check(context.Context) Response { return f.resp } + +// probe evaluates ch for the freeze, under a context of its own bounded at +// DefaultProbeTimeout, and flattens what it returns. +// +// The bound is per probe rather than one budget shared across the set, and that +// distinction is the whole of this function. A shared budget spent by an early +// slow checker leaves every checker after it running on a dead context, and a +// dead context does not yield "unknown": sqlite.SqlStore.Check, for one, turns +// it into NamedFail(name, "context deadline exceeded"). Responses sorts +// failures ahead of passes and then by name, and /health's top-level message is +// the first of them, so a subsystem that merely ran out of someone else's time +// could outrank and mask the failure the freeze was taken to preserve -- the +// exact drift Freeze exists to prevent, reintroduced by its own timeout. +// +// Bounding each probe separately costs a worst case of one DefaultProbeTimeout +// per registered check, reached only if every subsystem is wedged at once. A +// checker that ignores its context entirely (a bbolt View cannot be cancelled) +// is unbounded either way, so the shared budget never bought that back. +func probe(ctx context.Context, ch Checker) BasicResponse { + probeCtx, cancel := BoundDeadline(ctx, DefaultProbeTimeout) + defer cancel() + return snapshot(ch.Check(probeCtx)) +} + +// Freeze replaces every registered health and ready check with a static +// snapshot of what it reports now, so CheckHealth and CheckReady go on +// returning that same answer for the life of the process. +// +// It exists for terminal states. A process on its way out tears its subsystems +// down, and their checks then report that deliberate teardown as a fresh +// failure; because Responses sort failures first and then by name, a closed +// store can outrank -- and so mask -- the failure that made the process +// terminal. Freezing first preserves the report as it stood when that decision +// was made. +// +// Each snapshot is flattened into a BasicResponse so a live Response cannot +// keep moving inside the frozen set: a *FreshnessResponse ages into a +// staleness failure on its own once its prober stops. Both render the same +// JSON object, so a frozen body has the same shape as the one served a moment +// earlier. +// +// The registered set and its order are unchanged, so ReadyCheckNames reports +// what it did before. Only the values are pinned. +// +// Freeze is terminal and first-freeze-wins: a second call is a no-op, there is +// no thaw, and checks registered afterwards are ignored. A registration racing +// the freeze may or may not be captured, which is why the caller must be the +// one thing still running. +// +// Every probe is bounded on its own, at DefaultProbeTimeout, rather than out of +// one budget shared by the whole set; see probe. ctx is their parent, so a +// deadline on it still caps the freeze as a whole -- give it one only as a +// backstop, generous enough that a healthy freeze never reaches it. Once it +// expires the remaining probes run on a dead context, and a cancelled probe +// records the freeze itself rather than the state being frozen. For the same +// reason, pass a context that a signal cannot cancel. +func (c *Check) Freeze(ctx context.Context) { + c.mu.RLock() + frozen := c.frozen + health := append([]Checker(nil), c.healthChecks...) + ready := append([]Checker(nil), c.readyChecks...) + c.mu.RUnlock() + if frozen { + return + } + + // Evaluate with no lock held, for the reason evaluate documents: a checker + // can block on a network call and can re-enter registration. + frozenHealth := make([]Checker, len(health)) + for i, ch := range health { + frozenHealth[i] = frozenChecker{resp: probe(ctx, ch)} + } + // Ready names are rebuilt from the frozen responses rather than carried + // over, so a check registered in the gap between the two locks -- and + // therefore absent from the frozen set -- leaves both lists together. + frozenReady := make([]Checker, len(ready)) + readyNames := make([]string, len(ready)) + for i, ch := range ready { + resp := probe(ctx, ch) + frozenReady[i] = frozenChecker{resp: resp} + readyNames[i] = resp.Name() + } + + c.mu.Lock() + defer c.mu.Unlock() + if c.frozen { + return + } + c.healthChecks = frozenHealth + c.readyChecks = frozenReady + c.readyNames = readyNames + c.frozen = true +} diff --git a/kit/check/check_race_test.go b/kit/check/check_race_test.go index 8499defdbf8..dd39c951664 100644 --- a/kit/check/check_race_test.go +++ b/kit/check/check_race_test.go @@ -6,6 +6,7 @@ import ( "sync/atomic" "testing" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -98,3 +99,125 @@ func TestCheck_ConcurrentRegistrationAndEvaluation(t *testing.T) { resp = c.CheckReady(ctx) require.Len(t, resp.Checks(), wantReady) } + +// TestCheck_ConcurrentFreeze runs freezers against evaluators and registerers, +// all released together. Freeze reads the checker slices under RLock, evaluates +// with no lock held, and installs under Lock, so it is the one operation that +// spans both locks; under -race this fails if any of the three steps touches +// the slices unguarded. +// +// The assertion that matters is after the race settles: once frozen, two +// successive evaluations must be element-wise identical. Everything the +// evaluators check while the race is on is weaker than that -- it exists to +// catch a torn install being observed mid-flight, which the settled comparison +// could not see. +func TestCheck_ConcurrentFreeze(t *testing.T) { + const ( + numFreezers = 4 + numRegisterers = 8 + numEvaluators = 16 + numChecksEach = 32 + numEvaluations = 64 + + healthName = "h" + readyName = "r" + ) + + c := NewCheck() + ctx := context.Background() + + // Registered up front so every freezer has something to snapshot even if it + // wins the race against every registerer. + c.AddNamedHealthCheck(Named(healthName, mockPass(healthName))) + c.AddNamedReadyCheck(Named(readyName, mockPass(readyName))) + + var ( + startMu sync.RWMutex + concurrency atomic.Int64 + maxConcurrency atomic.Int64 + ) + + // enter blocks until the start gate opens, then records the observed + // overlap. The returned func is deferred by the caller to leave. + enter := func() func() { + startMu.RLock() + cur := concurrency.Add(1) + for { + old := maxConcurrency.Load() + if cur <= old || maxConcurrency.CompareAndSwap(old, cur) { + break + } + } + return func() { + concurrency.Add(-1) + startMu.RUnlock() + } + } + + var wg sync.WaitGroup + startMu.Lock() + + for range numFreezers { + wg.Add(1) + go func() { + defer wg.Done() + defer enter()() + c.Freeze(ctx) + }() + } + + for range numRegisterers { + wg.Add(1) + go func() { + defer wg.Done() + defer enter()() + for i := range numChecksEach { + if i%2 == 0 { + c.AddHealthCheck(mockPass(healthName)) + } else { + c.AddNamedReadyCheck(Named(readyName, mockPass(readyName))) + } + } + }() + } + + for range numEvaluators { + wg.Add(1) + go func() { + defer wg.Done() + defer enter()() + for range numEvaluations { + for _, resp := range []Response{c.CheckHealth(ctx), c.CheckReady(ctx)} { + // A torn or zero-valued frozen entry would show up as an + // empty name or an empty status, neither of which any + // registered checker here can produce. + for _, sub := range resp.Checks() { + assert.NotEmpty(t, sub.Name()) + assert.Contains(t, []Status{StatusPass, StatusFail}, sub.Status()) + } + } + } + }() + } + + startMu.Unlock() + wg.Wait() + + t.Logf("max concurrency: %d", maxConcurrency.Load()) + + // Frozen, and terminal: the set no longer moves, whichever freezer won. + first := c.CheckHealth(ctx) + second := c.CheckHealth(ctx) + require.Equal(t, first.Checks(), second.Checks()) + require.Equal(t, first.Status(), second.Status()) + + firstReady := c.CheckReady(ctx) + secondReady := c.CheckReady(ctx) + require.Equal(t, firstReady.Checks(), secondReady.Checks()) + require.Len(t, c.ReadyCheckNames(), len(firstReady.Checks())) + + // Nothing registered after the winning freeze survived, so both lists are + // bounded by what was registered before it. + require.LessOrEqual(t, len(first.Checks()), 1+numRegisterers*(numChecksEach/2)) + require.GreaterOrEqual(t, len(first.Checks()), 1) +} diff --git a/kit/check/checktest/checktest.go b/kit/check/checktest/checktest.go new file mode 100644 index 00000000000..7ec6d62daf8 --- /dev/null +++ b/kit/check/checktest/checktest.go @@ -0,0 +1,84 @@ +// Package checktest holds the helpers shared by tests that compare two +// renderings of a /health or /ready document. +// +// It is a package rather than a helper per test package because the callers -- +// kit/check, http, and cmd/influxd/launcher -- each need the same thing and +// cannot share a _test.go file across package boundaries. The thing they need +// is not incidental: two renderings of the same frozen report are equal only +// once the values that move on their own have been masked, and a caller that +// gets that set wrong writes an equality assertion that is flaky rather than +// one that fails. +package checktest + +import ( + "encoding/json" + "regexp" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// The fields of a rendered check document whose values move on their own, and +// the sentinels Normalize puts in their place. +const ( + // FieldStarted and FieldUp belong to /ready alone (the http package's + // readyBody). started is fixed at handler construction; up is the elapsed + // time since, recomputed per request, so it advances even across a frozen + // check set. /health carries neither. + FieldStarted = "started" + FieldUp = "up" + + // FieldMessage is common to both, and moves only while it carries a + // staleness message; see StaleMessagePattern. + FieldMessage = "message" + + SentinelStarted = "" + SentinelUp = "" + SentinelStale = "" +) + +// StaleMessagePattern matches the message a check.FreshnessResponse renders +// once its snapshot has aged out. The age in it advances between two renders of +// the same response, so a test comparing two renders must mask it. Tracks +// check.staleMessage; a change to that format belongs here too. +var StaleMessagePattern = regexp.MustCompile(`^stale: last probe .* ago \(threshold .*\)$`) + +// Normalize replaces every value in a decoded check document that moves on its +// own with a fixed sentinel, and returns doc, so two captures taken a moment +// apart compare equal on everything they actually pin. +// +// Each field is checked for shape before it is masked, so a normalized document +// still fails a comparison if one of them changes type -- masking must not +// become a way to stop noticing. Masking is conditional on presence, because +// /health carries neither started nor up, and a message is masked only while it +// matches StaleMessagePattern: every other message is content a caller is +// probably asserting on, and blanking it would quietly gut the assertion. +func Normalize(t testing.TB, doc map[string]any) map[string]any { + t.Helper() + + if v, ok := doc[FieldStarted]; ok { + s, isString := v.(string) + require.Truef(t, isString, "%s must be a string: %v", FieldStarted, v) + _, err := time.Parse(time.RFC3339Nano, s) + require.NoErrorf(t, err, "%s must be an RFC3339 timestamp: %q", FieldStarted, s) + doc[FieldStarted] = SentinelStarted + } + if v, ok := doc[FieldUp]; ok { + _, isString := v.(string) + require.Truef(t, isString, "%s must be a string: %v", FieldUp, v) + doc[FieldUp] = SentinelUp + } + if msg, ok := doc[FieldMessage].(string); ok && StaleMessagePattern.MatchString(msg) { + doc[FieldMessage] = SentinelStale + } + return doc +} + +// NormalizeJSON decodes a rendered check document and normalizes it. +func NormalizeJSON(t testing.TB, b []byte) map[string]any { + t.Helper() + var doc map[string]any + require.NoErrorf(t, json.Unmarshal(b, &doc), "body: %s", b) + return Normalize(t, doc) +} diff --git a/kit/check/freeze_test.go b/kit/check/freeze_test.go new file mode 100644 index 00000000000..4d630b3bd06 --- /dev/null +++ b/kit/check/freeze_test.go @@ -0,0 +1,261 @@ +package check + +import ( + "context" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// mutableCheck is a NamedChecker whose answer the test can change after +// registration, so a frozen set can be shown not to follow it. It counts its +// invocations, which is what pins Freeze as terminal: a second Freeze that +// re-evaluated the checkers would show up here even when the resulting +// snapshot happened to be identical. +type mutableCheck struct { + name string + resp atomic.Pointer[BasicResponse] + calls atomic.Int64 +} + +func newMutableCheck(name string, status Status, msg string) *mutableCheck { + m := &mutableCheck{name: name} + m.set(status, msg) + return m +} + +func (m *mutableCheck) set(status Status, msg string) { + r := NewBasicResponse(m.name, status, msg, nil) + m.resp.Store(&r) +} + +func (m *mutableCheck) CheckName() string { return m.name } + +func (m *mutableCheck) Check(context.Context) Response { + m.calls.Add(1) + return *m.resp.Load() +} + +// fixedChecker exposes an existing Response as a NamedChecker without copying +// or wrapping it, so a test can hand the check set a live *FreshnessResponse +// and still hold the pointer. This is how bolt.KVStore registers its own +// freshness-backed check. +type fixedChecker struct { + name string + resp Response +} + +func (f fixedChecker) CheckName() string { return f.name } +func (f fixedChecker) Check(context.Context) Response { return f.resp } + +// TestCheck_Freeze_PinsValues covers the whole point of the freeze: what the +// checks reported at the moment of the call is what they go on reporting, +// however the subsystems behind them behave afterwards. +func TestCheck_Freeze_PinsValues(t *testing.T) { + ctx := context.Background() + c := NewCheck() + h := newMutableCheck("h", StatusPass, "healthy") + r := newMutableCheck("r", StatusFail, "why it failed") + c.AddNamedHealthCheck(h) + c.AddNamedReadyCheck(r) + + c.Freeze(ctx) + + // Teardown, as far as these checkers are concerned: the passing one starts + // failing and the failing one starts passing. Both must be ignored. + h.set(StatusFail, "torn down") + r.set(StatusPass, "") + + health := c.CheckHealth(ctx) + require.Equal(t, StatusPass, health.Status()) + require.Len(t, health.Checks(), 1) + require.Equal(t, "h", health.Checks()[0].Name()) + require.Equal(t, StatusPass, health.Checks()[0].Status()) + require.Equal(t, "healthy", health.Checks()[0].Message()) + + ready := c.CheckReady(ctx) + require.Equal(t, StatusFail, ready.Status()) + require.Len(t, ready.Checks(), 1) + require.Equal(t, "r", ready.Checks()[0].Name()) + require.Equal(t, StatusFail, ready.Checks()[0].Status()) + require.Equal(t, "why it failed", ready.Checks()[0].Message()) +} + +// TestCheck_Freeze_FlattensFreshnessResponse is the flattening pin. A +// *FreshnessResponse retained by pointer inside the frozen set would age into +// a staleness failure on its own once its prober stopped, which is exactly the +// drift the freeze exists to prevent. +func TestCheck_Freeze_FlattensFreshnessResponse(t *testing.T) { + ctx := context.Background() + const staleness = 20 * time.Millisecond + + f := NewFreshnessResponse("probed", staleness) + f.Update(Pass()) + + c := NewCheck() + c.AddNamedHealthCheck(fixedChecker{name: "probed", resp: f}) + c.Freeze(ctx) + + time.Sleep(3 * staleness) + require.Equal(t, StatusFail, f.Status(), + "the live response must age out, or this test proves nothing") + + resp := c.CheckHealth(ctx) + require.Equal(t, StatusPass, resp.Status()) + require.Len(t, resp.Checks(), 1) + require.Equal(t, StatusPass, resp.Checks()[0].Status()) + require.NotContains(t, resp.Checks()[0].Message(), "stale:") +} + +// TestCheck_Freeze_FlattensNestedChecks covers the recursion: a live Response +// reached only through another response's Checks() ages just as readily as a +// top-level one. +func TestCheck_Freeze_FlattensNestedChecks(t *testing.T) { + ctx := context.Background() + const staleness = 20 * time.Millisecond + + inner := NewFreshnessResponse("inner", staleness) + inner.Update(Pass()) + outer := NewBasicResponse("outer", StatusPass, "", Responses{inner}) + + c := NewCheck() + c.AddNamedHealthCheck(fixedChecker{name: "outer", resp: outer}) + c.Freeze(ctx) + + time.Sleep(3 * staleness) + require.Equal(t, StatusFail, inner.Status(), + "the live nested response must age out, or this test proves nothing") + + resp := c.CheckHealth(ctx) + require.Len(t, resp.Checks(), 1) + nested := resp.Checks()[0].Checks() + require.Len(t, nested, 1) + require.Equal(t, "inner", nested[0].Name()) + require.Equal(t, StatusPass, nested[0].Status()) + require.NotContains(t, nested[0].Message(), "stale:") +} + +// TestCheck_Freeze_PreservesReadyCheckNames pins registration order across the +// freeze. ReadyCheckNames means registration order -- CheckReady's aggregate is +// sorted, and building the frozen set from that would silently reorder it. +func TestCheck_Freeze_PreservesReadyCheckNames(t *testing.T) { + ctx := context.Background() + c := NewCheck() + // Deliberately not alphabetical, and deliberately not sorted by status + // either, so a set rebuilt from the sorted aggregate could not match. + for _, name := range []string{"zulu", "alpha", "mike"} { + c.AddNamedReadyCheck(newMutableCheck(name, StatusFail, MsgNotReady)) + } + c.AddNamedReadyCheck(newMutableCheck("bravo", StatusPass, "")) + + before := c.ReadyCheckNames() + c.Freeze(ctx) + require.Equal(t, before, c.ReadyCheckNames()) + require.Equal(t, []string{"zulu", "alpha", "mike", "bravo"}, c.ReadyCheckNames()) +} + +// TestCheck_Freeze_IsTerminal pins first-freeze-wins and the registration +// guard. The call count is what makes "terminal" an assertion rather than an +// inference: a second Freeze that re-ran the checkers would be visible here +// even though the second snapshot would look the same. +func TestCheck_Freeze_IsTerminal(t *testing.T) { + ctx := context.Background() + c := NewCheck() + h := newMutableCheck("h", StatusPass, "first") + c.AddNamedHealthCheck(h) + + c.Freeze(ctx) + require.Equal(t, int64(1), h.calls.Load()) + + h.set(StatusFail, "second") + c.Freeze(ctx) + require.Equal(t, int64(1), h.calls.Load(), "a second Freeze re-evaluated the checkers") + + // All three registration paths: the named health check, the anonymous one, + // and the ready check. + c.AddNamedHealthCheck(newMutableCheck("late-named", StatusFail, "after the freeze")) + c.AddHealthCheck(CheckerFunc(func(context.Context) Response { + return NamedFail("late-anonymous", "after the freeze") + })) + c.AddNamedReadyCheck(newMutableCheck("late-ready", StatusFail, "after the freeze")) + + health := c.CheckHealth(ctx) + require.Equal(t, StatusPass, health.Status()) + require.Len(t, health.Checks(), 1) + require.Equal(t, "first", health.Checks()[0].Message()) + + require.Empty(t, c.CheckReady(ctx).Checks()) + require.Empty(t, c.ReadyCheckNames()) +} + +// statusByName indexes an aggregate's sub-checks so a test can assert about one +// of them without depending on the order Responses sorts them into. +func statusByName(rs Responses) map[string]Response { + out := make(map[string]Response, len(rs)) + for _, r := range rs { + out[r.Name()] = r + } + return out +} + +// TestCheck_Freeze_BoundsEachProbeSeparately pins that every probe gets a +// context of its own rather than a share of one budget spent in registration +// order. +// +// A shared budget leaves every checker after a slow one running on a dead +// context, and a dead context does not read as "unknown" downstream: a real +// checker turns it into a failure with the cancellation as its message +// (sqlite.SqlStore.Check does exactly this). Responses sorts failures ahead of +// passes and then by name, and /health's top-level message is the first of +// them, so a subsystem that merely ran out of someone else's time could outrank +// and mask the failure the freeze was taken to preserve -- the drift Freeze +// exists to prevent, reintroduced by its own timeout. +// +// The test fails three different ways, which is the point: unbounded probes +// hang on the first checker, a shared budget fails the assertions on the two +// after it, and only per-probe bounds pass. +func TestCheck_Freeze_BoundsEachProbeSeparately(t *testing.T) { + c := NewCheck() + + // Sorts first and spends its entire probe budget. Under one shared budget + // it would spend everyone else's with it. + c.AddNamedHealthCheck(NamedFunc("a-slow", func(ctx context.Context) Response { + <-ctx.Done() + return NamedFail("a-slow", ctx.Err().Error()) + })) + // Reports whether it was given any time of its own. + c.AddNamedHealthCheck(NamedFunc("b-fast", func(ctx context.Context) Response { + if err := ctx.Err(); err != nil { + return NamedFail("b-fast", err.Error()) + } + return NamedPass("b-fast") + })) + // The ready set is evaluated after the health set, so a shared budget is + // already gone by the time it is reached. + c.AddNamedReadyCheck(NamedFunc("c-ready", func(ctx context.Context) Response { + if err := ctx.Err(); err != nil { + return NamedFail("c-ready", err.Error()) + } + return NamedPass("c-ready") + })) + + // No deadline of its own: whatever bounds a probe here, Freeze applied. + c.Freeze(context.Background()) + + health := statusByName(c.CheckHealth(context.Background()).Checks()) + require.Len(t, health, 2) + require.Equal(t, StatusFail, health["a-slow"].Status(), + "the slow checker must have been bounded at all, or this proves nothing") + require.Equal(t, context.DeadlineExceeded.Error(), health["a-slow"].Message(), + "the bound must be a deadline of its own, not an inherited cancel") + require.Equal(t, StatusPass, health["b-fast"].Status(), + "a slow probe spent the budget of the check registered after it: %s", + health["b-fast"].Message()) + + ready := statusByName(c.CheckReady(context.Background()).Checks()) + require.Equal(t, StatusPass, ready["c-ready"].Status(), + "a slow health probe spent the budget of the ready checks: %s", + ready["c-ready"].Message()) +} diff --git a/kit/check/freshness.go b/kit/check/freshness.go index ffc1a9775a9..1acd2408550 100644 --- a/kit/check/freshness.go +++ b/kit/check/freshness.go @@ -31,9 +31,13 @@ type FreshnessResponse struct { snap atomic.Pointer[freshnessSnapshot] } +// msgNoProbe is what a FreshnessResponse reports before its first Update: +// the probe has not run yet, which is distinct from having run and aged out. +const msgNoProbe = "no probe completed yet" + // NewFreshnessResponse returns an empty FreshnessResponse with the given // name and staleness budget. Until Update is first called, Status() -// returns StatusFail and Message() reports "no probe completed yet". +// returns StatusFail and Message() reports msgNoProbe. func NewFreshnessResponse(name string, staleness time.Duration) *FreshnessResponse { return &FreshnessResponse{name: name, staleness: staleness} } @@ -65,7 +69,7 @@ func (f *FreshnessResponse) Status() Status { func (f *FreshnessResponse) Message() string { s := f.snap.Load() if s == nil { - return "no probe completed yet" + return msgNoProbe } if age := time.Since(s.at); age > f.staleness { return staleMessage(age, f.staleness) @@ -83,29 +87,32 @@ func (f *FreshnessResponse) Checks() Responses { return s.resp.Checks() } -// MarshalJSON emits a wireResponse derived from one atomic snapshot -// load. Reading every field through the four interface methods would -// be correct (each does its own atomic load) but could observe two -// different snapshots across the call sequence; a single load here -// guarantees the rendered JSON object reflects exactly one state. -func (f *FreshnessResponse) MarshalJSON() ([]byte, error) { - w := wireResponse{Name: f.name} +// Snapshot renders f from one atomic load, implementing Snapshotter. +// Reading every field through the four interface methods would be +// correct (each does its own atomic load) but could observe two +// different snapshots across the call sequence, yielding a combination +// that was never true: a stale status carried alongside the previous +// probe's empty message, which /health renders as the bare word "fail". +// +// The returned BasicResponse fixes f's own fields. Its Checks are the +// underlying probe's, which this type does not own and does not copy. +func (f *FreshnessResponse) Snapshot() BasicResponse { s := f.snap.Load() - switch { - case s == nil: - w.Status = StatusFail - w.Message = "no probe completed yet" - default: - if age := time.Since(s.at); age > f.staleness { - w.Status = StatusFail - w.Message = staleMessage(age, f.staleness) - } else { - w.Status = s.resp.Status() - w.Message = s.resp.Message() - w.Checks = s.resp.Checks() - } + if s == nil { + return NewBasicResponse(f.name, StatusFail, msgNoProbe, nil) } - return json.Marshal(w) + if age := time.Since(s.at); age > f.staleness { + return NewBasicResponse(f.name, StatusFail, staleMessage(age, f.staleness), nil) + } + return NewBasicResponse(f.name, s.resp.Status(), s.resp.Message(), s.resp.Checks()) +} + +// MarshalJSON emits the wire shape from a single snapshot, so the +// rendered JSON object reflects exactly one state. BasicResponse embeds +// wireResponse, whose exported fields encoding/json promotes, so this +// marshals byte-identically to building the wireResponse here. +func (f *FreshnessResponse) MarshalJSON() ([]byte, error) { + return json.Marshal(f.Snapshot()) } func staleMessage(age, threshold time.Duration) string { diff --git a/kit/check/freshness_test.go b/kit/check/freshness_test.go index 19399f7d248..795b7c7b1a9 100644 --- a/kit/check/freshness_test.go +++ b/kit/check/freshness_test.go @@ -8,6 +8,7 @@ import ( "testing" "time" + "github.com/influxdata/influxdb/v2/kit/check/checktest" "github.com/stretchr/testify/require" ) @@ -93,6 +94,119 @@ func TestFreshnessResponse_JSONMarshalEmitsDerivedValues(t *testing.T) { }) } +// TestFreshnessResponse_Snapshot covers the three states Snapshot renders and +// pins it against the accessors it replaces. +func TestFreshnessResponse_Snapshot(t *testing.T) { + const staleness = 50 * time.Millisecond + + t.Run("no probe", func(t *testing.T) { + f := NewFreshnessResponse("svc", staleness) + s := f.Snapshot() + require.Equal(t, "svc", s.Name()) + require.Equal(t, StatusFail, s.Status()) + require.Equal(t, msgNoProbe, s.Message()) + require.Nil(t, s.Checks()) + }) + + t.Run("fresh", func(t *testing.T) { + f := NewFreshnessResponse("svc", time.Second) + f.Update(NewBasicResponse("inner", StatusPass, "ok", Responses{NamedPass("nested")})) + s := f.Snapshot() + require.Equal(t, "svc", s.Name(), "the wrapper's name wins, not the probe's") + require.Equal(t, StatusPass, s.Status()) + require.Equal(t, "ok", s.Message()) + require.Len(t, s.Checks(), 1) + require.Equal(t, "nested", s.Checks()[0].Name()) + }) + + t.Run("stale", func(t *testing.T) { + f := NewFreshnessResponse("svc", staleness) + f.Update(Pass()) + time.Sleep(staleness + 50*time.Millisecond) + + s := f.Snapshot() + require.Equal(t, StatusFail, s.Status()) + require.Regexp(t, `^stale: last probe .* ago \(threshold 50ms\)$`, s.Message()) + require.Nil(t, s.Checks(), "an aged-out snapshot reports no nested checks") + }) + + // A snapshot is a value: it does not age, which is what makes it safe to + // hold in a frozen check set. + t.Run("does not age", func(t *testing.T) { + f := NewFreshnessResponse("svc", staleness) + f.Update(Pass()) + s := f.Snapshot() + time.Sleep(staleness + 50*time.Millisecond) + require.Equal(t, StatusFail, f.Status(), "the live response must age out") + require.Equal(t, StatusPass, s.Status()) + }) +} + +// TestFreshnessResponse_MarshalMatchesSnapshot pins the MarshalJSON rewrite as +// byte-compatible. MarshalJSON now marshals the BasicResponse Snapshot returns, +// which reaches the wire shape through an embedded unexported struct whose +// exported fields encoding/json promotes; this asserts that indirection emits +// what the hand-built wireResponse did. +func TestFreshnessResponse_MarshalMatchesSnapshot(t *testing.T) { + const staleness = 50 * time.Millisecond + + for _, tc := range []struct { + name string + build func() *FreshnessResponse + }{ + { + name: "no probe", + build: func() *FreshnessResponse { return NewFreshnessResponse("svc", staleness) }, + }, + { + name: "fresh with nested checks", + build: func() *FreshnessResponse { + f := NewFreshnessResponse("svc", time.Second) + f.Update(NewBasicResponse("inner", StatusPass, "ok", Responses{NamedFail("nested", "bad")})) + return f + }, + }, + { + name: "stale", + build: func() *FreshnessResponse { + f := NewFreshnessResponse("svc", staleness) + f.Update(Pass()) + time.Sleep(staleness + 50*time.Millisecond) + return f + }, + }, + } { + t.Run(tc.name, func(t *testing.T) { + f := tc.build() + fromResponse, err := json.Marshal(f) + require.NoError(t, err) + fromSnapshot, err := json.Marshal(f.Snapshot()) + require.NoError(t, err) + require.Equal(t, + checktest.NormalizeJSON(t, fromSnapshot), + checktest.NormalizeJSON(t, fromResponse)) + }) + } +} + +// TestRenamedResponse_MarshalMatchesSnapshot covers the same rewrite for the +// wrapper Rename puts around a stateful Response. A renamed FreshnessResponse +// must render the new name over the inner state, from one observation. +func TestRenamedResponse_MarshalMatchesSnapshot(t *testing.T) { + f := NewFreshnessResponse("inner", time.Second) + f.Update(Info("ok")) + r := Rename(f, "outer") + + b, err := json.Marshal(r) + require.NoError(t, err) + require.JSONEq(t, `{"name":"outer","status":"pass","message":"ok"}`, string(b)) + + s, ok := r.(Snapshotter) + require.True(t, ok, "a renamed Response must still offer a single coherent read") + require.Equal(t, "outer", s.Snapshot().Name()) + require.Equal(t, StatusPass, s.Snapshot().Status()) +} + // TestFreshnessResponse_ConcurrentUpdateAndRead exercises Update racing // with the four interface methods and MarshalJSON. Uses the RWMutex // start-gate so every goroutine contends simultaneously. diff --git a/kit/check/helpers.go b/kit/check/helpers.go index cb1e7a02057..8ac1a2d027b 100644 --- a/kit/check/helpers.go +++ b/kit/check/helpers.go @@ -67,13 +67,21 @@ func (r renamedResponse) Status() Status { return r.inner.Status() } func (r renamedResponse) Message() string { return r.inner.Message() } func (r renamedResponse) Checks() Responses { return r.inner.Checks() } +// Snapshot implements Snapshotter by delegating to the inner Response, so a +// rename does not cost the caller the single coherent read: without this, a +// renamed *FreshnessResponse would be flattened through the four accessors and +// could report a status from one observation with a message from the next. +func (r renamedResponse) Snapshot() BasicResponse { + if s, ok := r.inner.(Snapshotter); ok { + return s.Snapshot().WithName(r.name) + } + return NewBasicResponse(r.name, r.inner.Status(), r.inner.Message(), r.inner.Checks()) +} + +// MarshalJSON emits the renamed wire shape from a single snapshot, for the +// reason FreshnessResponse.MarshalJSON does. func (r renamedResponse) MarshalJSON() ([]byte, error) { - return json.Marshal(wireResponse{ - Name: r.name, - Status: r.inner.Status(), - Message: r.inner.Message(), - Checks: r.inner.Checks(), - }) + return json.Marshal(r.Snapshot()) } // Named returns a NamedChecker that delegates to checker and stamps name diff --git a/kit/check/response.go b/kit/check/response.go index 43c70723a0d..338fe8e068a 100644 --- a/kit/check/response.go +++ b/kit/check/response.go @@ -13,6 +13,10 @@ import ( // {"name","status","message"?,"checks"?}. BasicResponse gets this for // free by embedding wireResponse; FreshnessResponse and renamedResponse // build the same shape explicitly in MarshalJSON. +// +// A stateful implementation should also implement Snapshotter, so callers +// that need every field from one observation -- rendering, or freezing a +// terminal report -- can take it without reading each accessor separately. type Response interface { Name() string Status() Status @@ -20,6 +24,50 @@ type Response interface { Checks() Responses } +// Snapshotter is implemented by a Response that can render its entire state +// from a single coherent read. A Response whose fields derive from mutable +// state -- FreshnessResponse -- can otherwise report a torn combination, a +// status taken from one observation and a message from the next, because the +// four accessors each read independently. +// +// Implementations must return a BasicResponse whose own fields are fixed; +// nested Checks may still be live, and snapshot recurses into them. +type Snapshotter interface { + Snapshot() BasicResponse +} + +// snapshot flattens r into a value whose fields are fixed at the moment of the +// call, using Snapshot when r implements it and the four accessors otherwise, +// and recursing into nested checks so no live Response survives inside the +// result. +// +// Flattening matters wherever a Response outlives the thing it describes: a +// *FreshnessResponse held by pointer goes on aging into a staleness failure +// after its prober stops, so a set of them kept as-is would drift. Every +// Response marshals to the same wire shape, so the flattened value renders +// identical JSON to the one it replaces. +func snapshot(r Response) BasicResponse { + if s, ok := r.(Snapshotter); ok { + b := s.Snapshot() + return NewBasicResponse(b.Name(), b.Status(), b.Message(), snapshotAll(b.Checks())) + } + return NewBasicResponse(r.Name(), r.Status(), r.Message(), snapshotAll(r.Checks())) +} + +// snapshotAll flattens every element of rs. It returns nil for an empty input +// so the result keeps the shape omitempty gives Checks: an absent field rather +// than "checks":[]. +func snapshotAll(rs Responses) Responses { + if len(rs) == 0 { + return nil + } + out := make(Responses, len(rs)) + for i, r := range rs { + out[i] = snapshot(r) + } + return out +} + // wireResponse is the on-the-wire JSON shape shared by every Response // implementation, used for both marshal and unmarshal. Field tags MUST // match the legacy struct format so external clients of /health and From dc0cf0f66ae1c7161ae1ae2080d9009693edefbb Mon Sep 17 00:00:00 2001 From: davidby-influx <72418212+davidby-influx@users.noreply.github.com> Date: Tue, 1 Sep 2026 10:29:01 -0700 Subject: [PATCH 2/6] chore: correct documentation about PID file Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- HEALTH_READY.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/HEALTH_READY.md b/HEALTH_READY.md index 00cf7a62412..ed5dab5cd18 100644 --- a/HEALTH_READY.md +++ b/HEALTH_READY.md @@ -331,8 +331,8 @@ There are three honest mitigations and no fourth: accepting that it is then readable by anyone who can reach the port. Keeping the store open across the window would restore the detail, but it -would re-hold the flock and the PID file that the split teardown exists -to release, blocking the restart the operator is presumably attempting. +would continue holding the flock that the split teardown exists to +release. The PID file remains held independently until final shutdown. Note that this is the existing policy applied consistently, not a new hole: a startup failure *before* the authorization store opens already From 2b383f0dcb5504d5565c2e2116c34928e25a0599 Mon Sep 17 00:00:00 2001 From: davidby-influx <72418212+davidby-influx@users.noreply.github.com> Date: Tue, 1 Sep 2026 10:35:10 -0700 Subject: [PATCH 3/6] fix: snapshot for atomicity Code review sugestion Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- kit/check/freshness.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/kit/check/freshness.go b/kit/check/freshness.go index 1acd2408550..b3f1ee3b194 100644 --- a/kit/check/freshness.go +++ b/kit/check/freshness.go @@ -104,6 +104,9 @@ func (f *FreshnessResponse) Snapshot() BasicResponse { if age := time.Since(s.at); age > f.staleness { return NewBasicResponse(f.name, StatusFail, staleMessage(age, f.staleness), nil) } +if inner, ok := s.resp.(Snapshotter); ok { + return inner.Snapshot().WithName(f.name) + } return NewBasicResponse(f.name, s.resp.Status(), s.resp.Message(), s.resp.Checks()) } From aa883bf63bafbae2997c2ff32be7ce81e37b0216 Mon Sep 17 00:00:00 2001 From: davidby-influx <72418212+davidby-influx@users.noreply.github.com> Date: Tue, 1 Sep 2026 10:36:07 -0700 Subject: [PATCH 4/6] chore: correct comment on PID file Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- cmd/influxd/launcher/startup_failure_test.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/cmd/influxd/launcher/startup_failure_test.go b/cmd/influxd/launcher/startup_failure_test.go index dab879aa262..ad4b5be0a77 100644 --- a/cmd/influxd/launcher/startup_failure_test.go +++ b/cmd/influxd/launcher/startup_failure_test.go @@ -160,9 +160,9 @@ func fetchCheckDocuments(t *testing.T, l *launcher.TestLauncher) (health, ready } // TestLauncher_StartupFailure_LingerServesFrozenAttribution is the end-to-end -// shape of --startup-error-linger: a failed startup releases the PID file and -// the store locks a restart needs, and goes on serving the report it had at the -// moment of failure until the window closes. +// shape of --startup-error-linger: a failed startup releases the store locks a +// restart needs while retaining the PID file, and serves the failure report +// until the window closes; final shutdown then releases the PID file. func TestLauncher_StartupFailure_LingerServesFrozenAttribution(t *testing.T) { l := launcher.NewTestLauncherServer() From 497d1f71536fdd60030f43074842ad4610bd075c Mon Sep 17 00:00:00 2001 From: davidby-influx Date: Tue, 1 Sep 2026 11:12:56 -0700 Subject: [PATCH 5/6] chore: copilot does not go fmt! --- kit/check/freshness.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/kit/check/freshness.go b/kit/check/freshness.go index b3f1ee3b194..075053e6bcb 100644 --- a/kit/check/freshness.go +++ b/kit/check/freshness.go @@ -104,7 +104,7 @@ func (f *FreshnessResponse) Snapshot() BasicResponse { if age := time.Since(s.at); age > f.staleness { return NewBasicResponse(f.name, StatusFail, staleMessage(age, f.staleness), nil) } -if inner, ok := s.resp.(Snapshotter); ok { + if inner, ok := s.resp.(Snapshotter); ok { return inner.Snapshot().WithName(f.name) } return NewBasicResponse(f.name, s.resp.Status(), s.resp.Message(), s.resp.Checks()) From f85a4ab1cf35a87731a0b88eedc705b76e6a948b Mon Sep 17 00:00:00 2001 From: davidby-influx Date: Tue, 15 Sep 2026 21:02:31 -0700 Subject: [PATCH 6/6] chore: manual rebase --- cmd/influxd/launcher/cmd_test.go | 21 ++++++++------------- 1 file changed, 8 insertions(+), 13 deletions(-) diff --git a/cmd/influxd/launcher/cmd_test.go b/cmd/influxd/launcher/cmd_test.go index 4f1db5cf096..4a6150fb442 100644 --- a/cmd/influxd/launcher/cmd_test.go +++ b/cmd/influxd/launcher/cmd_test.go @@ -276,14 +276,16 @@ func TestNewInfluxdCommand_StartupErrorLinger(t *testing.T) { t.Run("absent", func(t *testing.T) { t.Parallel() - o := resolveOpts(t, viper.New()) + o, err := resolveOpts(t, viper.New()) + assert.NoError(t, err) assert.Zero(t, o.StartupErrorLinger, "the default must exit immediately, as before") }) t.Run("command line", func(t *testing.T) { t.Parallel() - o := resolveOpts(t, viper.New(), "--startup-error-linger=30s") + o, err := resolveOpts(t, viper.New(), "--startup-error-linger=30s") + assert.NoError(t, err) assert.Equal(t, 30*time.Second, o.StartupErrorLinger) }) @@ -294,7 +296,8 @@ func TestNewInfluxdCommand_StartupErrorLinger(t *testing.T) { v.SetConfigType("yaml") require.NoError(t, v.ReadConfig(strings.NewReader("startup-error-linger: 1m\n"))) - o := resolveOpts(t, v) + o, err := resolveOpts(t, v) + assert.NoError(t, err) assert.Equal(t, time.Minute, o.StartupErrorLinger) }) } @@ -305,15 +308,7 @@ func TestNewInfluxdCommand_StartupErrorLinger(t *testing.T) { func TestNewInfluxdCommand_StartupErrorLingerFromEnv(t *testing.T) { t.Setenv("INFLUXD_STARTUP_ERROR_LINGER", "45s") - o := resolveOpts(t, viper.New()) + o, err := resolveOpts(t, viper.New()) + assert.NoError(t, err) assert.Equal(t, 45*time.Second, o.StartupErrorLinger) } - -// TestPrintConfig_ReportsStartupErrorLinger keeps the option discoverable: an -// operator finds it by reading what print-config emits. -func TestPrintConfig_ReportsStartupErrorLinger(t *testing.T) { - t.Parallel() - - got := printConfig(t) - assert.Contains(t, got, "startup-error-linger: 0s") -}