config: drop primary/prefer and treat all backends equally #9

Merged
benvin merged 2 commits from benvin/drop-primary into main 2026-09-05 16:07:37 +10:00
12 changed files with 426 additions and 435 deletions
+24 -31
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
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
backends being merged during a migration:
moment, and consumers have to know which, or query each in turn. `pdbmux`
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`
- **new** — the PuppetDB nodes are moving on to, e.g. `http://puppetdb2.example.com:8080`
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.
All backends are equal — `pdbmux` is never told which one to favour. Backend
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.
## 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/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/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/*` (any other) | Transparently proxied to the **primary** backend, unmerged, streamed verbatim. |
| `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) | 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. |
Fan-out is concurrent. If one backend errors or times out, `pdbmux` serves the
@@ -44,25 +42,24 @@ unknown fields survive untouched.
## Merge semantics
- **`/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),
the **preferred** backend's record is kept.
`report_timestamp` wins. On a tie, the backend listed first in `backends`
supplies the record — a tie-break only, so the merged output is deterministic.
- **`/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
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
holds its newer `report_timestamp`. `pdbmux` derives this from a per-certname
freshness map built by querying `/nodes` from every backend, cached for
`freshness_ttl` (default 30s). Ties/fallbacks use `prefer`.
- **`static`** — always keep the `prefer` backend's facts for shared nodes.
No extra `/nodes` query.
`freshness_ttl` (default 30s).
- **`static`** — skip the extra `/nodes` query and take each shared node's
facts from the first backend in configured order that holds it.
- A node present in only one backend always appears (falls back to whichever
backend actually returned facts for it).
- **`/reports`, `/events`** — **union**, not a per-node winner. Reports are
immutable history, so a node that migrated legitimately has reports in the old
PuppetDB *and* the new one and both belong in the merged view. Reports dedupe
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
each).
immutable history, so a node's reports can legitimately exist in more than one
backend and all of them belong in the merged view. Reports dedupe on `hash`;
events, which carry no id of their own, dedupe on the verbatim record (a node
reporting to more than one backend stores identical records in each).
- **Aggregates** — `extract`/`group_by` rows are counts, not records, so each
backend returns a partial answer that has to be **added**, not deduped. This
covers `/event-counts`, `/aggregate-event-counts`, and a `/reports` query whose
@@ -86,7 +83,7 @@ Each backend applies `order_by`/`limit`/`offset` to its own slice only, so
`pdbmux` re-does all three over the union:
- `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`
— 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
@@ -112,14 +109,12 @@ or the paths it searched.
```yaml
listen: ":8080"
backends:
- name: old
backends: # order is a tie-break only, not a ranking
- name: pdb-a
url: http://puppetdb1.example.com:8080
- name: new
- name: pdb-b
url: https://puppetdb2.example.com
primary: new # backend used for non-merged /pdb/query/v4/* pass-through
merge: freshness # freshness | static
prefer: new # winner on ties / static merge / fallback
timeout: 10s # per-upstream request timeout
freshness_ttl: 30s # freshness-map cache TTL (freshness merge only)
```
@@ -131,14 +126,12 @@ the `/pdb/query/v4/...` path per request.
|---|---|
| `PDBMUX_CONFIG` | config file path (not a file key) |
| `PDBMUX_LISTEN` | `listen` |
| `PDBMUX_PRIMARY` | `primary` |
| `PDBMUX_MERGE` | `merge` |
| `PDBMUX_PREFER` | `prefer` |
| `PDBMUX_TIMEOUT` | `timeout` (Go duration, e.g. `10s`) |
| `PDBMUX_FRESHNESS_TTL` | `freshness_ttl` |
| `PDBMUX_BACKENDS` | whole backend list, as `name=url,name=url` |
Flags: `--config`, `--listen`, `--primary`, `--merge`.
Flags: `--config`, `--listen`, `--merge`.
`config init` writes to `--config`/`PDBMUX_CONFIG` when set, else to
`$XDG_CONFIG_HOME/pdbmux/config.yaml`.
@@ -150,7 +143,7 @@ Subcommands: `serve` (default), `config init`, `config show`, `version`. Run
base URL in place of a PuppetDB one.
```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 \
--data-urlencode 'query=["=","certname","host1.example.com"]'
```
+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
// 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
// first backend's value rather than being coerced to zero. results must be
// ordered by precedence; output keeps first-seen order.
// earlier backend's value rather than being coerced to zero. results come in
// 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 {
type slot struct {
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.
// 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
// field is distinct from any present value.
func groupKey(row map[string]json.RawMessage, keys []string) string {
+16 -16
View File
@@ -73,8 +73,8 @@ func TestParseAggregate_NoFunctionIsNotAggregate(t *testing.T) {
func TestSumRows_SharedKeysAreAdded(t *testing.T) {
spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
merged := sumRows([]backendResult{
{name: "new", records: rows(`{"count":3,"status":"changed"}`, `{"count":1,"status":"failed"}`)},
{name: "old", records: rows(`{"count":4,"status":"changed"}`, `{"count":2,"status":"failed"}`)},
{name: "a", records: rows(`{"count":3,"status":"changed"}`, `{"count":1,"status":"failed"}`)},
{name: "b", records: rows(`{"count":4,"status":"changed"}`, `{"count":2,"status":"failed"}`)},
}, spec.columns)
want := []map[string]any{
@@ -89,8 +89,8 @@ func TestSumRows_SharedKeysAreAdded(t *testing.T) {
func TestSumRows_DisjointKeysAreKept(t *testing.T) {
spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
merged := sumRows([]backendResult{
{name: "new", records: rows(`{"count":3,"status":"changed"}`)},
{name: "old", records: rows(`{"count":2,"status":"skipped"}`)},
{name: "a", records: rows(`{"count":3,"status":"changed"}`)},
{name: "b", records: rows(`{"count":2,"status":"skipped"}`)},
}, spec.columns)
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"]]`)
const raw = `{"count":3,"status":"changed","extra":{"kept":true}}`
merged := sumRows([]backendResult{
{name: "new", records: rows(raw)},
{name: "old", records: nil},
{name: "a", records: rows(raw)},
{name: "b", records: nil},
}, spec.columns)
if len(merged) != 1 || string(merged[0]) != raw {
@@ -118,8 +118,8 @@ func TestSumRows_SingleBackendRowKeepsUpstreamBytes(t *testing.T) {
func TestSumRows_NonNumericAggregateColumnIsNotZeroed(t *testing.T) {
spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
merged := sumRows([]backendResult{
{name: "new", records: rows(`{"count":5,"status":"changed"}`)},
{name: "old", records: rows(`{"count":null,"status":"changed"}`)},
{name: "a", records: rows(`{"count":5,"status":"changed"}`)},
{name: "b", records: rows(`{"count":null,"status":"changed"}`)},
}, spec.columns)
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) {
spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
merged := sumRows([]backendResult{
{name: "new", records: rows(`{"status":"changed"}`)},
{name: "old", records: rows(`{"count":6,"status":"changed"}`)},
{name: "a", records: rows(`{"status":"changed"}`)},
{name: "b", records: rows(`{"count":6,"status":"changed"}`)},
}, spec.columns)
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) {
spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`)
merged := sumRows([]backendResult{
{name: "new", records: rows(`"surprise"`)},
{name: "old", records: rows(`{"count":1,"status":"changed"}`)},
{name: "a", records: rows(`"surprise"`)},
{name: "b", records: rows(`{"count":1,"status":"changed"}`)},
}, spec.columns)
if len(merged) != 2 || string(merged[0]) != `"surprise"` {
@@ -161,8 +161,8 @@ func TestSumRows_NoKeyColumnsCollapseToOneRow(t *testing.T) {
t.Fatal("expected an aggregate spec")
}
merged := sumRows([]backendResult{
{name: "new", records: rows(`{"count":10}`)},
{name: "old", records: rows(`{"count":32}`)},
{name: "a", records: rows(`{"count":10}`)},
{name: "b", records: rows(`{"count":32}`)},
}, spec.columns)
want := []map[string]any{{"count": float64(42)}}
@@ -188,11 +188,11 @@ func TestInferredColumns_SplitsCountsFromIdentity(t *testing.T) {
func TestSumRows_EventCountsPerSubject(t *testing.T) {
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":"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}`,
)},
}, inferredColumns)
+5 -43
View File
@@ -28,8 +28,8 @@ const (
)
var exampleBackends = []Backend{
{Name: "primary", URL: "http://puppetdb1.example.com:8080"},
{Name: "secondary", URL: "http://puppetdb2.example.com:8080"},
{Name: "pdb-a", URL: "http://puppetdb1.example.com:8080"},
{Name: "pdb-b", URL: "http://puppetdb2.example.com:8080"},
}
type Backend struct {
@@ -39,10 +39,8 @@ type Backend struct {
type Config struct {
Listen string `yaml:"listen"`
Backends []Backend `yaml:"backends"`
Primary string `yaml:"primary"`
Backends []Backend `yaml:"backends"` // all equal; order is only a deterministic tie-break
Merge string `yaml:"merge"`
Prefer string `yaml:"prefer"` // wins under static merge, and breaks ties under freshness merge
Timeout time.Duration `yaml:"timeout"`
FreshnessTTL time.Duration `yaml:"freshness_ttl"`
@@ -69,8 +67,6 @@ func DefaultConfig() Config {
func ExampleConfig() Config {
cfg := DefaultConfig()
cfg.Backends = append([]Backend(nil), exampleBackends...)
cfg.Primary = exampleBackends[0].Name
cfg.Prefer = exampleBackends[0].Name
return cfg
}
@@ -143,7 +139,6 @@ func Load(flagPath string) (Config, error) {
}
applyEnv(&cfg, os.Getenv)
cfg.normalize()
return cfg, nil
}
@@ -151,15 +146,9 @@ func applyEnv(cfg *Config, getenv func(string) string) {
if v := getenv(envPrefix + "LISTEN"); v != "" {
cfg.Listen = v
}
if v := getenv(envPrefix + "PRIMARY"); v != "" {
cfg.Primary = v
}
if v := getenv(envPrefix + "MERGE"); v != "" {
cfg.Merge = v
}
if v := getenv(envPrefix + "PREFER"); v != "" {
cfg.Prefer = v
}
if v := getenv(envPrefix + "TIMEOUT"); v != "" {
if d, err := time.ParseDuration(v); err == nil {
cfg.Timeout = d
@@ -195,18 +184,6 @@ func parseBackends(s string) []Backend {
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
}
}
// configHint names the file a user should edit: the one actually loaded, else
// the default write target.
func (c Config) configHint() string {
@@ -231,32 +208,17 @@ func (c Config) Validate() error {
}
seen[b.Name] = true
}
if !seen[c.Primary] {
return fmt.Errorf("primary %q is not a configured backend", c.Primary)
}
switch c.Merge {
case mergeFreshness, mergeStatic:
default:
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 {
return fmt.Errorf("timeout must be positive")
}
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(path string) error {
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
return fmt.Errorf("creating config dir: %w", err)
@@ -268,8 +230,8 @@ func writeDefaultConfig(path string) error {
header := []byte("# pdbmux configuration\n" +
"# A merging proxy presenting one PuppetDB v4 query surface over several\n" +
"# PuppetDB backends. The backend URLs below are placeholders — edit them.\n" +
"# Env overrides: PDBMUX_LISTEN, PDBMUX_PRIMARY, PDBMUX_MERGE, PDBMUX_PREFER,\n" +
"# PDBMUX_TIMEOUT, PDBMUX_FRESHNESS_TTL, PDBMUX_BACKENDS (name=url,name=url).\n\n")
"# Env overrides: PDBMUX_LISTEN, PDBMUX_MERGE, PDBMUX_TIMEOUT,\n" +
"# PDBMUX_FRESHNESS_TTL, PDBMUX_BACKENDS (name=url,name=url).\n\n")
if err := os.WriteFile(path, append(header, data...), 0o644); err != nil {
return fmt.Errorf("writing config: %w", err)
}
+16 -26
View File
@@ -14,11 +14,9 @@ import (
func testConfigValid() Config {
cfg := DefaultConfig()
cfg.Backends = []Backend{
{Name: "old", URL: "http://localhost:18080"},
{Name: "new", URL: "http://localhost:18081"},
{Name: "a", URL: "http://localhost:18080"},
{Name: "b", URL: "http://localhost:18081"},
}
cfg.Primary = "new"
cfg.Prefer = "new"
return cfg
}
@@ -56,7 +54,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())
clearEnv(t)
t.Setenv(envPrefix+"BACKENDS", "a=http://localhost:18080,b=http://localhost:18081")
@@ -65,8 +63,11 @@ func TestLoad_PrimaryDefaultsToFirstBackend(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if cfg.Primary != "a" || cfg.Prefer != "a" {
t.Errorf("primary/prefer should default to first backend, got %q/%q", cfg.Primary, cfg.Prefer)
if len(cfg.Backends) != 2 || cfg.Backends[0].Name != "a" || cfg.Backends[1].Name != "b" {
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)
}
}
@@ -79,8 +80,8 @@ func TestLoad_FileAndEnvOverride(t *testing.T) {
if err := os.MkdirAll(cfgDir, 0o755); err != nil {
t.Fatal(err)
}
body := "listen: :9999\nmerge: static\nprimary: old\nprefer: old\n" +
"backends:\n - name: old\n url: http://localhost:18080\n - name: new\n url: http://localhost:18081\n"
body := "listen: :9999\nmerge: static\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 {
t.Fatal(err)
}
@@ -95,8 +96,8 @@ func TestLoad_FileAndEnvOverride(t *testing.T) {
if cfg.Listen != "127.0.0.1:1234" {
t.Errorf("env should beat file for listen, got %q", cfg.Listen)
}
if cfg.Merge != mergeStatic || cfg.Primary != "old" {
t.Errorf("file override failed: merge=%s primary=%s", cfg.Merge, cfg.Primary)
if cfg.Merge != mergeStatic {
t.Errorf("file override failed: merge=%s", cfg.Merge)
}
}
@@ -104,8 +105,6 @@ func TestApplyEnv_Backends(t *testing.T) {
cfg := testConfigValid()
env := map[string]string{
envPrefix + "BACKENDS": "a=http://a:8080,b=http://b:8080",
envPrefix + "PRIMARY": "a",
envPrefix + "PREFER": "a",
envPrefix + "TIMEOUT": "3s",
envPrefix + "FRESHNESS_TTL": "45s",
}
@@ -127,10 +126,8 @@ func TestValidate(t *testing.T) {
}{
{"ok", func(*Config) {}, false},
{"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},
{"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},
{"zero timeout", func(c *Config) { c.Timeout = 0 }, true},
}
@@ -156,13 +153,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) {
cfg := ExampleConfig()
if err := cfg.Validate(); err != nil {
@@ -175,8 +165,8 @@ func TestExampleConfig_IsValidAndNeutral(t *testing.T) {
}
}
const testConfigBody = "listen: \":9999\"\nmerge: static\nprimary: old\nprefer: old\n" +
"backends:\n - name: old\n url: http://localhost:18080\n - name: new\n url: http://localhost:18081\n"
const testConfigBody = "listen: \":9999\"\nmerge: static\n" +
"backends:\n - name: a\n url: http://localhost:18080\n - name: b\n url: http://localhost:18081\n"
func writeConfigFile(t *testing.T, path string) {
t.Helper()
@@ -401,7 +391,7 @@ func captureStdout(t *testing.T, f func()) string {
func clearEnv(t *testing.T) {
t.Helper()
for _, k := range []string{"CONFIG", "LISTEN", "PRIMARY", "MERGE", "PREFER", "TIMEOUT", "FRESHNESS_TTL", "BACKENDS"} {
for _, k := range []string{"CONFIG", "LISTEN", "MERGE", "TIMEOUT", "FRESHNESS_TTL", "BACKENDS"} {
t.Setenv(envPrefix+k, "")
}
}
+2 -9
View File
@@ -23,7 +23,6 @@ func main() {
cfg Config
configPath string
listen string
primary string
merge string
)
@@ -44,9 +43,6 @@ func main() {
if cmd.Flags().Changed("listen") {
cfg.Listen = listen
}
if cmd.Flags().Changed("primary") {
cfg.Primary = primary
}
if cmd.Flags().Changed("merge") {
cfg.Merge = merge
}
@@ -70,7 +66,6 @@ func main() {
pf := root.PersistentFlags()
pf.StringVar(&configPath, "config", "", "Config file path (overrides PDBMUX_CONFIG and the default search path)")
pf.StringVar(&listen, "listen", defaultListen, "HTTP listen address (overrides config and PDBMUX_LISTEN)")
pf.StringVar(&primary, "primary", "", "Primary backend name for non-merged pass-through")
pf.StringVar(&merge, "merge", mergeFreshness, "Facts merge strategy: freshness or static")
serveCmd := &cobra.Command{
@@ -132,8 +127,8 @@ func runServer(cfg Config) error {
ReadHeaderTimeout: 10 * time.Second,
}
logger.Printf("listening on %s (merge=%s primary=%s backends=%d)",
cfg.Listen, cfg.Merge, cfg.Primary, len(cfg.Backends))
logger.Printf("listening on %s (merge=%s backends=%d)",
cfg.Listen, cfg.Merge, len(cfg.Backends))
errCh := make(chan error, 1)
go func() {
@@ -163,9 +158,7 @@ func printConfig(cfg Config) {
fmt.Printf("config file : none loaded (searched %s)\n", strings.Join(configSearchPaths(), ", "))
}
fmt.Printf("listen : %s\n", cfg.Listen)
fmt.Printf("primary : %s\n", cfg.Primary)
fmt.Printf("merge : %s\n", cfg.Merge)
fmt.Printf("prefer : %s\n", cfg.Prefer)
fmt.Printf("timeout : %s\n", durationString(cfg.Timeout))
fmt.Printf("freshness_ttl: %s\n", durationString(cfg.FreshnessTTL))
fmt.Println("backends:")
+8 -5
View File
@@ -49,7 +49,7 @@ func parseTimestamp(s string) 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 {
type pick struct {
raw json.RawMessage
@@ -81,7 +81,7 @@ func mergeNodes(results []backendResult) []json.RawMessage {
// certname -> name of the backend holding that node's newest report.
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 {
type pick struct {
backend string
@@ -104,9 +104,9 @@ func buildFreshness(results []backendResult) freshness {
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 {
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{}
for _, res := range results {
for _, rec := range res.records {
@@ -133,7 +133,10 @@ func mergeFacts(results []backendResult, owner func(certname string) string) []j
out := []json.RawMessage{}
for _, cn := range order {
backends := present[cn]
chosen := owner(cn)
chosen := ""
if owner != nil {
chosen = owner(cn)
}
if !contains(backends, chosen) {
chosen = backends[0]
}
+59 -58
View File
@@ -89,10 +89,10 @@ func event(cn, reportHash, resource string) string {
}
func TestMergeNodes_NewerWins(t *testing.T) {
old := recs(t, "old", 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"))
a := recs(t, "a", node("h1", "2026-07-01T00:00:00Z"), node("h2", "2026-07-10T00: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{}
for _, r := range merged {
var m recordMeta
@@ -100,13 +100,13 @@ func TestMergeNodes_NewerWins(t *testing.T) {
got[m.Certname] = m.ReportTimestamp
}
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" {
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" {
t.Errorf("h3: only in new, got %s", got["h3"])
t.Errorf("h3: only in b, got %s", got["h3"])
}
if len(merged) != 3 {
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) {
old := recs(t, "old", node("h1", "2026-07-01T00:00:00Z"))
// new returned nothing (e.g. empty result).
nw := backendResult{name: "new"}
merged := mergeNodes([]backendResult{old, nw})
a := recs(t, "a", node("h1", "2026-07-01T00:00:00Z"))
// b returned nothing (e.g. empty result).
b := backendResult{name: "b"}
merged := mergeNodes([]backendResult{a, b})
if len(merged) != 1 || certnames(t, merged)[0] != "h1" {
t.Fatalf("expected only h1, got %v", certnames(t, merged))
}
}
func TestMergeNodes_TieKeepsEarlierBackend(t *testing.T) {
// Equal timestamps: the backend listed first (precedence) wins.
prefer := recs(t, "new", node("h1", "2026-07-01T00:00:00Z"))
other := recs(t, "old", node("h1", "2026-07-01T00:00:00Z"))
merged := mergeNodes([]backendResult{prefer, other})
// Equal timestamps: the backend listed first wins, as a tie-break.
first := recs(t, "b", node("h1", "2026-07-01T00:00:00Z"))
second := recs(t, "a", node("h1", "2026-07-01T00:00:00Z"))
merged := mergeNodes([]backendResult{first, second})
if len(merged) != 1 {
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) {
old := recs(t, "old", `{"certname":"h1","report_timestamp":"2026-07-01T00:00:00Z","extra":{"deep":42}}`)
merged := mergeNodes([]backendResult{old})
a := recs(t, "a", `{"certname":"h1","report_timestamp":"2026-07-01T00:00:00Z","extra":{"deep":42}}`)
merged := mergeNodes([]backendResult{a})
if len(merged) != 1 {
t.Fatalf("expected 1 record")
}
@@ -150,69 +150,70 @@ func TestMergeNodes_PreservesUnknownFields(t *testing.T) {
}
}
func TestMergeFacts_Static_PreferWins(t *testing.T) {
// h1 in both; static prefer=new -> new's facts kept, old's dropped.
old := recs(t, "old", fact("h1", "role", "web-old", ""), fact("h2", "role", "db-old", ""))
nw := recs(t, "new", fact("h1", "role", "web-new", ""))
func TestMergeFacts_NilOwnerUsesConfiguredOrder(t *testing.T) {
// Static merge passes no owner: h1 is in both, so the first backend in the
// slice supplies its facts.
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)
assertContains(t, got, "h1:role=web-new")
assertNotContains(t, got, "h1:role=web-old")
// h2 only in old -> falls back to old.
assertContains(t, got, "h2:role=db-old")
assertContains(t, got, "h1:role=web-b")
assertNotContains(t, got, "h1:role=web-a")
// h2 only in a -> still served from a.
assertContains(t, got, "h2:role=db-a")
}
func TestMergeFacts_Freshness_NewerBackendWins(t *testing.T) {
// owner map says h1 belongs to old (older backend has the newer report),
// h2 belongs to new. Multiple facts per node must all come from the winner.
old := recs(t, "old",
fact("h1", "role", "web-old", ""), fact("h1", "ip", "10.0.0.1", ""),
fact("h2", "role", "db-old", ""))
nw := recs(t, "new",
fact("h1", "role", "web-new", ""), fact("h1", "ip", "10.9.9.9", ""),
fact("h2", "role", "db-new", ""), fact("h2", "ip", "10.0.0.2", ""))
// owner map says h1 belongs to a and h2 to b. Multiple facts per node must
// all come from the winner.
a := recs(t, "a",
fact("h1", "role", "web-a", ""), fact("h1", "ip", "10.0.0.1", ""),
fact("h2", "role", "db-a", ""))
b := recs(t, "b",
fact("h1", "role", "web-b", ""), fact("h1", "ip", "10.9.9.9", ""),
fact("h2", "role", "db-b", ""), fact("h2", "ip", "10.0.0.2", ""))
owner := func(cn string) string {
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)
// h1 -> all old facts, no new facts.
assertContains(t, got, "h1:role=web-old")
// h1 -> all a facts, no b facts.
assertContains(t, got, "h1:role=web-a")
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")
// h2 -> all new facts.
assertContains(t, got, "h2:role=db-new")
// h2 -> all b facts.
assertContains(t, got, "h2:role=db-b")
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) {
// owner returns a backend with no facts for h1 -> fall back to first
// backend present (precedence order of the slice).
prefer := recs(t, "new", fact("h1", "role", "web-new", ""))
other := recs(t, "old", fact("h1", "role", "web-old", ""))
merged := mergeFacts([]backendResult{prefer, other}, func(string) string { return "ghost" })
func TestMergeFacts_OwnerMissingFallsBackToConfiguredOrder(t *testing.T) {
// owner returns a backend with no facts for h1 -> fall back to the first
// backend in the slice that has some.
first := recs(t, "b", fact("h1", "role", "web-b", ""))
second := recs(t, "a", fact("h1", "role", "web-a", ""))
merged := mergeFacts([]backendResult{first, second}, func(string) string { return "ghost" })
got := factValues(t, merged)
assertContains(t, got, "h1:role=web-new") // new is first in slice
assertNotContains(t, got, "h1:role=web-old")
assertContains(t, got, "h1:role=web-b") // b is first in slice
assertNotContains(t, got, "h1:role=web-a")
}
func TestBuildFreshness(t *testing.T) {
// old has newer report for h1; new has newer for h2.
old := recs(t, "old", 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"))
f := buildFreshness([]backendResult{old, nw})
if f["h1"] != "old" {
t.Errorf("h1 should belong to old, got %q", f["h1"])
// a holds the newer report for h1; b holds the newer one for h2.
a := recs(t, "a", node("h1", "2026-07-20T00:00:00Z"), node("h2", "2026-07-01T00:00:00Z"))
b := recs(t, "b", node("h1", "2026-07-01T00:00:00Z"), node("h2", "2026-07-20T00:00:00Z"))
f := buildFreshness([]backendResult{a, b})
if f["h1"] != "a" {
t.Errorf("h1 should belong to a, got %q", f["h1"])
}
if f["h2"] != "new" {
t.Errorf("h2 should belong to new, got %q", f["h2"])
if f["h2"] != "b" {
t.Errorf("h2 should belong to b, got %q", f["h2"])
}
}
+2 -2
View File
@@ -9,7 +9,7 @@ import (
"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 {
seen := make(map[string]bool)
out := []json.RawMessage{}
@@ -65,7 +65,7 @@ func parseOrderBy(s string) ([]orderField, error) {
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) {
if len(order) == 0 || len(recs) < 2 {
return
+10 -10
View File
@@ -8,17 +8,17 @@ import (
)
func TestMergeUnion_KeepsBothBackendsHistory(t *testing.T) {
old := recs(t, "old", report("h1", "r1", "2026-07-01T00:00:00Z"))
nw := recs(t, "new", report("h1", "r2", "2026-07-02T00:00:00Z"))
merged := mergeUnion([]backendResult{nw, old}, reportKey)
a := recs(t, "a", report("h1", "r1", "2026-07-01T00:00:00Z"))
b := recs(t, "b", report("h1", "r2", "2026-07-02T00:00:00Z"))
merged := mergeUnion([]backendResult{b, a}, reportKey)
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) {
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"}) {
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) {
// extract/group_by queries return synthetic rows with no hash; dropping the
// second backend's rows as "duplicates" would silently lose half the data.
old := recs(t, "old", `{"status":"changed","count":3}`)
nw := recs(t, "new", `{"status":"changed","count":5}`)
merged := mergeUnion([]backendResult{old, nw}, reportKey)
a := recs(t, "a", `{"status":"changed","count":3}`)
b := recs(t, "b", `{"status":"changed","count":5}`)
merged := mergeUnion([]backendResult{a, b}, reportKey)
if len(merged) != 2 {
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
// them as duplicates undercounts the merged result.
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 {
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) {
same := event("h1", "r1", "Package[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 {
t.Errorf("expected 2 distinct events, got %d: %v", len(merged), merged)
}
+58 -43
View File
@@ -8,7 +8,6 @@ import (
"log"
"net/http"
"net/url"
"sort"
"strconv"
"strings"
"sync"
@@ -82,7 +81,7 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) {
s.serveFirstHolder(w, r)
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
}
merged := mergeUnion(s.byPrecedence(alive), key)
merged := mergeUnion(alive, key)
sortRecords(merged, page.order)
if page.wantTotal {
if total := sumTotals(alive); total >= 0 {
@@ -158,7 +157,7 @@ func (s *Server) serveSummed(w http.ResponseWriter, r *http.Request, path string
return
}
merged := sumRows(s.byPrecedence(alive), columns)
merged := sumRows(alive, columns)
sortRecords(merged, page.order)
if page.wantTotal {
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)
return
}
for _, res := range s.byPrecedence(alive) {
for _, res := range alive {
if len(res.records) > 0 {
writeJSON(w, rawRecords(res.records))
return
@@ -226,33 +225,15 @@ func rawRecords(recs []record) []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 {
ordered := s.byPrecedence(results)
if s.cfg.Merge == mergeStatic {
prefer := s.cfg.Prefer
return mergeFacts(ordered, func(string) string { return prefer })
return mergeFacts(results, nil)
}
fresh := s.freshnessMap(context.Background(), ordered)
prefer := s.cfg.Prefer
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
fresh := s.freshnessMap(context.Background(), results)
return mergeFacts(results, func(cn string) string { return fresh[cn] })
}
// 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)
}
f := buildFreshness(s.byPrecedence(alive))
f := buildFreshness(alive)
s.mu.Lock()
s.freshData = f
@@ -329,29 +310,63 @@ func (s *Server) queryBackend(ctx context.Context, b Backend, path string, param
return recs, total, err
}
func (s *Server) proxyPrimary(w http.ResponseWriter, r *http.Request) {
b := s.cfg.PrimaryBackend()
// 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.
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
if r.URL.RawQuery != "" {
target += "?" + r.URL.RawQuery
}
req, err := http.NewRequestWithContext(r.Context(), http.MethodGet, target, nil)
if err != nil {
http.Error(w, err.Error(), http.StatusBadGateway)
return
return nil, err
}
resp, err := 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)
return
return s.client.Do(req)
}
func setContentType(w http.ResponseWriter, contentType string) {
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 {
+222 -189
View File
@@ -119,13 +119,11 @@ func truncate(t *testing.T, body, limit string) string {
return string(out)
}
func testConfig(oldURL, newURL, merge string) Config {
func testConfig(aURL, bURL, merge string) Config {
return Config{
Listen: ":0",
Backends: []Backend{{Name: "old", URL: oldURL}, {Name: "new", URL: newURL}},
Primary: "new",
Backends: []Backend{{Name: "a", URL: aURL}, {Name: "b", URL: bURL}},
Merge: merge,
Prefer: "new",
Timeout: 2 * 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) {
old := newFakeBackend(t,
a := newFakeBackend(t,
`[`+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")+`]`, `[]`)
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"]`)
if rec.Code != http.StatusOK {
@@ -167,76 +165,78 @@ func TestHandler_NodesMerged(t *testing.T) {
}
for _, m := range got {
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) {
old := newFakeBackend(t, `[]`, `[]`)
nw := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
q := `["=","certname","abc.example.net"]`
doGet(t, srv.Handler(), factsPath, q)
if old.gotQuery(factsPath) != q {
t.Errorf("old backend got query %q, want %q", old.gotQuery(factsPath), q)
if a.gotQuery(factsPath) != q {
t.Errorf("a backend got query %q, want %q", a.gotQuery(factsPath), q)
}
if nw.gotQuery(factsPath) != q {
t.Errorf("new backend got query %q, want %q", nw.gotQuery(factsPath), q)
if b.gotQuery(factsPath) != q {
t.Errorf("b backend got query %q, want %q", b.gotQuery(factsPath), q)
}
}
func TestHandler_FactsStaticMerge(t *testing.T) {
old := newFakeBackend(t, `[]`,
`[`+fact("h1", "role", "web-old", "")+`,`+fact("h2", "role", "db-old", "")+`]`)
nw := newFakeBackend(t, `[]`,
`[`+fact("h1", "role", "web-new", "")+`]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
// Static merge ignores timestamps: a shared certname resolves to the first
// backend in configured order that holds it.
a := newFakeBackend(t, `[]`,
`[`+fact("h1", "role", "web-a", "")+`,`+fact("h2", "role", "db-a", "")+`]`)
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"]`)
if rec.Code != http.StatusOK {
t.Fatalf("status %d", rec.Code)
}
body := rec.Body.String()
if !strings.Contains(body, "web-new") || strings.Contains(body, "web-old") {
t.Errorf("static prefer=new should keep web-new, drop web-old: %s", body)
if !strings.Contains(body, "web-a") || strings.Contains(body, "web-b") {
t.Errorf("h1 should resolve to the first backend holding it: %s", body)
}
if !strings.Contains(body, "db-old") {
t.Errorf("h2 only in old should survive: %s", body)
if !strings.Contains(body, "db-a") || !strings.Contains(body, "db-b") {
t.Errorf("nodes held by only one backend must all survive: %s", body)
}
}
func TestHandler_FactsFreshnessMerge(t *testing.T) {
// Freshness: old holds h1's newer report; new holds h2's newer report.
old := newFakeBackend(t,
// Freshness: a holds h1's newer report; b holds h2's newer report.
a := newFakeBackend(t,
`[`+node("h1", "2026-07-20T00:00:00Z")+`,`+node("h2", "2026-07-01T00:00:00Z")+`]`,
`[`+fact("h1", "role", "web-old", "")+`,`+fact("h2", "role", "db-old", "")+`]`)
nw := newFakeBackend(t,
`[`+fact("h1", "role", "web-a", "")+`,`+fact("h2", "role", "db-a", "")+`]`)
b := newFakeBackend(t,
`[`+node("h1", "2026-07-01T00:00:00Z")+`,`+node("h2", "2026-07-20T00:00:00Z")+`]`,
`[`+fact("h1", "role", "web-new", "")+`,`+fact("h2", "role", "db-new", "")+`]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeFreshness))
`[`+fact("h1", "role", "web-b", "")+`,`+fact("h2", "role", "db-b", "")+`]`)
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness))
rec := doGet(t, srv.Handler(), factsPath, `["=","name","role"]`)
if rec.Code != http.StatusOK {
t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
}
body := rec.Body.String()
// h1 -> old (newer report there); h2 -> new.
if !strings.Contains(body, "web-old") || strings.Contains(body, "web-new") {
t.Errorf("h1 should resolve to old: %s", body)
// h1 -> a (newer report there); h2 -> b.
if !strings.Contains(body, "web-a") || strings.Contains(body, "web-b") {
t.Errorf("h1 should resolve to a: %s", body)
}
if !strings.Contains(body, "db-new") || strings.Contains(body, "db-old") {
t.Errorf("h2 should resolve to new: %s", body)
if !strings.Contains(body, "db-b") || strings.Contains(body, "db-a") {
t.Errorf("h2 should resolve to b: %s", body)
}
}
func TestHandler_OneBackendDown(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`)
old.fail = true
nw := newFakeBackend(t,
a := newFakeBackend(t, `[]`, `[]`)
a.fail = true
b := newFakeBackend(t,
`[`+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, "")
if rec.Code != http.StatusOK {
@@ -248,10 +248,10 @@ func TestHandler_OneBackendDown(t *testing.T) {
}
func TestHandler_BothBackendsDown(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`)
nw := newFakeBackend(t, `[]`, `[]`)
old.fail, nw.fail = true, true
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
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(), nodesPath, "")
if rec.Code != http.StatusBadGateway {
@@ -259,11 +259,12 @@ func TestHandler_BothBackendsDown(t *testing.T) {
}
}
func TestHandler_PassThroughToPrimary(t *testing.T) {
// A non-merged v4 path (e.g. /resources) goes only to the primary (new).
old := newFakeBackend(t, `[]`, `[]`)
nw := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
func TestHandler_PassThroughFirstAnswer(t *testing.T) {
// A path with no merge rule (e.g. /resources) is served by the first backend
// that answers; the rest are not asked at all.
a := newFakeBackend(t, `[]`, `[]`)
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, `["=","certname","h1"]`)
@@ -273,19 +274,51 @@ func TestHandler_PassThroughToPrimary(t *testing.T) {
if !strings.Contains(rec.Body.String(), path) {
t.Errorf("expected pass-through body, got %s", rec.Body.String())
}
// Only primary (new) should have been queried.
if _, hit := old.params(path); hit {
t.Errorf("non-primary backend should not be queried for pass-through")
if _, hit := a.params(path); !hit {
t.Errorf("first backend should be queried for pass-through")
}
if _, hit := nw.params(path); !hit {
t.Errorf("primary backend should be queried for pass-through")
if _, hit := b.params(path); hit {
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) {
old := newFakeBackend(t, `[]`, `[]`)
nw := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
req := httptest.NewRequest(http.MethodPost, factsPath, nil)
rec := httptest.NewRecorder()
srv.Handler().ServeHTTP(rec, req)
@@ -295,9 +328,9 @@ func TestHandler_PostRejected(t *testing.T) {
}
func TestHandler_Health(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`)
nw := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGet(t, srv.Handler(), "/healthz", "")
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 {
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)
}
}
func TestHandler_HealthDegradedAndDown(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`)
nw := newFakeBackend(t, `[]`, `[]`)
old.fail = true
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.fail = true
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGet(t, srv.Handler(), "/healthz", "")
var hr healthReport
@@ -328,7 +361,7 @@ func TestHandler_HealthDegradedAndDown(t *testing.T) {
t.Errorf("degraded should still be 200, got %d", rec.Code)
}
nw.fail = true
b.fail = true
rec = doGet(t, srv.Handler(), "/healthz", "")
_ = json.Unmarshal(rec.Body.Bytes(), &hr)
if hr.Status != "down" || rec.Code != http.StatusServiceUnavailable {
@@ -337,21 +370,21 @@ func TestHandler_HealthDegradedAndDown(t *testing.T) {
}
func TestFreshnessCache_Reused(t *testing.T) {
old := newFakeBackend(t,
a := newFakeBackend(t,
`[`+node("h1", "2026-07-20T00:00:00Z")+`]`,
`[`+fact("h1", "role", "web-old", "")+`]`)
nw := newFakeBackend(t,
`[`+fact("h1", "role", "web-a", "")+`]`)
b := newFakeBackend(t,
`[`+node("h1", "2026-07-01T00:00:00Z")+`]`,
`[`+fact("h1", "role", "web-new", "")+`]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeFreshness))
`[`+fact("h1", "role", "web-b", "")+`]`)
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness))
// Two facts queries; the freshness /nodes probe should be cached after the
// 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++ {
rec := doGet(t, srv.Handler(), factsPath, `["=","name","role"]`)
if !strings.Contains(rec.Body.String(), "web-old") {
t.Fatalf("iteration %d: expected h1->old, got %s", i, rec.Body.String())
if !strings.Contains(rec.Body.String(), "web-a") {
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"}]`
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.
old := newFakeBackend(t, `[]`, `[]`)
old.bodies[reportsPath] = `[` + report("h1", "r2", "2026-07-10T00:00:00Z") + `,` +
a := newFakeBackend(t, `[]`, `[]`)
a.bodies[reportsPath] = `[` + report("h1", "r2", "2026-07-10T00:00:00Z") + `,` +
report("h1", "r1", "2026-07-01T00:00:00Z") + `]`
nw := newFakeBackend(t, `[]`, `[]`)
nw.bodies[reportsPath] = `[` + report("h1", "r4", "2026-07-30T00:00:00Z") + `,` +
b := newFakeBackend(t, `[]`, `[]`)
b.bodies[reportsPath] = `[` + report("h1", "r4", "2026-07-30T00: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{
"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
// hash in each; the merged view must show it once.
dup := report("h1", "r1", "2026-07-01T00:00:00Z")
old := newFakeBackend(t, `[]`, `[]`)
old.bodies[reportsPath] = `[` + dup + `]`
nw := newFakeBackend(t, `[]`, `[]`)
nw.bodies[reportsPath] = `[` + dup + `]`
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
a := newFakeBackend(t, `[]`, `[]`)
a.bodies[reportsPath] = `[` + dup + `]`
b := newFakeBackend(t, `[]`, `[]`)
b.bodies[reportsPath] = `[` + dup + `]`
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), reportsPath, nil)
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) {
old := newFakeBackend(t, `[]`, `[]`)
old.bodies[reportsPath] = `[` + report("h1", "r5", "2026-07-05T00:00:00Z") + `,` +
a := newFakeBackend(t, `[]`, `[]`)
a.bodies[reportsPath] = `[` + report("h1", "r5", "2026-07-05T00:00:00Z") + `,` +
report("h1", "r3", "2026-07-03T00:00:00Z") + `,` +
report("h1", "r1", "2026-07-01T00:00:00Z") + `]`
nw := newFakeBackend(t, `[]`, `[]`)
nw.bodies[reportsPath] = `[` + report("h1", "r6", "2026-07-06T00:00:00Z") + `,` +
b := newFakeBackend(t, `[]`, `[]`)
b.bodies[reportsPath] = `[` + report("h1", "r6", "2026-07-06T00:00:00Z") + `,` +
report("h1", "r4", "2026-07-04T00: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{
"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
// 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)
if !ok {
t.Fatalf("%s backend was not queried", name)
@@ -463,13 +496,13 @@ func TestHandler_ReportsPagedAcrossBackends(t *testing.T) {
}
func TestHandler_ReportsIncludeTotalSummed(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`)
old.bodies[reportsPath] = `[` + report("h1", "r1", "2026-07-01T00:00:00Z") + `]`
old.totals[reportsPath] = 40
nw := newFakeBackend(t, `[]`, `[]`)
nw.bodies[reportsPath] = `[` + report("h1", "r2", "2026-07-02T00:00:00Z") + `]`
nw.totals[reportsPath] = 60
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
a := newFakeBackend(t, `[]`, `[]`)
a.bodies[reportsPath] = `[` + report("h1", "r1", "2026-07-01T00:00:00Z") + `]`
a.totals[reportsPath] = 40
b := newFakeBackend(t, `[]`, `[]`)
b.bodies[reportsPath] = `[` + report("h1", "r2", "2026-07-02T00:00:00Z") + `]`
b.totals[reportsPath] = 60
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{
"include_total": {"true"},
@@ -481,13 +514,13 @@ func TestHandler_ReportsIncludeTotalSummed(t *testing.T) {
}
func TestHandler_ReportsNoTotalWhenNotRequested(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`)
old.bodies[reportsPath] = `[]`
old.totals[reportsPath] = 40
nw := newFakeBackend(t, `[]`, `[]`)
nw.bodies[reportsPath] = `[]`
nw.totals[reportsPath] = 60
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
a := newFakeBackend(t, `[]`, `[]`)
a.bodies[reportsPath] = `[]`
a.totals[reportsPath] = 40
b := newFakeBackend(t, `[]`, `[]`)
b.bodies[reportsPath] = `[]`
b.totals[reportsPath] = 60
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), reportsPath, nil)
if got := rec.Header().Get(recordsHeader); got != "" {
@@ -496,9 +529,9 @@ func TestHandler_ReportsNoTotalWhenNotRequested(t *testing.T) {
}
func TestHandler_ReportsBadPagingParam(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`)
nw := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
for _, params := range []url.Values{
{"limit": {"lots"}},
@@ -515,11 +548,11 @@ func TestHandler_ReportsBadPagingParam(t *testing.T) {
func TestHandler_EventsUnioned(t *testing.T) {
// Puppetboard fetches a report's events as /events?query=["=","report",hash],
// and the report may live in either backend.
old := newFakeBackend(t, `[]`, `[]`)
old.bodies[eventsPath] = `[` + event("h1", "r1", "Package[nginx]") + `]`
nw := newFakeBackend(t, `[]`, `[]`)
nw.bodies[eventsPath] = `[` + event("h1", "r2", "Service[nginx]") + `]`
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
a := newFakeBackend(t, `[]`, `[]`)
a.bodies[eventsPath] = `[` + event("h1", "r1", "Package[nginx]") + `]`
b := newFakeBackend(t, `[]`, `[]`)
b.bodies[eventsPath] = `[` + event("h1", "r2", "Service[nginx]") + `]`
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGet(t, srv.Handler(), eventsPath, `["=","certname","h1"]`)
if rec.Code != http.StatusOK {
@@ -533,11 +566,11 @@ func TestHandler_EventsUnioned(t *testing.T) {
func TestHandler_EventsDedupedByIdentity(t *testing.T) {
dup := event("h1", "r1", "Package[nginx]")
old := newFakeBackend(t, `[]`, `[]`)
old.bodies[eventsPath] = `[` + dup + `]`
nw := newFakeBackend(t, `[]`, `[]`)
nw.bodies[eventsPath] = `[` + dup + `]`
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
a := newFakeBackend(t, `[]`, `[]`)
a.bodies[eventsPath] = `[` + dup + `]`
b := newFakeBackend(t, `[]`, `[]`)
b.bodies[eventsPath] = `[` + dup + `]`
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGet(t, srv.Handler(), eventsPath, "")
var got []json.RawMessage
@@ -550,27 +583,27 @@ func TestHandler_EventsDedupedByIdentity(t *testing.T) {
}
func TestHandler_ReportSubResourceFromHoldingBackend(t *testing.T) {
// Only old holds report r1, so its logs must come from old even though new
// is the primary — a pass-through would have 404'd.
// Only a holds report r1, so its logs come from a; an unmerged pass-through
// to whichever backend answered first could have 404'd.
const path = reportsPath + "/r1/logs"
old := newFakeBackend(t, `[]`, `[]`)
old.bodies[path] = `[{"level":"notice","message":"from-old"}]`
nw := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
a := newFakeBackend(t, `[]`, `[]`)
a.bodies[path] = `[{"level":"notice","message":"from-a"}]`
b := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGet(t, srv.Handler(), path, "")
if rec.Code != http.StatusOK {
t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
}
if !strings.Contains(rec.Body.String(), "from-old") {
t.Errorf("expected old's logs, got %s", rec.Body.String())
if !strings.Contains(rec.Body.String(), "from-a") {
t.Errorf("expected the holding backend's logs, got %s", rec.Body.String())
}
}
func TestHandler_ReportSubResourceMissingEverywhere(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`)
nw := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGet(t, srv.Handler(), reportsPath+"/nope/events", "")
if rec.Code != http.StatusNotFound {
@@ -579,11 +612,11 @@ func TestHandler_ReportSubResourceMissingEverywhere(t *testing.T) {
}
func TestHandler_ReportsOneBackendDown(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`)
old.fail = true
nw := newFakeBackend(t, `[]`, `[]`)
nw.bodies[reportsPath] = `[` + report("h1", "r1", "2026-07-01T00:00:00Z") + `]`
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
a := newFakeBackend(t, `[]`, `[]`)
a.fail = true
b := newFakeBackend(t, `[]`, `[]`)
b.bodies[reportsPath] = `[` + report("h1", "r1", "2026-07-01T00:00:00Z") + `]`
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{"order_by": {receiveDesc}})
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) {
// A node reporting to both PuppetDBs has its run counted in each; the
// merged view is the sum, not two rows.
old := newFakeBackend(t, `[]`, `[]`)
old.bodies[eventCountsPath] = `[` + eventCount("h1", 4, 3, 1, 0) + `]`
nw := newFakeBackend(t, `[]`, `[]`)
nw.bodies[eventCountsPath] = `[` + eventCount("h1", 2, 1, 0, 0) + `,` + eventCount("h2", 5, 0, 0, 0) + `]`
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
a := newFakeBackend(t, `[]`, `[]`)
a.bodies[eventCountsPath] = `[` + eventCount("h1", 4, 3, 1, 0) + `]`
b := newFakeBackend(t, `[]`, `[]`)
b.bodies[eventCountsPath] = `[` + eventCount("h1", 2, 1, 0, 0) + `,` + eventCount("h2", 5, 0, 0, 0) + `]`
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), eventCountsPath, url.Values{
"query": {`["=","certname","h1"]`},
@@ -634,7 +667,7 @@ func TestHandler_EventCountsSummedPerSubject(t *testing.T) {
if rec.Code != http.StatusOK {
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}) {
t.Errorf("successes = %v, want [6 5]", got)
}
@@ -644,18 +677,18 @@ func TestHandler_EventCountsSummedPerSubject(t *testing.T) {
}
func TestHandler_EventCountsDisjointSubjectsPassThrough(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`)
old.bodies[eventCountsPath] = `[` + eventCount("h1", 1, 0, 0, 0) + `]`
nw := newFakeBackend(t, `[]`, `[]`)
nw.bodies[eventCountsPath] = `[` + eventCount("h2", 2, 0, 0, 0) + `]`
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
a := newFakeBackend(t, `[]`, `[]`)
a.bodies[eventCountsPath] = `[` + eventCount("h1", 1, 0, 0, 0) + `]`
b := newFakeBackend(t, `[]`, `[]`)
b.bodies[eventCountsPath] = `[` + eventCount("h2", 2, 0, 0, 0) + `]`
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
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}) {
t.Errorf("successes = %v, want [2 1] (both nodes, untouched)", got)
if got := counts(t, rec.Body.Bytes(), "successes"); !slices.Equal(got, []float64{1, 2}) {
t.Errorf("successes = %v, want [1 2] (both nodes, untouched)", got)
}
// 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)
if p.Get("summarize_by") != "certname" {
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) {
// 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.
old := newFakeBackend(t, `[]`, `[]`)
old.bodies[eventCountsPath] = `[` + eventCount("h1", 1, 0, 0, 0) + `]`
old.totals[eventCountsPath] = 1
nw := newFakeBackend(t, `[]`, `[]`)
nw.bodies[eventCountsPath] = `[` + eventCount("h1", 2, 0, 0, 0) + `]`
nw.totals[eventCountsPath] = 1
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
a := newFakeBackend(t, `[]`, `[]`)
a.bodies[eventCountsPath] = `[` + eventCount("h1", 1, 0, 0, 0) + `]`
a.totals[eventCountsPath] = 1
b := newFakeBackend(t, `[]`, `[]`)
b.bodies[eventCountsPath] = `[` + eventCount("h1", 2, 0, 0, 0) + `]`
b.totals[eventCountsPath] = 1
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), eventCountsPath, url.Values{
"summarize_by": {"certname"},
@@ -684,13 +717,13 @@ func TestHandler_EventCountsRecordsIsMergedRowCount(t *testing.T) {
}
func TestHandler_AggregateEventCountsSummed(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`)
old.bodies[aggregateEventCountsPath] =
a := newFakeBackend(t, `[]`, `[]`)
a.bodies[aggregateEventCountsPath] =
`[{"successes":2,"failures":1,"noops":0,"skips":3,"total":6,"summarize_by":"certname"}]`
nw := newFakeBackend(t, `[]`, `[]`)
nw.bodies[aggregateEventCountsPath] =
b := newFakeBackend(t, `[]`, `[]`)
b.bodies[aggregateEventCountsPath] =
`[{"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"}})
if rec.Code != http.StatusOK {
@@ -715,13 +748,13 @@ func TestHandler_AggregateEventCountsSummed(t *testing.T) {
func TestHandler_AggregateEventCountsNullColumnSurvives(t *testing.T) {
// PuppetDB returns null totals for an empty result set; summing must not
// crash or blank out the backend that does have numbers.
old := newFakeBackend(t, `[]`, `[]`)
old.bodies[aggregateEventCountsPath] =
a := newFakeBackend(t, `[]`, `[]`)
a.bodies[aggregateEventCountsPath] =
`[{"successes":null,"failures":null,"total":null,"summarize_by":"certname"}]`
nw := newFakeBackend(t, `[]`, `[]`)
nw.bodies[aggregateEventCountsPath] =
b := newFakeBackend(t, `[]`, `[]`)
b.bodies[aggregateEventCountsPath] =
`[{"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"}})
if rec.Code != http.StatusOK {
@@ -737,11 +770,11 @@ const statusCountQuery = `["extract",[["function","count"],"status"],["~","certn
func TestHandler_ReportsAggregateSummed(t *testing.T) {
// Puppetboard's daily-reports chart: each backend counts only its own
// reports, so the merged chart needs the per-status sums.
old := newFakeBackend(t, `[]`, `[]`)
old.bodies[reportsPath] = `[{"count":4,"status":"changed"},{"count":2,"status":"failed"}]`
nw := newFakeBackend(t, `[]`, `[]`)
nw.bodies[reportsPath] = `[{"count":3,"status":"changed"},{"count":9,"status":"unchanged"}]`
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
a := newFakeBackend(t, `[]`, `[]`)
a.bodies[reportsPath] = `[{"count":4,"status":"changed"},{"count":2,"status":"failed"}]`
b := newFakeBackend(t, `[]`, `[]`)
b.bodies[reportsPath] = `[{"count":3,"status":"changed"},{"count":9,"status":"unchanged"}]`
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{"query": {statusCountQuery}})
if rec.Code != http.StatusOK {
@@ -764,13 +797,13 @@ func TestHandler_ReportsAggregateSummed(t *testing.T) {
}
func TestHandler_ReportsAggregateRecordsIsMergedRowCount(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`)
old.bodies[reportsPath] = `[{"count":4,"status":"changed"}]`
old.totals[reportsPath] = 1
nw := newFakeBackend(t, `[]`, `[]`)
nw.bodies[reportsPath] = `[{"count":3,"status":"changed"}]`
nw.totals[reportsPath] = 1
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
a := newFakeBackend(t, `[]`, `[]`)
a.bodies[reportsPath] = `[{"count":4,"status":"changed"}]`
a.totals[reportsPath] = 1
b := newFakeBackend(t, `[]`, `[]`)
b.bodies[reportsPath] = `[{"count":3,"status":"changed"}]`
b.totals[reportsPath] = 1
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{
"query": {statusCountQuery},
@@ -784,11 +817,11 @@ func TestHandler_ReportsAggregateRecordsIsMergedRowCount(t *testing.T) {
func TestHandler_ReportsNonAggregateStillUnioned(t *testing.T) {
// An extract with no function is a projection of real reports, so the
// union — not a sum — is still the right merge.
old := newFakeBackend(t, `[]`, `[]`)
old.bodies[reportsPath] = `[` + report("h1", "r1", "2026-07-01T00:00:00Z") + `]`
nw := newFakeBackend(t, `[]`, `[]`)
nw.bodies[reportsPath] = `[` + report("h1", "r2", "2026-07-02T00:00:00Z") + `]`
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
a := newFakeBackend(t, `[]`, `[]`)
a.bodies[reportsPath] = `[` + report("h1", "r1", "2026-07-01T00:00:00Z") + `]`
b := newFakeBackend(t, `[]`, `[]`)
b.bodies[reportsPath] = `[` + report("h1", "r2", "2026-07-02T00:00:00Z") + `]`
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{
"query": {`["extract",["hash","certname"],["=","certname","h1"]]`},
@@ -800,11 +833,11 @@ func TestHandler_ReportsNonAggregateStillUnioned(t *testing.T) {
}
func TestHandler_EventCountsOneBackendDown(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`)
old.fail = true
nw := newFakeBackend(t, `[]`, `[]`)
nw.bodies[eventCountsPath] = `[` + eventCount("h1", 2, 0, 0, 0) + `]`
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
a := newFakeBackend(t, `[]`, `[]`)
a.fail = true
b := newFakeBackend(t, `[]`, `[]`)
b.bodies[eventCountsPath] = `[` + eventCount("h1", 2, 0, 0, 0) + `]`
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), eventCountsPath, url.Values{"summarize_by": {"certname"}})
if rec.Code != http.StatusOK {
@@ -816,9 +849,9 @@ func TestHandler_EventCountsOneBackendDown(t *testing.T) {
}
func TestHandler_EventCountsBadPagingParam(t *testing.T) {
old := newFakeBackend(t, `[]`, `[]`)
nw := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), eventCountsPath, url.Values{"limit": {"lots"}})
if rec.Code != http.StatusBadRequest {