diff --git a/README.md b/README.md index d60e7b0..344f012 100644 --- a/README.md +++ b/README.md @@ -38,15 +38,17 @@ not PQL) is forwarded verbatim. | `GET /metrics/v2/read/` | Fan out to all and merge the Jolokia response; numeric attributes are **summed** by default (see merge semantics). | | `GET /metrics/v2/list` | Fan out to all and serve the **union** of the backends' MBean trees. | | `GET /metrics/v1/mbeans[/]` | Same merge, applied to the legacy envelope-less body. | -| `GET /healthz` | Per-backend reachability plus cache state. `200 {"status":"ok"}` if all reachable, `200 degraded` if some fail, `503 down` if all fail. | +| `GET /healthz` | Per-backend reachability, probe state and cache state. `200 {"status":"ok"}` if all reachable, `200 degraded` if some fail, `503 down` if all fail. | -Fan-out is concurrent. If one backend errors or times out, `pdbmux` serves the -surviving backends' results and logs a warning; a merged endpoint only returns `502` when -**every** backend fails. Response records are passed through as raw JSON so -unknown fields survive untouched. +Fan-out is concurrent, and goes only to the backends the health prober currently +believes are up — see [Backend health](#backend-health). If one backend errors or +times out, `pdbmux` serves the surviving backends' results and logs a warning; a +merged endpoint only returns `502` when **every** backend fails. Response records +are passed through as raw JSON so unknown fields survive untouched. Responses carry PuppetDB's `X-Records` when the query asked for a total, and on -the cached paths two headers `pdbmux` adds itself, `X-Cache` and `Age` — see +the merged paths `X-Backends` (see [Backend health](#backend-health)). Cached +paths add two more headers `pdbmux` sets itself, `X-Cache` and `Age` — see [Caching](#caching). ## Merge semantics @@ -212,6 +214,79 @@ Each backend applies `order_by`/`limit`/`offset` to its own slice only, so - A malformed `limit`, `offset` or `order_by` gets a `400` rather than being forwarded. +## Backend health + +A backend that is down otherwise costs a full `timeout` stall on **every** +request, since fan-out has no way to know before it asks. `pdbmux` polls each +backend's status endpoint in the background instead, and skips the ones that are +not answering. + +- **Endpoint** — `health_probe_path`, default `/status/v1/services`, PuppetDB's + trapperkeeper status service (unauthenticated by default). A backend is healthy + when it answers `200` **and** every service in the body reports + `"state": "running"` — a `200` whose body says `starting`, `stopping`, `error` + or `unknown` counts as a failure. A body that is not in that shape is judged on + its status code alone, so pointing `health_probe_path` at some other endpoint + still works. +- **A refused probe is not a sick backend.** Only evidence about the *backend* + takes it out of the pool: a transport failure (connection refused, DNS, TLS, + timeout), a `5xx` — `503` included, since trapperkeeper answers `503` exactly + when its services are not nominal — or a `429`, which is the backend reporting + its own capacity rather than judging the request, so an overloaded backend gets + backed off instead of kept at full traffic. A reply that refuses the *probe + request* is evidence about the probe instead: the other `4xx` are the backend + answering that our request is the problem (`404`/`410` the path is not there, + `405` it does not take a `GET`, `401`/`403` we are not allowed to ask), and + `501` says it does not implement the + endpoint. A backend answering that way is **left in service** — still queried, + still contributing records — and reported as `probe_unsupported` rather than + `healthy`, so an operator can tell "verified healthy" from "not actually being + checked". A probe path that is wrong for one backend alone can therefore never + strand a working backend. The misconfiguration is logged once per transition, + naming the backend, the probe path and the status. +- **Thresholds** — a healthy backend leaves the pool after + `health_probe_failures` (default 3) **consecutive** unsuccessful probes; a down + one comes back after `health_probe_successes` (default 2) consecutive + successes, and the same thresholds gate `probe_unsupported` in and out. One + blip cannot flap a backend out, and one lucky reply cannot flap it back in. + The run counts every unsuccessful probe whatever its kind, so a backend that + fails every probe in mixed ways — a `503`, then a `404`, then a timeout — + still trips the threshold; only a success resets the run. The kinds decide + *which* state the run enters: a real failure anywhere in the run outranks a + refusal, so the backend goes `unhealthy` and is skipped, and only a run of + nothing but refusals enters `probe_unsupported` and stays in service. A down + backend keeps being probed, so recovery is automatic. +- **Fails open** — if the prober has marked **every** backend down, `pdbmux` + queries them all anyway. A wrong `health_probe_path`, a broken prober or a + partition that only the prober sees can therefore never black-hole traffic; + the worst case is today's behaviour. +- **Serves immediately** — the listener never waits for a first probe round, and + a backend nobody has probed yet counts as healthy, so a restart drops nothing. +- **Quiet** — only *transitions* (up→down, down→up, in and out of + `probe_unsupported`) are logged, never individual probes. +- **Partial responses stay partial.** Health state changes which backends are + asked, never what a merged answer means: a response built from a subset is + still served, as before. Every merged response carries `X-Backends: + /` naming how many backends' records went into it, so + a client can tell a full answer from a partial one. On a cache hit the header + describes the stored body, not the current backend count. +- **`/healthz`** gives each backend a `state` (`healthy`, `unhealthy`, + `probe_unsupported`, `unprobed`, or `unmonitored` when probing is off), + `consecutive_failures`, + `consecutive_successes`, `last_probe` and `last_error`, alongside the + `reachable` check `/healthz` runs itself — which always asks **every** + backend, so a backend queries are skipping is still reported. Read the two + together: `state` is the prober's verdict and `probe_unsupported` means "not + being verified", *not* "well", so `reachable` is the field that says whether + the backend is answering right now. A proxy that `404`s everything because the + backend behind it is dead shows `state: probe_unsupported` with `reachable` + carrying the query error, and the overall `status` drops to `degraded` or + `down` accordingly. A `query` object reports the last merged fan-out: + `partial`, `contributed`, `configured`, `partial_rounds` and `last_partial`. +- **`health_probe_enabled: false`** turns the whole thing off: no probing + goroutines, no backend ever skipped, every backend queried on every request. + `X-Backends` still reports how many answered. + ## Caching `pdbmux` caches merged `/nodes` and `/facts` record sets **in memory** so a busy @@ -298,6 +373,12 @@ facts_ttl: 30s # /facts + /nodes response cache TTL; 0 disables, c facts_cache_bytes: 67108864 # byte budget for that cache (64 MiB), LRU-evicted source_fact: pdbmux_source # name of the synthetic provenance fact source_fact_enabled: true # false serves backends' records untouched +health_probe_enabled: true # false queries every backend on every request +health_probe_path: /status/v1/services # backend health endpoint +health_probe_interval: 10s # how often each backend is probed +health_probe_timeout: 5s # per-probe timeout +health_probe_failures: 3 # consecutive failures before a backend is skipped +health_probe_successes: 2 # consecutive successes before it is used again ``` `backends[*].url` is a **base** URL (`scheme://host[:port]`); `pdbmux` appends @@ -315,8 +396,14 @@ the `/pdb/query/v4/...` path per request. | `PDBMUX_BACKENDS` | whole backend list, as `name=url,name=url` | | `PDBMUX_SOURCE_FACT` | `source_fact` (default `pdbmux_source`) | | `PDBMUX_SOURCE_FACT_ENABLED` | `source_fact_enabled` (default `true`); `false` disables injection | +| `PDBMUX_HEALTH_PROBE_ENABLED` | `health_probe_enabled` (default `true`) | +| `PDBMUX_HEALTH_PROBE_PATH` | `health_probe_path` (default `/status/v1/services`) | +| `PDBMUX_HEALTH_PROBE_INTERVAL` | `health_probe_interval` (Go duration) | +| `PDBMUX_HEALTH_PROBE_TIMEOUT` | `health_probe_timeout` (Go duration) | +| `PDBMUX_HEALTH_PROBE_FAILURES` | `health_probe_failures` (plain integer, minimum 1) | +| `PDBMUX_HEALTH_PROBE_SUCCESSES` | `health_probe_successes` (plain integer, minimum 1) | -Flags: `--config`, `--listen`, `--merge`. +Flags: `--config`, `--listen`, `--merge`, `--health-probe`. `config init` writes to `--config`/`PDBMUX_CONFIG` when set, else to `$XDG_CONFIG_HOME/pdbmux/config.yaml`. diff --git a/config.go b/config.go index ae12542..1666ffd 100644 --- a/config.go +++ b/config.go @@ -34,6 +34,13 @@ const ( maxFactsTTL = 30 * time.Second defaultFactsTTL = 30 * time.Second defaultCacheSize = int64(64 << 20) + + // PuppetDB serves its trapperkeeper status service here, unauthenticated. + defaultHealthProbePath = "/status/v1/services" + defaultHealthProbeInterval = 10 * time.Second + defaultHealthProbeTimeout = 5 * time.Second + defaultHealthProbeFailures = 3 + defaultHealthProbeSuccesses = 2 ) var exampleBackends = []Backend{ @@ -58,6 +65,13 @@ type Config struct { SourceFact string `yaml:"source_fact"` SourceFactEnabled bool `yaml:"source_fact_enabled"` + HealthProbe bool `yaml:"health_probe_enabled"` + HealthProbePath string `yaml:"health_probe_path"` + HealthProbeInterval time.Duration `yaml:"health_probe_interval"` + HealthProbeTimeout time.Duration `yaml:"health_probe_timeout"` + HealthProbeFailures int `yaml:"health_probe_failures"` + HealthProbeSuccesses int `yaml:"health_probe_successes"` + sourcePath string // file this config was read from, empty if none was found factsTTLClamped time.Duration // pre-clamp facts_ttl, zero when nothing was clamped } @@ -80,6 +94,13 @@ func DefaultConfig() Config { CacheBytes: defaultCacheSize, SourceFact: defaultSourceFact, SourceFactEnabled: true, + + HealthProbe: true, + HealthProbePath: defaultHealthProbePath, + HealthProbeInterval: defaultHealthProbeInterval, + HealthProbeTimeout: defaultHealthProbeTimeout, + HealthProbeFailures: defaultHealthProbeFailures, + HealthProbeSuccesses: defaultHealthProbeSuccesses, } } @@ -206,6 +227,34 @@ func applyEnv(cfg *Config, getenv func(string) string) { cfg.CacheBytes = n } } + if v := getenv(envPrefix + "HEALTH_PROBE_ENABLED"); v != "" { + if b, err := strconv.ParseBool(v); err == nil { + cfg.HealthProbe = b + } + } + if v := getenv(envPrefix + "HEALTH_PROBE_PATH"); v != "" { + cfg.HealthProbePath = v + } + if v := getenv(envPrefix + "HEALTH_PROBE_INTERVAL"); v != "" { + if d, err := time.ParseDuration(v); err == nil { + cfg.HealthProbeInterval = d + } + } + if v := getenv(envPrefix + "HEALTH_PROBE_TIMEOUT"); v != "" { + if d, err := time.ParseDuration(v); err == nil { + cfg.HealthProbeTimeout = d + } + } + if v := getenv(envPrefix + "HEALTH_PROBE_FAILURES"); v != "" { + if n, err := strconv.Atoi(v); err == nil { + cfg.HealthProbeFailures = n + } + } + if v := getenv(envPrefix + "HEALTH_PROBE_SUCCESSES"); v != "" { + if n, err := strconv.Atoi(v); err == nil { + cfg.HealthProbeSuccesses = n + } + } if v := getenv(envPrefix + "BACKENDS"); v != "" { if bs := parseBackends(v); len(bs) > 0 { cfg.Backends = bs @@ -272,6 +321,26 @@ func (c Config) Validate() error { if c.CacheBytes < 0 { return fmt.Errorf("facts_cache_bytes must not be negative") } + if c.HealthProbe { + if c.HealthProbePath == "" { + return fmt.Errorf("health_probe_path must be non-empty, or set health_probe_enabled to false") + } + if !strings.HasPrefix(c.HealthProbePath, "/") { + return fmt.Errorf("health_probe_path must start with /, got %q", c.HealthProbePath) + } + if c.HealthProbeInterval <= 0 { + return fmt.Errorf("health_probe_interval must be positive") + } + if c.HealthProbeTimeout <= 0 { + return fmt.Errorf("health_probe_timeout must be positive") + } + if c.HealthProbeFailures < 1 { + return fmt.Errorf("health_probe_failures must be at least 1") + } + if c.HealthProbeSuccesses < 1 { + return fmt.Errorf("health_probe_successes must be at least 1") + } + } return nil } @@ -293,9 +362,14 @@ func writeDefaultConfig(path string) error { "# Env overrides: PDBMUX_LISTEN, PDBMUX_MERGE, PDBMUX_TIMEOUT,\n" + "# PDBMUX_FRESHNESS_TTL, PDBMUX_FACTS_TTL, PDBMUX_FACTS_CACHE_BYTES,\n" + "# PDBMUX_BACKENDS (name=url,name=url),\n" + - "# PDBMUX_SOURCE_FACT, PDBMUX_SOURCE_FACT_ENABLED.\n" + + "# PDBMUX_SOURCE_FACT, PDBMUX_SOURCE_FACT_ENABLED,\n" + + "# PDBMUX_HEALTH_PROBE_ENABLED, PDBMUX_HEALTH_PROBE_PATH,\n" + + "# PDBMUX_HEALTH_PROBE_INTERVAL, PDBMUX_HEALTH_PROBE_TIMEOUT,\n" + + "# PDBMUX_HEALTH_PROBE_FAILURES, PDBMUX_HEALTH_PROBE_SUCCESSES.\n" + "# facts_ttl caches merged /facts and /nodes in memory; it is capped at 30s\n" + - "# (a larger value is clamped) and 0 disables the cache.\n\n") + "# (a larger value is clamped) and 0 disables the cache.\n" + + "# health_probe_* polls each backend's status endpoint so queries skip a\n" + + "# backend that is down; when every backend is down all are queried anyway.\n\n") if err := os.WriteFile(path, append(header, data...), 0o644); err != nil { return fmt.Errorf("writing config: %w", err) } diff --git a/config_test.go b/config_test.go index ef89ea4..89c2d9d 100644 --- a/config_test.go +++ b/config_test.go @@ -118,6 +118,46 @@ func TestApplyEnv_Backends(t *testing.T) { } } +func TestDefaultConfig_HealthProbe(t *testing.T) { + cfg := DefaultConfig() + if !cfg.HealthProbe || cfg.HealthProbePath != defaultHealthProbePath { + t.Errorf("health probe defaults to enabled=%v path=%q, want true %q", + cfg.HealthProbe, cfg.HealthProbePath, defaultHealthProbePath) + } + if cfg.HealthProbeInterval != defaultHealthProbeInterval || cfg.HealthProbeTimeout != defaultHealthProbeTimeout { + t.Errorf("probe interval/timeout = %v/%v", cfg.HealthProbeInterval, cfg.HealthProbeTimeout) + } + if cfg.HealthProbeFailures != defaultHealthProbeFailures || cfg.HealthProbeSuccesses != defaultHealthProbeSuccesses { + t.Errorf("probe thresholds = %d/%d", cfg.HealthProbeFailures, cfg.HealthProbeSuccesses) + } +} + +func TestApplyEnv_HealthProbe(t *testing.T) { + cfg := testConfigValid() + env := map[string]string{ + envPrefix + "HEALTH_PROBE_ENABLED": "false", + envPrefix + "HEALTH_PROBE_PATH": "/status/v1/simple", + envPrefix + "HEALTH_PROBE_INTERVAL": "45s", + envPrefix + "HEALTH_PROBE_TIMEOUT": "2s", + envPrefix + "HEALTH_PROBE_FAILURES": "5", + envPrefix + "HEALTH_PROBE_SUCCESSES": "1", + } + applyEnv(&cfg, func(k string) string { return env[k] }) + + if cfg.HealthProbe { + t.Error("PDBMUX_HEALTH_PROBE_ENABLED=false did not disable probing") + } + if cfg.HealthProbePath != "/status/v1/simple" { + t.Errorf("path = %q", cfg.HealthProbePath) + } + if cfg.HealthProbeInterval != 45*time.Second || cfg.HealthProbeTimeout != 2*time.Second { + t.Errorf("interval/timeout = %v/%v", cfg.HealthProbeInterval, cfg.HealthProbeTimeout) + } + if cfg.HealthProbeFailures != 5 || cfg.HealthProbeSuccesses != 1 { + t.Errorf("thresholds = %d/%d", cfg.HealthProbeFailures, cfg.HealthProbeSuccesses) + } +} + func TestDefaultConfig_SourceFact(t *testing.T) { cfg := DefaultConfig() if cfg.SourceFact != defaultSourceFact || !cfg.SourceFactEnabled { @@ -195,6 +235,16 @@ func TestValidate(t *testing.T) { {"zero timeout", func(c *Config) { c.Timeout = 0 }, true}, {"empty source fact while enabled", func(c *Config) { c.SourceFact = "" }, true}, {"empty source fact while disabled", func(c *Config) { c.SourceFact = ""; c.SourceFactEnabled = false }, false}, + {"empty health probe path", func(c *Config) { c.HealthProbePath = "" }, true}, + {"relative health probe path", func(c *Config) { c.HealthProbePath = "status/v1/services" }, true}, + {"zero health probe interval", func(c *Config) { c.HealthProbeInterval = 0 }, true}, + {"zero health probe timeout", func(c *Config) { c.HealthProbeTimeout = 0 }, true}, + {"zero health probe failures", func(c *Config) { c.HealthProbeFailures = 0 }, true}, + {"zero health probe successes", func(c *Config) { c.HealthProbeSuccesses = 0 }, true}, + {"bad health probe settings while disabled", func(c *Config) { + c.HealthProbe = false + c.HealthProbePath, c.HealthProbeInterval, c.HealthProbeFailures = "", 0, 0 + }, false}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { @@ -456,7 +506,9 @@ func captureStdout(t *testing.T, f func()) string { func clearEnv(t *testing.T) { t.Helper() - for _, k := range []string{"CONFIG", "LISTEN", "MERGE", "TIMEOUT", "FRESHNESS_TTL", "FACTS_TTL", "FACTS_CACHE_BYTES", "BACKENDS"} { + for _, k := range []string{"CONFIG", "LISTEN", "MERGE", "TIMEOUT", "FRESHNESS_TTL", "FACTS_TTL", "FACTS_CACHE_BYTES", "BACKENDS", + "HEALTH_PROBE_ENABLED", "HEALTH_PROBE_PATH", "HEALTH_PROBE_INTERVAL", "HEALTH_PROBE_TIMEOUT", + "HEALTH_PROBE_FAILURES", "HEALTH_PROBE_SUCCESSES"} { t.Setenv(envPrefix+k, "") } } diff --git a/health.go b/health.go new file mode 100644 index 0000000..23d9b47 --- /dev/null +++ b/health.go @@ -0,0 +1,373 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "log" + "net/http" + "strings" + "sync" + "time" +) + +// probeBodyLimit caps the status body read; only the per-service state is used. +const probeBodyLimit = 1 << 20 + +// runningState is the healthy state reported by PuppetDB's status service. +const runningState = "running" + +const ( + stateHealthy = "healthy" + stateUnhealthy = "unhealthy" + stateUnprobed = "unprobed" + stateUnmonitored = "unmonitored" + stateProbeUnsupported = "probe_unsupported" +) + +// probeOutcome is what one probe reply said. A rejected probe is deliberately +// neither a success nor a failure: it is evidence about the probe request, not +// about the backend behind it. +type probeOutcome int + +const ( + outcomeOK probeOutcome = iota + outcomeFailure + outcomeRejected +) + +type probeState struct { + healthy bool + unsupported bool + probed bool + // runFailed records whether the current run of not-OK probes contains at + // least one real failure, which decides which non-healthy state it enters. + runFailed bool + failures int + successes int + lastProbe time.Time + lastErr string +} + +// backendHealth is a copy of one backend's probe state, safe to read outside the +// prober's lock. +type backendHealth struct { + Healthy bool + Unsupported bool + Probed bool + Failures int + Successes int + LastProbe time.Time + LastErr string +} + +func (h backendHealth) stateName() string { + switch { + case !h.Probed: + return stateUnprobed + case h.Unsupported: + return stateProbeUnsupported + case h.Healthy: + return stateHealthy + default: + return stateUnhealthy + } +} + +// probeRejectedError marks a reply that refused the probe request itself rather +// than reporting the backend unwell. +type probeRejectedError struct { + status int + path string +} + +func (e *probeRejectedError) Error() string { + return fmt.Sprintf("HTTP %d: probe path %q is not usable on this backend", e.status, e.path) +} + +// probeRejects reports whether a status code refuses the probe request rather +// than reporting the service unhealthy. A 4xx is the backend answering that our +// request is the problem, and 501 says it does not implement the endpoint. +// 429 is excluded: it is the backend reporting its own capacity, not a verdict +// on the request, so an overloaded backend must get backed off rather than kept +// at full traffic. Other 5xx stay real evidence too: trapperkeeper answers 503 +// exactly when its services are not nominal. +func probeRejects(code int) bool { + if code == http.StatusTooManyRequests { + return false + } + return (code >= 400 && code < 500) || code == http.StatusNotImplemented +} + +func classifyProbe(err error) probeOutcome { + switch { + case err == nil: + return outcomeOK + case errors.As(err, new(*probeRejectedError)): + return outcomeRejected + default: + return outcomeFailure + } +} + +// prober polls each backend's health endpoint on an interval and tracks whether +// it is currently answering. A nil *prober means probing is disabled and every +// method is a no-op reporting every backend healthy. +type prober struct { + backends []Backend + path string + interval time.Duration + failures int + successes int + client *http.Client + log *log.Logger + now func() time.Time + + mu sync.Mutex + states map[string]*probeState + + wg sync.WaitGroup + cancel context.CancelFunc + stopOnce sync.Once +} + +func newProber(cfg Config, logger *log.Logger) *prober { + if !cfg.HealthProbe { + return nil + } + p := &prober{ + backends: append([]Backend(nil), cfg.Backends...), + path: cfg.HealthProbePath, + interval: cfg.HealthProbeInterval, + failures: cfg.HealthProbeFailures, + successes: cfg.HealthProbeSuccesses, + client: &http.Client{Timeout: cfg.HealthProbeTimeout}, + log: logger, + now: time.Now, + states: make(map[string]*probeState, len(cfg.Backends)), + } + // A backend nobody has probed yet counts as healthy, so a restart never + // drops traffic while the first round runs. + for _, b := range cfg.Backends { + p.states[b.Name] = &probeState{healthy: true} + } + return p +} + +// Start launches one polling goroutine per backend and returns immediately. +func (p *prober) Start(ctx context.Context) { + if p == nil { + return + } + ctx, p.cancel = context.WithCancel(ctx) + for _, b := range p.backends { + p.wg.Add(1) + go p.loop(ctx, b) + } +} + +// Stop cancels the polling goroutines and waits for them to exit. +func (p *prober) Stop() { + if p == nil || p.cancel == nil { + return + } + p.stopOnce.Do(p.cancel) + p.wg.Wait() + p.client.CloseIdleConnections() +} + +func (p *prober) loop(ctx context.Context, b Backend) { + defer p.wg.Done() + ticker := time.NewTicker(p.interval) + defer ticker.Stop() + for { + if ctx.Err() != nil { + return + } + p.probeOne(ctx, b) + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } +} + +// healthy reports whether queries should go to this backend. An unknown name is +// healthy so a config the prober does not know about is never black-holed. +func (p *prober) healthy(name string) bool { + if p == nil { + return true + } + p.mu.Lock() + defer p.mu.Unlock() + st, ok := p.states[name] + if !ok { + return true + } + return st.healthy +} + +func (p *prober) snapshot() map[string]backendHealth { + if p == nil { + return nil + } + p.mu.Lock() + defer p.mu.Unlock() + out := make(map[string]backendHealth, len(p.states)) + for name, st := range p.states { + out[name] = backendHealth{ + Healthy: st.healthy, + Unsupported: st.unsupported, + Probed: st.probed, + Failures: st.failures, + Successes: st.successes, + LastProbe: st.lastProbe, + LastErr: st.lastErr, + } + } + return out +} + +func (p *prober) probeOne(ctx context.Context, b Backend) { + err := p.probe(ctx, b) + // A probe cut short by shutdown says nothing about the backend. + if ctx.Err() != nil { + return + } + p.record(b.Name, err) +} + +// record applies one probe outcome, flipping state only once the matching +// threshold is reached so a single blip cannot flap a backend in either +// direction. One run of consecutive not-OK probes makes the down decision, no +// matter how the kinds are mixed; the kinds only pick which non-healthy state +// the run enters, and a real failure anywhere in the run outranks a rejection. +func (p *prober) record(name string, err error) { + p.mu.Lock() + st, ok := p.states[name] + if !ok { + p.mu.Unlock() + return + } + outcome := classifyProbe(err) + st.probed = true + st.lastProbe = p.now() + + var transition string + if outcome == outcomeOK { + st.lastErr = "" + st.failures, st.runFailed = 0, false + st.successes++ + if st.successes >= p.successes { + switch { + case st.unsupported: + st.unsupported = false + transition = fmt.Sprintf("info: backend %q health probe %q is usable again", name, p.path) + case !st.healthy: + st.healthy = true + transition = fmt.Sprintf("info: backend %q is up after %d consecutive probe successes", name, st.successes) + } + } + } else { + st.lastErr = err.Error() + st.successes = 0 + st.failures++ + st.runFailed = st.runFailed || outcome == outcomeFailure + if st.failures >= p.failures { + switch { + case st.runFailed: + // Some probe in this run was evidence about the backend itself, so the + // run is a failure run whatever else it also contains. + st.unsupported = false + if st.healthy { + st.healthy = false + transition = fmt.Sprintf("warning: backend %q is down after %d consecutive failed probes: %v", name, st.failures, err) + } + case !st.unsupported: + // Nothing but refusals of the probe request, which says nothing about the + // backend's health: it keeps serving queries, unverified. + st.unsupported, st.healthy = true, true + transition = fmt.Sprintf("warning: backend %q rejected the health probe %d times: %v; leaving it in service but unverified, fix health_probe_path or its authorization", name, st.failures, err) + } + } + } + p.mu.Unlock() + + // Only transitions are logged: this loop runs for the life of the process. + if transition != "" { + p.log.Print(transition) + } +} + +func (p *prober) probe(ctx context.Context, b Backend) error { + target := strings.TrimRight(b.URL, "/") + p.path + req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil) + if err != nil { + return err + } + resp, err := p.client.Do(req) + if err != nil { + return err + } + defer func() { _ = resp.Body.Close() }() + body, err := io.ReadAll(io.LimitReader(resp.Body, probeBodyLimit)) + if err != nil { + return err + } + if probeRejects(resp.StatusCode) { + return &probeRejectedError{status: resp.StatusCode, path: p.path} + } + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) + } + return statusBodyHealthy(body) +} + +// statusBodyHealthy rejects a 200 whose body reports a service that is starting, +// stopping or in error. A body that does not speak the status service's shape is +// accepted on its status code alone, so a custom health path still works. +func statusBodyHealthy(body []byte) error { + var services map[string]struct { + State string `json:"state"` + } + if err := json.Unmarshal(body, &services); err != nil { + return nil + } + for name, svc := range services { + if svc.State != "" && svc.State != runningState { + return fmt.Errorf("service %q is %q, not %q", name, svc.State, runningState) + } + } + return nil +} + +// partialTracker records whether the last merged fan-out heard from every +// configured backend. +type partialTracker struct { + mu sync.Mutex + seen bool + contributed int + configured int + rounds uint64 + last time.Time +} + +func (t *partialTracker) record(contributed, configured int, now time.Time) { + t.mu.Lock() + defer t.mu.Unlock() + t.seen = true + t.contributed, t.configured = contributed, configured + if contributed < configured { + t.rounds++ + t.last = now + } +} + +func (t *partialTracker) snapshot() (seen bool, contributed, configured int, rounds uint64, last time.Time) { + t.mu.Lock() + defer t.mu.Unlock() + return t.seen, t.contributed, t.configured, t.rounds, t.last +} diff --git a/health_test.go b/health_test.go new file mode 100644 index 0000000..4a3e040 --- /dev/null +++ b/health_test.go @@ -0,0 +1,866 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "io" + "log" + "net/http" + "net/http/httptest" + "runtime" + "strings" + "sync" + "testing" + "time" +) + +const runningStatusBody = `{"puppetdb-status":{"service_version":"8.0.0","service_status_version":1,` + + `"detail_level":"info","state":"running","status":{},"active_alerts":[]},` + + `"status-service":{"service_version":"1.1.1","service_status_version":1,` + + `"detail_level":"info","state":"running","status":{},"active_alerts":[]}}` + +func degradedStatusBody(state string) string { + return `{"puppetdb-status":{"service_version":"8.0.0","service_status_version":1,` + + `"detail_level":"info","state":"` + state + `","status":{},"active_alerts":[]}}` +} + +// statusBackend is a PuppetDB status endpoint whose reply the test can change +// between probes. +type statusBackend struct { + srv *httptest.Server + + mu sync.Mutex + status int + body string +} + +func newStatusBackend(t *testing.T) *statusBackend { + t.Helper() + sb := &statusBackend{status: http.StatusOK, body: runningStatusBody} + sb.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + sb.mu.Lock() + code, body := sb.status, sb.body + sb.mu.Unlock() + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(code) + _, _ = io.WriteString(w, body) + })) + t.Cleanup(sb.srv.Close) + return sb +} + +func (sb *statusBackend) reply(status int, body string) { + sb.mu.Lock() + defer sb.mu.Unlock() + sb.status, sb.body = status, body +} + +func newTestProber(t *testing.T, cfg Config) *prober { + t.Helper() + p := newProber(cfg, log.New(io.Discard, "", 0)) + if p == nil { + t.Fatal("newProber returned nil for an enabled config") + } + return p +} + +func healthConfig(aURL, bURL string) Config { + cfg := testConfig(aURL, bURL, mergeStatic) + cfg.HealthProbe = true + cfg.HealthProbePath = defaultHealthProbePath + cfg.HealthProbeInterval = 5 * time.Millisecond + cfg.HealthProbeTimeout = 2 * time.Second + cfg.HealthProbeFailures = 2 + cfg.HealthProbeSuccesses = 2 + return cfg +} + +func markDown(t *testing.T, p *prober, name string) { + t.Helper() + for i := 0; i < p.failures; i++ { + p.record(name, errors.New("probe failed")) + } + if p.healthy(name) { + t.Fatalf("backend %q still healthy after %d probe failures", name, p.failures) + } +} + +func rejectedProbe(p *prober) error { + return &probeRejectedError{status: http.StatusNotFound, path: p.path} +} + +func markUnsupported(t *testing.T, p *prober, name string) { + t.Helper() + for i := 0; i < p.failures; i++ { + p.record(name, rejectedProbe(p)) + } + if !p.healthy(name) { + t.Fatalf("backend %q was excluded by a rejected probe", name) + } + if state := p.snapshot()[name].stateName(); state != stateProbeUnsupported { + t.Fatalf("state = %q, want %q", state, stateProbeUnsupported) + } +} + +func TestProber_FailureThresholdDebouncesABlip(t *testing.T) { + cfg := healthConfig("http://a.invalid", "http://b.invalid") + cfg.HealthProbeFailures = 3 + p := newTestProber(t, cfg) + + for i := 1; i < 3; i++ { + p.record("a", errors.New("boom")) + if !p.healthy("a") { + t.Fatalf("backend went down after %d of 3 failures", i) + } + } + // A success inside the window resets the run, so the blip never flaps it out. + p.record("a", nil) + p.record("a", errors.New("boom")) + p.record("a", errors.New("boom")) + if !p.healthy("a") { + t.Fatal("a failure run interrupted by a success should not mark the backend down") + } + p.record("a", errors.New("boom")) + if p.healthy("a") { + t.Fatal("backend should be down after 3 consecutive failures") + } +} + +func TestProber_SuccessThresholdDebouncesRecovery(t *testing.T) { + cfg := healthConfig("http://a.invalid", "http://b.invalid") + cfg.HealthProbeSuccesses = 2 + p := newTestProber(t, cfg) + markDown(t, p, "a") + + p.record("a", nil) + if p.healthy("a") { + t.Fatal("one success should not bring a down backend back") + } + // A failure in between restarts the recovery run. + p.record("a", errors.New("boom")) + p.record("a", nil) + if p.healthy("a") { + t.Fatal("recovery run should restart after an interleaved failure") + } + p.record("a", nil) + if !p.healthy("a") { + t.Fatal("backend should be back up after 2 consecutive successes") + } +} + +func TestProber_UnprobedBackendIsHealthy(t *testing.T) { + p := newTestProber(t, healthConfig("http://a.invalid", "http://b.invalid")) + if !p.healthy("a") || !p.healthy("b") { + t.Fatal("a backend nobody has probed yet must count as healthy") + } + if !p.healthy("not-configured") { + t.Fatal("an unknown backend must count as healthy") + } + if state := p.snapshot()["a"].stateName(); state != stateUnprobed { + t.Fatalf("state = %q, want %q", state, stateUnprobed) + } +} + +func TestProber_ProbeAcceptsRunningStatus(t *testing.T) { + sb := newStatusBackend(t) + p := newTestProber(t, healthConfig(sb.srv.URL, "http://b.invalid")) + if err := p.probe(t.Context(), Backend{Name: "a", URL: sb.srv.URL}); err != nil { + t.Fatalf("running status rejected: %v", err) + } +} + +func TestProber_ProbeRejectsDegradedBodyOnA200(t *testing.T) { + sb := newStatusBackend(t) + p := newTestProber(t, healthConfig(sb.srv.URL, "http://b.invalid")) + for _, state := range []string{"starting", "error", "stopping", "unknown"} { + sb.reply(http.StatusOK, degradedStatusBody(state)) + err := p.probe(t.Context(), Backend{Name: "a", URL: sb.srv.URL}) + if err == nil { + t.Fatalf("state %q on a 200 was accepted as healthy", state) + } + if !strings.Contains(err.Error(), state) { + t.Errorf("error %q does not name the state %q", err, state) + } + } +} + +func TestProber_ProbeRejectsNon200(t *testing.T) { + sb := newStatusBackend(t) + sb.reply(http.StatusServiceUnavailable, degradedStatusBody("error")) + p := newTestProber(t, healthConfig(sb.srv.URL, "http://b.invalid")) + if err := p.probe(t.Context(), Backend{Name: "a", URL: sb.srv.URL}); err == nil { + t.Fatal("503 was accepted as healthy") + } +} + +// A deployment pointing health_probe_path at something that is not the status +// service still works: the status code decides. +func TestProber_ProbeAcceptsNonStatusBody(t *testing.T) { + sb := newStatusBackend(t) + sb.reply(http.StatusOK, `running`) + p := newTestProber(t, healthConfig(sb.srv.URL, "http://b.invalid")) + if err := p.probe(t.Context(), Backend{Name: "a", URL: sb.srv.URL}); err != nil { + t.Fatalf("non-status 200 body rejected: %v", err) + } +} + +func TestProber_ProbesUseTheConfiguredPath(t *testing.T) { + var gotPath string + var mu sync.Mutex + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + gotPath = r.URL.Path + mu.Unlock() + _, _ = io.WriteString(w, runningStatusBody) + })) + t.Cleanup(srv.Close) + + cfg := healthConfig(srv.URL, "http://b.invalid") + cfg.HealthProbePath = "/custom/health" + p := newTestProber(t, cfg) + if err := p.probe(t.Context(), Backend{Name: "a", URL: srv.URL + "/"}); err != nil { + t.Fatalf("probe: %v", err) + } + mu.Lock() + defer mu.Unlock() + if gotPath != "/custom/health" { + t.Fatalf("probed %q, want /custom/health", gotPath) + } +} + +// The loop keeps probing a backend it has marked down, so recovery needs no +// operator action. +func TestProber_RecoversWithoutIntervention(t *testing.T) { + sb := newStatusBackend(t) + sb.reply(http.StatusServiceUnavailable, degradedStatusBody("error")) + p := newTestProber(t, healthConfig(sb.srv.URL, "http://b.invalid")) + + p.Start(t.Context()) + defer p.Stop() + + waitFor(t, func() bool { return !p.healthy("a") }) + sb.reply(http.StatusOK, runningStatusBody) + waitFor(t, func() bool { return p.healthy("a") }) +} + +func TestProber_StopLeavesNoGoroutines(t *testing.T) { + sb := newStatusBackend(t) + cfg := healthConfig(sb.srv.URL, sb.srv.URL) + cfg.Backends = []Backend{ + {Name: "a", URL: sb.srv.URL}, + {Name: "b", URL: sb.srv.URL}, + {Name: "c", URL: sb.srv.URL}, + } + p := newTestProber(t, cfg) + + settle := func() int { + runtime.GC() + time.Sleep(20 * time.Millisecond) + return runtime.NumGoroutine() + } + before := settle() + + p.Start(context.Background()) + waitFor(t, func() bool { + snap := p.snapshot() + return snap["a"].Probed && snap["b"].Probed && snap["c"].Probed + }) + p.Stop() + + deadline := time.Now().Add(5 * time.Second) + var after int + for time.Now().Before(deadline) { + if after = settle(); after <= before { + return + } + } + t.Fatalf("goroutines did not settle after Stop: %d before, %d after", before, after) +} + +func TestProber_StopIsIdempotentAndSafeWhenNeverStarted(t *testing.T) { + p := newTestProber(t, healthConfig("http://a.invalid", "http://b.invalid")) + p.Stop() + + p.Start(context.Background()) + p.Stop() + p.Stop() +} + +func TestFanOut_SkipsUnhealthyBackend(t *testing.T) { + a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`) + b := newFakeBackend(t, `[`+node("h2", "2026-07-20T00:00:00Z")+`]`, `[]`) + srv := newTestServer(healthConfig(a.srv.URL, b.srv.URL)) + markDown(t, srv.health, "b") + + rec := doGet(t, srv.Handler(), nodesPath, "") + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + if _, asked := b.params(nodesPath); asked { + t.Fatal("an unhealthy backend was still queried") + } + if _, asked := a.params(nodesPath); !asked { + t.Fatal("the healthy backend was not queried") + } + if !strings.Contains(rec.Body.String(), "h1") || strings.Contains(rec.Body.String(), "h2") { + t.Fatalf("unexpected merged body: %s", rec.Body.String()) + } + if got := rec.Header().Get(backendsHeader); got != "1/2" { + t.Errorf("%s = %q, want 1/2", backendsHeader, got) + } +} + +func TestFanOut_RecoveredBackendIsQueriedAgain(t *testing.T) { + a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`) + b := newFakeBackend(t, `[`+node("h2", "2026-07-20T00:00:00Z")+`]`, `[]`) + srv := newTestServer(healthConfig(a.srv.URL, b.srv.URL)) + markDown(t, srv.health, "b") + doGet(t, srv.Handler(), nodesPath, "") + + for i := 0; i < srv.health.successes; i++ { + srv.health.record("b", nil) + } + if !srv.health.healthy("b") { + t.Fatal("b should be healthy again") + } + + rec := doGet(t, srv.Handler(), nodesPath, "") + if _, asked := b.params(nodesPath); !asked { + t.Fatal("a recovered backend was not queried again") + } + if !strings.Contains(rec.Body.String(), "h2") { + t.Fatalf("recovered backend's records missing: %s", rec.Body.String()) + } + if got := rec.Header().Get(backendsHeader); got != "2/2" { + t.Errorf("%s = %q, want 2/2", backendsHeader, got) + } +} + +// Failing open matters more than skipping dead backends: a broken prober must +// not black-hole every query. +func TestFanOut_FailsOpenWhenEveryBackendIsUnhealthy(t *testing.T) { + a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`) + b := newFakeBackend(t, `[`+node("h2", "2026-07-20T00:00:00Z")+`]`, `[]`) + srv := newTestServer(healthConfig(a.srv.URL, b.srv.URL)) + markDown(t, srv.health, "a") + markDown(t, srv.health, "b") + + rec := doGet(t, srv.Handler(), nodesPath, "") + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + if _, asked := a.params(nodesPath); !asked { + t.Error("a was not queried despite being the only kind of backend left") + } + if _, asked := b.params(nodesPath); !asked { + t.Error("b was not queried despite being the only kind of backend left") + } + body := rec.Body.String() + if !strings.Contains(body, "h1") || !strings.Contains(body, "h2") { + t.Fatalf("fail-open should have merged both backends: %s", body) + } +} + +func TestFanOut_UnprobedBackendsAreQueried(t *testing.T) { + a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`) + b := newFakeBackend(t, `[`+node("h2", "2026-07-20T00:00:00Z")+`]`, `[]`) + srv := newTestServer(healthConfig(a.srv.URL, b.srv.URL)) + + doGet(t, srv.Handler(), nodesPath, "") + if _, asked := a.params(nodesPath); !asked { + t.Error("a was skipped before its first probe") + } + if _, asked := b.params(nodesPath); !asked { + t.Error("b was skipped before its first probe") + } +} + +func TestHealthProbe_DisabledIsANoOp(t *testing.T) { + a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`) + b := newFakeBackend(t, `[`+node("h2", "2026-07-20T00:00:00Z")+`]`, `[]`) + cfg := healthConfig(a.srv.URL, b.srv.URL) + cfg.HealthProbe = false + srv := newTestServer(cfg) + + if srv.health != nil { + t.Fatal("probing is disabled but a prober was built") + } + srv.StartProbes(context.Background()) + srv.StopProbes() + + rec := doGet(t, srv.Handler(), nodesPath, "") + for name, fb := range map[string]*fakeBackend{"a": a, "b": b} { + if _, asked := fb.params(nodesPath); !asked { + t.Errorf("backend %s was not queried", name) + } + if _, probed := fb.params(defaultHealthProbePath); probed { + t.Errorf("backend %s was probed with probing disabled", name) + } + } + body := rec.Body.String() + if !strings.Contains(body, "h1") || !strings.Contains(body, "h2") { + t.Fatalf("unexpected merged body: %s", body) + } +} + +func TestHealthz_ReportsProbeState(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + b := newFakeBackend(t, `[]`, `[]`) + srv := newTestServer(healthConfig(a.srv.URL, b.srv.URL)) + srv.health.record("a", nil) + markDown(t, srv.health, "b") + + rec := doGet(t, srv.Handler(), "/healthz", "") + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + var hr healthReport + if err := json.Unmarshal(rec.Body.Bytes(), &hr); err != nil { + t.Fatal(err) + } + + if got := hr.Backends["a"].State; got != stateHealthy { + t.Errorf("a state = %q, want %q", got, stateHealthy) + } + down := hr.Backends["b"] + if down.State != stateUnhealthy { + t.Errorf("b state = %q, want %q", down.State, stateUnhealthy) + } + if down.Failures != srv.health.failures { + t.Errorf("b consecutive_failures = %d, want %d", down.Failures, srv.health.failures) + } + if down.LastError == "" || down.LastProbe == "" { + t.Errorf("b is missing last_error/last_probe: %+v", down) + } + // A backend the prober has down is still reported, and still reachable. + if down.Reachable != "ok" { + t.Errorf("b reachable = %q, want ok", down.Reachable) + } +} + +func TestHealthz_ReportsUnmonitoredWhenProbingIsDisabled(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + b := newFakeBackend(t, `[]`, `[]`) + cfg := healthConfig(a.srv.URL, b.srv.URL) + cfg.HealthProbe = false + srv := newTestServer(cfg) + + rec := doGet(t, srv.Handler(), "/healthz", "") + var hr healthReport + if err := json.Unmarshal(rec.Body.Bytes(), &hr); err != nil { + t.Fatal(err) + } + if got := hr.Backends["a"].State; got != stateUnmonitored { + t.Errorf("a state = %q, want %q", got, stateUnmonitored) + } +} + +func TestHealthz_ReportsPartialRounds(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + b := newFakeBackend(t, `[]`, `[]`) + b.fail = true + srv := newTestServer(healthConfig(a.srv.URL, b.srv.URL)) + + if q := srv.queryHealth(); q.Partial { + t.Fatalf("no query has run yet, but partial is set: %+v", q) + } + + doGet(t, srv.Handler(), nodesPath, "") + var hr healthReport + rec := doGet(t, srv.Handler(), "/healthz", "") + if err := json.Unmarshal(rec.Body.Bytes(), &hr); err != nil { + t.Fatal(err) + } + if !hr.Query.Partial || hr.Query.Contributed != 1 || hr.Query.Configured != 2 { + t.Fatalf("partial round not reported: %+v", hr.Query) + } + if hr.Query.PartialRounds != 1 || hr.Query.LastPartial == "" { + t.Fatalf("partial round not counted: %+v", hr.Query) + } + + b.fail = false + doGet(t, srv.Handler(), nodesPath, "") + rec = doGet(t, srv.Handler(), "/healthz", "") + if err := json.Unmarshal(rec.Body.Bytes(), &hr); err != nil { + t.Fatal(err) + } + if hr.Query.Partial || hr.Query.Contributed != 2 { + t.Fatalf("full round still reported as partial: %+v", hr.Query) + } +} + +func TestBackendsHeader_CountsContributors(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + b := newFakeBackend(t, `[]`, `[]`) + srv := newTestServer(healthConfig(a.srv.URL, b.srv.URL)) + + if got := doGet(t, srv.Handler(), nodesPath, "").Header().Get(backendsHeader); got != "2/2" { + t.Errorf("%s = %q, want 2/2", backendsHeader, got) + } + + b.fail = true + if got := doGet(t, srv.Handler(), nodesPath, "").Header().Get(backendsHeader); got != "1/2" { + t.Errorf("%s = %q, want 1/2", backendsHeader, got) + } +} + +// A reply that refuses the probe request says nothing about the backend, so the +// backend must keep serving queries instead of being stranded out of the pool. +func TestProber_RejectedProbeKeepsBackendInService(t *testing.T) { + codes := []int{ + http.StatusBadRequest, http.StatusUnauthorized, http.StatusForbidden, + http.StatusNotFound, http.StatusMethodNotAllowed, http.StatusGone, + http.StatusNotImplemented, + } + for _, code := range codes { + sb := newStatusBackend(t) + sb.reply(code, "nope") + p := newTestProber(t, healthConfig(sb.srv.URL, "http://b.invalid")) + + err := p.probe(t.Context(), Backend{Name: "a", URL: sb.srv.URL}) + if classifyProbe(err) != outcomeRejected { + t.Fatalf("HTTP %d classified as %v, want a rejected probe (err %v)", code, classifyProbe(err), err) + } + for i := 0; i < p.failures+3; i++ { + p.record("a", err) + if !p.healthy("a") { + t.Fatalf("HTTP %d excluded backend a after %d probes", code, i+1) + } + } + snap := p.snapshot()["a"] + if got := snap.stateName(); got != stateProbeUnsupported { + t.Errorf("HTTP %d state = %q, want %q", code, got, stateProbeUnsupported) + } + if !strings.Contains(snap.LastErr, defaultHealthProbePath) { + t.Errorf("HTTP %d last_error %q does not name the probe path", code, snap.LastErr) + } + } +} + +// 503 is trapperkeeper's answer when its services are not nominal, so it stays a +// real failure; so does anything that never got an HTTP reply at all. +func TestProber_ServerErrorsAndTransportFailuresStillMarkDown(t *testing.T) { + // 429 says the backend is out of capacity, which is evidence about the backend. + for _, code := range []int{ + http.StatusInternalServerError, http.StatusBadGateway, + http.StatusServiceUnavailable, http.StatusGatewayTimeout, + http.StatusTooManyRequests, + } { + sb := newStatusBackend(t) + sb.reply(code, degradedStatusBody("error")) + p := newTestProber(t, healthConfig(sb.srv.URL, "http://b.invalid")) + + err := p.probe(t.Context(), Backend{Name: "a", URL: sb.srv.URL}) + if classifyProbe(err) != outcomeFailure { + t.Fatalf("HTTP %d classified as %v, want a real failure", code, classifyProbe(err)) + } + for i := 0; i < p.failures; i++ { + p.record("a", err) + } + if p.healthy("a") { + t.Errorf("HTTP %d left backend a in service", code) + } + if got := p.snapshot()["a"].stateName(); got != stateUnhealthy { + t.Errorf("HTTP %d state = %q, want %q", code, got, stateUnhealthy) + } + } + + dead := httptest.NewServer(http.NotFoundHandler()) + deadURL := dead.URL + dead.Close() + p := newTestProber(t, healthConfig(deadURL, "http://b.invalid")) + err := p.probe(t.Context(), Backend{Name: "a", URL: deadURL}) + if classifyProbe(err) != outcomeFailure { + t.Fatalf("a refused connection classified as %v, want a real failure (err %v)", classifyProbe(err), err) + } + for i := 0; i < p.failures; i++ { + p.record("a", err) + } + if p.healthy("a") { + t.Error("a refused connection left the backend in service") + } +} + +// A real failure landing in a rejected run outranks the rejections: the run is +// one run of unsuccessful probes, and its worst evidence decides the state. +func TestProber_RealFailureInARejectedRunMarksDown(t *testing.T) { + p := newTestProber(t, healthConfig("http://a.invalid", "http://b.invalid")) + markUnsupported(t, p, "a") + for i := 0; i < 20; i++ { + p.record("a", rejectedProbe(p)) + } + + p.record("a", errors.New("boom")) + if p.healthy("a") { + t.Fatal("a real failure inside a rejected run left the backend in service") + } + if got := p.snapshot()["a"].stateName(); got != stateUnhealthy { + t.Errorf("state = %q, want %q", got, stateUnhealthy) + } +} + +// The regression this guards: a backend failing every probe but flipping the +// kind of failure — a fronting proxy alternating 404 and 503, say — never +// answers a probe successfully, so it must go down at the threshold like any +// other permanently failing backend. +func TestProber_AlternatingOutcomeKindsStillMarkDown(t *testing.T) { + rejected := func(p *prober) error { return rejectedProbe(p) } + failed := func(*prober) error { return errors.New("boom") } + ok := func(*prober) error { return nil } + + cycles := map[string][]func(*prober) error{ + "rejected/failure": {rejected, failed}, + "failure/rejected": {failed, rejected}, + "failure-heavy 3-cycle": {rejected, failed, failed}, + "reject-heavy 3-cycle": {rejected, rejected, failed}, + } + for name, cycle := range cycles { + t.Run(name, func(t *testing.T) { + p := newTestProber(t, healthConfig("http://a.invalid", "http://b.invalid")) + for i := 0; i < 200; i++ { + p.record("a", cycle[i%len(cycle)](p)) + } + if p.healthy("a") { + t.Fatal("a backend that failed 200 probes is still in service") + } + if got := p.snapshot()["a"].stateName(); got != stateUnhealthy { + t.Errorf("state = %q, want %q", got, stateUnhealthy) + } + }) + } + + // The same kinds never trip while a success keeps ending the run. + p := newTestProber(t, healthConfig("http://a.invalid", "http://b.invalid")) + interleaved := []func(*prober) error{rejected, ok, failed, ok} + for i := 0; i < 200; i++ { + p.record("a", interleaved[i%len(interleaved)](p)) + if !p.healthy("a") { + t.Fatal("a success between every unsuccessful probe still tripped the threshold") + } + } +} + +// Reaching the threshold on nothing but refusals is the one case that stays in +// service: it is evidence about the probe request, not about the backend. +func TestProber_PurelyRejectedRunStaysInService(t *testing.T) { + p := newTestProber(t, healthConfig("http://a.invalid", "http://b.invalid")) + for i := 0; i < p.failures*10; i++ { + p.record("a", rejectedProbe(p)) + if !p.healthy("a") { + t.Fatalf("a purely rejected run excluded the backend after %d probes", i+1) + } + } + if got := p.snapshot()["a"].stateName(); got != stateProbeUnsupported { + t.Fatalf("state = %q, want %q", got, stateProbeUnsupported) + } +} + +// A single success resets the run whatever it was made of. +func TestProber_SuccessResetsTheRunInEveryDirection(t *testing.T) { + for _, name := range []string{"rejected", "failure", "mixed"} { + t.Run(name, func(t *testing.T) { + p := newTestProber(t, healthConfig("http://a.invalid", "http://b.invalid")) + cfgFailures := p.failures + next := func(i int) error { + switch name { + case "rejected": + return rejectedProbe(p) + case "failure": + return errors.New("boom") + default: + if i%2 == 0 { + return rejectedProbe(p) + } + return errors.New("boom") + } + } + for i := 0; i < cfgFailures-1; i++ { + p.record("a", next(i)) + } + p.record("a", nil) + snap := p.snapshot()["a"] + if snap.Failures != 0 { + t.Fatalf("consecutive_failures = %d after a success, want 0", snap.Failures) + } + // One short of the threshold again, so the pre-success run is gone. + for i := 0; i < cfgFailures-1; i++ { + p.record("a", next(i)) + } + if !p.healthy("a") { + t.Fatal("the run survived a success") + } + if got := p.snapshot()["a"].stateName(); got == stateProbeUnsupported { + t.Fatal("a sub-threshold run reached probe_unsupported") + } + }) + } +} + +func TestProber_RecoversFromProbeUnsupported(t *testing.T) { + p := newTestProber(t, healthConfig("http://a.invalid", "http://b.invalid")) + markUnsupported(t, p, "a") + + for i := 0; i < p.successes; i++ { + p.record("a", nil) + } + if !p.healthy("a") { + t.Fatal("backend left the pool while recovering") + } + if got := p.snapshot()["a"].stateName(); got != stateHealthy { + t.Fatalf("state = %q, want %q", got, stateHealthy) + } +} + +// Rejections cannot talk a down backend back into the pool while its failure run +// is unbroken — the failures are still the run's best evidence. A successful +// probe ends the run, and rejections after that mean the probe stopped working +// rather than the backend. +func TestProber_RejectedProbeReadmitsADownBackendOnlyAfterASuccess(t *testing.T) { + p := newTestProber(t, healthConfig("http://a.invalid", "http://b.invalid")) + markDown(t, p, "a") + + for i := 0; i < p.failures*5; i++ { + p.record("a", rejectedProbe(p)) + if p.healthy("a") { + t.Fatal("rejections re-admitted a backend whose failure run never ended") + } + } + if got := p.snapshot()["a"].stateName(); got != stateUnhealthy { + t.Fatalf("state = %q, want %q", got, stateUnhealthy) + } + + p.record("a", nil) + markUnsupported(t, p, "a") +} + +func TestProber_LogsProbeRejectionOncePerTransition(t *testing.T) { + var buf strings.Builder + p := newProber(healthConfig("http://a.invalid", "http://b.invalid"), log.New(&buf, "", 0)) + if p == nil { + t.Fatal("newProber returned nil for an enabled config") + } + + rejected := &probeRejectedError{status: http.StatusNotFound, path: p.path} + for i := 0; i < 50; i++ { + p.record("a", rejected) + } + if n := strings.Count(buf.String(), "rejected the health probe"); n != 1 { + t.Fatalf("logged the misconfiguration %d times, want once:\n%s", n, buf.String()) + } + if !strings.Contains(buf.String(), `backend "a"`) || !strings.Contains(buf.String(), defaultHealthProbePath) { + t.Errorf("log line names neither the backend nor the probe path: %s", buf.String()) + } + if !strings.Contains(buf.String(), "404") { + t.Errorf("log line does not name the status: %s", buf.String()) + } + + // Recovery logs its own single line, and a second rejected run logs again. + for i := 0; i < p.successes+5; i++ { + p.record("a", nil) + } + if n := strings.Count(buf.String(), "usable again"); n != 1 { + t.Fatalf("logged recovery %d times, want once:\n%s", n, buf.String()) + } + for i := 0; i < 50; i++ { + p.record("a", rejected) + } + if n := strings.Count(buf.String(), "rejected the health probe"); n != 2 { + t.Fatalf("second rejected run logged %d times in total, want 2:\n%s", n, buf.String()) + } +} + +// The regression this guards: a backend whose probe path is wrong for it alone +// still answers queries perfectly, and must keep contributing records. +func TestFanOut_ProbeUnsupportedBackendStillContributes(t *testing.T) { + a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`) + b := newFakeBackend(t, `[`+node("h2", "2026-07-20T00:00:00Z")+`]`, `[]`) + srv := newTestServer(healthConfig(a.srv.URL, b.srv.URL)) + markUnsupported(t, srv.health, "b") + + rec := doGet(t, srv.Handler(), nodesPath, "") + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + if _, asked := b.params(nodesPath); !asked { + t.Fatal("a backend with an unusable probe path was not queried") + } + body := rec.Body.String() + if !strings.Contains(body, "h1") || !strings.Contains(body, "h2") { + t.Fatalf("both backends should have contributed: %s", body) + } + if got := rec.Header().Get(backendsHeader); got != "2/2" { + t.Errorf("%s = %q, want 2/2", backendsHeader, got) + } +} + +func TestHealthz_ReportsProbeUnsupported(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + b := newFakeBackend(t, `[]`, `[]`) + srv := newTestServer(healthConfig(a.srv.URL, b.srv.URL)) + srv.health.record("a", nil) + markUnsupported(t, srv.health, "b") + + rec := doGet(t, srv.Handler(), "/healthz", "") + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + var hr healthReport + if err := json.Unmarshal(rec.Body.Bytes(), &hr); err != nil { + t.Fatal(err) + } + if got := hr.Backends["a"].State; got != stateHealthy { + t.Errorf("a state = %q, want %q", got, stateHealthy) + } + // Distinct from both "verified healthy" and "excluded". + unsupported := hr.Backends["b"] + if unsupported.State != stateProbeUnsupported { + t.Errorf("b state = %q, want %q", unsupported.State, stateProbeUnsupported) + } + if unsupported.Reachable != "ok" { + t.Errorf("b reachable = %q, want ok", unsupported.Reachable) + } + if unsupported.LastError == "" || unsupported.LastProbe == "" { + t.Errorf("b is missing last_error/last_probe: %+v", unsupported) + } +} + +// probe_unsupported alone cannot say whether a backend is alive: a proxy that +// 404s everything because the backend behind it is dead looks exactly like a +// misconfigured probe path. reachable is the field that separates them. +func TestHealthz_ProbeUnsupportedSeparatesReachableFromDead(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + b := newFakeBackend(t, `[]`, `[]`) + srv := newTestServer(healthConfig(a.srv.URL, b.srv.URL)) + markUnsupported(t, srv.health, "a") + markUnsupported(t, srv.health, "b") + b.fail = true + + rec := doGet(t, srv.Handler(), "/healthz", "") + var hr healthReport + if err := json.Unmarshal(rec.Body.Bytes(), &hr); err != nil { + t.Fatal(err) + } + for name, got := range map[string]string{"a": hr.Backends["a"].State, "b": hr.Backends["b"].State} { + if got != stateProbeUnsupported { + t.Fatalf("%s state = %q, want %q", name, got, stateProbeUnsupported) + } + } + if hr.Backends["a"].Reachable != "ok" { + t.Errorf("a reachable = %q, want ok", hr.Backends["a"].Reachable) + } + if hr.Backends["b"].Reachable == "ok" { + t.Error("a dead backend behind an unusable probe path still reports reachable=ok") + } + if hr.Status != "degraded" { + t.Errorf("status = %q, want degraded", hr.Status) + } +} + +func TestStatusBodyHealthy_IgnoresBodiesWithoutStates(t *testing.T) { + for _, body := range []string{`{}`, `[]`, `null`, `{"svc":{}}`, `not json`} { + if err := statusBodyHealthy([]byte(body)); err != nil { + t.Errorf("body %q rejected: %v", body, err) + } + } +} diff --git a/main.go b/main.go index ebe0521..3ab60cb 100644 --- a/main.go +++ b/main.go @@ -20,10 +20,11 @@ var version = "dev" func main() { var ( - cfg Config - configPath string - listen string - merge string + cfg Config + configPath string + listen string + merge string + healthProbe bool ) // Loaded lazily: --config is only known once cobra has parsed flags. @@ -46,6 +47,9 @@ func main() { if cmd.Flags().Changed("merge") { cfg.Merge = merge } + if cmd.Flags().Changed("health-probe") { + cfg.HealthProbe = healthProbe + } if err := cfg.Validate(); err != nil { return err } @@ -67,6 +71,7 @@ func main() { pf.StringVar(&configPath, "config", "", "Config file path (overrides PDBMUX_CONFIG and the default search path)") pf.StringVar(&listen, "listen", defaultListen, "HTTP listen address (overrides config and PDBMUX_LISTEN)") pf.StringVar(&merge, "merge", mergeFreshness, "Facts merge strategy: freshness or static") + pf.BoolVar(&healthProbe, "health-probe", true, "Probe backend health and skip backends that are down") serveCmd := &cobra.Command{ Use: "serve", @@ -120,6 +125,8 @@ func main() { func runServer(cfg Config) error { logger := log.New(os.Stderr, "pdbmux: ", log.LstdFlags) srv := NewServer(cfg, logger) + srv.StartProbes(context.Background()) + defer srv.StopProbes() httpSrv := &http.Server{ Addr: cfg.Listen, @@ -163,6 +170,15 @@ func factsTTLString(cfg Config) string { return s } +func healthProbeString(cfg Config) string { + if !cfg.HealthProbe { + return "disabled" + } + return fmt.Sprintf("%s every %s (timeout %s, %d failures down / %d successes up)", + cfg.HealthProbePath, durationString(cfg.HealthProbeInterval), + durationString(cfg.HealthProbeTimeout), cfg.HealthProbeFailures, cfg.HealthProbeSuccesses) +} + func printConfig(cfg Config) { if p := cfg.SourcePath(); p != "" { fmt.Printf("config file : %s (loaded)\n", p) @@ -180,6 +196,7 @@ func printConfig(cfg Config) { } fmt.Printf("facts_ttl : %s\n", factsTTLString(cfg)) fmt.Printf("facts_cache : %d bytes\n", cfg.CacheBytes) + fmt.Printf("health_probe : %s\n", healthProbeString(cfg)) fmt.Println("backends:") for _, b := range cfg.Backends { fmt.Printf(" - %-8s %s\n", b.Name, b.URL) diff --git a/server.go b/server.go index 5e9fbb6..7bc4447 100644 --- a/server.go +++ b/server.go @@ -32,6 +32,9 @@ const ( // and how old the served copy is. cacheStatusHeader = "X-Cache" ageHeader = "Age" + + // Set by pdbmux: "/" backends behind a merged response. + backendsHeader = "X-Backends" ) type backendResult struct { @@ -53,6 +56,10 @@ type Server struct { flights flightGroup stale staleTracker + // health is nil when probing is disabled, which makes every backend healthy. + health *prober + partial partialTracker + // now is shared with the cache's clock so Age matches the stored timestamp. now func() time.Time @@ -73,9 +80,17 @@ func NewServer(cfg Config, logger *log.Logger) *Server { if cfg.cacheEnabled() { s.factsCache = newMemoryCache(cfg.FactsTTL, cfg.CacheBytes) } + s.health = newProber(cfg, logger) return s } +// StartProbes begins background health probing; it never blocks on a first +// round, so the listener serves straight away. +func (s *Server) StartProbes(ctx context.Context) { s.health.Start(ctx) } + +// StopProbes stops the probing goroutines and waits for them to exit. +func (s *Server) StopProbes() { s.health.Stop() } + // cacheFor picks the cache backing a request. Merged /facts and /nodes record // sets share the in-memory cache; every other path is uncached until the reports // cache lands, and a new backend is a case here rather than a change to any @@ -156,10 +171,18 @@ func (s *Server) serveMerged(w http.ResponseWriter, r *http.Request, path string if err != nil { return cachedResponse{}, err } - return cachedResponse{Body: encodeRecords(merge(alive)), Records: -1}, nil + resp := cachedResponse{Body: encodeRecords(merge(alive)), Records: -1} + s.countBackends(&resp, alive) + return resp, nil }) } +// countBackends stamps a response with how many backends it was built from, of +// how many configured. +func (s *Server) countBackends(resp *cachedResponse, alive []backendResult) { + resp.Backends, resp.Configured = len(alive), len(s.cfg.Backends) +} + // Reports and events are immutable history, so both backends' records belong in the merged view. func (s *Server) serveUnion(w http.ResponseWriter, r *http.Request, path string, key func(record) (string, bool)) { in := r.URL.Query() @@ -179,6 +202,7 @@ func (s *Server) serveUnion(w http.ResponseWriter, r *http.Request, path string, merged := mergeUnion(alive, key) sortRecords(merged, page.order) resp := cachedResponse{Body: encodeRecords(page.apply(merged)), Records: -1} + s.countBackends(&resp, alive) if page.wantTotal { if total := sumTotals(alive); total >= 0 { resp.Records = total @@ -232,6 +256,7 @@ func (s *Server) serveSummed(w http.ResponseWriter, r *http.Request, path string merged := sumRows(alive, columns) sortRecords(merged, page.order) resp := cachedResponse{Body: encodeRecords(page.apply(merged)), Records: -1} + s.countBackends(&resp, alive) if page.wantTotal { resp.Records = len(merged) } @@ -276,17 +301,21 @@ func (s *Server) aliveResults(ctx context.Context, path string, params url.Value } alive = append(alive, res) } + s.partial.record(len(alive), len(s.cfg.Backends), s.now()) if len(alive) == 0 { return nil, errAllBackendsFailed } return alive, nil } -// cachedResponse is the stored form of a merged response: the JSON body plus the -// X-Records value it carried, so a cache hit reproduces both. +// cachedResponse is the stored form of a merged response: the JSON body, the +// X-Records value it carried and how many backends it was built from, so a cache +// hit reproduces all three. type cachedResponse struct { - Body json.RawMessage `json:"body"` - Records int `json:"records"` // -1 when the response sets no X-Records + Body json.RawMessage `json:"body"` + Records int `json:"records"` // -1 when the response sets no X-Records + Backends int `json:"backends"` // backends that contributed records + Configured int `json:"configured"` // backends configured at build time } // serveCached answers from the cache when the entry is fresh, otherwise runs @@ -413,6 +442,9 @@ func writeCached(w http.ResponseWriter, resp cachedResponse) { if resp.Records >= 0 { w.Header().Set(recordsHeader, strconv.Itoa(resp.Records)) } + if resp.Configured > 0 { + w.Header().Set(backendsHeader, strconv.Itoa(resp.Backends)+"/"+strconv.Itoa(resp.Configured)) + } // resp.Body is shared with the cache and with every caller of a single // flight, so it is written, never appended to. body := []byte(resp.Body) @@ -499,11 +531,40 @@ func (s *Server) freshnessMap(ctx context.Context, _ []backendResult) freshness return f } -// Returns one result per backend, in config order. +// Returns one result per queried backend, in config order. Backends the prober +// currently has down are skipped so a known-dead backend costs no timeout. func (s *Server) fanOut(ctx context.Context, path string, params url.Values) []backendResult { - results := make([]backendResult, len(s.cfg.Backends)) + return s.fanOutTo(ctx, s.liveBackends(), path, params) +} + +// fanOutAll ignores health state and asks every configured backend. +func (s *Server) fanOutAll(ctx context.Context, path string, params url.Values) []backendResult { + return s.fanOutTo(ctx, s.cfg.Backends, path, params) +} + +// liveBackends drops the backends currently marked unhealthy, but falls open to +// the full list when that would leave none: a broken prober, a wrong health +// path or a partition seen only by the prober must never black-hole traffic. +func (s *Server) liveBackends() []Backend { + if s.health == nil { + return s.cfg.Backends + } + live := make([]Backend, 0, len(s.cfg.Backends)) + for _, b := range s.cfg.Backends { + if s.health.healthy(b.Name) { + live = append(live, b) + } + } + if len(live) == 0 { + return s.cfg.Backends + } + return live +} + +func (s *Server) fanOutTo(ctx context.Context, backends []Backend, path string, params url.Values) []backendResult { + results := make([]backendResult, len(backends)) var wg sync.WaitGroup - for i, b := range s.cfg.Backends { + for i, b := range backends { wg.Add(1) go func(i int, b Backend) { defer wg.Done() @@ -605,9 +666,51 @@ func setContentType(w http.ResponseWriter, contentType string) { } type healthReport struct { - Status string `json:"status"` - Backends map[string]string `json:"backends"` // name -> "ok" | error text - Cache cacheHealth `json:"cache"` + Status string `json:"status"` + Backends map[string]backendReport `json:"backends"` + Query queryReport `json:"query"` + Cache cacheHealth `json:"cache"` +} + +// backendReport pairs this request's own reachability check with the background +// prober's running state for the same backend. The two answer different +// questions and must be read together: probe_unsupported means "not being +// verified", not "well", so reachable is the field that says whether the backend +// is actually answering queries right now. +type backendReport struct { + // Reachable is this request's own live query to the backend, run against + // every configured backend regardless of probe state: "ok" or the error text. + Reachable string `json:"reachable"` + // State is the background prober's verdict. probe_unsupported means the probe + // path is unusable on this backend, so its health is unknown — read Reachable + // to find out whether it is up. + State string `json:"state"` // healthy | unhealthy | probe_unsupported | unprobed | unmonitored + Failures int `json:"consecutive_failures"` + Successes int `json:"consecutive_successes"` + LastProbe string `json:"last_probe,omitempty"` + LastError string `json:"last_error,omitempty"` +} + +// queryReport describes the most recent merged fan-out. +type queryReport struct { + Partial bool `json:"partial"` + Contributed int `json:"contributed"` + Configured int `json:"configured"` + PartialRounds uint64 `json:"partial_rounds"` + LastPartial string `json:"last_partial,omitempty"` +} + +func (s *Server) queryHealth() queryReport { + seen, contributed, configured, rounds, last := s.partial.snapshot() + q := queryReport{Configured: len(s.cfg.Backends), PartialRounds: rounds} + if seen { + q.Contributed, q.Configured = contributed, configured + q.Partial = contributed < configured + } + if !last.IsZero() { + q.LastPartial = last.UTC().Format(time.RFC3339) + } + return q } type cacheHealth struct { @@ -646,17 +749,33 @@ func (s *Server) cacheHealth() cacheHealth { func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { probe := `["=","certname","pdbmux-healthz-probe"]` - results := s.fanOut(r.Context(), nodesPath, queryParams(probe)) + // Every backend is checked, including ones the prober has down, so the + // report never hides a backend queries are currently skipping. + results := s.fanOutAll(r.Context(), nodesPath, queryParams(probe)) + states := s.health.snapshot() - report := healthReport{Backends: map[string]string{}, Cache: s.cacheHealth()} + report := healthReport{ + Backends: map[string]backendReport{}, + Query: s.queryHealth(), + Cache: s.cacheHealth(), + } healthy := 0 for _, res := range results { + b := backendReport{Reachable: "ok", State: stateUnmonitored} if res.err != nil { - report.Backends[res.name] = res.err.Error() - continue + b.Reachable = res.err.Error() + } else { + healthy++ } - report.Backends[res.name] = "ok" - healthy++ + if st, ok := states[res.name]; ok { + b.State = st.stateName() + b.Failures, b.Successes = st.Failures, st.Successes + b.LastError = st.LastErr + if !st.LastProbe.IsZero() { + b.LastProbe = st.LastProbe.UTC().Format(time.RFC3339) + } + } + report.Backends[res.name] = b } switch { case healthy == len(results): diff --git a/server_test.go b/server_test.go index a426837..73b038f 100644 --- a/server_test.go +++ b/server_test.go @@ -354,7 +354,7 @@ func TestHandler_Health(t *testing.T) { if err := json.Unmarshal(rec.Body.Bytes(), &hr); err != nil { t.Fatal(err) } - if hr.Status != "ok" || hr.Backends["a"] != "ok" || hr.Backends["b"] != "ok" { + if hr.Status != "ok" || hr.Backends["a"].Reachable != "ok" || hr.Backends["b"].Reachable != "ok" { t.Fatalf("unexpected health: %+v", hr) } }