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

config: drop primary/prefer and treat all backends equally

Browse Source 
- unmerged /pdb/query/v4/* paths now go to the first backend that answers, not a designated primary
This commit is contained in:
Unkin Agent
2026-09-05 11:53:45 +10:00
parent d724cf0a5e
commit ddfa47e438
12 changed files with 424 additions and 432 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+22 -28
View File
@@ -9,15 +9,13 @@ PuppetDB and it sees one consistent view spanning all of them.
Running more than one PuppetDB — during a migration between two of them, or Running more than one PuppetDB — during a migration between two of them, or
across regions — means a given node's current data lives in exactly one at any across regions — means a given node's current data lives in exactly one at any
moment, and consumers have to know which, or query each in turn. Consider two moment, and consumers have to know which, or query each in turn. `pdbmux`
backends being merged during a migration: merges them all so consumers don't have to know (or query twice) which PuppetDB
a node currently lives in.
- **old** — the PuppetDB nodes are moving off, e.g. `http://puppetdb1.example.com:8080` All backends are equal — `pdbmux` is never told which one to favour. Backend
- **new** — the PuppetDB nodes are moving on to, e.g. `http://puppetdb2.example.com:8080` names are arbitrary labels and there is no fixed number of them. The configured
order is used only as a tie-break, so output is reproducible.
Nodes move from old to new as they migrate. `pdbmux` merges both so consumers
don't have to know (or query twice) which PuppetDB a node currently lives in.
The backend names are arbitrary labels; there is no fixed number of backends.
## Endpoints ## Endpoints
@@ -32,8 +30,8 @@ not PQL) is forwarded verbatim.
| `GET /pdb/query/v4/events` | Fan out to all and serve the **union**, deduped by record identity, re-ordered and re-paged. | | `GET /pdb/query/v4/events` | Fan out to all and serve the **union**, deduped by record identity, re-ordered and re-paged. |
| `GET /pdb/query/v4/event-counts` | Fan out to all and **sum** each subject's counts into one row per subject. | | `GET /pdb/query/v4/event-counts` | Fan out to all and **sum** each subject's counts into one row per subject. |
| `GET /pdb/query/v4/aggregate-event-counts` | Fan out to all and **sum** the summary object's counts. | | `GET /pdb/query/v4/aggregate-event-counts` | Fan out to all and **sum** the summary object's counts. |
| `GET /pdb/query/v4/reports/<hash>/{events,logs,metrics}` | Ask every backend; serve the answer from whichever backend actually holds that report. `404` when neither does. | | `GET /pdb/query/v4/reports/<hash>/{events,logs,metrics}` | Ask every backend; serve the answer from whichever backend actually holds that report. `404` when none does. |
| `GET /pdb/query/v4/*` (any other) | Transparently proxied to the **primary** backend, unmerged, streamed verbatim. | | `GET /pdb/query/v4/*` (any other) | No merge rule, so backends are tried in configured order and the first success is streamed back verbatim; if all reject it, the first upstream error response is replayed. |
| `GET /healthz` | Per-backend reachability. `200 {"status":"ok"}` if all reachable, `200 degraded` if some fail, `503 down` if all fail. | | `GET /healthz` | Per-backend reachability. `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 Fan-out is concurrent. If one backend errors or times out, `pdbmux` serves the
@@ -44,22 +42,22 @@ unknown fields survive untouched.
## Merge semantics ## Merge semantics
- **`/nodes`** — dedupe by `certname`; the record with the strictly-newer - **`/nodes`** — dedupe by `certname`; the record with the strictly-newer
`report_timestamp` wins. On a tie (or when a node exists in only one backend), `report_timestamp` wins. On a tie, the backend listed first in `backends`
the **preferred** backend's record is kept. supplies the record — a tie-break only, so the merged output is deterministic.
- **`/facts`** — node-level granularity. For a `certname` present in more than - **`/facts`** — node-level granularity. For a `certname` present in more than
one backend, `pdbmux` keeps **all** of that node's facts from **one** backend and one backend, `pdbmux` keeps **all** of that node's facts from **one** backend and
drops the other's, chosen by the merge strategy: drops the others', chosen by the merge strategy:
- **`freshness`** (default) — attribute each `certname` to whichever backend - **`freshness`** (default) — attribute each `certname` to whichever backend
holds its newer `report_timestamp`. `pdbmux` derives this from a per-certname holds its newer `report_timestamp`. `pdbmux` derives this from a per-certname
freshness map built by querying `/nodes` from every backend, cached for freshness map built by querying `/nodes` from every backend, cached for
`freshness_ttl` (default 30s). Ties/fallbacks use `prefer`. `freshness_ttl` (default 30s).
- **`static`** — always keep the `prefer` backend's facts for shared nodes. - **`static`** — skip the extra `/nodes` query and take each shared node's
No extra `/nodes` query. facts from the first backend in configured order that holds it.
- A node present in only one backend always appears (falls back to whichever - A node present in only one backend always appears (falls back to whichever
backend actually returned facts for it). backend actually returned facts for it).
- **`/reports`, `/events`** — **union**, not a per-node winner. Reports are - **`/reports`, `/events`** — **union**, not a per-node winner. Reports are
immutable history, so a node that migrated legitimately has reports in the old immutable history, so a node that moved between PuppetDBs legitimately has
PuppetDB *and* the new one and both belong in the merged view. Reports dedupe reports in both and both belong in the merged view. Reports dedupe
on `hash`; events, which carry no id of their own, dedupe on the verbatim on `hash`; events, which carry no id of their own, dedupe on the verbatim
record (a node briefly reporting to both PuppetDBs stores identical records in record (a node briefly reporting to both PuppetDBs stores identical records in
each). each).
@@ -86,7 +84,7 @@ Each backend applies `order_by`/`limit`/`offset` to its own slice only, so
`pdbmux` re-does all three over the union: `pdbmux` re-does all three over the union:
- `order_by` is parsed and the merged set re-sorted by those fields (ties keep - `order_by` is parsed and the merged set re-sorted by those fields (ties keep
backend precedence). A record missing an ordered field sorts first. the merged set's existing order). A record missing an ordered field sorts first.
- Backends are asked for the first `offset + limit` records — never an `offset` - Backends are asked for the first `offset + limit` records — never an `offset`
— and the requested window is then cut from the merged, re-sorted set. — and the requested window is then cut from the merged, re-sorted set.
- `include_total=true` on a union endpoint makes `pdbmux` sum each backend's - `include_total=true` on a union endpoint makes `pdbmux` sum each backend's
@@ -105,14 +103,12 @@ config file — everything comes from `PDBMUX_*` env vars.
```yaml ```yaml
listen: ":8080" listen: ":8080"
backends: backends: # order is a tie-break only, not a ranking
- name: old - name: pdb-a
url: http://puppetdb1.example.com:8080 url: http://puppetdb1.example.com:8080
- name: new - name: pdb-b
url: https://puppetdb2.example.com url: https://puppetdb2.example.com
primary: new # backend used for non-merged /pdb/query/v4/* pass-through
merge: freshness # freshness | static merge: freshness # freshness | static
prefer: new # winner on ties / static merge / fallback
timeout: 10s # per-upstream request timeout timeout: 10s # per-upstream request timeout
freshness_ttl: 30s # freshness-map cache TTL (freshness merge only) freshness_ttl: 30s # freshness-map cache TTL (freshness merge only)
``` ```
@@ -123,14 +119,12 @@ the `/pdb/query/v4/...` path per request.
| Env var | Overrides | | Env var | Overrides |
|---|---| |---|---|
| `PDBMUX_LISTEN` | `listen` | | `PDBMUX_LISTEN` | `listen` |
| `PDBMUX_PRIMARY` | `primary` |
| `PDBMUX_MERGE` | `merge` | | `PDBMUX_MERGE` | `merge` |
| `PDBMUX_PREFER` | `prefer` |
| `PDBMUX_TIMEOUT` | `timeout` (Go duration, e.g. `10s`) | | `PDBMUX_TIMEOUT` | `timeout` (Go duration, e.g. `10s`) |
| `PDBMUX_FRESHNESS_TTL` | `freshness_ttl` | | `PDBMUX_FRESHNESS_TTL` | `freshness_ttl` |
| `PDBMUX_BACKENDS` | whole backend list, as `name=url,name=url` | | `PDBMUX_BACKENDS` | whole backend list, as `name=url,name=url` |
Flags: `--listen`, `--primary`, `--merge`. Flags: `--listen`, `--merge`.
## Running ## Running
@@ -139,7 +133,7 @@ Subcommands: `serve` (default), `config init`, `config show`, `version`. Run
base URL in place of a PuppetDB one. base URL in place of a PuppetDB one.
```bash ```bash
PDBMUX_BACKENDS='old=http://puppetdb1.example.com:8080,new=http://puppetdb2.example.com:8080' pdbmux PDBMUX_BACKENDS='pdb-a=http://puppetdb1.example.com:8080,pdb-b=http://puppetdb2.example.com:8080' pdbmux
curl -s --get http://localhost:8080/pdb/query/v4/nodes \ curl -s --get http://localhost:8080/pdb/query/v4/nodes \
--data-urlencode 'query=["=","certname","host1.example.com"]' --data-urlencode 'query=["=","certname","host1.example.com"]'
``` ```
aggregate.go
+4 -3
View File
@@ -159,8 +159,9 @@ type sumGroup struct {
// A row that is not a JSON object passes through untouched, as does the sole row // A row that is not a JSON object passes through untouched, as does the sole row
// of a key only one backend reported — those keep their upstream bytes. An // of a key only one backend reported — those keep their upstream bytes. An
// aggregate column that is absent or non-numeric in a later row is left at the // aggregate column that is absent or non-numeric in a later row is left at the
// first backend's value rather than being coerced to zero. results must be // earlier backend's value rather than being coerced to zero. results come in
// ordered by precedence; output keeps first-seen order. // configured backend order and the output keeps first-seen order, a tie-break
// only.
func sumRows(results []backendResult, columns func(map[string]json.RawMessage) ([]string, []string)) []json.RawMessage { func sumRows(results []backendResult, columns func(map[string]json.RawMessage) ([]string, []string)) []json.RawMessage {
type slot struct { type slot struct {
raw json.RawMessage // passthrough row, when group is nil raw json.RawMessage // passthrough row, when group is nil
@@ -237,7 +238,7 @@ func (g *sumGroup) encode() json.RawMessage {
} }
// groupKey builds a row's identity from the named fields' verbatim JSON values. // groupKey builds a row's identity from the named fields' verbatim JSON values.
// Both backends run the same PuppetDB serialiser, so byte equality is a sound // Every backend runs the same PuppetDB serialiser, so byte equality is a sound
// comparison for object-valued keys such as event-counts' subject. An absent // comparison for object-valued keys such as event-counts' subject. An absent
// field is distinct from any present value. // field is distinct from any present value.
func groupKey(row map[string]json.RawMessage, keys []string) string { func groupKey(row map[string]json.RawMessage, keys []string) string {
aggregate_test.go
+16 -16
View File
@@ -73,8 +73,8 @@ func TestParseAggregate_NoFunctionIsNotAggregate(t *testing.T) {
func TestSumRows_SharedKeysAreAdded(t *testing.T) { func TestSumRows_SharedKeysAreAdded(t *testing.T) {
spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`) spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
merged := sumRows([]backendResult{ merged := sumRows([]backendResult{
{name: "new", records: rows(`{"count":3,"status":"changed"}`, `{"count":1,"status":"failed"}`)}, {name: "a", records: rows(`{"count":3,"status":"changed"}`, `{"count":1,"status":"failed"}`)},
{name: "old", records: rows(`{"count":4,"status":"changed"}`, `{"count":2,"status":"failed"}`)}, {name: "b", records: rows(`{"count":4,"status":"changed"}`, `{"count":2,"status":"failed"}`)},
}, spec.columns) }, spec.columns)
want := []map[string]any{ want := []map[string]any{
@@ -89,8 +89,8 @@ func TestSumRows_SharedKeysAreAdded(t *testing.T) {
func TestSumRows_DisjointKeysAreKept(t *testing.T) { func TestSumRows_DisjointKeysAreKept(t *testing.T) {
spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`) spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
merged := sumRows([]backendResult{ merged := sumRows([]backendResult{
{name: "new", records: rows(`{"count":3,"status":"changed"}`)}, {name: "a", records: rows(`{"count":3,"status":"changed"}`)},
{name: "old", records: rows(`{"count":2,"status":"skipped"}`)}, {name: "b", records: rows(`{"count":2,"status":"skipped"}`)},
}, spec.columns) }, spec.columns)
want := []map[string]any{ want := []map[string]any{
@@ -106,8 +106,8 @@ func TestSumRows_SingleBackendRowKeepsUpstreamBytes(t *testing.T) {
spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`) spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
const raw = `{"count":3,"status":"changed","extra":{"kept":true}}` const raw = `{"count":3,"status":"changed","extra":{"kept":true}}`
merged := sumRows([]backendResult{ merged := sumRows([]backendResult{
{name: "new", records: rows(raw)}, {name: "a", records: rows(raw)},
{name: "old", records: nil}, {name: "b", records: nil},
}, spec.columns) }, spec.columns)
if len(merged) != 1 || string(merged[0]) != raw { if len(merged) != 1 || string(merged[0]) != raw {
@@ -118,8 +118,8 @@ func TestSumRows_SingleBackendRowKeepsUpstreamBytes(t *testing.T) {
func TestSumRows_NonNumericAggregateColumnIsNotZeroed(t *testing.T) { func TestSumRows_NonNumericAggregateColumnIsNotZeroed(t *testing.T) {
spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`) spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
merged := sumRows([]backendResult{ merged := sumRows([]backendResult{
{name: "new", records: rows(`{"count":5,"status":"changed"}`)}, {name: "a", records: rows(`{"count":5,"status":"changed"}`)},
{name: "old", records: rows(`{"count":null,"status":"changed"}`)}, {name: "b", records: rows(`{"count":null,"status":"changed"}`)},
}, spec.columns) }, spec.columns)
want := []map[string]any{{"count": float64(5), "status": "changed"}} want := []map[string]any{{"count": float64(5), "status": "changed"}}
@@ -131,8 +131,8 @@ func TestSumRows_NonNumericAggregateColumnIsNotZeroed(t *testing.T) {
func TestSumRows_MissingAggregateColumnStartsFromTheNumericRow(t *testing.T) { func TestSumRows_MissingAggregateColumnStartsFromTheNumericRow(t *testing.T) {
spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`) spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
merged := sumRows([]backendResult{ merged := sumRows([]backendResult{
{name: "new", records: rows(`{"status":"changed"}`)}, {name: "a", records: rows(`{"status":"changed"}`)},
{name: "old", records: rows(`{"count":6,"status":"changed"}`)}, {name: "b", records: rows(`{"count":6,"status":"changed"}`)},
}, spec.columns) }, spec.columns)
want := []map[string]any{{"count": float64(6), "status": "changed"}} want := []map[string]any{{"count": float64(6), "status": "changed"}}
@@ -144,8 +144,8 @@ func TestSumRows_MissingAggregateColumnStartsFromTheNumericRow(t *testing.T) {
func TestSumRows_NonObjectRowsPassThrough(t *testing.T) { func TestSumRows_NonObjectRowsPassThrough(t *testing.T) {
spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`) spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
merged := sumRows([]backendResult{ merged := sumRows([]backendResult{
{name: "new", records: rows(`"surprise"`)}, {name: "a", records: rows(`"surprise"`)},
{name: "old", records: rows(`{"count":1,"status":"changed"}`)}, {name: "b", records: rows(`{"count":1,"status":"changed"}`)},
}, spec.columns) }, spec.columns)
if len(merged) != 2 || string(merged[0]) != `"surprise"` { if len(merged) != 2 || string(merged[0]) != `"surprise"` {
@@ -161,8 +161,8 @@ func TestSumRows_NoKeyColumnsCollapseToOneRow(t *testing.T) {
t.Fatal("expected an aggregate spec") t.Fatal("expected an aggregate spec")
} }
merged := sumRows([]backendResult{ merged := sumRows([]backendResult{
{name: "new", records: rows(`{"count":10}`)}, {name: "a", records: rows(`{"count":10}`)},
{name: "old", records: rows(`{"count":32}`)}, {name: "b", records: rows(`{"count":32}`)},
}, spec.columns) }, spec.columns)
want := []map[string]any{{"count": float64(42)}} want := []map[string]any{{"count": float64(42)}}
@@ -188,11 +188,11 @@ func TestInferredColumns_SplitsCountsFromIdentity(t *testing.T) {
func TestSumRows_EventCountsPerSubject(t *testing.T) { func TestSumRows_EventCountsPerSubject(t *testing.T) {
merged := sumRows([]backendResult{ merged := sumRows([]backendResult{
{name: "new", records: rows( {name: "a", records: rows(
`{"subject_type":"certname","subject":{"title":"h1"},"failures":1,"successes":2,"noops":0,"skips":0}`, `{"subject_type":"certname","subject":{"title":"h1"},"failures":1,"successes":2,"noops":0,"skips":0}`,
`{"subject_type":"certname","subject":{"title":"h2"},"failures":0,"successes":5,"noops":0,"skips":0}`, `{"subject_type":"certname","subject":{"title":"h2"},"failures":0,"successes":5,"noops":0,"skips":0}`,
)}, )},
{name: "old", records: rows( {name: "b", records: rows(
`{"subject_type":"certname","subject":{"title":"h1"},"failures":3,"successes":4,"noops":1,"skips":0}`, `{"subject_type":"certname","subject":{"title":"h1"},"failures":3,"successes":4,"noops":1,"skips":0}`,
)}, )},
}, inferredColumns) }, inferredColumns)
config.go
+5 -43
View File
@@ -23,8 +23,8 @@ const (
) )
var exampleBackends = []Backend{ var exampleBackends = []Backend{
{Name: "primary", URL: "http://puppetdb1.example.com:8080"}, {Name: "pdb-a", URL: "http://puppetdb1.example.com:8080"},
{Name: "secondary", URL: "http://puppetdb2.example.com:8080"}, {Name: "pdb-b", URL: "http://puppetdb2.example.com:8080"},
} }
type Backend struct { type Backend struct {
@@ -34,10 +34,8 @@ type Backend struct {
type Config struct { type Config struct {
Listen string `yaml:"listen"` Listen string `yaml:"listen"`
Backends []Backend `yaml:"backends"` Backends []Backend `yaml:"backends"` // all equal; order is only a deterministic tie-break
Primary string `yaml:"primary"`
Merge string `yaml:"merge"` Merge string `yaml:"merge"`
Prefer string `yaml:"prefer"` // wins under static merge, and breaks ties under freshness merge
Timeout time.Duration `yaml:"timeout"` Timeout time.Duration `yaml:"timeout"`
FreshnessTTL time.Duration `yaml:"freshness_ttl"` FreshnessTTL time.Duration `yaml:"freshness_ttl"`
} }
@@ -59,8 +57,6 @@ func DefaultConfig() Config {
func ExampleConfig() Config { func ExampleConfig() Config {
cfg := DefaultConfig() cfg := DefaultConfig()
cfg.Backends = append([]Backend(nil), exampleBackends...) cfg.Backends = append([]Backend(nil), exampleBackends...)
cfg.Primary = exampleBackends[0].Name
cfg.Prefer = exampleBackends[0].Name
return cfg return cfg
} }
@@ -93,7 +89,6 @@ func Load() (Config, error) {
} }
applyEnv(&cfg, os.Getenv) applyEnv(&cfg, os.Getenv)
cfg.normalize()
return cfg, nil return cfg, nil
} }
@@ -101,15 +96,9 @@ func applyEnv(cfg *Config, getenv func(string) string) {
if v := getenv(envPrefix + "LISTEN"); v != "" { if v := getenv(envPrefix + "LISTEN"); v != "" {
cfg.Listen = v cfg.Listen = v
} }
if v := getenv(envPrefix + "PRIMARY"); v != "" {
cfg.Primary = v
}
if v := getenv(envPrefix + "MERGE"); v != "" { if v := getenv(envPrefix + "MERGE"); v != "" {
cfg.Merge = v cfg.Merge = v
} }
if v := getenv(envPrefix + "PREFER"); v != "" {
cfg.Prefer = v
}
if v := getenv(envPrefix + "TIMEOUT"); v != "" { if v := getenv(envPrefix + "TIMEOUT"); v != "" {
if d, err := time.ParseDuration(v); err == nil { if d, err := time.ParseDuration(v); err == nil {
cfg.Timeout = d cfg.Timeout = d
@@ -145,18 +134,6 @@ func parseBackends(s string) []Backend {
return out return out
} }
func (c *Config) normalize() {
if len(c.Backends) == 0 {
return
}
if c.Primary == "" {
c.Primary = c.Backends[0].Name
}
if c.Prefer == "" {
c.Prefer = c.Backends[0].Name
}
}
func (c Config) Validate() error { func (c Config) Validate() error {
if len(c.Backends) == 0 { if len(c.Backends) == 0 {
return fmt.Errorf("no backends configured: set %sBACKENDS to \"name=url,name=url\" or add a backends list to %s", return fmt.Errorf("no backends configured: set %sBACKENDS to \"name=url,name=url\" or add a backends list to %s",
@@ -172,32 +149,17 @@ func (c Config) Validate() error {
} }
seen[b.Name] = true seen[b.Name] = true
} }
if !seen[c.Primary] {
return fmt.Errorf("primary %q is not a configured backend", c.Primary)
}
switch c.Merge { switch c.Merge {
case mergeFreshness, mergeStatic: case mergeFreshness, mergeStatic:
default: default:
return fmt.Errorf("merge must be %q or %q, got %q", mergeFreshness, mergeStatic, c.Merge) return fmt.Errorf("merge must be %q or %q, got %q", mergeFreshness, mergeStatic, c.Merge)
} }
if !seen[c.Prefer] {
return fmt.Errorf("prefer %q is not a configured backend", c.Prefer)
}
if c.Timeout <= 0 { if c.Timeout <= 0 {
return fmt.Errorf("timeout must be positive") return fmt.Errorf("timeout must be positive")
} }
return nil return nil
} }
func (c Config) PrimaryBackend() Backend {
for _, b := range c.Backends {
if b.Name == c.Primary {
return b
}
}
return c.Backends[0]
}
func writeDefaultConfig() error { func writeDefaultConfig() error {
dir := ConfigDir() dir := ConfigDir()
if err := os.MkdirAll(dir, 0o755); err != nil { if err := os.MkdirAll(dir, 0o755); err != nil {
@@ -211,8 +173,8 @@ func writeDefaultConfig() error {
header := []byte("# pdbmux configuration\n" + header := []byte("# pdbmux configuration\n" +
"# A merging proxy presenting one PuppetDB v4 query surface over several\n" + "# A merging proxy presenting one PuppetDB v4 query surface over several\n" +
"# PuppetDB backends. The backend URLs below are placeholders — edit them.\n" + "# PuppetDB backends. The backend URLs below are placeholders — edit them.\n" +
"# Env overrides: PDBMUX_LISTEN, PDBMUX_PRIMARY, PDBMUX_MERGE, PDBMUX_PREFER,\n" + "# Env overrides: PDBMUX_LISTEN, PDBMUX_MERGE, PDBMUX_TIMEOUT,\n" +
"# PDBMUX_TIMEOUT, PDBMUX_FRESHNESS_TTL, PDBMUX_BACKENDS (name=url,name=url).\n\n") "# PDBMUX_FRESHNESS_TTL, PDBMUX_BACKENDS (name=url,name=url).\n\n")
if err := os.WriteFile(path, append(header, data...), 0o644); err != nil { if err := os.WriteFile(path, append(header, data...), 0o644); err != nil {
return fmt.Errorf("writing config: %w", err) return fmt.Errorf("writing config: %w", err)
} }
config_test.go
+14 -24
View File
@@ -13,11 +13,9 @@ import (
func testConfigValid() Config { func testConfigValid() Config {
cfg := DefaultConfig() cfg := DefaultConfig()
cfg.Backends = []Backend{ cfg.Backends = []Backend{
{Name: "old", URL: "http://localhost:18080"}, {Name: "a", URL: "http://localhost:18080"},
{Name: "new", URL: "http://localhost:18081"}, {Name: "b", URL: "http://localhost:18081"},
} }
cfg.Primary = "new"
cfg.Prefer = "new"
return cfg return cfg
} }
@@ -55,7 +53,7 @@ func TestLoad_NoBackendsLoadsButFailsValidation(t *testing.T) {
} }
} }
func TestLoad_PrimaryDefaultsToFirstBackend(t *testing.T) { func TestLoad_BackendsKeepConfiguredOrder(t *testing.T) {
t.Setenv("XDG_CONFIG_HOME", t.TempDir()) t.Setenv("XDG_CONFIG_HOME", t.TempDir())
clearEnv(t) clearEnv(t)
t.Setenv(envPrefix+"BACKENDS", "a=http://localhost:18080,b=http://localhost:18081") t.Setenv(envPrefix+"BACKENDS", "a=http://localhost:18080,b=http://localhost:18081")
@@ -64,8 +62,11 @@ func TestLoad_PrimaryDefaultsToFirstBackend(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
if cfg.Primary != "a" || cfg.Prefer != "a" { if len(cfg.Backends) != 2 || cfg.Backends[0].Name != "a" || cfg.Backends[1].Name != "b" {
t.Errorf("primary/prefer should default to first backend, got %q/%q", cfg.Primary, cfg.Prefer) t.Errorf("backends should keep the configured order, got %+v", cfg.Backends)
}
if err := cfg.Validate(); err != nil {
t.Errorf("a bare backend list must validate: %v", err)
} }
} }
@@ -78,8 +79,8 @@ func TestLoad_FileAndEnvOverride(t *testing.T) {
if err := os.MkdirAll(cfgDir, 0o755); err != nil { if err := os.MkdirAll(cfgDir, 0o755); err != nil {
t.Fatal(err) t.Fatal(err)
} }
body := "listen: :9999\nmerge: static\nprimary: old\nprefer: old\n" + body := "listen: :9999\nmerge: static\n" +
"backends:\n - name: old\n url: http://localhost:18080\n - name: new\n url: http://localhost:18081\n" "backends:\n - name: a\n url: http://localhost:18080\n - name: b\n url: http://localhost:18081\n"
if err := os.WriteFile(filepath.Join(cfgDir, configFileName), []byte(body), 0o644); err != nil { if err := os.WriteFile(filepath.Join(cfgDir, configFileName), []byte(body), 0o644); err != nil {
t.Fatal(err) t.Fatal(err)
} }
@@ -94,8 +95,8 @@ func TestLoad_FileAndEnvOverride(t *testing.T) {
if cfg.Listen != "127.0.0.1:1234" { if cfg.Listen != "127.0.0.1:1234" {
t.Errorf("env should beat file for listen, got %q", cfg.Listen) t.Errorf("env should beat file for listen, got %q", cfg.Listen)
} }
if cfg.Merge != mergeStatic || cfg.Primary != "old" { if cfg.Merge != mergeStatic {
t.Errorf("file override failed: merge=%s primary=%s", cfg.Merge, cfg.Primary) t.Errorf("file override failed: merge=%s", cfg.Merge)
} }
} }
@@ -103,8 +104,6 @@ func TestApplyEnv_Backends(t *testing.T) {
cfg := testConfigValid() cfg := testConfigValid()
env := map[string]string{ env := map[string]string{
envPrefix + "BACKENDS": "a=http://a:8080,b=http://b:8080", envPrefix + "BACKENDS": "a=http://a:8080,b=http://b:8080",
envPrefix + "PRIMARY": "a",
envPrefix + "PREFER": "a",
envPrefix + "TIMEOUT": "3s", envPrefix + "TIMEOUT": "3s",
envPrefix + "FRESHNESS_TTL": "45s", envPrefix + "FRESHNESS_TTL": "45s",
} }
@@ -126,10 +125,8 @@ func TestValidate(t *testing.T) {
}{ }{
{"ok", func(*Config) {}, false}, {"ok", func(*Config) {}, false},
{"no backends", func(c *Config) { c.Backends = nil }, true}, {"no backends", func(c *Config) { c.Backends = nil }, true},
{"dup name", func(c *Config) { c.Backends = append(c.Backends, Backend{Name: "old", URL: "x"}) }, true}, {"dup name", func(c *Config) { c.Backends = append(c.Backends, Backend{Name: "a", URL: "x"}) }, true},
{"missing url", func(c *Config) { c.Backends[0].URL = "" }, true}, {"missing url", func(c *Config) { c.Backends[0].URL = "" }, true},
{"primary not a backend", func(c *Config) { c.Primary = "ghost" }, true},
{"prefer not a backend", func(c *Config) { c.Prefer = "ghost" }, true},
{"bad merge", func(c *Config) { c.Merge = "wrong" }, true}, {"bad merge", func(c *Config) { c.Merge = "wrong" }, true},
{"zero timeout", func(c *Config) { c.Timeout = 0 }, true}, {"zero timeout", func(c *Config) { c.Timeout = 0 }, true},
} }
@@ -155,13 +152,6 @@ func TestParseBackends(t *testing.T) {
} }
} }
func TestPrimaryBackend(t *testing.T) {
cfg := testConfigValid()
if got, want := cfg.PrimaryBackend().URL, cfg.Backends[1].URL; got != want {
t.Errorf("primary backend URL = %q, want %q", got, want)
}
}
func TestExampleConfig_IsValidAndNeutral(t *testing.T) { func TestExampleConfig_IsValidAndNeutral(t *testing.T) {
cfg := ExampleConfig() cfg := ExampleConfig()
if err := cfg.Validate(); err != nil { if err := cfg.Validate(); err != nil {
@@ -176,7 +166,7 @@ func TestExampleConfig_IsValidAndNeutral(t *testing.T) {
func clearEnv(t *testing.T) { func clearEnv(t *testing.T) {
t.Helper() t.Helper()
for _, k := range []string{"LISTEN", "PRIMARY", "MERGE", "PREFER", "TIMEOUT", "FRESHNESS_TTL", "BACKENDS"} { for _, k := range []string{"LISTEN", "MERGE", "TIMEOUT", "FRESHNESS_TTL", "BACKENDS"} {
t.Setenv(envPrefix+k, "") t.Setenv(envPrefix+k, "")
} }
} }
main.go
+4 -11
View File
@@ -25,18 +25,14 @@ func main() {
} }
var ( var (
listen string listen string
primary string merge string
merge string
) )
serve := func(cmd *cobra.Command) error { serve := func(cmd *cobra.Command) error {
if cmd.Flags().Changed("listen") { if cmd.Flags().Changed("listen") {
cfg.Listen = listen cfg.Listen = listen
} }
if cmd.Flags().Changed("primary") {
cfg.Primary = primary
}
if cmd.Flags().Changed("merge") { if cmd.Flags().Changed("merge") {
cfg.Merge = merge cfg.Merge = merge
} }
@@ -59,7 +55,6 @@ func main() {
pf := root.PersistentFlags() pf := root.PersistentFlags()
pf.StringVar(&listen, "listen", cfg.Listen, "HTTP listen address (overrides config and PDBMUX_LISTEN)") pf.StringVar(&listen, "listen", cfg.Listen, "HTTP listen address (overrides config and PDBMUX_LISTEN)")
pf.StringVar(&primary, "primary", cfg.Primary, "Primary backend name for non-merged pass-through")
pf.StringVar(&merge, "merge", cfg.Merge, "Facts merge strategy: freshness or static") pf.StringVar(&merge, "merge", cfg.Merge, "Facts merge strategy: freshness or static")
serveCmd := &cobra.Command{ serveCmd := &cobra.Command{
@@ -112,8 +107,8 @@ func runServer(cfg Config) error {
ReadHeaderTimeout: 10 * time.Second, ReadHeaderTimeout: 10 * time.Second,
} }
logger.Printf("listening on %s (merge=%s primary=%s backends=%d)", logger.Printf("listening on %s (merge=%s backends=%d)",
cfg.Listen, cfg.Merge, cfg.Primary, len(cfg.Backends)) cfg.Listen, cfg.Merge, len(cfg.Backends))
errCh := make(chan error, 1) errCh := make(chan error, 1)
go func() { go func() {
@@ -139,9 +134,7 @@ func runServer(cfg Config) error {
func printConfig(cfg Config) { func printConfig(cfg Config) {
fmt.Printf("config file : %s\n", ConfigPath()) fmt.Printf("config file : %s\n", ConfigPath())
fmt.Printf("listen : %s\n", cfg.Listen) fmt.Printf("listen : %s\n", cfg.Listen)
fmt.Printf("primary : %s\n", cfg.Primary)
fmt.Printf("merge : %s\n", cfg.Merge) fmt.Printf("merge : %s\n", cfg.Merge)
fmt.Printf("prefer : %s\n", cfg.Prefer)
fmt.Printf("timeout : %s\n", durationString(cfg.Timeout)) fmt.Printf("timeout : %s\n", durationString(cfg.Timeout))
fmt.Printf("freshness_ttl: %s\n", durationString(cfg.FreshnessTTL)) fmt.Printf("freshness_ttl: %s\n", durationString(cfg.FreshnessTTL))
fmt.Println("backends:") fmt.Println("backends:")
merge.go
+8 -5
View File
@@ -49,7 +49,7 @@ func parseTimestamp(s string) time.Time {
return time.Time{} return time.Time{}
} }
// results must be ordered by precedence: ties keep the earlier backend's record. // Ties keep the earlier backend's record — a deterministic tie-break, not a preference.
func mergeNodes(results []backendResult) []json.RawMessage { func mergeNodes(results []backendResult) []json.RawMessage {
type pick struct { type pick struct {
raw json.RawMessage raw json.RawMessage
@@ -81,7 +81,7 @@ func mergeNodes(results []backendResult) []json.RawMessage {
// certname -> name of the backend holding that node's newest report. // certname -> name of the backend holding that node's newest report.
type freshness map[string]string type freshness map[string]string
// results must be ordered by precedence: ties keep the earlier backend. // Ties keep the earlier backend — a deterministic tie-break, not a preference.
func buildFreshness(results []backendResult) freshness { func buildFreshness(results []backendResult) freshness {
type pick struct { type pick struct {
backend string backend string
@@ -104,9 +104,9 @@ func buildFreshness(results []backendResult) freshness {
return f return f
} }
// owner names the winning backend per certname; when it holds no facts for that certname, precedence order wins. // owner names the winning backend per certname; a nil owner (static merge), or one holding no facts for that certname, falls back to configured order.
func mergeFacts(results []backendResult, owner func(certname string) string) []json.RawMessage { func mergeFacts(results []backendResult, owner func(certname string) string) []json.RawMessage {
present := map[string][]string{} // certname -> backend names, in precedence order present := map[string][]string{} // certname -> backend names, in configured order
byKey := map[string][]json.RawMessage{} byKey := map[string][]json.RawMessage{}
for _, res := range results { for _, res := range results {
for _, rec := range res.records { for _, rec := range res.records {
@@ -133,7 +133,10 @@ func mergeFacts(results []backendResult, owner func(certname string) string) []j
out := []json.RawMessage{} out := []json.RawMessage{}
for _, cn := range order { for _, cn := range order {
backends := present[cn] backends := present[cn]
chosen := owner(cn) chosen := ""
if owner != nil {
chosen = owner(cn)
}
if !contains(backends, chosen) { if !contains(backends, chosen) {
chosen = backends[0] chosen = backends[0]
} }
merge_test.go
+59 -58
View File
@@ -89,10 +89,10 @@ func event(cn, reportHash, resource string) string {
} }
func TestMergeNodes_NewerWins(t *testing.T) { func TestMergeNodes_NewerWins(t *testing.T) {
old := recs(t, "old", node("h1", "2026-07-01T00:00:00Z"), node("h2", "2026-07-10T00:00:00Z")) a := recs(t, "a", node("h1", "2026-07-01T00:00:00Z"), node("h2", "2026-07-10T00:00:00Z"))
nw := recs(t, "new", node("h1", "2026-07-20T00:00:00Z"), node("h3", "2026-07-05T00:00:00Z")) b := recs(t, "b", node("h1", "2026-07-20T00:00:00Z"), node("h3", "2026-07-05T00:00:00Z"))
merged := mergeNodes([]backendResult{old, nw}) merged := mergeNodes([]backendResult{a, b})
got := map[string]string{} got := map[string]string{}
for _, r := range merged { for _, r := range merged {
var m recordMeta var m recordMeta
@@ -100,13 +100,13 @@ func TestMergeNodes_NewerWins(t *testing.T) {
got[m.Certname] = m.ReportTimestamp got[m.Certname] = m.ReportTimestamp
} }
if got["h1"] != "2026-07-20T00:00:00Z" { if got["h1"] != "2026-07-20T00:00:00Z" {
t.Errorf("h1: newer (new) should win, got %s", got["h1"]) t.Errorf("h1: newer (b) should win, got %s", got["h1"])
} }
if got["h2"] != "2026-07-10T00:00:00Z" { if got["h2"] != "2026-07-10T00:00:00Z" {
t.Errorf("h2: only in old, got %s", got["h2"]) t.Errorf("h2: only in a, got %s", got["h2"])
} }
if got["h3"] != "2026-07-05T00:00:00Z" { if got["h3"] != "2026-07-05T00:00:00Z" {
t.Errorf("h3: only in new, got %s", got["h3"]) t.Errorf("h3: only in b, got %s", got["h3"])
} }
if len(merged) != 3 { if len(merged) != 3 {
t.Errorf("expected 3 deduped nodes, got %d", len(merged)) t.Errorf("expected 3 deduped nodes, got %d", len(merged))
@@ -114,20 +114,20 @@ func TestMergeNodes_NewerWins(t *testing.T) {
} }
func TestMergeNodes_OneBackendOnly(t *testing.T) { func TestMergeNodes_OneBackendOnly(t *testing.T) {
old := recs(t, "old", node("h1", "2026-07-01T00:00:00Z")) a := recs(t, "a", node("h1", "2026-07-01T00:00:00Z"))
// new returned nothing (e.g. empty result). // b returned nothing (e.g. empty result).
nw := backendResult{name: "new"} b := backendResult{name: "b"}
merged := mergeNodes([]backendResult{old, nw}) merged := mergeNodes([]backendResult{a, b})
if len(merged) != 1 || certnames(t, merged)[0] != "h1" { if len(merged) != 1 || certnames(t, merged)[0] != "h1" {
t.Fatalf("expected only h1, got %v", certnames(t, merged)) t.Fatalf("expected only h1, got %v", certnames(t, merged))
} }
} }
func TestMergeNodes_TieKeepsEarlierBackend(t *testing.T) { func TestMergeNodes_TieKeepsEarlierBackend(t *testing.T) {
// Equal timestamps: the backend listed first (precedence) wins. // Equal timestamps: the backend listed first wins, as a tie-break.
prefer := recs(t, "new", node("h1", "2026-07-01T00:00:00Z")) first := recs(t, "b", node("h1", "2026-07-01T00:00:00Z"))
other := recs(t, "old", node("h1", "2026-07-01T00:00:00Z")) second := recs(t, "a", node("h1", "2026-07-01T00:00:00Z"))
merged := mergeNodes([]backendResult{prefer, other}) merged := mergeNodes([]backendResult{first, second})
if len(merged) != 1 { if len(merged) != 1 {
t.Fatalf("expected 1 record, got %d", len(merged)) t.Fatalf("expected 1 record, got %d", len(merged))
} }
@@ -138,8 +138,8 @@ func TestMergeNodes_TieKeepsEarlierBackend(t *testing.T) {
} }
func TestMergeNodes_PreservesUnknownFields(t *testing.T) { func TestMergeNodes_PreservesUnknownFields(t *testing.T) {
old := recs(t, "old", `{"certname":"h1","report_timestamp":"2026-07-01T00:00:00Z","extra":{"deep":42}}`) a := recs(t, "a", `{"certname":"h1","report_timestamp":"2026-07-01T00:00:00Z","extra":{"deep":42}}`)
merged := mergeNodes([]backendResult{old}) merged := mergeNodes([]backendResult{a})
if len(merged) != 1 { if len(merged) != 1 {
t.Fatalf("expected 1 record") t.Fatalf("expected 1 record")
} }
@@ -150,69 +150,70 @@ func TestMergeNodes_PreservesUnknownFields(t *testing.T) {
} }
} }
func TestMergeFacts_Static_PreferWins(t *testing.T) { func TestMergeFacts_NilOwnerUsesConfiguredOrder(t *testing.T) {
// h1 in both; static prefer=new -> new's facts kept, old's dropped. // Static merge passes no owner: h1 is in both, so the first backend in the
old := recs(t, "old", fact("h1", "role", "web-old", ""), fact("h2", "role", "db-old", "")) // slice supplies its facts.
nw := recs(t, "new", fact("h1", "role", "web-new", "")) a := recs(t, "a", fact("h1", "role", "web-a", ""), fact("h2", "role", "db-a", ""))
b := recs(t, "b", fact("h1", "role", "web-b", ""))
merged := mergeFacts([]backendResult{nw, old}, func(string) string { return "new" }) merged := mergeFacts([]backendResult{b, a}, nil)
got := factValues(t, merged) got := factValues(t, merged)
assertContains(t, got, "h1:role=web-new") assertContains(t, got, "h1:role=web-b")
assertNotContains(t, got, "h1:role=web-old") assertNotContains(t, got, "h1:role=web-a")
// h2 only in old -> falls back to old. // h2 only in a -> still served from a.
assertContains(t, got, "h2:role=db-old") assertContains(t, got, "h2:role=db-a")
} }
func TestMergeFacts_Freshness_NewerBackendWins(t *testing.T) { func TestMergeFacts_Freshness_NewerBackendWins(t *testing.T) {
// owner map says h1 belongs to old (older backend has the newer report), // owner map says h1 belongs to a and h2 to b. Multiple facts per node must
// h2 belongs to new. Multiple facts per node must all come from the winner. // all come from the winner.
old := recs(t, "old", a := recs(t, "a",
fact("h1", "role", "web-old", ""), fact("h1", "ip", "10.0.0.1", ""), fact("h1", "role", "web-a", ""), fact("h1", "ip", "10.0.0.1", ""),
fact("h2", "role", "db-old", "")) fact("h2", "role", "db-a", ""))
nw := recs(t, "new", b := recs(t, "b",
fact("h1", "role", "web-new", ""), fact("h1", "ip", "10.9.9.9", ""), fact("h1", "role", "web-b", ""), fact("h1", "ip", "10.9.9.9", ""),
fact("h2", "role", "db-new", ""), fact("h2", "ip", "10.0.0.2", "")) fact("h2", "role", "db-b", ""), fact("h2", "ip", "10.0.0.2", ""))
owner := func(cn string) string { owner := func(cn string) string {
if cn == "h1" { if cn == "h1" {
return "old" return "a"
} }
return "new" return "b"
} }
merged := mergeFacts([]backendResult{nw, old}, owner) merged := mergeFacts([]backendResult{b, a}, owner)
got := factValues(t, merged) got := factValues(t, merged)
// h1 -> all old facts, no new facts. // h1 -> all a facts, no b facts.
assertContains(t, got, "h1:role=web-old") assertContains(t, got, "h1:role=web-a")
assertContains(t, got, "h1:ip=10.0.0.1") assertContains(t, got, "h1:ip=10.0.0.1")
assertNotContains(t, got, "h1:role=web-new") assertNotContains(t, got, "h1:role=web-b")
assertNotContains(t, got, "h1:ip=10.9.9.9") assertNotContains(t, got, "h1:ip=10.9.9.9")
// h2 -> all new facts. // h2 -> all b facts.
assertContains(t, got, "h2:role=db-new") assertContains(t, got, "h2:role=db-b")
assertContains(t, got, "h2:ip=10.0.0.2") assertContains(t, got, "h2:ip=10.0.0.2")
assertNotContains(t, got, "h2:role=db-old") assertNotContains(t, got, "h2:role=db-a")
} }
func TestMergeFacts_OwnerMissingFallsBackToPrecedence(t *testing.T) { func TestMergeFacts_OwnerMissingFallsBackToConfiguredOrder(t *testing.T) {
// owner returns a backend with no facts for h1 -> fall back to first // owner returns a backend with no facts for h1 -> fall back to the first
// backend present (precedence order of the slice). // backend in the slice that has some.
prefer := recs(t, "new", fact("h1", "role", "web-new", "")) first := recs(t, "b", fact("h1", "role", "web-b", ""))
other := recs(t, "old", fact("h1", "role", "web-old", "")) second := recs(t, "a", fact("h1", "role", "web-a", ""))
merged := mergeFacts([]backendResult{prefer, other}, func(string) string { return "ghost" }) merged := mergeFacts([]backendResult{first, second}, func(string) string { return "ghost" })
got := factValues(t, merged) got := factValues(t, merged)
assertContains(t, got, "h1:role=web-new") // new is first in slice assertContains(t, got, "h1:role=web-b") // b is first in slice
assertNotContains(t, got, "h1:role=web-old") assertNotContains(t, got, "h1:role=web-a")
} }
func TestBuildFreshness(t *testing.T) { func TestBuildFreshness(t *testing.T) {
// old has newer report for h1; new has newer for h2. // a holds the newer report for h1; b holds the newer one for h2.
old := recs(t, "old", node("h1", "2026-07-20T00:00:00Z"), node("h2", "2026-07-01T00:00:00Z")) a := recs(t, "a", node("h1", "2026-07-20T00:00:00Z"), node("h2", "2026-07-01T00:00:00Z"))
nw := recs(t, "new", node("h1", "2026-07-01T00:00:00Z"), node("h2", "2026-07-20T00:00:00Z")) b := recs(t, "b", node("h1", "2026-07-01T00:00:00Z"), node("h2", "2026-07-20T00:00:00Z"))
f := buildFreshness([]backendResult{old, nw}) f := buildFreshness([]backendResult{a, b})
if f["h1"] != "old" { if f["h1"] != "a" {
t.Errorf("h1 should belong to old, got %q", f["h1"]) t.Errorf("h1 should belong to a, got %q", f["h1"])
} }
if f["h2"] != "new" { if f["h2"] != "b" {
t.Errorf("h2 should belong to new, got %q", f["h2"]) t.Errorf("h2 should belong to b, got %q", f["h2"])
} }
} }
reports.go
+2 -2
View File
@@ -9,7 +9,7 @@ import (
"strings" "strings"
) )
// results must be ordered by precedence; a key func returning ok=false means the record has no identity and is always kept. // The first backend in results holding a key supplies the record; a key func returning ok=false means the record has no identity and is always kept.
func mergeUnion(results []backendResult, key func(record) (string, bool)) []json.RawMessage { func mergeUnion(results []backendResult, key func(record) (string, bool)) []json.RawMessage {
seen := make(map[string]bool) seen := make(map[string]bool)
out := []json.RawMessage{} out := []json.RawMessage{}
@@ -65,7 +65,7 @@ func parseOrderBy(s string) ([]orderField, error) {
return out, nil return out, nil
} }
// Each backend ordered only its own slice, so the union is re-sorted here; stable, so ties keep backend precedence. // Each backend ordered only its own slice, so the union is re-sorted here; stable, so ties keep the merged set's existing order.
func sortRecords(recs []json.RawMessage, order []orderField) { func sortRecords(recs []json.RawMessage, order []orderField) {
if len(order) == 0 || len(recs) < 2 { if len(order) == 0 || len(recs) < 2 {
return return
reports_test.go
+10 -10
View File
@@ -8,17 +8,17 @@ import (
) )
func TestMergeUnion_KeepsBothBackendsHistory(t *testing.T) { func TestMergeUnion_KeepsBothBackendsHistory(t *testing.T) {
old := recs(t, "old", report("h1", "r1", "2026-07-01T00:00:00Z")) a := recs(t, "a", report("h1", "r1", "2026-07-01T00:00:00Z"))
nw := recs(t, "new", report("h1", "r2", "2026-07-02T00:00:00Z")) b := recs(t, "b", report("h1", "r2", "2026-07-02T00:00:00Z"))
merged := mergeUnion([]backendResult{nw, old}, reportKey) merged := mergeUnion([]backendResult{b, a}, reportKey)
if got := hashesOf(t, merged); !slices.Equal(got, []string{"r2", "r1"}) { if got := hashesOf(t, merged); !slices.Equal(got, []string{"r2", "r1"}) {
t.Errorf("union = %v, want both reports in precedence order", got) t.Errorf("union = %v, want both reports in configured order", got)
} }
} }
func TestMergeUnion_DedupesSharedHash(t *testing.T) { func TestMergeUnion_DedupesSharedHash(t *testing.T) {
dup := report("h1", "r1", "2026-07-01T00:00:00Z") dup := report("h1", "r1", "2026-07-01T00:00:00Z")
merged := mergeUnion([]backendResult{recs(t, "new", dup), recs(t, "old", dup)}, reportKey) merged := mergeUnion([]backendResult{recs(t, "b", dup), recs(t, "a", dup)}, reportKey)
if got := hashesOf(t, merged); !slices.Equal(got, []string{"r1"}) { if got := hashesOf(t, merged); !slices.Equal(got, []string{"r1"}) {
t.Errorf("union = %v, want a single r1", got) t.Errorf("union = %v, want a single r1", got)
} }
@@ -27,9 +27,9 @@ func TestMergeUnion_DedupesSharedHash(t *testing.T) {
func TestMergeUnion_HashlessRowsAreAllKept(t *testing.T) { func TestMergeUnion_HashlessRowsAreAllKept(t *testing.T) {
// extract/group_by queries return synthetic rows with no hash; dropping the // extract/group_by queries return synthetic rows with no hash; dropping the
// second backend's rows as "duplicates" would silently lose half the data. // second backend's rows as "duplicates" would silently lose half the data.
old := recs(t, "old", `{"status":"changed","count":3}`) a := recs(t, "a", `{"status":"changed","count":3}`)
nw := recs(t, "new", `{"status":"changed","count":5}`) b := recs(t, "b", `{"status":"changed","count":5}`)
merged := mergeUnion([]backendResult{old, nw}, reportKey) merged := mergeUnion([]backendResult{a, b}, reportKey)
if len(merged) != 2 { if len(merged) != 2 {
t.Errorf("expected both aggregate rows, got %d: %v", len(merged), merged) t.Errorf("expected both aggregate rows, got %d: %v", len(merged), merged)
} }
@@ -39,7 +39,7 @@ func TestMergeUnion_IdenticalHashlessRowsAreNotCollapsed(t *testing.T) {
// Two backends can legitimately produce the same aggregate row; collapsing // Two backends can legitimately produce the same aggregate row; collapsing
// them as duplicates undercounts the merged result. // them as duplicates undercounts the merged result.
same := `{"status":"changed","count":1}` same := `{"status":"changed","count":1}`
merged := mergeUnion([]backendResult{recs(t, "old", same), recs(t, "new", same)}, reportKey) merged := mergeUnion([]backendResult{recs(t, "a", same), recs(t, "b", same)}, reportKey)
if len(merged) != 2 { if len(merged) != 2 {
t.Errorf("expected both backends' aggregate rows, got %d: %v", len(merged), merged) t.Errorf("expected both backends' aggregate rows, got %d: %v", len(merged), merged)
} }
@@ -48,7 +48,7 @@ func TestMergeUnion_IdenticalHashlessRowsAreNotCollapsed(t *testing.T) {
func TestMergeUnion_EventsDedupeOnRawIdentity(t *testing.T) { func TestMergeUnion_EventsDedupeOnRawIdentity(t *testing.T) {
same := event("h1", "r1", "Package[nginx]") same := event("h1", "r1", "Package[nginx]")
other := event("h1", "r1", "Service[nginx]") other := event("h1", "r1", "Service[nginx]")
merged := mergeUnion([]backendResult{recs(t, "new", same, other), recs(t, "old", same)}, rawKey) merged := mergeUnion([]backendResult{recs(t, "b", same, other), recs(t, "a", same)}, rawKey)
if len(merged) != 2 { if len(merged) != 2 {
t.Errorf("expected 2 distinct events, got %d: %v", len(merged), merged) t.Errorf("expected 2 distinct events, got %d: %v", len(merged), merged)
} }
server.go
+58 -43
View File
@@ -8,7 +8,6 @@ import (
"log" "log"
"net/http" "net/http"
"net/url" "net/url"
"sort"
"strconv" "strconv"
"strings" "strings"
"sync" "sync"
@@ -82,7 +81,7 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) {
s.serveFirstHolder(w, r) s.serveFirstHolder(w, r)
return return
} }
s.proxyPrimary(w, r) s.proxyUnmerged(w, r)
} }
} }
@@ -125,7 +124,7 @@ func (s *Server) serveUnion(w http.ResponseWriter, r *http.Request, path string,
return return
} }
merged := mergeUnion(s.byPrecedence(alive), key) merged := mergeUnion(alive, key)
sortRecords(merged, page.order) sortRecords(merged, page.order)
if page.wantTotal { if page.wantTotal {
if total := sumTotals(alive); total >= 0 { if total := sumTotals(alive); total >= 0 {
@@ -158,7 +157,7 @@ func (s *Server) serveSummed(w http.ResponseWriter, r *http.Request, path string
return return
} }
merged := sumRows(s.byPrecedence(alive), columns) merged := sumRows(alive, columns)
sortRecords(merged, page.order) sortRecords(merged, page.order)
if page.wantTotal { if page.wantTotal {
w.Header().Set(recordsHeader, strconv.Itoa(len(merged))) w.Header().Set(recordsHeader, strconv.Itoa(len(merged)))
@@ -182,7 +181,7 @@ func (s *Server) serveFirstHolder(w http.ResponseWriter, r *http.Request) {
http.Error(w, "no backend holds this report", http.StatusNotFound) http.Error(w, "no backend holds this report", http.StatusNotFound)
return return
} }
for _, res := range s.byPrecedence(alive) { for _, res := range alive {
if len(res.records) > 0 { if len(res.records) > 0 {
writeJSON(w, rawRecords(res.records)) writeJSON(w, rawRecords(res.records))
return return
@@ -226,33 +225,15 @@ func rawRecords(recs []record) []json.RawMessage {
} }
func (s *Server) mergeNodesResponse(results []backendResult) []json.RawMessage { func (s *Server) mergeNodesResponse(results []backendResult) []json.RawMessage {
return mergeNodes(s.byPrecedence(results)) return mergeNodes(results)
} }
func (s *Server) mergeFactsResponse(results []backendResult) []json.RawMessage { func (s *Server) mergeFactsResponse(results []backendResult) []json.RawMessage {
ordered := s.byPrecedence(results)
if s.cfg.Merge == mergeStatic { if s.cfg.Merge == mergeStatic {
prefer := s.cfg.Prefer return mergeFacts(results, nil)
return mergeFacts(ordered, func(string) string { return prefer })
} }
fresh := s.freshnessMap(context.Background(), ordered) fresh := s.freshnessMap(context.Background(), results)
prefer := s.cfg.Prefer return mergeFacts(results, func(cn string) string { return fresh[cn] })
return mergeFacts(ordered, func(cn string) string {
if b, ok := fresh[cn]; ok {
return b
}
return prefer
})
}
// Puts Prefer first so it wins ties; the rest keep config order.
func (s *Server) byPrecedence(results []backendResult) []backendResult {
ordered := make([]backendResult, len(results))
copy(ordered, results)
sort.SliceStable(ordered, func(i, j int) bool {
return ordered[i].name == s.cfg.Prefer && ordered[j].name != s.cfg.Prefer
})
return ordered
} }
// Queries /nodes unfiltered rather than reusing the request's results, because a /facts query's certname set can differ. // Queries /nodes unfiltered rather than reusing the request's results, because a /facts query's certname set can differ.
@@ -274,7 +255,7 @@ func (s *Server) freshnessMap(ctx context.Context, _ []backendResult) freshness
} }
alive = append(alive, res) alive = append(alive, res)
} }
f := buildFreshness(s.byPrecedence(alive)) f := buildFreshness(alive)
s.mu.Lock() s.mu.Lock()
s.freshData = f s.freshData = f
@@ -329,29 +310,63 @@ func (s *Server) queryBackend(ctx context.Context, b Backend, path string, param
return recs, total, err return recs, total, err
} }
func (s *Server) proxyPrimary(w http.ResponseWriter, r *http.Request) { // The record shape is unknown, so a union would be guesswork: the first 2xx wins and the first error response is replayed when none succeeds.
b := s.cfg.PrimaryBackend() func (s *Server) proxyUnmerged(w http.ResponseWriter, r *http.Request) {
var fallback *bufferedResponse
for _, b := range s.cfg.Backends {
resp, err := s.passThrough(r, b)
if err != nil {
s.log.Printf("warning: backend %q pass-through failed for %s: %v", b.Name, r.URL.Path, err)
continue
}
if resp.StatusCode >= 200 && resp.StatusCode < 300 {
setContentType(w, resp.Header.Get("Content-Type"))
w.WriteHeader(resp.StatusCode)
_, _ = io.Copy(w, resp.Body)
_ = resp.Body.Close()
return
}
body, _ := io.ReadAll(resp.Body)
_ = resp.Body.Close()
if fallback == nil {
fallback = &bufferedResponse{
status: resp.StatusCode,
contentType: resp.Header.Get("Content-Type"),
body: body,
}
}
}
if fallback == nil {
http.Error(w, "all backends failed", http.StatusBadGateway)
return
}
setContentType(w, fallback.contentType)
w.WriteHeader(fallback.status)
_, _ = w.Write(fallback.body)
}
type bufferedResponse struct {
status int
contentType string
body []byte
}
func (s *Server) passThrough(r *http.Request, b Backend) (*http.Response, error) {
target := strings.TrimRight(b.URL, "/") + r.URL.Path target := strings.TrimRight(b.URL, "/") + r.URL.Path
if r.URL.RawQuery != "" { if r.URL.RawQuery != "" {
target += "?" + r.URL.RawQuery target += "?" + r.URL.RawQuery
} }
req, err := http.NewRequestWithContext(r.Context(), http.MethodGet, target, nil) req, err := http.NewRequestWithContext(r.Context(), http.MethodGet, target, nil)
if err != nil { if err != nil {
http.Error(w, err.Error(), http.StatusBadGateway) return nil, err
return
} }
resp, err := s.client.Do(req) return s.client.Do(req)
if err != nil { }
s.log.Printf("warning: primary %q pass-through failed for %s: %v", b.Name, r.URL.Path, err)
http.Error(w, "primary backend failed", http.StatusBadGateway) func setContentType(w http.ResponseWriter, contentType string) {
return if contentType != "" {
w.Header().Set("Content-Type", contentType)
} }
defer func() { _ = resp.Body.Close() }()
if ct := resp.Header.Get("Content-Type"); ct != "" {
w.Header().Set("Content-Type", ct)
}
w.WriteHeader(resp.StatusCode)
_, _ = io.Copy(w, resp.Body)
} }
type healthReport struct { type healthReport struct {
server_test.go
+222 -189
View File
@@ -119,13 +119,11 @@ func truncate(t *testing.T, body, limit string) string {
return string(out) return string(out)
} }
func testConfig(oldURL, newURL, merge string) Config { func testConfig(aURL, bURL, merge string) Config {
return Config{ return Config{
Listen: ":0", Listen: ":0",
Backends: []Backend{{Name: "old", URL: oldURL}, {Name: "new", URL: newURL}}, Backends: []Backend{{Name: "a", URL: aURL}, {Name: "b", URL: bURL}},
Primary: "new",
Merge: merge, Merge: merge,
Prefer: "new",
Timeout: 2 * time.Second, Timeout: 2 * time.Second,
FreshnessTTL: 30 * time.Second, FreshnessTTL: 30 * time.Second,
} }
@@ -148,11 +146,11 @@ func doGet(t *testing.T, h http.Handler, path, query string) *httptest.ResponseR
} }
func TestHandler_NodesMerged(t *testing.T) { func TestHandler_NodesMerged(t *testing.T) {
old := newFakeBackend(t, a := newFakeBackend(t,
`[`+node("h1", "2026-07-01T00:00:00Z")+`,`+node("h2", "2026-07-10T00:00:00Z")+`]`, `[]`) `[`+node("h1", "2026-07-01T00:00:00Z")+`,`+node("h2", "2026-07-10T00:00:00Z")+`]`, `[]`)
nw := newFakeBackend(t, b := newFakeBackend(t,
`[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`) `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGet(t, srv.Handler(), nodesPath, `["=","certname","h1"]`) rec := doGet(t, srv.Handler(), nodesPath, `["=","certname","h1"]`)
if rec.Code != http.StatusOK { if rec.Code != http.StatusOK {
@@ -167,76 +165,78 @@ func TestHandler_NodesMerged(t *testing.T) {
} }
for _, m := range got { for _, m := range got {
if m.Certname == "h1" && m.ReportTimestamp != "2026-07-20T00:00:00Z" { if m.Certname == "h1" && m.ReportTimestamp != "2026-07-20T00:00:00Z" {
t.Errorf("h1 should be new's newer record, got %s", m.ReportTimestamp) t.Errorf("h1 should be the newer record, got %s", m.ReportTimestamp)
} }
} }
} }
func TestHandler_QueryPassthrough(t *testing.T) { func TestHandler_QueryPassthrough(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
q := `["=","certname","abc.example.net"]` q := `["=","certname","abc.example.net"]`
doGet(t, srv.Handler(), factsPath, q) doGet(t, srv.Handler(), factsPath, q)
if old.gotQuery(factsPath) != q { if a.gotQuery(factsPath) != q {
t.Errorf("old backend got query %q, want %q", old.gotQuery(factsPath), q) t.Errorf("a backend got query %q, want %q", a.gotQuery(factsPath), q)
} }
if nw.gotQuery(factsPath) != q { if b.gotQuery(factsPath) != q {
t.Errorf("new backend got query %q, want %q", nw.gotQuery(factsPath), q) t.Errorf("b backend got query %q, want %q", b.gotQuery(factsPath), q)
} }
} }
func TestHandler_FactsStaticMerge(t *testing.T) { func TestHandler_FactsStaticMerge(t *testing.T) {
old := newFakeBackend(t, `[]`, // Static merge ignores timestamps: a shared certname resolves to the first
`[`+fact("h1", "role", "web-old", "")+`,`+fact("h2", "role", "db-old", "")+`]`) // backend in configured order that holds it.
nw := newFakeBackend(t, `[]`, a := newFakeBackend(t, `[]`,
`[`+fact("h1", "role", "web-new", "")+`]`) `[`+fact("h1", "role", "web-a", "")+`,`+fact("h2", "role", "db-a", "")+`]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) b := newFakeBackend(t, `[]`,
`[`+fact("h1", "role", "web-b", "")+`,`+fact("h3", "role", "db-b", "")+`]`)
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGet(t, srv.Handler(), factsPath, `["=","name","role"]`) rec := doGet(t, srv.Handler(), factsPath, `["=","name","role"]`)
if rec.Code != http.StatusOK { if rec.Code != http.StatusOK {
t.Fatalf("status %d", rec.Code) t.Fatalf("status %d", rec.Code)
} }
body := rec.Body.String() body := rec.Body.String()
if !strings.Contains(body, "web-new") || strings.Contains(body, "web-old") { if !strings.Contains(body, "web-a") || strings.Contains(body, "web-b") {
t.Errorf("static prefer=new should keep web-new, drop web-old: %s", body) t.Errorf("h1 should resolve to the first backend holding it: %s", body)
} }
if !strings.Contains(body, "db-old") { if !strings.Contains(body, "db-a") || !strings.Contains(body, "db-b") {
t.Errorf("h2 only in old should survive: %s", body) t.Errorf("nodes held by only one backend must all survive: %s", body)
} }
} }
func TestHandler_FactsFreshnessMerge(t *testing.T) { func TestHandler_FactsFreshnessMerge(t *testing.T) {
// Freshness: old holds h1's newer report; new holds h2's newer report. // Freshness: a holds h1's newer report; b holds h2's newer report.
old := newFakeBackend(t, a := newFakeBackend(t,
`[`+node("h1", "2026-07-20T00:00:00Z")+`,`+node("h2", "2026-07-01T00:00:00Z")+`]`, `[`+node("h1", "2026-07-20T00:00:00Z")+`,`+node("h2", "2026-07-01T00:00:00Z")+`]`,
`[`+fact("h1", "role", "web-old", "")+`,`+fact("h2", "role", "db-old", "")+`]`) `[`+fact("h1", "role", "web-a", "")+`,`+fact("h2", "role", "db-a", "")+`]`)
nw := newFakeBackend(t, b := newFakeBackend(t,
`[`+node("h1", "2026-07-01T00:00:00Z")+`,`+node("h2", "2026-07-20T00:00:00Z")+`]`, `[`+node("h1", "2026-07-01T00:00:00Z")+`,`+node("h2", "2026-07-20T00:00:00Z")+`]`,
`[`+fact("h1", "role", "web-new", "")+`,`+fact("h2", "role", "db-new", "")+`]`) `[`+fact("h1", "role", "web-b", "")+`,`+fact("h2", "role", "db-b", "")+`]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeFreshness)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness))
rec := doGet(t, srv.Handler(), factsPath, `["=","name","role"]`) rec := doGet(t, srv.Handler(), factsPath, `["=","name","role"]`)
if rec.Code != http.StatusOK { if rec.Code != http.StatusOK {
t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
} }
body := rec.Body.String() body := rec.Body.String()
// h1 -> old (newer report there); h2 -> new. // h1 -> a (newer report there); h2 -> b.
if !strings.Contains(body, "web-old") || strings.Contains(body, "web-new") { if !strings.Contains(body, "web-a") || strings.Contains(body, "web-b") {
t.Errorf("h1 should resolve to old: %s", body) t.Errorf("h1 should resolve to a: %s", body)
} }
if !strings.Contains(body, "db-new") || strings.Contains(body, "db-old") { if !strings.Contains(body, "db-b") || strings.Contains(body, "db-a") {
t.Errorf("h2 should resolve to new: %s", body) t.Errorf("h2 should resolve to b: %s", body)
} }
} }
func TestHandler_OneBackendDown(t *testing.T) { func TestHandler_OneBackendDown(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
old.fail = true a.fail = true
nw := newFakeBackend(t, b := newFakeBackend(t,
`[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`) `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGet(t, srv.Handler(), nodesPath, "") rec := doGet(t, srv.Handler(), nodesPath, "")
if rec.Code != http.StatusOK { if rec.Code != http.StatusOK {
@@ -248,10 +248,10 @@ func TestHandler_OneBackendDown(t *testing.T) {
} }
func TestHandler_BothBackendsDown(t *testing.T) { func TestHandler_BothBackendsDown(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
old.fail, nw.fail = true, true a.fail, b.fail = true, true
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGet(t, srv.Handler(), nodesPath, "") rec := doGet(t, srv.Handler(), nodesPath, "")
if rec.Code != http.StatusBadGateway { if rec.Code != http.StatusBadGateway {
@@ -259,11 +259,12 @@ func TestHandler_BothBackendsDown(t *testing.T) {
} }
} }
func TestHandler_PassThroughToPrimary(t *testing.T) { func TestHandler_PassThroughFirstAnswer(t *testing.T) {
// A non-merged v4 path (e.g. /resources) goes only to the primary (new). // A path with no merge rule (e.g. /resources) is served by the first backend
old := newFakeBackend(t, `[]`, `[]`) // that answers; the rest are not asked at all.
nw := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) b := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
const path = "/pdb/query/v4/resources" const path = "/pdb/query/v4/resources"
rec := doGet(t, srv.Handler(), path, `["=","certname","h1"]`) rec := doGet(t, srv.Handler(), path, `["=","certname","h1"]`)
@@ -273,19 +274,51 @@ func TestHandler_PassThroughToPrimary(t *testing.T) {
if !strings.Contains(rec.Body.String(), path) { if !strings.Contains(rec.Body.String(), path) {
t.Errorf("expected pass-through body, got %s", rec.Body.String()) t.Errorf("expected pass-through body, got %s", rec.Body.String())
} }
// Only primary (new) should have been queried. if _, hit := a.params(path); !hit {
if _, hit := old.params(path); hit { t.Errorf("first backend should be queried for pass-through")
t.Errorf("non-primary backend should not be queried for pass-through")
} }
if _, hit := nw.params(path); !hit { if _, hit := b.params(path); hit {
t.Errorf("primary backend should be queried for pass-through") t.Errorf("later backends should not be queried once one answers")
}
}
func TestHandler_PassThroughFallsBackToNextBackend(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
a.fail = true
b := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
const path = "/pdb/query/v4/resources"
rec := doGet(t, srv.Handler(), path, "")
if rec.Code != http.StatusOK {
t.Fatalf("expected the surviving backend to serve it, got %d: %s", rec.Code, rec.Body.String())
}
if !strings.Contains(rec.Body.String(), path) {
t.Errorf("expected pass-through body, got %s", rec.Body.String())
}
}
func TestHandler_PassThroughReplaysUpstreamError(t *testing.T) {
// Every backend rejects it, so PuppetDB's own status reaches the client
// rather than a synthetic 502.
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.fail, b.fail = true, true
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGet(t, srv.Handler(), "/pdb/query/v4/resources", "")
if rec.Code != http.StatusInternalServerError {
t.Fatalf("expected the upstream 500 replayed, got %d", rec.Code)
}
if !strings.Contains(rec.Body.String(), "boom") {
t.Errorf("expected the upstream body, got %s", rec.Body.String())
} }
} }
func TestHandler_PostRejected(t *testing.T) { func TestHandler_PostRejected(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
req := httptest.NewRequest(http.MethodPost, factsPath, nil) req := httptest.NewRequest(http.MethodPost, factsPath, nil)
rec := httptest.NewRecorder() rec := httptest.NewRecorder()
srv.Handler().ServeHTTP(rec, req) srv.Handler().ServeHTTP(rec, req)
@@ -295,9 +328,9 @@ func TestHandler_PostRejected(t *testing.T) {
} }
func TestHandler_Health(t *testing.T) { func TestHandler_Health(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGet(t, srv.Handler(), "/healthz", "") rec := doGet(t, srv.Handler(), "/healthz", "")
if rec.Code != http.StatusOK { if rec.Code != http.StatusOK {
@@ -307,16 +340,16 @@ func TestHandler_Health(t *testing.T) {
if err := json.Unmarshal(rec.Body.Bytes(), &hr); err != nil { if err := json.Unmarshal(rec.Body.Bytes(), &hr); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if hr.Status != "ok" || hr.Backends["old"] != "ok" || hr.Backends["new"] != "ok" { if hr.Status != "ok" || hr.Backends["a"] != "ok" || hr.Backends["b"] != "ok" {
t.Fatalf("unexpected health: %+v", hr) t.Fatalf("unexpected health: %+v", hr)
} }
} }
func TestHandler_HealthDegradedAndDown(t *testing.T) { func TestHandler_HealthDegradedAndDown(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
old.fail = true a.fail = true
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGet(t, srv.Handler(), "/healthz", "") rec := doGet(t, srv.Handler(), "/healthz", "")
var hr healthReport var hr healthReport
@@ -328,7 +361,7 @@ func TestHandler_HealthDegradedAndDown(t *testing.T) {
t.Errorf("degraded should still be 200, got %d", rec.Code) t.Errorf("degraded should still be 200, got %d", rec.Code)
} }
nw.fail = true b.fail = true
rec = doGet(t, srv.Handler(), "/healthz", "") rec = doGet(t, srv.Handler(), "/healthz", "")
_ = json.Unmarshal(rec.Body.Bytes(), &hr) _ = json.Unmarshal(rec.Body.Bytes(), &hr)
if hr.Status != "down" || rec.Code != http.StatusServiceUnavailable { if hr.Status != "down" || rec.Code != http.StatusServiceUnavailable {
@@ -337,21 +370,21 @@ func TestHandler_HealthDegradedAndDown(t *testing.T) {
} }
func TestFreshnessCache_Reused(t *testing.T) { func TestFreshnessCache_Reused(t *testing.T) {
old := newFakeBackend(t, a := newFakeBackend(t,
`[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`,
`[`+fact("h1", "role", "web-old", "")+`]`) `[`+fact("h1", "role", "web-a", "")+`]`)
nw := newFakeBackend(t, b := newFakeBackend(t,
`[`+node("h1", "2026-07-01T00:00:00Z")+`]`, `[`+node("h1", "2026-07-01T00:00:00Z")+`]`,
`[`+fact("h1", "role", "web-new", "")+`]`) `[`+fact("h1", "role", "web-b", "")+`]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeFreshness)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness))
// Two facts queries; the freshness /nodes probe should be cached after the // Two facts queries; the freshness /nodes probe should be cached after the
// first, so query recording only reflects the last observed nodes query but // first, so query recording only reflects the last observed nodes query but
// results stay consistent (h1 -> old). // results stay consistent (h1 -> a).
for i := 0; i < 2; i++ { for i := 0; i < 2; i++ {
rec := doGet(t, srv.Handler(), factsPath, `["=","name","role"]`) rec := doGet(t, srv.Handler(), factsPath, `["=","name","role"]`)
if !strings.Contains(rec.Body.String(), "web-old") { if !strings.Contains(rec.Body.String(), "web-a") {
t.Fatalf("iteration %d: expected h1->old, got %s", i, rec.Body.String()) t.Fatalf("iteration %d: expected h1->a, got %s", i, rec.Body.String())
} }
} }
} }
@@ -383,15 +416,15 @@ func hashes(t *testing.T, body []byte) []string {
const receiveDesc = `[{"field":"receive_time","order":"desc"}]` const receiveDesc = `[{"field":"receive_time","order":"desc"}]`
func TestHandler_ReportsUnioned(t *testing.T) { func TestHandler_ReportsUnioned(t *testing.T) {
// h1 migrated: its pre-migration reports are in old, later ones in new. // h1 moved between backends: earlier reports are in a, later ones in b.
// Both must show up, unlike /facts where one backend wins the node. // Both must show up, unlike /facts where one backend wins the node.
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
old.bodies[reportsPath] = `[` + report("h1", "r2", "2026-07-10T00:00:00Z") + `,` + a.bodies[reportsPath] = `[` + report("h1", "r2", "2026-07-10T00:00:00Z") + `,` +
report("h1", "r1", "2026-07-01T00:00:00Z") + `]` report("h1", "r1", "2026-07-01T00:00:00Z") + `]`
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
nw.bodies[reportsPath] = `[` + report("h1", "r4", "2026-07-30T00:00:00Z") + `,` + b.bodies[reportsPath] = `[` + report("h1", "r4", "2026-07-30T00:00:00Z") + `,` +
report("h1", "r3", "2026-07-20T00:00:00Z") + `]` report("h1", "r3", "2026-07-20T00:00:00Z") + `]`
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{ rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{
"query": {`["=","certname","h1"]`}, "query": {`["=","certname","h1"]`},
@@ -411,11 +444,11 @@ func TestHandler_ReportsDedupedByHash(t *testing.T) {
// A node reporting to both PuppetDBs mid-migration stores the same report // A node reporting to both PuppetDBs mid-migration stores the same report
// hash in each; the merged view must show it once. // hash in each; the merged view must show it once.
dup := report("h1", "r1", "2026-07-01T00:00:00Z") dup := report("h1", "r1", "2026-07-01T00:00:00Z")
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
old.bodies[reportsPath] = `[` + dup + `]` a.bodies[reportsPath] = `[` + dup + `]`
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
nw.bodies[reportsPath] = `[` + dup + `]` b.bodies[reportsPath] = `[` + dup + `]`
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), reportsPath, nil) rec := doGetParams(t, srv.Handler(), reportsPath, nil)
if got := hashes(t, rec.Body.Bytes()); !slices.Equal(got, []string{"r1"}) { if got := hashes(t, rec.Body.Bytes()); !slices.Equal(got, []string{"r1"}) {
@@ -424,15 +457,15 @@ func TestHandler_ReportsDedupedByHash(t *testing.T) {
} }
func TestHandler_ReportsPagedAcrossBackends(t *testing.T) { func TestHandler_ReportsPagedAcrossBackends(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
old.bodies[reportsPath] = `[` + report("h1", "r5", "2026-07-05T00:00:00Z") + `,` + a.bodies[reportsPath] = `[` + report("h1", "r5", "2026-07-05T00:00:00Z") + `,` +
report("h1", "r3", "2026-07-03T00:00:00Z") + `,` + report("h1", "r3", "2026-07-03T00:00:00Z") + `,` +
report("h1", "r1", "2026-07-01T00:00:00Z") + `]` report("h1", "r1", "2026-07-01T00:00:00Z") + `]`
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
nw.bodies[reportsPath] = `[` + report("h1", "r6", "2026-07-06T00:00:00Z") + `,` + b.bodies[reportsPath] = `[` + report("h1", "r6", "2026-07-06T00:00:00Z") + `,` +
report("h1", "r4", "2026-07-04T00:00:00Z") + `,` + report("h1", "r4", "2026-07-04T00:00:00Z") + `,` +
report("h1", "r2", "2026-07-02T00:00:00Z") + `]` report("h1", "r2", "2026-07-02T00:00:00Z") + `]`
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{ rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{
"order_by": {receiveDesc}, "order_by": {receiveDesc},
@@ -448,7 +481,7 @@ func TestHandler_ReportsPagedAcrossBackends(t *testing.T) {
} }
// Each backend must be asked for the first offset+limit records so the // Each backend must be asked for the first offset+limit records so the
// merged window is fully covered. // merged window is fully covered.
for name, fb := range map[string]*fakeBackend{"old": old, "new": nw} { for name, fb := range map[string]*fakeBackend{"a": a, "b": b} {
p, ok := fb.params(reportsPath) p, ok := fb.params(reportsPath)
if !ok { if !ok {
t.Fatalf("%s backend was not queried", name) t.Fatalf("%s backend was not queried", name)
@@ -463,13 +496,13 @@ func TestHandler_ReportsPagedAcrossBackends(t *testing.T) {
} }
func TestHandler_ReportsIncludeTotalSummed(t *testing.T) { func TestHandler_ReportsIncludeTotalSummed(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
old.bodies[reportsPath] = `[` + report("h1", "r1", "2026-07-01T00:00:00Z") + `]` a.bodies[reportsPath] = `[` + report("h1", "r1", "2026-07-01T00:00:00Z") + `]`
old.totals[reportsPath] = 40 a.totals[reportsPath] = 40
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
nw.bodies[reportsPath] = `[` + report("h1", "r2", "2026-07-02T00:00:00Z") + `]` b.bodies[reportsPath] = `[` + report("h1", "r2", "2026-07-02T00:00:00Z") + `]`
nw.totals[reportsPath] = 60 b.totals[reportsPath] = 60
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{ rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{
"include_total": {"true"}, "include_total": {"true"},
@@ -481,13 +514,13 @@ func TestHandler_ReportsIncludeTotalSummed(t *testing.T) {
} }
func TestHandler_ReportsNoTotalWhenNotRequested(t *testing.T) { func TestHandler_ReportsNoTotalWhenNotRequested(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
old.bodies[reportsPath] = `[]` a.bodies[reportsPath] = `[]`
old.totals[reportsPath] = 40 a.totals[reportsPath] = 40
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
nw.bodies[reportsPath] = `[]` b.bodies[reportsPath] = `[]`
nw.totals[reportsPath] = 60 b.totals[reportsPath] = 60
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), reportsPath, nil) rec := doGetParams(t, srv.Handler(), reportsPath, nil)
if got := rec.Header().Get(recordsHeader); got != "" { if got := rec.Header().Get(recordsHeader); got != "" {
@@ -496,9 +529,9 @@ func TestHandler_ReportsNoTotalWhenNotRequested(t *testing.T) {
} }
func TestHandler_ReportsBadPagingParam(t *testing.T) { func TestHandler_ReportsBadPagingParam(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
for _, params := range []url.Values{ for _, params := range []url.Values{
{"limit": {"lots"}}, {"limit": {"lots"}},
@@ -515,11 +548,11 @@ func TestHandler_ReportsBadPagingParam(t *testing.T) {
func TestHandler_EventsUnioned(t *testing.T) { func TestHandler_EventsUnioned(t *testing.T) {
// Puppetboard fetches a report's events as /events?query=["=","report",hash], // Puppetboard fetches a report's events as /events?query=["=","report",hash],
// and the report may live in either backend. // and the report may live in either backend.
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
old.bodies[eventsPath] = `[` + event("h1", "r1", "Package[nginx]") + `]` a.bodies[eventsPath] = `[` + event("h1", "r1", "Package[nginx]") + `]`
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
nw.bodies[eventsPath] = `[` + event("h1", "r2", "Service[nginx]") + `]` b.bodies[eventsPath] = `[` + event("h1", "r2", "Service[nginx]") + `]`
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGet(t, srv.Handler(), eventsPath, `["=","certname","h1"]`) rec := doGet(t, srv.Handler(), eventsPath, `["=","certname","h1"]`)
if rec.Code != http.StatusOK { if rec.Code != http.StatusOK {
@@ -533,11 +566,11 @@ func TestHandler_EventsUnioned(t *testing.T) {
func TestHandler_EventsDedupedByIdentity(t *testing.T) { func TestHandler_EventsDedupedByIdentity(t *testing.T) {
dup := event("h1", "r1", "Package[nginx]") dup := event("h1", "r1", "Package[nginx]")
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
old.bodies[eventsPath] = `[` + dup + `]` a.bodies[eventsPath] = `[` + dup + `]`
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
nw.bodies[eventsPath] = `[` + dup + `]` b.bodies[eventsPath] = `[` + dup + `]`
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGet(t, srv.Handler(), eventsPath, "") rec := doGet(t, srv.Handler(), eventsPath, "")
var got []json.RawMessage var got []json.RawMessage
@@ -550,27 +583,27 @@ func TestHandler_EventsDedupedByIdentity(t *testing.T) {
} }
func TestHandler_ReportSubResourceFromHoldingBackend(t *testing.T) { func TestHandler_ReportSubResourceFromHoldingBackend(t *testing.T) {
// Only old holds report r1, so its logs must come from old even though new // Only a holds report r1, so its logs come from a; an unmerged pass-through
// is the primary — a pass-through would have 404'd. // to whichever backend answered first could have 404'd.
const path = reportsPath + "/r1/logs" const path = reportsPath + "/r1/logs"
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
old.bodies[path] = `[{"level":"notice","message":"from-old"}]` a.bodies[path] = `[{"level":"notice","message":"from-a"}]`
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGet(t, srv.Handler(), path, "") rec := doGet(t, srv.Handler(), path, "")
if rec.Code != http.StatusOK { if rec.Code != http.StatusOK {
t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
} }
if !strings.Contains(rec.Body.String(), "from-old") { if !strings.Contains(rec.Body.String(), "from-a") {
t.Errorf("expected old's logs, got %s", rec.Body.String()) t.Errorf("expected the holding backend's logs, got %s", rec.Body.String())
} }
} }
func TestHandler_ReportSubResourceMissingEverywhere(t *testing.T) { func TestHandler_ReportSubResourceMissingEverywhere(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGet(t, srv.Handler(), reportsPath+"/nope/events", "") rec := doGet(t, srv.Handler(), reportsPath+"/nope/events", "")
if rec.Code != http.StatusNotFound { if rec.Code != http.StatusNotFound {
@@ -579,11 +612,11 @@ func TestHandler_ReportSubResourceMissingEverywhere(t *testing.T) {
} }
func TestHandler_ReportsOneBackendDown(t *testing.T) { func TestHandler_ReportsOneBackendDown(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
old.fail = true a.fail = true
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
nw.bodies[reportsPath] = `[` + report("h1", "r1", "2026-07-01T00:00:00Z") + `]` b.bodies[reportsPath] = `[` + report("h1", "r1", "2026-07-01T00:00:00Z") + `]`
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{"order_by": {receiveDesc}}) rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{"order_by": {receiveDesc}})
if rec.Code != http.StatusOK { if rec.Code != http.StatusOK {
@@ -621,11 +654,11 @@ func counts(t *testing.T, body []byte, field string) []float64 {
func TestHandler_EventCountsSummedPerSubject(t *testing.T) { func TestHandler_EventCountsSummedPerSubject(t *testing.T) {
// A node reporting to both PuppetDBs has its run counted in each; the // A node reporting to both PuppetDBs has its run counted in each; the
// merged view is the sum, not two rows. // merged view is the sum, not two rows.
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
old.bodies[eventCountsPath] = `[` + eventCount("h1", 4, 3, 1, 0) + `]` a.bodies[eventCountsPath] = `[` + eventCount("h1", 4, 3, 1, 0) + `]`
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
nw.bodies[eventCountsPath] = `[` + eventCount("h1", 2, 1, 0, 0) + `,` + eventCount("h2", 5, 0, 0, 0) + `]` b.bodies[eventCountsPath] = `[` + eventCount("h1", 2, 1, 0, 0) + `,` + eventCount("h2", 5, 0, 0, 0) + `]`
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), eventCountsPath, url.Values{ rec := doGetParams(t, srv.Handler(), eventCountsPath, url.Values{
"query": {`["=","certname","h1"]`}, "query": {`["=","certname","h1"]`},
@@ -634,7 +667,7 @@ func TestHandler_EventCountsSummedPerSubject(t *testing.T) {
if rec.Code != http.StatusOK { if rec.Code != http.StatusOK {
t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
} }
// Precedence puts new (prefer) first, so h1 leads. // Rows come out in configured backend order, so h1 leads.
if got := counts(t, rec.Body.Bytes(), "successes"); !slices.Equal(got, []float64{6, 5}) { if got := counts(t, rec.Body.Bytes(), "successes"); !slices.Equal(got, []float64{6, 5}) {
t.Errorf("successes = %v, want [6 5]", got) t.Errorf("successes = %v, want [6 5]", got)
} }
@@ -644,18 +677,18 @@ func TestHandler_EventCountsSummedPerSubject(t *testing.T) {
} }
func TestHandler_EventCountsDisjointSubjectsPassThrough(t *testing.T) { func TestHandler_EventCountsDisjointSubjectsPassThrough(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
old.bodies[eventCountsPath] = `[` + eventCount("h1", 1, 0, 0, 0) + `]` a.bodies[eventCountsPath] = `[` + eventCount("h1", 1, 0, 0, 0) + `]`
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
nw.bodies[eventCountsPath] = `[` + eventCount("h2", 2, 0, 0, 0) + `]` b.bodies[eventCountsPath] = `[` + eventCount("h2", 2, 0, 0, 0) + `]`
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), eventCountsPath, url.Values{"summarize_by": {"certname"}}) rec := doGetParams(t, srv.Handler(), eventCountsPath, url.Values{"summarize_by": {"certname"}})
if got := counts(t, rec.Body.Bytes(), "successes"); !slices.Equal(got, []float64{2, 1}) { if got := counts(t, rec.Body.Bytes(), "successes"); !slices.Equal(got, []float64{1, 2}) {
t.Errorf("successes = %v, want [2 1] (both nodes, untouched)", got) t.Errorf("successes = %v, want [1 2] (both nodes, untouched)", got)
} }
// summarize_by must reach the backends verbatim. // summarize_by must reach the backends verbatim.
for name, fb := range map[string]*fakeBackend{"old": old, "new": nw} { for name, fb := range map[string]*fakeBackend{"a": a, "b": b} {
p, _ := fb.params(eventCountsPath) p, _ := fb.params(eventCountsPath)
if p.Get("summarize_by") != "certname" { if p.Get("summarize_by") != "certname" {
t.Errorf("%s backend got summarize_by=%q, want certname", name, p.Get("summarize_by")) t.Errorf("%s backend got summarize_by=%q, want certname", name, p.Get("summarize_by"))
@@ -666,13 +699,13 @@ func TestHandler_EventCountsDisjointSubjectsPassThrough(t *testing.T) {
func TestHandler_EventCountsRecordsIsMergedRowCount(t *testing.T) { func TestHandler_EventCountsRecordsIsMergedRowCount(t *testing.T) {
// Each backend reports one row; they share a subject, so the merged total // Each backend reports one row; they share a subject, so the merged total
// is one — not the two the backends' own X-Records add up to. // is one — not the two the backends' own X-Records add up to.
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
old.bodies[eventCountsPath] = `[` + eventCount("h1", 1, 0, 0, 0) + `]` a.bodies[eventCountsPath] = `[` + eventCount("h1", 1, 0, 0, 0) + `]`
old.totals[eventCountsPath] = 1 a.totals[eventCountsPath] = 1
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
nw.bodies[eventCountsPath] = `[` + eventCount("h1", 2, 0, 0, 0) + `]` b.bodies[eventCountsPath] = `[` + eventCount("h1", 2, 0, 0, 0) + `]`
nw.totals[eventCountsPath] = 1 b.totals[eventCountsPath] = 1
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), eventCountsPath, url.Values{ rec := doGetParams(t, srv.Handler(), eventCountsPath, url.Values{
"summarize_by": {"certname"}, "summarize_by": {"certname"},
@@ -684,13 +717,13 @@ func TestHandler_EventCountsRecordsIsMergedRowCount(t *testing.T) {
} }
func TestHandler_AggregateEventCountsSummed(t *testing.T) { func TestHandler_AggregateEventCountsSummed(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
old.bodies[aggregateEventCountsPath] = a.bodies[aggregateEventCountsPath] =
`[{"successes":2,"failures":1,"noops":0,"skips":3,"total":6,"summarize_by":"certname"}]` `[{"successes":2,"failures":1,"noops":0,"skips":3,"total":6,"summarize_by":"certname"}]`
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
nw.bodies[aggregateEventCountsPath] = b.bodies[aggregateEventCountsPath] =
`[{"successes":5,"failures":4,"noops":1,"skips":0,"total":10,"summarize_by":"certname"}]` `[{"successes":5,"failures":4,"noops":1,"skips":0,"total":10,"summarize_by":"certname"}]`
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), aggregateEventCountsPath, url.Values{"summarize_by": {"certname"}}) rec := doGetParams(t, srv.Handler(), aggregateEventCountsPath, url.Values{"summarize_by": {"certname"}})
if rec.Code != http.StatusOK { if rec.Code != http.StatusOK {
@@ -715,13 +748,13 @@ func TestHandler_AggregateEventCountsSummed(t *testing.T) {
func TestHandler_AggregateEventCountsNullColumnSurvives(t *testing.T) { func TestHandler_AggregateEventCountsNullColumnSurvives(t *testing.T) {
// PuppetDB returns null totals for an empty result set; summing must not // PuppetDB returns null totals for an empty result set; summing must not
// crash or blank out the backend that does have numbers. // crash or blank out the backend that does have numbers.
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
old.bodies[aggregateEventCountsPath] = a.bodies[aggregateEventCountsPath] =
`[{"successes":null,"failures":null,"total":null,"summarize_by":"certname"}]` `[{"successes":null,"failures":null,"total":null,"summarize_by":"certname"}]`
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
nw.bodies[aggregateEventCountsPath] = b.bodies[aggregateEventCountsPath] =
`[{"successes":3,"failures":0,"total":3,"summarize_by":"certname"}]` `[{"successes":3,"failures":0,"total":3,"summarize_by":"certname"}]`
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), aggregateEventCountsPath, url.Values{"summarize_by": {"certname"}}) rec := doGetParams(t, srv.Handler(), aggregateEventCountsPath, url.Values{"summarize_by": {"certname"}})
if rec.Code != http.StatusOK { if rec.Code != http.StatusOK {
@@ -737,11 +770,11 @@ const statusCountQuery = `["extract",[["function","count"],"status"],["~","certn
func TestHandler_ReportsAggregateSummed(t *testing.T) { func TestHandler_ReportsAggregateSummed(t *testing.T) {
// Puppetboard's daily-reports chart: each backend counts only its own // Puppetboard's daily-reports chart: each backend counts only its own
// reports, so the merged chart needs the per-status sums. // reports, so the merged chart needs the per-status sums.
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
old.bodies[reportsPath] = `[{"count":4,"status":"changed"},{"count":2,"status":"failed"}]` a.bodies[reportsPath] = `[{"count":4,"status":"changed"},{"count":2,"status":"failed"}]`
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
nw.bodies[reportsPath] = `[{"count":3,"status":"changed"},{"count":9,"status":"unchanged"}]` b.bodies[reportsPath] = `[{"count":3,"status":"changed"},{"count":9,"status":"unchanged"}]`
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{"query": {statusCountQuery}}) rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{"query": {statusCountQuery}})
if rec.Code != http.StatusOK { if rec.Code != http.StatusOK {
@@ -764,13 +797,13 @@ func TestHandler_ReportsAggregateSummed(t *testing.T) {
} }
func TestHandler_ReportsAggregateRecordsIsMergedRowCount(t *testing.T) { func TestHandler_ReportsAggregateRecordsIsMergedRowCount(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
old.bodies[reportsPath] = `[{"count":4,"status":"changed"}]` a.bodies[reportsPath] = `[{"count":4,"status":"changed"}]`
old.totals[reportsPath] = 1 a.totals[reportsPath] = 1
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
nw.bodies[reportsPath] = `[{"count":3,"status":"changed"}]` b.bodies[reportsPath] = `[{"count":3,"status":"changed"}]`
nw.totals[reportsPath] = 1 b.totals[reportsPath] = 1
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{ rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{
"query": {statusCountQuery}, "query": {statusCountQuery},
@@ -784,11 +817,11 @@ func TestHandler_ReportsAggregateRecordsIsMergedRowCount(t *testing.T) {
func TestHandler_ReportsNonAggregateStillUnioned(t *testing.T) { func TestHandler_ReportsNonAggregateStillUnioned(t *testing.T) {
// An extract with no function is a projection of real reports, so the // An extract with no function is a projection of real reports, so the
// union — not a sum — is still the right merge. // union — not a sum — is still the right merge.
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
old.bodies[reportsPath] = `[` + report("h1", "r1", "2026-07-01T00:00:00Z") + `]` a.bodies[reportsPath] = `[` + report("h1", "r1", "2026-07-01T00:00:00Z") + `]`
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
nw.bodies[reportsPath] = `[` + report("h1", "r2", "2026-07-02T00:00:00Z") + `]` b.bodies[reportsPath] = `[` + report("h1", "r2", "2026-07-02T00:00:00Z") + `]`
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{ rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{
"query": {`["extract",["hash","certname"],["=","certname","h1"]]`}, "query": {`["extract",["hash","certname"],["=","certname","h1"]]`},
@@ -800,11 +833,11 @@ func TestHandler_ReportsNonAggregateStillUnioned(t *testing.T) {
} }
func TestHandler_EventCountsOneBackendDown(t *testing.T) { func TestHandler_EventCountsOneBackendDown(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
old.fail = true a.fail = true
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
nw.bodies[eventCountsPath] = `[` + eventCount("h1", 2, 0, 0, 0) + `]` b.bodies[eventCountsPath] = `[` + eventCount("h1", 2, 0, 0, 0) + `]`
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), eventCountsPath, url.Values{"summarize_by": {"certname"}}) rec := doGetParams(t, srv.Handler(), eventCountsPath, url.Values{"summarize_by": {"certname"}})
if rec.Code != http.StatusOK { if rec.Code != http.StatusOK {
@@ -816,9 +849,9 @@ func TestHandler_EventCountsOneBackendDown(t *testing.T) {
} }
func TestHandler_EventCountsBadPagingParam(t *testing.T) { func TestHandler_EventCountsBadPagingParam(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`) a := newFakeBackend(t, `[]`, `[]`)
nw := newFakeBackend(t, `[]`, `[]`) b := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), eventCountsPath, url.Values{"limit": {"lots"}}) rec := doGetParams(t, srv.Handler(), eventCountsPath, url.Values{"limit": {"lots"}})
if rec.Code != http.StatusBadRequest { if rec.Code != http.StatusBadRequest {