diff --git a/README.md b/README.md index d60e7b0..4538f24 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,50 @@ 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. +- **Thresholds** — a healthy backend goes down after `health_probe_failures` + (default 3) **consecutive** failures; a down one comes back after + `health_probe_successes` (default 2) consecutive successes. One blip cannot + flap a backend out, and one lucky reply cannot flap it back in. 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) 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`, + `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. 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 +344,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 +367,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..c37c1d0 --- /dev/null +++ b/health.go @@ -0,0 +1,289 @@ +package main + +import ( + "context" + "encoding/json" + "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" +) + +type probeState struct { + healthy bool + probed 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 + Probed bool + Failures int + Successes int + LastProbe time.Time + LastErr string +} + +func (h backendHealth) stateName() string { + switch { + case !h.Probed: + return stateUnprobed + case h.Healthy: + return stateHealthy + default: + return stateUnhealthy + } +} + +// 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, + 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. +func (p *prober) record(name string, err error) { + p.mu.Lock() + st, ok := p.states[name] + if !ok { + p.mu.Unlock() + return + } + st.probed = true + st.lastProbe = p.now() + var transition string + if err != nil { + st.lastErr = err.Error() + st.successes = 0 + st.failures++ + if st.healthy && st.failures >= p.failures { + st.healthy = false + transition = fmt.Sprintf("warning: backend %q is down after %d consecutive probe failures: %v", name, st.failures, err) + } + } else { + st.lastErr = "" + st.failures = 0 + st.successes++ + if !st.healthy && st.successes >= p.successes { + st.healthy = true + transition = fmt.Sprintf("info: backend %q is up after %d consecutive probe successes", name, st.successes) + } + } + 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 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..f1c820e --- /dev/null +++ b/health_test.go @@ -0,0 +1,497 @@ +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 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) + } +} + +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..7b32a2f 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,43 @@ 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. +type backendReport struct { + Reachable string `json:"reachable"` // "ok" | error text + State string `json:"state"` // healthy | unhealthy | 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 +741,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) } }