From 49ce1293de5034ef5decefc5854161adf7c28e46 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 6 Sep 2026 15:08:33 +1000 Subject: [PATCH 1/2] Sum /facts aggregates across backends A `/facts` aggregate row carries no certname, so the per-certname fact merge collapsed every backend's rows into one bucket and served a single backend's numbers. - Route a `/facts` query whose `extract` carries a `function` column to serveSummed, as /nodes, /resources and /reports already do - Document which extract functions combine correctly across backends --- README.md | 37 ++++++++++++---------- cache_test.go | 19 ++++++++++++ e2e_query_test.go | 58 ++++++++++++++++++++++++++-------- server.go | 11 ++++++- server_test.go | 79 +++++++++++++++++++++++++++++++++++++++++++++++ source_test.go | 20 ++++-------- 6 files changed, 180 insertions(+), 44 deletions(-) diff --git a/README.md b/README.md index cda7844..21d44b6 100644 --- a/README.md +++ b/README.md @@ -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,22 @@ 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**; query a single + backend directly for those. - `/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 +479,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/` 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/` 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 diff --git a/cache_test.go b/cache_test.go index 02cf087..9f29901 100644 --- a/cache_test.go +++ b/cache_test.go @@ -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}]`}) diff --git a/e2e_query_test.go b/e2e_query_test.go index 11968e7..dffde52 100644 --- a/e2e_query_test.go +++ b/e2e_query_test.go @@ -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 diff --git a/server.go b/server.go index a16b96a..da39330 100644 --- a/server.go +++ b/server.go @@ -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 { diff --git a/server_test.go b/server_test.go index 73b038f..8b2454f 100644 --- a/server_test.go +++ b/server_test.go @@ -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. diff --git a/source_test.go b/source_test.go index ae63bf0..f33d287 100644 --- a/source_test.go +++ b/source_test.go @@ -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) } } From 71b823fe71d6a9385824ac3b824e7650cb19b22c Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 6 Sep 2026 15:10:08 +1000 Subject: [PATCH 2/2] Name the extract functions that do not merge --- README.md | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 21d44b6..3fa9cf0 100644 --- a/README.md +++ b/README.md @@ -89,8 +89,10 @@ paths add two more headers `pdbmux` sets itself, `X-Cache` and `Age` — see - 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**; query a single - backend directly for those. + 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