unkin/pdbmux
4
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Health-check backends and skip the ones that are down #14

Merged
benvin merged 3 commits from benvin/backend-health into main 2026-09-06 00:18:36 +10:00
Conversation 0 Commits 3 Files Changed 8
+1620 -32
8 changed files with 1620 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+94 -7
View File
@@ -38,15 +38,17 @@ not PQL) is forwarded verbatim.
| `GET /metrics/v2/read/<mbean>` | 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[/<mbean>]` | 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:
<contributed>/<configured>` 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`.
config.go
+76 -2
View File
@@ -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)
}
config_test.go
+53 -1
View File
@@ -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, "")
}
}
health.go
+373
View File
@@ -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
}
health_test.go
+866
View File
@@ -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)
}
}
}
main.go
+21 -4
View File
@@ -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)
server.go
+136 -17
View File
@@ -32,6 +32,9 @@ const (
// and how old the served copy is.
cacheStatusHeader = "X-Cache"
ageHeader = "Age"
// Set by pdbmux: "<contributed>/<configured>" 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):
server_test.go
+1 -1
View File
@@ -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)
}
}