Sum /facts aggregates across backends
ci/woodpecker/pr/build Pipeline was successful
ci/woodpecker/pr/test Pipeline was successful
ci/woodpecker/pr/pre-commit Pipeline was canceled

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
This commit is contained in:
2026-09-06 15:08:33 +10:00
parent c87ecf65e8
commit 49ce1293de
6 changed files with 180 additions and 44 deletions
+21 -16
View File
@@ -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/<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
+19
View File
@@ -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}]`})
+45 -13
View File
@@ -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
+10 -1
View File
@@ -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 {
+79
View File
@@ -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.
+6 -14
View File
@@ -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)
}
}