Merge pull request 'Sum /facts aggregates across backends' (#17) from benvin/facts-aggregates into main

Reviewed-on: #17
2026-09-06 16:10:39 +10:00
parent 0720d0930b 71b823fe71
commit 1410f9f603
6 changed files with 182 additions and 44 deletions
@@ -25,7 +25,7 @@ not PQL) is forwarded verbatim.
 | Path | Behaviour |
 |---|---|
 | `GET /pdb/query/v4/nodes` | Fan out to all backends, dedupe by `certname`, keep the record with the newer `report_timestamp`, stamped with the winning backend's name (see provenance). An `extract`/`count` query is **summed** instead. |
-| `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), plus a synthetic `pdbmux_source` fact naming it. |
+| `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), plus a synthetic `pdbmux_source` fact naming it. An `extract`/`count` query is **summed** instead. |
 | `GET /pdb/query/v4/resources` | An `extract`/`count` query is fanned out and **summed**; any other query is an unmerged pass-through. |
 | `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. |
@@ -75,17 +75,24 @@ paths add two more headers `pdbmux` sets itself, `X-Cache` and `Age` — see
 - **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 any `/reports`,
-  `/nodes` or `/resources` query whose `extract` carries a `["function", ...]`
-  column.
+  `/nodes`, `/facts` or `/resources` query whose `extract` carries a
+  `["function", ...]` column.
  - The grouping key is the row's non-aggregate fields: for `/reports`,
-    `/nodes` and `/resources` 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.
-  - On `/nodes` this takes precedence over the `certname` merge: a count row has
-    no `certname`, so deduping would collapse every backend's count into one
-    backend's number. A `/nodes` query with no `function` column — including a
-    plain `extract` projection — still merges by `certname`.
+    `/nodes`, `/facts` and `/resources` 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.
+  - On `/nodes` and `/facts` this takes precedence over the `certname` merge: an
+    aggregate row has no `certname`, so deduping would collapse every backend's
+    rows into one backend's numbers. A query with no `function` column —
+    including a plain `extract` projection — still merges by `certname`.
+  - PuppetDB accepts `count`, `sum`, `avg`, `min`, `max`, `to_string` and
+    `jsonb_typeof` as `extract` functions. Only `count` and `sum` are additive,
+    so only those two merge correctly. `avg`, `min` and `max` are folded like any
+    other numeric column and their merged value is **wrong**; `to_string` and
+    `jsonb_typeof` are row functions whose non-numeric column is carried through
+    from the first backend to report the key. Query a single backend directly
+    for any of those five.
  - `/resources` has no cross-backend record identity to dedupe on, so only its
    aggregate queries merge; everything else stays an unmerged pass-through.
  - Rows sharing a key collapse into one with their numeric columns summed. A key
@@ -474,11 +481,11 @@ have drained.
 | `PDBMUX_E2E_TUNNEL_IMAGE` | `docker.io/library/alpine:3` |
 | `PDBMUX_E2E_NODE_LOOKUP` | path to a `node-lookup` binary (else `PATH`, else skipped) |

-Three tests skip rather than assert, each naming a known gap and failing if that
-gap closes: `/facts` aggregates are not summed, `/pdb/query/v4/facts/<name>` is
-served unmerged (so Puppetboard's single-fact drilldown silently loses the other
-backend's nodes), and `/fact-names` is likewise unmerged (so the facts overview
-lists only the first backend's fact names).
+Two tests skip rather than assert, each naming a known gap and failing if that
+gap closes: `/pdb/query/v4/facts/<name>` is served unmerged (so Puppetboard's
+single-fact drilldown silently loses the other backend's nodes), and
+`/fact-names` is likewise unmerged (so the facts overview lists only the first
+backend's fact names).

 ## Deployment

@@ -1241,6 +1241,25 @@ func TestHandler_NodesAggregateNotCached(t *testing.T) {
 	}
 }

+func TestHandler_FactsAggregateNotCached(t *testing.T) {
+	const q = `["extract",[["function","count"]]]`
+	a := newCountingBackend(t, map[string]string{factsPath: `[{"count":7}]`})
+	b := newCountingBackend(t, map[string]string{factsPath: `[{"count":10}]`})
+	srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
+
+	first := doGet(t, srv.Handler(), factsPath, q)
+	second := doGet(t, srv.Handler(), factsPath, q)
+	if first.Code != http.StatusOK || second.Code != http.StatusOK {
+		t.Fatalf("statuses %d/%d", first.Code, second.Code)
+	}
+	if got := a.hitCount(factsPath); got != 2 {
+		t.Errorf("/facts aggregates are uncached: %d requests, want 2", got)
+	}
+	if got := counts(t, second.Body.Bytes(), "count"); !slices.Equal(got, []float64{17}) {
+		t.Errorf("count = %v, want [17]", got)
+	}
+}
+
 func TestHandler_ResourcesAggregateNotCached(t *testing.T) {
 	a := newCountingBackend(t, map[string]string{resourcesPath: `[{"count":7}]`})
 	b := newCountingBackend(t, map[string]string{resourcesPath: `[{"count":5}]`})
@@ -6,6 +6,7 @@ import (
 	"context"
 	"encoding/json"
 	"net/url"
+	"reflect"
 	"sort"
 	"testing"
 )
@@ -156,28 +157,59 @@ func TestResourcesAggregatesAreSummed(t *testing.T) {
 	}
 }

-// Known gap: handleQuery's `case factsPath:` goes straight to serveMerged, with
-// none of the parseAggregate branch /nodes, /resources and /reports have, so a
-// /facts aggregate is fed to the certname-keyed merge. Count rows carry an empty
-// certname, collapse into one bucket, and the response is whichever backend owns
-// that bucket rather than the sum. Tracked separately; this test records the gap
-// and fails once it closes so it can be turned into a real assertion.
-func TestFactsAggregatesAreNotSummed(t *testing.T) {
+// A fact count row carries no certname, so the per-certname fact merge would
+// keep one backend's rows and drop the other's; only adding the numbers is right.
+func TestFactsAggregatesAreSummed(t *testing.T) {
 	ctx := context.Background()
 	const q = `["extract",[["function","count"]]]`

 	wantA := backendCount(ctx, t, h.a, factsPath, q)
 	wantB := backendCount(ctx, t, h.b, factsPath, q)
 	if wantA == wantB {
-		t.Fatalf("the fixture gives both backends %d facts, so this test cannot tell a sum from one backend's number", wantA)
+		t.Fatalf("the fixture gives both backends %d facts, so a sum is indistinguishable from one backend's number", wantA)
 	}

-	got := countOf(t, get(t, factsPath, query(q)).rows(t))
-	if got == wantA+wantB {
-		t.Fatalf("/facts count = %d, which is the correct sum: the aggregate gap has closed, so assert this properly and drop the skip", got)
+	resp := get(t, factsPath, query(q))
+	if got := countOf(t, resp.rows(t)); got != wantA+wantB {
+		t.Fatalf("/facts count = %d, want %d (%s=%d + %s=%d)", got, wantA+wantB, h.a.name, wantA, h.b.name, wantB)
 	}
-	t.Skipf("known gap: /facts aggregates do not route to serveSummed, so the count is %d (backend %s alone) instead of %d",
-		got, h.a.name, wantA+wantB)
+	if got := resp.header.Get(backendsHeader); got != "2/2" {
+		t.Errorf("%s = %q, want %q", backendsHeader, got, "2/2")
+	}
+
+	const grouped = `["extract",[["function","count"],"name"],["group_by","name"]]`
+	want := map[string]int{}
+	for _, b := range []*backend{h.a, h.b} {
+		for name, n := range countsByName(t, b.query(ctx, t, factsPath, query(grouped))) {
+			want[name] += n
+		}
+	}
+	got := countsByName(t, get(t, factsPath, query(grouped)).rows(t))
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("grouped /facts counts = %v, want %v", got, want)
+	}
+	// The aggregate path must not inject provenance, which a group_by on name would expose.
+	if n, ok := got[defaultSourceFact]; ok {
+		t.Errorf("grouped /facts counts include %d synthetic %s rows", n, defaultSourceFact)
+	}
+}
+
+// countsByName reads a ["function","count"] + group_by "name" result set.
+func countsByName(t *testing.T, rows []map[string]any) map[string]int {
+	t.Helper()
+	out := map[string]int{}
+	for _, row := range rows {
+		name, ok := row["name"].(string)
+		if !ok {
+			t.Fatalf("grouped aggregate row has no name: %v", row)
+		}
+		n, ok := row["count"].(float64)
+		if !ok {
+			t.Fatalf("grouped aggregate row has no numeric count: %v", row)
+		}
+		out[name] = int(n)
+	}
+	return out
 }

 // The provenance fact must name the backend whose data won, and must be absent
@@ -131,7 +131,7 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) {
 	case resourcesPath:
 		s.serveResources(w, r)
 	case factsPath:
-		s.serveMerged(w, r, factsPath, s.mergeFactsResponse(r))
+		s.serveFacts(w, r)
 	case reportsPath:
 		s.serveReports(w, r)
 	case eventsPath:
@@ -212,6 +212,15 @@ func (s *Server) serveUnion(w http.ResponseWriter, r *http.Request, path string,
 	})
 }

+// Aggregate rows carry no certname, so the per-certname fact merge would keep only one backend's; they take the summing path instead.
+func (s *Server) serveFacts(w http.ResponseWriter, r *http.Request) {
+	if spec := parseAggregate(r.URL.Query().Get("query")); spec != nil {
+		s.serveSummed(w, r, factsPath, spec.columns)
+		return
+	}
+	s.serveMerged(w, r, factsPath, s.mergeFactsResponse(r))
+}
+
 // A count row carries no certname, so the certname-keyed merge would collapse every backend's count into one backend's; aggregates take the summing path instead.
 func (s *Server) serveNodes(w http.ResponseWriter, r *http.Request) {
 	if spec := parseAggregate(r.URL.Query().Get("query")); spec != nil {
@@ -945,6 +945,85 @@ func TestHandler_NodesNonAggregateStillMergedByCertname(t *testing.T) {
 	}
 }

+func TestHandler_FactsAggregateSummed(t *testing.T) {
+	// A count row carries no certname, so the per-certname fact merge would have
+	// kept whichever backend owned the empty-certname bucket.
+	a := newFakeBackend(t, `[]`, `[{"count":7}]`)
+	b := newFakeBackend(t, `[]`, `[{"count":10}]`)
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), factsPath, `["extract",[["function","count"]]]`)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
+	}
+	if got := counts(t, rec.Body.Bytes(), "count"); !slices.Equal(got, []float64{17}) {
+		t.Errorf("count = %v, want [17]", got)
+	}
+	if got := rec.Header().Get(backendsHeader); got != "2/2" {
+		t.Errorf("%s = %q, want 2/2", backendsHeader, got)
+	}
+	if _, ok := b.params(factsPath); !ok {
+		t.Error("second backend was never asked for the fact count")
+	}
+}
+
+func TestHandler_FactsAggregateGroupedSummed(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[{"count":4,"name":"osfamily"},{"count":1,"name":"only_a"}]`)
+	b := newFakeBackend(t, `[]`, `[{"count":3,"name":"osfamily"}]`)
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), factsPath,
+		`["extract",[["function","count"],"name"],["~","certname",".*"],["group_by","name"]]`)
+	var got []map[string]any
+	if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil {
+		t.Fatal(err)
+	}
+	byName := map[string]float64{}
+	for _, row := range got {
+		n, _ := row["name"].(string)
+		byName[n], _ = row["count"].(float64)
+	}
+	want := map[string]float64{"osfamily": 7, "only_a": 1}
+	if !reflect.DeepEqual(byName, want) {
+		t.Errorf("counts = %v, want %v", byName, want)
+	}
+}
+
+// include_total on a summed response reports merged rows, not the backends' own totals.
+func TestHandler_FactsAggregateRecordsIsMergedRowCount(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[{"count":4,"name":"osfamily"}]`)
+	a.totals[factsPath] = 1
+	b := newFakeBackend(t, `[]`, `[{"count":3,"name":"osfamily"}]`)
+	b.totals[factsPath] = 1
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGetParams(t, srv.Handler(), factsPath, url.Values{
+		"query":         {`["extract",[["function","count"],"name"],["group_by","name"]]`},
+		"include_total": {"true"},
+	})
+	if got := rec.Header().Get(recordsHeader); got != "1" {
+		t.Errorf("%s = %q, want 1", recordsHeader, got)
+	}
+}
+
+func TestHandler_FactsNonAggregateStillMergedByCertname(t *testing.T) {
+	// Regression: routing aggregates to the summing path must not divert plain
+	// queries, including an extract projection that carries no function column.
+	a := newFakeBackend(t, `[]`, `[`+fact("h1", "role", "web", "")+`]`)
+	b := newFakeBackend(t, `[]`, `[`+fact("h2", "role", "db", "")+`]`)
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	for _, q := range []string{
+		`["~","certname",".*"]`,
+		`["extract",["certname","name","value"],["~","certname",".*"]]`,
+	} {
+		rec := doGet(t, srv.Handler(), factsPath, q)
+		if got := rec.Body.String(); !strings.Contains(got, `"h1"`) || !strings.Contains(got, `"h2"`) {
+			t.Errorf("query %s: body = %s, want both backends' facts merged", q, got)
+		}
+	}
+}
+
 func TestHandler_ResourcesAggregateSummed(t *testing.T) {
 	// /resources is otherwise an unmerged pass-through, so before this the
 	// landing page's resource total was whichever backend answered first.
@@ -328,10 +328,8 @@ func TestHandler_ProjectionWithoutNameColumnCarriesUpstreamValue(t *testing.T) {
 	}
 }

-// Covers only that an aggregate /facts response gains no synthetic record and no
-// rewritten row. It does not cover whether the aggregate rows are correct:
-// /facts has no parseAggregate branch, so its rows take the certname merge
-// instead of serveSummed and are not summed across backends.
+// A /facts aggregate is summed, and the summed row must gain no synthetic record
+// and no stamp.
 func TestHandler_SourceNotInjectedOnFactsAggregate(t *testing.T) {
 	a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`)
 	b := newFakeBackend(t, `[]`, `[]`)
@@ -343,17 +341,11 @@ func TestHandler_SourceNotInjectedOnFactsAggregate(t *testing.T) {
 	if _, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact); n != 0 {
 		t.Errorf("aggregate response gained %d synthetic records: %s", n, rec.Body.String())
 	}
-	// Every row must be one an upstream actually returned: an aggregate row pdbmux
-	// invented or rewrote would change the count the client sees.
-	var raws []json.RawMessage
-	if err := json.Unmarshal(rec.Body.Bytes(), &raws); err != nil {
-		t.Fatal(err)
+	if strings.Contains(rec.Body.String(), defaultSourceFact) {
+		t.Errorf("aggregate rows were stamped: %s", rec.Body.String())
 	}
-	upstream := []string{`{"count":3}`, `{"count":2}`}
-	for _, raw := range raws {
-		if !slices.Contains(upstream, string(raw)) {
-			t.Errorf("aggregate row %s is not an upstream row", raw)
-		}
+	if got := counts(t, rec.Body.Bytes(), "count"); !slices.Equal(got, []float64{5}) {
+		t.Errorf("count = %v, want [5]", got)
 	}
 }