diff --git a/README.md b/README.md index 43f68ff..10ac5e6 100644 --- a/README.md +++ b/README.md @@ -1,9 +1,9 @@ # pdbmux — merging PuppetDB proxy `pdbmux` is a small HTTP daemon that fronts **several** PuppetDB backends and -serves a single, merged PuppetDB v4 query surface on one address. Point any -PuppetDB API client at `pdbmux` instead of a raw PuppetDB and it sees one -consistent view spanning all of them. +serves a single, merged PuppetDB v4 query surface on one address. Point +Puppetboard, or any other PuppetDB API client, at `pdbmux` instead of a raw +PuppetDB and it sees one consistent view spanning all of them. ## Why @@ -26,13 +26,16 @@ not PQL) is forwarded verbatim. | Path | Behaviour | |---|---| -| `GET /pdb/query/v4/nodes` | Fan out to both backends, dedupe by `certname`, keep the record with the newer `report_timestamp`. | -| `GET /pdb/query/v4/facts` | Fan out to both, and per `certname` keep **all** facts from the backend that owns that node (see merge semantics). | +| `GET /pdb/query/v4/nodes` | Fan out to all backends, dedupe by `certname`, keep the record with the newer `report_timestamp`. | +| `GET /pdb/query/v4/facts` | Fan out to all, and per `certname` keep **all** facts from the backend that owns that node (see merge semantics). | +| `GET /pdb/query/v4/reports` | Fan out to all and serve the **union**, deduped by report `hash`, re-ordered and re-paged across backends. | +| `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/reports//{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 /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 -survivor's results and logs a warning; a merged endpoint only returns `502` when +surviving backends' results and logs a warning; a merged endpoint only returns `502` when **every** backend fails. Response records are passed through as raw JSON so unknown fields survive untouched. @@ -41,17 +44,40 @@ unknown fields survive untouched. - **`/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. -- **`/facts`** — node-level granularity. For a `certname` present in both - backends, `pdbmux` keeps **all** of that node's facts from **one** backend and +- **`/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: - **`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 both backends, cached for + 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. - 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). Records the merge cannot key — `extract`/`group_by` aggregate rows — are + never deduped, so every backend's rows pass through even when byte-identical; + summing those aggregates across backends is not implemented yet. + +### Paging and ordering on the merged endpoints + +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. +- 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` makes `pdbmux` sum each backend's `X-Records` header into + one merged header. Deduped records are counted once per backend, so the total + is an upper bound. +- A malformed `limit`, `offset` or `order_by` gets a `400` rather than being + forwarded. ## Config @@ -106,7 +132,8 @@ pdbmux version ``` Point a consumer at it — any PuppetDB v4 client works, it just needs the -`pdbmux` address in place of a PuppetDB one: +`pdbmux` base URL in place of a PuppetDB one. For Puppetboard, that is its +PuppetDB host/port setting; for a raw query: ```bash curl -s --get http://localhost:8080/pdb/query/v4/nodes \ diff --git a/main.go b/main.go index 76e569b..42584b1 100644 --- a/main.go +++ b/main.go @@ -1,21 +1,27 @@ // Command pdbmux is a small merging HTTP proxy over several PuppetDB backends. // -// When an estate runs more than one PuppetDB — during a migration, or across -// regions — a given node's current data lives in exactly one of them. pdbmux -// presents a single merged PuppetDB v4 query surface so clients see one -// consistent view: +// Running more than one PuppetDB — during a migration between them, or across +// regions — means a given node's current data lives in exactly one of them at +// any moment. pdbmux presents a single merged PuppetDB v4 query surface so +// clients see one consistent view: // // - GET /pdb/query/v4/nodes — fan out to all backends, dedupe by certname, // keep the record with the newer report_timestamp. // - GET /pdb/query/v4/facts — fan out to all, and for a certname present in // more than one keep ALL facts from the backend holding that node's newer // report (freshness merge) or a static preferred backend (static merge). +// - GET /pdb/query/v4/reports and /events — fan out to all and serve the +// deduped union, re-ordered and re-paged across backends, because reports +// are immutable history and a moved node has some in each. +// - GET /pdb/query/v4/reports//{events,logs,metrics} — served by +// whichever backend actually holds that report. // - any other GET /pdb/query/v4/* — transparently proxied to the primary. // - GET /healthz — per-backend reachability. // -// The query param is forwarded verbatim (PuppetDB AST JSON). If one backend -// errors/times out, the others' results are served; only if every backend fails -// does a merged endpoint return 502. +// The query param is forwarded verbatim (PuppetDB AST JSON); order_by, limit, +// offset and include_total are re-applied over the merged result set. If one +// backend errors/times out, the others' results are served; only if every +// backend fails does a merged endpoint return 502. package main import ( @@ -67,9 +73,9 @@ func main() { Use: appName, Short: "Merging HTTP proxy over several PuppetDB backends.", Long: "pdbmux presents a single merged PuppetDB v4 query surface over several\n" + - "PuppetDB backends, so clients see one consistent view of nodes and facts\n" + - "spanning all of them. Running pdbmux with no subcommand (or `pdbmux serve`)\n" + - "starts the proxy.", + "PuppetDB backends, so clients see one consistent view of nodes, facts and\n" + + "reports spanning all of them. Running pdbmux with no subcommand (or\n" + + "`pdbmux serve`) starts the proxy.", SilenceUsage: true, RunE: func(cmd *cobra.Command, args []string) error { return serve(cmd) }, } diff --git a/merge.go b/merge.go index 9240733..0db0d07 100644 --- a/merge.go +++ b/merge.go @@ -6,19 +6,21 @@ import ( ) // record is a single PuppetDB result element kept as raw JSON so unknown fields -// survive the merge untouched. certname/report_timestamp are decoded only for -// merge decisions. +// survive the merge untouched. certname/report_timestamp/hash are decoded only +// for merge decisions. type record struct { Raw json.RawMessage Certname string ReportTimestamp string // only populated for /nodes records + Hash string // only populated for /reports records } -// recordMeta is the subset we decode from any /nodes or /facts element to drive -// merge decisions. +// recordMeta is the subset we decode from any /nodes, /facts or /reports element +// to drive merge decisions. type recordMeta struct { Certname string `json:"certname"` ReportTimestamp string `json:"report_timestamp"` + Hash string `json:"hash"` } // decodeRecords turns a raw PuppetDB JSON array into records, preserving each @@ -36,6 +38,7 @@ func decodeRecords(body []byte) ([]record, error) { Raw: raw, Certname: m.Certname, ReportTimestamp: m.ReportTimestamp, + Hash: m.Hash, }) } return out, nil diff --git a/merge_test.go b/merge_test.go index 59c4a55..3e3050c 100644 --- a/merge_test.go +++ b/merge_test.go @@ -74,6 +74,20 @@ func fact(cn, name, val, ts string) string { return `{"certname":"` + cn + `","name":"` + name + `","value":"` + val + `","report_timestamp":"` + ts + `"}` } +// report builds a /reports record with the fields the merge and ordering paths +// care about. +func report(cn, hash, receive string) string { + return `{"certname":"` + cn + `","hash":"` + hash + `","receive_time":"` + receive + + `","end_time":"` + receive + `","status":"changed","environment":"production"}` +} + +// event builds an /events record, which carries its report's hash but no id of +// its own. +func event(cn, reportHash, resource string) string { + return `{"certname":"` + cn + `","report":"` + reportHash + `","resource_title":"` + resource + + `","status":"success","timestamp":"2026-07-01T00:00:00Z"}` +} + 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")) diff --git a/reports.go b/reports.go new file mode 100644 index 0000000..24c8d87 --- /dev/null +++ b/reports.go @@ -0,0 +1,247 @@ +package main + +import ( + "encoding/json" + "fmt" + "net/url" + "sort" + "strconv" + "strings" +) + +// mergeUnion concatenates every backend's records and drops duplicates by key. +// Reports and events are immutable history, so a certname that migrated between +// PuppetDBs legitimately has records in both and the union — not a per-node +// winner — is the correct merged view. results must be ordered by precedence; +// the first backend holding a key supplies the record. A key func returning +// ok=false means the record has no dedupe identity and is always kept. +func mergeUnion(results []backendResult, key func(record) (string, bool)) []json.RawMessage { + seen := make(map[string]bool) + out := []json.RawMessage{} + for _, res := range results { + for _, rec := range res.records { + if k, ok := key(rec); ok { + if seen[k] { + continue + } + seen[k] = true + } + out = append(out, rec.Raw) + } + } + return out +} + +// reportKey identifies a report by its content hash, which PuppetDB guarantees +// is unique per report. An `extract`/`group_by` query returns synthetic rows +// with no hash and no identity — two backends can emit byte-identical aggregate +// rows that both count — so those are never deduped. +func reportKey(rec record) (string, bool) { + if rec.Hash == "" { + return "", false + } + return "hash\x00" + rec.Hash, true +} + +// rawKey identifies a record by its verbatim JSON. Events carry no unique id, +// but two byte-identical events from the same PuppetDB serialiser describe the +// same resource change, so raw equality is a safe dedupe key. +func rawKey(rec record) (string, bool) { return "raw\x00" + string(rec.Raw), true } + +// orderField is one entry of PuppetDB's order_by param. +type orderField struct { + Field string + Desc bool +} + +// parseOrderBy decodes PuppetDB's order_by param, a JSON array of +// {"field":..., "order":"asc"|"desc"} objects. An empty param yields no fields. +func parseOrderBy(s string) ([]orderField, error) { + if strings.TrimSpace(s) == "" { + return nil, nil + } + var raw []struct { + Field string `json:"field"` + Order string `json:"order"` + } + if err := json.Unmarshal([]byte(s), &raw); err != nil { + return nil, fmt.Errorf("order_by is not a JSON array: %w", err) + } + out := make([]orderField, 0, len(raw)) + for _, r := range raw { + if r.Field == "" { + return nil, fmt.Errorf("order_by entry is missing a field") + } + out = append(out, orderField{Field: r.Field, Desc: strings.EqualFold(r.Order, "desc")}) + } + return out, nil +} + +// sortRecords re-sorts a merged record set by order. Each backend only ordered +// its own slice, so the union has to be ordered again here. The sort is stable, +// so ties keep backend precedence order. +func sortRecords(recs []json.RawMessage, order []orderField) { + if len(order) == 0 || len(recs) < 2 { + return + } + objs := make([]map[string]any, len(recs)) + for i, raw := range recs { + _ = json.Unmarshal(raw, &objs[i]) // non-objects sort as all-missing fields + } + idx := make([]int, len(recs)) + for i := range idx { + idx[i] = i + } + sort.SliceStable(idx, func(a, b int) bool { + oa, ob := objs[idx[a]], objs[idx[b]] + for _, f := range order { + c := compareValues(oa[f.Field], ob[f.Field]) + if c == 0 { + continue + } + if f.Desc { + return c > 0 + } + return c < 0 + } + return false + }) + sorted := make([]json.RawMessage, len(recs)) + for i, j := range idx { + sorted[i] = recs[j] + } + copy(recs, sorted) +} + +// compareValues orders two decoded JSON values. Unlike types are ordered by +// kind (null < bool < number < string) so a missing field always sorts first. +func compareValues(a, b any) int { + ra, rb := valueRank(a), valueRank(b) + if ra != rb { + if ra < rb { + return -1 + } + return 1 + } + switch av := a.(type) { + case bool: + bv := b.(bool) + switch { + case av == bv: + return 0 + case bv: + return -1 + default: + return 1 + } + case float64: + bv := b.(float64) + switch { + case av < bv: + return -1 + case av > bv: + return 1 + default: + return 0 + } + case string: + return strings.Compare(av, b.(string)) + } + return 0 +} + +func valueRank(v any) int { + switch v.(type) { + case nil: + return 0 + case bool: + return 1 + case float64: + return 2 + case string: + return 3 + default: + return 4 + } +} + +// paging holds the PuppetDB paging params a merged endpoint has to re-apply +// itself: each backend applies limit/offset to its own result set only, so the +// proxy must page the union instead. +type paging struct { + limit int // -1 when unset + offset int + order []orderField + wantTotal bool +} + +// parsePaging reads limit, offset, order_by and include_total from a request's +// query params. +func parsePaging(v url.Values) (paging, error) { + p := paging{limit: -1} + if s := v.Get("limit"); s != "" { + n, err := strconv.Atoi(s) + if err != nil || n < 0 { + return p, fmt.Errorf("limit must be a non-negative integer, got %q", s) + } + p.limit = n + } + if s := v.Get("offset"); s != "" { + n, err := strconv.Atoi(s) + if err != nil || n < 0 { + return p, fmt.Errorf("offset must be a non-negative integer, got %q", s) + } + p.offset = n + } + order, err := parseOrderBy(v.Get("order_by")) + if err != nil { + return p, err + } + p.order = order + p.wantTotal = v.Get("include_total") == "true" + return p, nil +} + +// upstreamParams rewrites the client's params for the fan-out. A backend must +// return everything that could land in the merged page, so it is asked for the +// first offset+limit records and the offset is applied locally instead. +func (p paging) upstreamParams(in url.Values) url.Values { + out := url.Values{} + for k, vs := range in { + out[k] = append([]string(nil), vs...) + } + out.Del("offset") + if p.limit >= 0 { + out.Set("limit", strconv.Itoa(p.limit+p.offset)) + } + return out +} + +// apply slices the merged, ordered record set down to the requested page. +func (p paging) apply(recs []json.RawMessage) []json.RawMessage { + if p.offset >= len(recs) { + return []json.RawMessage{} + } + recs = recs[p.offset:] + if p.limit >= 0 && p.limit < len(recs) { + recs = recs[:p.limit] + } + return recs +} + +// sumTotals adds up the X-Records counts the backends reported, ignoring any +// backend that did not send one. It returns -1 when no backend reported a count. +// Deduped records are counted once per backend, so the total is an upper bound. +func sumTotals(results []backendResult) int { + total := -1 + for _, res := range results { + if res.total < 0 { + continue + } + if total < 0 { + total = 0 + } + total += res.total + } + return total +} diff --git a/reports_test.go b/reports_test.go new file mode 100644 index 0000000..911a755 --- /dev/null +++ b/reports_test.go @@ -0,0 +1,254 @@ +package main + +import ( + "encoding/json" + "net/url" + "slices" + "testing" +) + +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) + if got := hashesOf(t, merged); !slices.Equal(got, []string{"r2", "r1"}) { + t.Errorf("union = %v, want both reports in precedence 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) + if got := hashesOf(t, merged); !slices.Equal(got, []string{"r1"}) { + t.Errorf("union = %v, want a single r1", got) + } +} + +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) + if len(merged) != 2 { + t.Errorf("expected both aggregate rows, got %d: %v", len(merged), merged) + } +} + +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) + if len(merged) != 2 { + t.Errorf("expected both backends' aggregate rows, got %d: %v", len(merged), merged) + } +} + +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) + if len(merged) != 2 { + t.Errorf("expected 2 distinct events, got %d: %v", len(merged), merged) + } +} + +func TestParseOrderBy(t *testing.T) { + got, err := parseOrderBy(`[{"field":"receive_time","order":"desc"},{"field":"certname"}]`) + if err != nil { + t.Fatal(err) + } + want := []orderField{{Field: "receive_time", Desc: true}, {Field: "certname"}} + if !slices.Equal(got, want) { + t.Errorf("parseOrderBy = %v, want %v", got, want) + } + + if got, err := parseOrderBy(" "); err != nil || got != nil { + t.Errorf("empty order_by = %v, %v; want nil, nil", got, err) + } + for _, bad := range []string{`receive_time`, `[{"order":"desc"}]`} { + if _, err := parseOrderBy(bad); err == nil { + t.Errorf("parseOrderBy(%q) should have failed", bad) + } + } +} + +func TestSortRecords_MultipleFieldsAndStability(t *testing.T) { + raws := rawsOf(t, + `{"certname":"b","status":"failed","hash":"r1"}`, + `{"certname":"a","status":"changed","hash":"r2"}`, + `{"certname":"a","status":"changed","hash":"r3"}`, + `{"certname":"a","status":"failed","hash":"r4"}`, + ) + sortRecords(raws, []orderField{{Field: "certname"}, {Field: "status", Desc: true}}) + // certname asc, then status desc; r2/r3 tie fully and keep input order. + if got := hashesOf(t, raws); !slices.Equal(got, []string{"r4", "r2", "r3", "r1"}) { + t.Errorf("sorted = %v, want [r4 r2 r3 r1]", got) + } +} + +func TestSortRecords_MissingFieldSortsFirst(t *testing.T) { + raws := rawsOf(t, + `{"hash":"r1","receive_time":"2026-07-01T00:00:00Z"}`, + `{"hash":"r2"}`, + ) + sortRecords(raws, []orderField{{Field: "receive_time"}}) + if got := hashesOf(t, raws); !slices.Equal(got, []string{"r2", "r1"}) { + t.Errorf("sorted = %v, want the record missing the field first", got) + } +} + +func TestSortRecords_NoOrderLeavesInputOrder(t *testing.T) { + raws := rawsOf(t, `{"hash":"r1"}`, `{"hash":"r2"}`) + sortRecords(raws, nil) + if got := hashesOf(t, raws); !slices.Equal(got, []string{"r1", "r2"}) { + t.Errorf("sorted = %v, want unchanged", got) + } +} + +func TestCompareValues_AcrossKinds(t *testing.T) { + cases := []struct { + a, b any + want int + }{ + {nil, false, -1}, + {false, true, -1}, + {true, 1.0, -1}, + {1.0, 2.0, -1}, + {2.0, 2.0, 0}, + {2.0, "x", -1}, + {"a", "b", -1}, + {"b", "a", 1}, + } + for _, c := range cases { + if got := compareValues(c.a, c.b); got != c.want { + t.Errorf("compareValues(%v, %v) = %d, want %d", c.a, c.b, got, c.want) + } + if got := compareValues(c.b, c.a); got != -c.want { + t.Errorf("compareValues(%v, %v) = %d, want %d (antisymmetry)", c.b, c.a, got, -c.want) + } + } +} + +func TestParsePaging(t *testing.T) { + p, err := parsePaging(url.Values{ + "limit": {"25"}, + "offset": {"50"}, + "include_total": {"true"}, + "order_by": {`[{"field":"receive_time","order":"desc"}]`}, + }) + if err != nil { + t.Fatal(err) + } + if p.limit != 25 || p.offset != 50 || !p.wantTotal || len(p.order) != 1 { + t.Fatalf("parsePaging = %+v", p) + } + + if p, err := parsePaging(nil); err != nil || p.limit != -1 || p.offset != 0 || p.wantTotal { + t.Errorf("empty params = %+v, %v; want limit=-1 and no paging", p, err) + } + for _, bad := range []url.Values{{"limit": {"-1"}}, {"limit": {"x"}}, {"offset": {"x"}}} { + if _, err := parsePaging(bad); err == nil { + t.Errorf("parsePaging(%v) should have failed", bad) + } + } +} + +func TestPagingUpstreamParams(t *testing.T) { + in := url.Values{ + "query": {`["=","certname","h1"]`}, + "limit": {"25"}, + "offset": {"50"}, + } + p, err := parsePaging(in) + if err != nil { + t.Fatal(err) + } + out := p.upstreamParams(in) + if out.Get("limit") != "75" { + t.Errorf("upstream limit = %q, want 75 (offset+limit)", out.Get("limit")) + } + if out.Has("offset") { + t.Errorf("upstream offset = %q, want it dropped", out.Get("offset")) + } + if out.Get("query") != in.Get("query") { + t.Errorf("query should pass through verbatim, got %q", out.Get("query")) + } + if in.Get("limit") != "25" { + t.Errorf("upstreamParams must not mutate the caller's params, limit is now %q", in.Get("limit")) + } +} + +func TestPagingUpstreamParams_NoLimitLeavesQueryUnbounded(t *testing.T) { + in := url.Values{"offset": {"5"}} + p, err := parsePaging(in) + if err != nil { + t.Fatal(err) + } + out := p.upstreamParams(in) + if out.Has("limit") || out.Has("offset") { + t.Errorf("upstream params = %v, want neither limit nor offset", out) + } +} + +func TestPagingApply(t *testing.T) { + raws := rawsOf(t, `{"hash":"r1"}`, `{"hash":"r2"}`, `{"hash":"r3"}`) + cases := []struct { + name string + page paging + want []string + }{ + {name: "window", page: paging{limit: 1, offset: 1}, want: []string{"r2"}}, + {name: "limit past end", page: paging{limit: 10}, want: []string{"r1", "r2", "r3"}}, + {name: "offset past end", page: paging{limit: 2, offset: 9}, want: nil}, + {name: "unset limit", page: paging{limit: -1, offset: 2}, want: []string{"r3"}}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + got := hashesOf(t, c.page.apply(raws)) + if len(got) == 0 && len(c.want) == 0 { + return + } + if !slices.Equal(got, c.want) { + t.Errorf("apply = %v, want %v", got, c.want) + } + }) + } +} + +func TestSumTotals(t *testing.T) { + if got := sumTotals([]backendResult{{total: 40}, {total: 60}}); got != 100 { + t.Errorf("sumTotals = %d, want 100", got) + } + if got := sumTotals([]backendResult{{total: -1}, {total: 7}}); got != 7 { + t.Errorf("sumTotals should skip backends without a count, got %d", got) + } + if got := sumTotals([]backendResult{{total: -1}, {total: -1}}); got != -1 { + t.Errorf("sumTotals with no counts = %d, want -1", got) + } +} + +// rawsOf builds a raw record slice from literal JSON elements. +func rawsOf(t *testing.T, elems ...string) []json.RawMessage { + t.Helper() + out := make([]json.RawMessage, 0, len(elems)) + for _, e := range elems { + out = append(out, json.RawMessage(e)) + } + return out +} + +// hashesOf extracts the hash field from raw records, in order. +func hashesOf(t *testing.T, raws []json.RawMessage) []string { + t.Helper() + out := make([]string, 0, len(raws)) + for _, r := range raws { + var m recordMeta + if err := json.Unmarshal(r, &m); err != nil { + t.Fatalf("unmarshal %s: %v", r, err) + } + out = append(out, m.Hash) + } + return out +} diff --git a/server.go b/server.go index 1763a1a..9efeb66 100644 --- a/server.go +++ b/server.go @@ -9,15 +9,22 @@ import ( "net/http" "net/url" "sort" + "strconv" "strings" "sync" "time" ) const ( - factsPath = "/pdb/query/v4/facts" - nodesPath = "/pdb/query/v4/nodes" - queryV4 = "/pdb/query/v4/" + factsPath = "/pdb/query/v4/facts" + nodesPath = "/pdb/query/v4/nodes" + reportsPath = "/pdb/query/v4/reports" + eventsPath = "/pdb/query/v4/events" + queryV4 = "/pdb/query/v4/" + + // recordsHeader is PuppetDB's total-result-count header, returned when a + // request carries include_total=true. + recordsHeader = "X-Records" ) // backendResult is one backend's decoded response for a query. err is non-nil @@ -26,6 +33,7 @@ const ( type backendResult struct { name string records []record + total int // upstream X-Records count, or -1 when the backend sent none err error } @@ -59,7 +67,9 @@ func (s *Server) Handler() http.Handler { } // handleQuery dispatches /pdb/query/v4/* requests: /facts and /nodes are merged -// across backends; every other v4 path is transparently proxied to the primary. +// per node, /reports and /events are unioned across backends, a report's +// sub-resources resolve to whichever backend stores that report, and every other +// v4 path is transparently proxied to the primary. func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "only GET is supported", http.StatusMethodNotAllowed) @@ -70,17 +80,109 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) { s.serveMerged(w, r, nodesPath, s.mergeNodesResponse) case factsPath: s.serveMerged(w, r, factsPath, s.mergeFactsResponse) + case reportsPath: + s.serveUnion(w, r, reportsPath, reportKey) + case eventsPath: + s.serveUnion(w, r, eventsPath, rawKey) default: + if isReportSubResource(r.URL.Path) { + s.serveFirstHolder(w, r) + return + } s.proxyPrimary(w, r) } } +// isReportSubResource reports whether path is a per-report child endpoint — +// /pdb/query/v4/reports//{events,logs,metrics} — whose data lives in +// exactly one backend. +func isReportSubResource(path string) bool { + rest, ok := strings.CutPrefix(path, reportsPath+"/") + if !ok { + return false + } + hash, sub, ok := strings.Cut(rest, "/") + if !ok || hash == "" { + return false + } + switch sub { + case "events", "logs", "metrics": + return true + } + return false +} + // serveMerged fans out the request to all backends, then hands the per-backend // results to merge to produce the response body. If every backend fails it // returns 502; if some fail it serves the survivors and logs a warning. func (s *Server) serveMerged(w http.ResponseWriter, r *http.Request, path string, merge func([]backendResult) []json.RawMessage) { - query := r.URL.Query().Get("query") - results := s.fanOut(r.Context(), path, query) + alive, ok := s.aliveResults(w, r, path, queryParams(r.URL.Query().Get("query"))) + if !ok { + return + } + writeJSON(w, merge(alive)) +} + +// serveUnion fans out a request whose records are immutable history — reports +// and events — and serves the deduped union of every backend. Because each +// backend ordered and paged only its own slice, the union is re-ordered and +// re-paged here from the client's order_by/limit/offset. +func (s *Server) serveUnion(w http.ResponseWriter, r *http.Request, path string, key func(record) (string, bool)) { + in := r.URL.Query() + page, err := parsePaging(in) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + alive, ok := s.aliveResults(w, r, path, page.upstreamParams(in)) + if !ok { + return + } + + merged := mergeUnion(s.byPrecedence(alive), key) + sortRecords(merged, page.order) + if page.wantTotal { + if total := sumTotals(alive); total >= 0 { + w.Header().Set(recordsHeader, strconv.Itoa(total)) + } + } + writeJSON(w, page.apply(merged)) +} + +// serveFirstHolder answers a per-report sub-resource request. The report lives +// in exactly one backend, so all are asked concurrently and the first one (in +// precedence order) that actually holds it wins. Backends that do not have the +// report answer 404, which is indistinguishable here from any other failure, so +// an empty result is only served once every backend has been consulted. +func (s *Server) serveFirstHolder(w http.ResponseWriter, r *http.Request) { + results := s.fanOut(r.Context(), r.URL.Path, r.URL.Query()) + + var alive []backendResult + for _, res := range results { + if res.err != nil { + s.log.Printf("info: backend %q has no %s: %v", res.name, r.URL.Path, res.err) + continue + } + alive = append(alive, res) + } + if len(alive) == 0 { + http.Error(w, "no backend holds this report", http.StatusNotFound) + return + } + for _, res := range s.byPrecedence(alive) { + if len(res.records) > 0 { + writeJSON(w, rawRecords(res.records)) + return + } + } + writeJSON(w, nil) +} + +// aliveResults fans out to every backend and returns the successful results. +// It writes a 502 and returns ok=false when every backend failed. +func (s *Server) aliveResults(w http.ResponseWriter, r *http.Request, path string, params url.Values) ([]backendResult, bool) { + results := s.fanOut(r.Context(), path, params) var alive []backendResult for _, res := range results { @@ -92,11 +194,27 @@ func (s *Server) serveMerged(w http.ResponseWriter, r *http.Request, path string } if len(alive) == 0 { http.Error(w, "all backends failed", http.StatusBadGateway) - return + return nil, false } + return alive, true +} - merged := merge(alive) - writeJSON(w, merged) +// queryParams builds the upstream param set for a merged endpoint that only +// forwards the PuppetDB query. +func queryParams(query string) url.Values { + if query == "" { + return nil + } + return url.Values{"query": []string{query}} +} + +// rawRecords strips decoded metadata back down to the verbatim JSON elements. +func rawRecords(recs []record) []json.RawMessage { + out := make([]json.RawMessage, 0, len(recs)) + for _, rec := range recs { + out = append(out, rec.Raw) + } + return out } // mergeNodesResponse merges /nodes results (dedupe by certname, newer wins). @@ -154,7 +272,7 @@ func (s *Server) freshnessMap(ctx context.Context, _ []backendResult) freshness s.mu.Unlock() // Empty query = all nodes; cheap enough for a short-TTL cache. - nodeResults := s.fanOut(ctx, nodesPath, "") + nodeResults := s.fanOut(ctx, nodesPath, nil) var alive []backendResult for _, res := range nodeResults { if res.err != nil { @@ -172,48 +290,52 @@ func (s *Server) freshnessMap(ctx context.Context, _ []backendResult) freshness return f } -// fanOut queries every backend concurrently for path?query=... and returns one -// backendResult per backend, in config order. -func (s *Server) fanOut(ctx context.Context, path, query string) []backendResult { +// fanOut queries every backend concurrently for path with the given params and +// returns one backendResult per backend, in config order. +func (s *Server) fanOut(ctx context.Context, path string, params url.Values) []backendResult { results := make([]backendResult, len(s.cfg.Backends)) var wg sync.WaitGroup for i, b := range s.cfg.Backends { wg.Add(1) go func(i int, b Backend) { defer wg.Done() - recs, err := s.queryBackend(ctx, b, path, query) - results[i] = backendResult{name: b.Name, records: recs, err: err} + recs, total, err := s.queryBackend(ctx, b, path, params) + results[i] = backendResult{name: b.Name, records: recs, total: total, err: err} }(i, b) } wg.Wait() return results } -// queryBackend performs one GET b.URL+path?query=... and decodes the JSON array. -func (s *Server) queryBackend(ctx context.Context, b Backend, path, query string) ([]record, error) { +// queryBackend performs one GET b.URL+path?params and decodes the JSON array. +// It also returns the upstream X-Records count, or -1 when the backend sent none. +func (s *Server) queryBackend(ctx context.Context, b Backend, path string, params url.Values) ([]record, int, error) { target := strings.TrimRight(b.URL, "/") + path req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil) if err != nil { - return nil, err + return nil, -1, err } - if query != "" { - q := url.Values{} - q.Set("query", query) - req.URL.RawQuery = q.Encode() + if len(params) > 0 { + req.URL.RawQuery = params.Encode() } resp, err := s.client.Do(req) if err != nil { - return nil, err + return nil, -1, err } defer func() { _ = resp.Body.Close() }() body, err := io.ReadAll(resp.Body) if err != nil { - return nil, err + return nil, -1, err } if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) + return nil, -1, fmt.Errorf("HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) } - return decodeRecords(body) + total := -1 + if n, err := strconv.Atoi(resp.Header.Get(recordsHeader)); err == nil && n >= 0 { + total = n + } + recs, err := decodeRecords(body) + return recs, total, err } // proxyPrimary transparently forwards a non-merged /pdb/query/v4/* request to @@ -254,7 +376,7 @@ type healthReport struct { // reachable, "degraded" if some fail, "down" if all fail (503 in that case). func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { probe := `["=","certname","pdbmux-healthz-probe"]` - results := s.fanOut(r.Context(), nodesPath, probe) + results := s.fanOut(r.Context(), nodesPath, queryParams(probe)) report := healthReport{Backends: map[string]string{}} healthy := 0 diff --git a/server_test.go b/server_test.go index 18243e2..ff6cdc7 100644 --- a/server_test.go +++ b/server_test.go @@ -7,7 +7,10 @@ import ( "net/http" "net/http/httptest" "net/url" + "slices" + "strconv" "strings" + "sync" "testing" "time" ) @@ -15,26 +18,55 @@ import ( // fakeBackend is an httptest PuppetDB that returns canned bodies per path and // records the query params it received. type fakeBackend struct { - srv *httptest.Server - nodesBody string - factsBody string - fail bool // return 500 for everything - delay time.Duration // artificial latency - gotQueries map[string]string + srv *httptest.Server + nodesBody string + factsBody string + // bodies holds extra canned responses keyed by path (reports, events, a + // report's sub-resources). A path under /reports/ that is absent from bodies + // answers 404, like a PuppetDB that does not hold that report. + bodies map[string]string + // totals is the X-Records count advertised per path when the request asks + // for include_total. + totals map[string]int + fail bool // return 500 for everything + delay time.Duration // artificial latency + + mu sync.Mutex + gotParams map[string]url.Values } func newFakeBackend(t *testing.T, nodesBody, factsBody string) *fakeBackend { t.Helper() - fb := &fakeBackend{nodesBody: nodesBody, factsBody: factsBody, gotQueries: map[string]string{}} + fb := &fakeBackend{ + nodesBody: nodesBody, + factsBody: factsBody, + bodies: map[string]string{}, + totals: map[string]int{}, + gotParams: map[string]url.Values{}, + } fb.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if fb.delay > 0 { time.Sleep(fb.delay) } - fb.gotQueries[r.URL.Path] = r.URL.Query().Get("query") + fb.mu.Lock() + fb.gotParams[r.URL.Path] = r.URL.Query() + fb.mu.Unlock() if fb.fail { http.Error(w, "boom", http.StatusInternalServerError) return } + if body, ok := fb.bodies[r.URL.Path]; ok { + if n, ok := fb.totals[r.URL.Path]; ok && r.URL.Query().Get("include_total") == "true" { + w.Header().Set(recordsHeader, strconv.Itoa(n)) + } + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, truncate(t, body, r.URL.Query().Get("limit"))) + return + } + if strings.HasPrefix(r.URL.Path, reportsPath+"/") { + http.Error(w, "no report with that hash", http.StatusNotFound) + return + } w.Header().Set("Content-Type", "application/json") switch r.URL.Path { case nodesPath: @@ -49,6 +81,43 @@ func newFakeBackend(t *testing.T, nodesBody, factsBody string) *fakeBackend { return fb } +// params returns the query params the backend saw for a path, and whether it was +// asked for that path at all. +func (fb *fakeBackend) params(path string) (url.Values, bool) { + fb.mu.Lock() + defer fb.mu.Unlock() + v, ok := fb.gotParams[path] + return v, ok +} + +// gotQuery returns the PuppetDB query param the backend saw for a path. +func (fb *fakeBackend) gotQuery(path string) string { + v, _ := fb.params(path) + return v.Get("query") +} + +// truncate applies an upstream limit param to a canned JSON array body, the way +// a real PuppetDB would, so paging tests exercise the proxy's re-paging. +func truncate(t *testing.T, body, limit string) string { + t.Helper() + n, err := strconv.Atoi(limit) + if err != nil { + return body + } + var raws []json.RawMessage + if err := json.Unmarshal([]byte(body), &raws); err != nil { + return body + } + if n < len(raws) { + raws = raws[:n] + } + out, err := json.Marshal(raws) + if err != nil { + t.Fatalf("re-marshal truncated body: %v", err) + } + return string(out) +} + func testConfig(oldURL, newURL, merge string) Config { return Config{ Listen: ":0", @@ -109,11 +178,11 @@ func TestHandler_QueryPassthrough(t *testing.T) { q := `["=","certname","abc.example.net"]` doGet(t, srv.Handler(), factsPath, q) - if old.gotQueries[factsPath] != q { - t.Errorf("old backend got query %q, want %q", old.gotQueries[factsPath], q) + if old.gotQuery(factsPath) != q { + t.Errorf("old backend got query %q, want %q", old.gotQuery(factsPath), q) } - if nw.gotQueries[factsPath] != q { - t.Errorf("new backend got query %q, want %q", nw.gotQueries[factsPath], q) + if nw.gotQuery(factsPath) != q { + t.Errorf("new backend got query %q, want %q", nw.gotQuery(factsPath), q) } } @@ -190,23 +259,24 @@ func TestHandler_BothBackendsDown(t *testing.T) { } func TestHandler_PassThroughToPrimary(t *testing.T) { - // A non-merged v4 path (e.g. /reports) goes only to the primary (new). + // 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)) - rec := doGet(t, srv.Handler(), "/pdb/query/v4/reports", `["=","certname","h1"]`) + const path = "/pdb/query/v4/resources" + rec := doGet(t, srv.Handler(), path, `["=","certname","h1"]`) if rec.Code != http.StatusOK { t.Fatalf("status %d", rec.Code) } - if !strings.Contains(rec.Body.String(), "/pdb/query/v4/reports") { + 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.gotQueries["/pdb/query/v4/reports"]; hit { + if _, hit := old.params(path); hit { t.Errorf("non-primary backend should not be queried for pass-through") } - if _, hit := nw.gotQueries["/pdb/query/v4/reports"]; !hit { + if _, hit := nw.params(path); !hit { t.Errorf("primary backend should be queried for pass-through") } } @@ -284,3 +354,241 @@ func TestFreshnessCache_Reused(t *testing.T) { } } } + +// doGetParams issues a GET with an arbitrary param set, for the paging/ordering +// params the reports endpoints accept. +func doGetParams(t *testing.T, h http.Handler, path string, params url.Values) *httptest.ResponseRecorder { + t.Helper() + target := path + if len(params) > 0 { + target += "?" + params.Encode() + } + req := httptest.NewRequest(http.MethodGet, target, nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + return rec +} + +// hashes extracts report hashes from a merged response body, in order. +func hashes(t *testing.T, body []byte) []string { + t.Helper() + var raws []json.RawMessage + if err := json.Unmarshal(body, &raws); err != nil { + t.Fatalf("unmarshal %s: %v", body, err) + } + return hashesOf(t, raws) +} + +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. + // 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") + `,` + + report("h1", "r1", "2026-07-01T00:00:00Z") + `]` + nw := newFakeBackend(t, `[]`, `[]`) + nw.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)) + + rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{ + "query": {`["=","certname","h1"]`}, + "order_by": {receiveDesc}, + }) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + got := hashes(t, rec.Body.Bytes()) + want := []string{"r4", "r3", "r2", "r1"} + if !slices.Equal(got, want) { + t.Errorf("merged reports = %v, want %v (union re-sorted by receive_time desc)", got, want) + } +} + +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)) + + rec := doGetParams(t, srv.Handler(), reportsPath, nil) + if got := hashes(t, rec.Body.Bytes()); !slices.Equal(got, []string{"r1"}) { + t.Errorf("merged reports = %v, want one r1", got) + } +} + +func TestHandler_ReportsPagedAcrossBackends(t *testing.T) { + old := newFakeBackend(t, `[]`, `[]`) + old.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") + `,` + + 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)) + + rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{ + "order_by": {receiveDesc}, + "limit": {"2"}, + "offset": {"2"}, + }) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + // Globally-ordered page 2 of the union, not each backend's own page 2. + if got := hashes(t, rec.Body.Bytes()); !slices.Equal(got, []string{"r4", "r3"}) { + t.Errorf("page = %v, want [r4 r3]", got) + } + // 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} { + p, ok := fb.params(reportsPath) + if !ok { + t.Fatalf("%s backend was not queried", name) + } + if p.Get("limit") != "4" { + t.Errorf("%s backend got limit=%q, want 4 (offset+limit)", name, p.Get("limit")) + } + if p.Has("offset") { + t.Errorf("%s backend got offset=%q, want it applied locally instead", name, p.Get("offset")) + } + } +} + +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)) + + rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{ + "include_total": {"true"}, + "limit": {"1"}, + }) + if got := rec.Header().Get(recordsHeader); got != "100" { + t.Errorf("%s = %q, want 100 (sum of both backends)", recordsHeader, got) + } +} + +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)) + + rec := doGetParams(t, srv.Handler(), reportsPath, nil) + if got := rec.Header().Get(recordsHeader); got != "" { + t.Errorf("%s = %q, want it unset without include_total", recordsHeader, got) + } +} + +func TestHandler_ReportsBadPagingParam(t *testing.T) { + old := newFakeBackend(t, `[]`, `[]`) + nw := newFakeBackend(t, `[]`, `[]`) + srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) + + for _, params := range []url.Values{ + {"limit": {"lots"}}, + {"offset": {"-1"}}, + {"order_by": {"receive_time"}}, + } { + rec := doGetParams(t, srv.Handler(), reportsPath, params) + if rec.Code != http.StatusBadRequest { + t.Errorf("%v: expected 400, got %d", params, rec.Code) + } + } +} + +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)) + + rec := doGet(t, srv.Handler(), eventsPath, `["=","certname","h1"]`) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + body := rec.Body.String() + if !strings.Contains(body, "Package[nginx]") || !strings.Contains(body, "Service[nginx]") { + t.Errorf("expected both backends' events: %s", body) + } +} + +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)) + + rec := doGet(t, srv.Handler(), eventsPath, "") + var got []json.RawMessage + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + if len(got) != 1 { + t.Errorf("expected the duplicate event once, got %d: %s", len(got), rec.Body.String()) + } +} + +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. + 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)) + + 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()) + } +} + +func TestHandler_ReportSubResourceMissingEverywhere(t *testing.T) { + old := newFakeBackend(t, `[]`, `[]`) + nw := newFakeBackend(t, `[]`, `[]`) + srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), reportsPath+"/nope/events", "") + if rec.Code != http.StatusNotFound { + t.Fatalf("expected 404 when no backend holds the report, got %d", rec.Code) + } +} + +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)) + + rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{"order_by": {receiveDesc}}) + if rec.Code != http.StatusOK { + t.Fatalf("expected 200 serving the survivor, got %d", rec.Code) + } + if got := hashes(t, rec.Body.Bytes()); !slices.Equal(got, []string{"r1"}) { + t.Errorf("merged reports = %v, want [r1]", got) + } +}