diff --git a/README.md b/README.md index ffffbdd..bfc7870 100644 --- a/README.md +++ b/README.md @@ -30,6 +30,8 @@ not PQL) is forwarded verbatim. | `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/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//{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. | @@ -60,9 +62,23 @@ unknown fields survive untouched. 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. + 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 + `extract` carries a `["function", ...]` column. + - The grouping key is the row's non-aggregate fields: for `/reports` they come + from the query — the plain `extract` fields plus any `group_by` clause — and + for the event-count endpoints from the row itself (`subject_type`/`subject`, + or `summarize_by`), whose remaining fields are all counts. + - Rows sharing a key collapse into one with their numeric columns summed. A key + only one backend reported is passed through byte-for-byte. An aggregate column + that is absent or non-numeric in a row is skipped, never zeroed, so the + backends that did report a number still count. + - A `/reports` query with no `function` column is a projection of real reports, + not an aggregate, and stays on the union path. + - `include_total=true` on a summed endpoint reports the **merged** row count, + not the sum of the backends' `X-Records`, since shared keys collapse. ### Paging and ordering on the merged endpoints @@ -73,9 +89,10 @@ Each backend applies `order_by`/`limit`/`offset` to its own slice only, so 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. +- `include_total=true` on a union endpoint 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. Summed endpoints report the merged row + count instead. - A malformed `limit`, `offset` or `order_by` gets a `400` rather than being forwarded. diff --git a/aggregate.go b/aggregate.go new file mode 100644 index 0000000..ba63ac3 --- /dev/null +++ b/aggregate.go @@ -0,0 +1,269 @@ +package main + +import ( + "encoding/json" + "sort" + "strconv" + "strings" +) + +// aggregateSpec names the columns of an extract/group_by result set: keys +// identify a row across backends, sums are the numeric aggregate columns added +// together. +type aggregateSpec struct { + keys []string + sums []string +} + +// parseAggregate reads a PuppetDB AST query and returns the aggregate shape of +// its response, or nil when the query is not an aggregate — only a top-level +// `extract` carrying at least one `["function", ...]` column produces summable +// rows. Key columns are the plain (non-function) extract fields, unioned with an +// explicit `group_by` clause when the query has one. +func parseAggregate(query string) *aggregateSpec { + if strings.TrimSpace(query) == "" { + return nil + } + var ast []json.RawMessage + if json.Unmarshal([]byte(query), &ast) != nil || len(ast) < 2 { + return nil + } + var op string + if json.Unmarshal(ast[0], &op) != nil || op != "extract" { + return nil + } + var cols []json.RawMessage + if json.Unmarshal(ast[1], &cols) != nil { + return nil + } + + spec := &aggregateSpec{} + for _, col := range cols { + var name string + if json.Unmarshal(col, &name) == nil { + spec.keys = appendUnique(spec.keys, name) + continue + } + if fn, ok := functionName(col); ok { + spec.sums = appendUnique(spec.sums, fn) + } + } + if len(spec.sums) == 0 { + return nil + } + for _, node := range ast[2:] { + for _, f := range groupByFields(node) { + spec.keys = appendUnique(spec.keys, f) + } + } + return spec +} + +// functionName returns the response column an extract function produces, which +// PuppetDB names after the function itself: ["function","count","certname"] +// yields a "count" column. +func functionName(col json.RawMessage) (string, bool) { + var parts []json.RawMessage + if json.Unmarshal(col, &parts) != nil || len(parts) < 2 { + return "", false + } + var head, name string + if json.Unmarshal(parts[0], &head) != nil || head != "function" { + return "", false + } + if json.Unmarshal(parts[1], &name) != nil || name == "" { + return "", false + } + return name, true +} + +// groupByFields returns the field names of a ["group_by", ...] AST node, or nil +// for any other node. +func groupByFields(node json.RawMessage) []string { + var parts []json.RawMessage + if json.Unmarshal(node, &parts) != nil || len(parts) < 2 { + return nil + } + var head string + if json.Unmarshal(parts[0], &head) != nil || head != "group_by" { + return nil + } + var out []string + for _, p := range parts[1:] { + var name string + if json.Unmarshal(p, &name) == nil { + out = append(out, name) + } + } + return out +} + +func appendUnique(s []string, v string) []string { + if contains(s, v) { + return s + } + return append(s, v) +} + +// columns reports which fields of a row form its grouping key and which are +// summed. A spec is fixed by the query, so the row is ignored. +func (a *aggregateSpec) columns(map[string]json.RawMessage) ([]string, []string) { + return a.keys, a.sums +} + +// inferredColumns derives an event-counts row's shape from the row itself: the +// counts to add (successes, failures, noops, skips, total) are its numeric +// fields, plus any null one — PuppetDB nulls an aggregate column when a backend +// matched nothing — and everything else, subject_type/subject/summarize_by, +// identifies the row. Those endpoints have a fixed response shape with no +// numeric key field, so nothing summable is mistaken for identity. +func inferredColumns(row map[string]json.RawMessage) ([]string, []string) { + var keys, sums []string + for name, val := range row { + if isJSONNumber(val) || isJSONNull(val) { + sums = append(sums, name) + continue + } + keys = append(keys, name) + } + sort.Strings(keys) + sort.Strings(sums) + return keys, sums +} + +// isJSONNumber reports whether a raw JSON value is a number. +func isJSONNumber(raw json.RawMessage) bool { + v := strings.TrimSpace(string(raw)) + if v == "" { + return false + } + return v[0] == '-' || (v[0] >= '0' && v[0] <= '9') +} + +func isJSONNull(raw json.RawMessage) bool { + return strings.TrimSpace(string(raw)) == "null" +} + +// sumGroup accumulates the rows sharing one grouping key. +type sumGroup struct { + raw json.RawMessage // first contributing row, verbatim + row map[string]json.RawMessage // its decoded fields + totals map[string]float64 // running sum per aggregate column + merged bool // a second row was folded in +} + +// sumRows folds each backend's aggregate rows into one row per grouping key, +// adding the numeric aggregate columns. columns decides, per row, which fields +// are the key and which are summed. +// +// 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. +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 + group *sumGroup + } + var order []slot + groups := map[string]*sumGroup{} + + for _, res := range results { + for _, rec := range res.records { + var row map[string]json.RawMessage + if json.Unmarshal(rec.Raw, &row) != nil { + order = append(order, slot{raw: rec.Raw}) + continue + } + keys, sums := columns(row) + k := groupKey(row, keys) + g, ok := groups[k] + if !ok { + g = &sumGroup{raw: rec.Raw, row: row, totals: map[string]float64{}} + for _, s := range sums { + if n, ok := numberOf(row[s]); ok { + g.totals[s] = n + } + } + groups[k] = g + order = append(order, slot{group: g}) + continue + } + for _, s := range sums { + n, ok := numberOf(row[s]) + if !ok { + continue + } + if _, seen := g.totals[s]; !seen { + // First numeric value for a column the earlier row lacked. + g.totals[s] = 0 + } + g.totals[s] += n + g.merged = true + } + } + } + + out := make([]json.RawMessage, 0, len(order)) + for _, sl := range order { + if sl.group == nil { + out = append(out, sl.raw) + continue + } + out = append(out, sl.group.encode()) + } + return out +} + +// encode renders a group back to JSON, reusing the first row's bytes when +// nothing was added to it. +func (g *sumGroup) encode() json.RawMessage { + if !g.merged { + return g.raw + } + row := make(map[string]json.RawMessage, len(g.row)) + for k, v := range g.row { + row[k] = v + } + for col, total := range g.totals { + row[col] = json.RawMessage(strconv.FormatFloat(total, 'f', -1, 64)) + } + raw, err := json.Marshal(row) + if err != nil { + return g.raw + } + return raw +} + +// 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 +// 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 { + var b strings.Builder + for _, k := range keys { + b.WriteString(k) + b.WriteByte(0) + if v, ok := row[k]; ok { + b.Write(v) + } else { + b.WriteByte(1) + } + b.WriteByte(0) + } + return b.String() +} + +// numberOf decodes a raw JSON number, reporting ok=false for anything else so +// non-numeric aggregate columns are carried through instead of summed. +func numberOf(raw json.RawMessage) (float64, bool) { + if !isJSONNumber(raw) { + return 0, false + } + var n float64 + if json.Unmarshal(raw, &n) != nil { + return 0, false + } + return n, true +} diff --git a/aggregate_test.go b/aggregate_test.go new file mode 100644 index 0000000..f294743 --- /dev/null +++ b/aggregate_test.go @@ -0,0 +1,210 @@ +package main + +import ( + "encoding/json" + "reflect" + "slices" + "testing" +) + +func rows(raws ...string) []record { + out := make([]record, 0, len(raws)) + for _, r := range raws { + out = append(out, record{Raw: json.RawMessage(r)}) + } + return out +} + +// decodeRows turns a merged result set into comparable maps. +func decodeRows(t *testing.T, raws []json.RawMessage) []map[string]any { + t.Helper() + out := make([]map[string]any, 0, len(raws)) + for _, raw := range raws { + var m map[string]any + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatalf("unmarshal %s: %v", raw, err) + } + out = append(out, m) + } + return out +} + +func TestParseAggregate_ExtractWithGroupBy(t *testing.T) { + spec := parseAggregate(`["extract",[["function","count"],"status"],["=","certname","h1"],["group_by","status"]]`) + if spec == nil { + t.Fatal("expected an aggregate spec") + } + if !slices.Equal(spec.keys, []string{"status"}) { + t.Errorf("keys = %v, want [status]", spec.keys) + } + if !slices.Equal(spec.sums, []string{"count"}) { + t.Errorf("sums = %v, want [count]", spec.sums) + } +} + +func TestParseAggregate_GroupByAddsUnextractedField(t *testing.T) { + spec := parseAggregate(`["extract",[["function","count","certname"]],["~","certname",".*"],["group_by","status"]]`) + if spec == nil { + t.Fatal("expected an aggregate spec") + } + if !slices.Equal(spec.keys, []string{"status"}) { + t.Errorf("keys = %v, want [status] from the group_by clause", spec.keys) + } + if !slices.Equal(spec.sums, []string{"count"}) { + t.Errorf("sums = %v, want [count]", spec.sums) + } +} + +func TestParseAggregate_NoFunctionIsNotAggregate(t *testing.T) { + for _, q := range []string{ + ``, + `["=","certname","h1"]`, + `["extract",["certname","hash"],["=","certname","h1"]]`, // projection, still real reports + `not json`, + `["extract"]`, + `{"not":"an array"}`, + } { + if spec := parseAggregate(q); spec != nil { + t.Errorf("parseAggregate(%q) = %+v, want nil", q, spec) + } + } +} + +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"}`)}, + }, spec.columns) + + want := []map[string]any{ + {"count": float64(7), "status": "changed"}, + {"count": float64(3), "status": "failed"}, + } + if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) { + t.Errorf("merged = %v, want %v", got, want) + } +} + +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"}`)}, + }, spec.columns) + + want := []map[string]any{ + {"count": float64(3), "status": "changed"}, + {"count": float64(2), "status": "skipped"}, + } + if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) { + t.Errorf("merged = %v, want %v", got, want) + } +} + +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}, + }, spec.columns) + + if len(merged) != 1 || string(merged[0]) != raw { + t.Errorf("merged = %s, want the row verbatim %s", merged, raw) + } +} + +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"}`)}, + }, spec.columns) + + want := []map[string]any{{"count": float64(5), "status": "changed"}} + if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) { + t.Errorf("merged = %v, want the numeric value preserved %v", got, want) + } +} + +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"}`)}, + }, spec.columns) + + want := []map[string]any{{"count": float64(6), "status": "changed"}} + if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) { + t.Errorf("merged = %v, want %v", got, want) + } +} + +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"}`)}, + }, spec.columns) + + if len(merged) != 2 || string(merged[0]) != `"surprise"` { + t.Fatalf("merged = %s, want the non-object row kept as-is", merged) + } +} + +func TestSumRows_NoKeyColumnsCollapseToOneRow(t *testing.T) { + // ["extract",[["function","count"]],...] is a whole-estate count: one row + // per backend, and the merged answer is their sum. + spec := parseAggregate(`["extract",[["function","count"]],["=","certname","h1"]]`) + if spec == nil { + t.Fatal("expected an aggregate spec") + } + merged := sumRows([]backendResult{ + {name: "new", records: rows(`{"count":10}`)}, + {name: "old", records: rows(`{"count":32}`)}, + }, spec.columns) + + want := []map[string]any{{"count": float64(42)}} + if got := decodeRows(t, merged); !reflect.DeepEqual(got, want) { + t.Errorf("merged = %v, want %v", got, want) + } +} + +func TestInferredColumns_SplitsCountsFromIdentity(t *testing.T) { + var row map[string]json.RawMessage + if err := json.Unmarshal([]byte(`{"subject_type":"certname","subject":{"title":"h1"},"failures":1,"successes":2,"skips":null}`), &row); err != nil { + t.Fatal(err) + } + keys, sums := inferredColumns(row) + if !slices.Equal(keys, []string{"subject", "subject_type"}) { + t.Errorf("keys = %v, want [subject subject_type]", keys) + } + // A null count is an empty aggregate, not part of the row's identity. + if !slices.Equal(sums, []string{"failures", "skips", "successes"}) { + t.Errorf("sums = %v, want [failures skips successes]", sums) + } +} + +func TestSumRows_EventCountsPerSubject(t *testing.T) { + merged := sumRows([]backendResult{ + {name: "new", 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( + `{"subject_type":"certname","subject":{"title":"h1"},"failures":3,"successes":4,"noops":1,"skips":0}`, + )}, + }, inferredColumns) + + got := decodeRows(t, merged) + want := []map[string]any{ + {"subject_type": "certname", "subject": map[string]any{"title": "h1"}, + "failures": float64(4), "successes": float64(6), "noops": float64(1), "skips": float64(0)}, + {"subject_type": "certname", "subject": map[string]any{"title": "h2"}, + "failures": float64(0), "successes": float64(5), "noops": float64(0), "skips": float64(0)}, + } + if !reflect.DeepEqual(got, want) { + t.Errorf("merged = %v, want %v", got, want) + } +} diff --git a/server.go b/server.go index d0c61e6..516802c 100644 --- a/server.go +++ b/server.go @@ -16,11 +16,13 @@ import ( ) const ( - factsPath = "/pdb/query/v4/facts" - nodesPath = "/pdb/query/v4/nodes" - reportsPath = "/pdb/query/v4/reports" - eventsPath = "/pdb/query/v4/events" - queryV4 = "/pdb/query/v4/" + factsPath = "/pdb/query/v4/facts" + nodesPath = "/pdb/query/v4/nodes" + reportsPath = "/pdb/query/v4/reports" + eventsPath = "/pdb/query/v4/events" + eventCountsPath = "/pdb/query/v4/event-counts" + aggregateEventCountsPath = "/pdb/query/v4/aggregate-event-counts" + queryV4 = "/pdb/query/v4/" // PuppetDB only sends this when the request carries include_total=true. recordsHeader = "X-Records" @@ -70,9 +72,11 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) { case factsPath: s.serveMerged(w, r, factsPath, s.mergeFactsResponse) case reportsPath: - s.serveUnion(w, r, reportsPath, reportKey) + s.serveReports(w, r) case eventsPath: s.serveUnion(w, r, eventsPath, rawKey) + case eventCountsPath, aggregateEventCountsPath: + s.serveSummed(w, r, r.URL.Path, inferredColumns) default: if isReportSubResource(r.URL.Path) { s.serveFirstHolder(w, r) @@ -131,6 +135,37 @@ func (s *Server) serveUnion(w http.ResponseWriter, r *http.Request, path string, writeJSON(w, page.apply(merged)) } +// An `extract` query with a `function` column returns synthetic aggregate rows that carry no identity, so they are summed rather than unioned. +func (s *Server) serveReports(w http.ResponseWriter, r *http.Request) { + if spec := parseAggregate(r.URL.Query().Get("query")); spec != nil { + s.serveSummed(w, r, reportsPath, spec.columns) + return + } + s.serveUnion(w, r, reportsPath, reportKey) +} + +// Merged rows are fewer than the backends' combined records, so include_total reports the merged count rather than a sum of X-Records. +func (s *Server) serveSummed(w http.ResponseWriter, r *http.Request, path string, columns func(map[string]json.RawMessage) ([]string, []string)) { + 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 := sumRows(s.byPrecedence(alive), columns) + sortRecords(merged, page.order) + if page.wantTotal { + w.Header().Set(recordsHeader, strconv.Itoa(len(merged))) + } + writeJSON(w, page.apply(merged)) +} + // A backend without the report answers 404, indistinguishable from a failure, so every backend is consulted before serving empty. func (s *Server) serveFirstHolder(w http.ResponseWriter, r *http.Request) { results := s.fanOut(r.Context(), r.URL.Path, r.URL.Query()) diff --git a/server_test.go b/server_test.go index ff6cdc7..8bf7800 100644 --- a/server_test.go +++ b/server_test.go @@ -7,6 +7,7 @@ import ( "net/http" "net/http/httptest" "net/url" + "reflect" "slices" "strconv" "strings" @@ -592,3 +593,235 @@ func TestHandler_ReportsOneBackendDown(t *testing.T) { t.Errorf("merged reports = %v, want [r1]", got) } } + +// eventCount builds one /event-counts row for a certname. +func eventCount(certname string, successes, failures, noops, skips int) string { + return `{"subject_type":"certname","subject":{"title":"` + certname + `"},` + + `"successes":` + strconv.Itoa(successes) + + `,"failures":` + strconv.Itoa(failures) + + `,"noops":` + strconv.Itoa(noops) + + `,"skips":` + strconv.Itoa(skips) + `}` +} + +// counts decodes a numeric column out of a merged aggregate body, in order. +func counts(t *testing.T, body []byte, field string) []float64 { + t.Helper() + var rows []map[string]any + if err := json.Unmarshal(body, &rows); err != nil { + t.Fatalf("unmarshal %s: %v", body, err) + } + out := make([]float64, 0, len(rows)) + for _, r := range rows { + n, _ := r[field].(float64) + out = append(out, n) + } + return out +} + +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)) + + rec := doGetParams(t, srv.Handler(), eventCountsPath, url.Values{ + "query": {`["=","certname","h1"]`}, + "summarize_by": {"certname"}, + }) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + // Precedence puts new (prefer) first, 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) + } + if got := counts(t, rec.Body.Bytes(), "failures"); !slices.Equal(got, []float64{4, 0}) { + t.Errorf("failures = %v, want [4 0]", got) + } +} + +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)) + + 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) + } + // summarize_by must reach the backends verbatim. + for name, fb := range map[string]*fakeBackend{"old": old, "new": nw} { + 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")) + } + } +} + +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)) + + rec := doGetParams(t, srv.Handler(), eventCountsPath, url.Values{ + "summarize_by": {"certname"}, + "include_total": {"true"}, + }) + if got := rec.Header().Get(recordsHeader); got != "1" { + t.Errorf("%s = %q, want 1 (merged rows, not 2)", recordsHeader, got) + } +} + +func TestHandler_AggregateEventCountsSummed(t *testing.T) { + old := newFakeBackend(t, `[]`, `[]`) + old.bodies[aggregateEventCountsPath] = + `[{"successes":2,"failures":1,"noops":0,"skips":3,"total":6,"summarize_by":"certname"}]` + nw := newFakeBackend(t, `[]`, `[]`) + nw.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)) + + rec := doGetParams(t, srv.Handler(), aggregateEventCountsPath, url.Values{"summarize_by": {"certname"}}) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + var got []map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + if len(got) != 1 { + t.Fatalf("expected one summary object, got %d: %s", len(got), rec.Body.String()) + } + want := map[string]any{ + "successes": float64(7), "failures": float64(5), "noops": float64(1), + "skips": float64(3), "total": float64(16), "summarize_by": "certname", + } + if !reflect.DeepEqual(got[0], want) { + t.Errorf("summary = %v, want %v", got[0], want) + } +} + +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] = + `[{"successes":null,"failures":null,"total":null,"summarize_by":"certname"}]` + nw := newFakeBackend(t, `[]`, `[]`) + nw.bodies[aggregateEventCountsPath] = + `[{"successes":3,"failures":0,"total":3,"summarize_by":"certname"}]` + srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) + + rec := doGetParams(t, srv.Handler(), aggregateEventCountsPath, url.Values{"summarize_by": {"certname"}}) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + if got := counts(t, rec.Body.Bytes(), "total"); !slices.Equal(got, []float64{3}) { + t.Errorf("total = %v, want [3]", got) + } +} + +const statusCountQuery = `["extract",[["function","count"],"status"],["~","certname",".*"],["group_by","status"]]` + +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)) + + rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{"query": {statusCountQuery}}) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + var got []map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + byStatus := map[string]float64{} + for _, row := range got { + s, _ := row["status"].(string) + n, _ := row["count"].(float64) + byStatus[s] = n + } + want := map[string]float64{"changed": 7, "failed": 2, "unchanged": 9} + if !reflect.DeepEqual(byStatus, want) { + t.Errorf("counts = %v, want %v", byStatus, want) + } +} + +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)) + + rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{ + "query": {statusCountQuery}, + "include_total": {"true"}, + }) + if got := rec.Header().Get(recordsHeader); got != "1" { + t.Errorf("%s = %q, want 1 (one merged status row)", recordsHeader, got) + } +} + +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)) + + rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{ + "query": {`["extract",["hash","certname"],["=","certname","h1"]]`}, + "order_by": {receiveDesc}, + }) + if got := hashes(t, rec.Body.Bytes()); !slices.Equal(got, []string{"r2", "r1"}) { + t.Errorf("merged reports = %v, want [r2 r1]", got) + } +} + +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)) + + rec := doGetParams(t, srv.Handler(), eventCountsPath, url.Values{"summarize_by": {"certname"}}) + if rec.Code != http.StatusOK { + t.Fatalf("expected 200 serving the survivor, got %d", rec.Code) + } + if got := counts(t, rec.Body.Bytes(), "successes"); !slices.Equal(got, []float64{2}) { + t.Errorf("successes = %v, want [2]", got) + } +} + +func TestHandler_EventCountsBadPagingParam(t *testing.T) { + old := newFakeBackend(t, `[]`, `[]`) + nw := newFakeBackend(t, `[]`, `[]`) + srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) + + rec := doGetParams(t, srv.Handler(), eventCountsPath, url.Values{"limit": {"lots"}}) + if rec.Code != http.StatusBadRequest { + t.Errorf("expected 400 for a malformed limit, got %d", rec.Code) + } +}