Merge pull request 'feat: sum aggregate rows across backends on event-counts and /reports' (#8) from benvin/aggregate-sums into main

Reviewed-on: #8

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/<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 /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
+  each).
-  never deduped, so every backend's rows pass through even when byte-identical;
+- **Aggregates** — `extract`/`group_by` rows are counts, not records, so each
-  summing those aggregates across backends is not implemented yet.
+  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
+- `include_total=true` on a union endpoint makes `pdbmux` sum each backend's
-  one merged header. Deduped records are counted once per backend, so the total
+  `X-Records` header into one merged header. Deduped records are counted once per
-  is an upper bound.
+  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.
 

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
+}

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)
+	}
+}

server.go

@@ -16,11 +16,13 @@ import (
 )
 
 const (
-	factsPath   = "/pdb/query/v4/facts"
+	factsPath                = "/pdb/query/v4/facts"
-	nodesPath   = "/pdb/query/v4/nodes"
+	nodesPath                = "/pdb/query/v4/nodes"
-	reportsPath = "/pdb/query/v4/reports"
+	reportsPath              = "/pdb/query/v4/reports"
-	eventsPath  = "/pdb/query/v4/events"
+	eventsPath               = "/pdb/query/v4/events"
-	queryV4     = "/pdb/query/v4/"
+	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())

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)
+	}
+}