From c886617d72fb01f721880fce664084c2cf29ffe7 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Mon, 7 Sep 2026 17:58:37 +1000 Subject: [PATCH 1/2] Sum /events aggregates instead of keeping one backend's row An extract carrying a ["function", ...] column returns counts, not events, so the union's verbatim-record key folded two backends' identical rows into one number. - Route /events through the central aggregate guard with its own fan-out path - Drop its unsummed opt-out so the route-table property tests cover it - Document the combined path on /events --- README.md | 21 ++++++++------- e2e_query_test.go | 66 +++++++++++++++++++++++++++++++++++++++++++++++ routes_test.go | 1 - server.go | 12 ++++----- server_test.go | 38 +++++++++++++++++++++++++++ 5 files changed, 121 insertions(+), 17 deletions(-) diff --git a/README.md b/README.md index 8c935c4..5cf41b7 100644 --- a/README.md +++ b/README.md @@ -30,7 +30,7 @@ not PQL) is forwarded verbatim. | `GET /pdb/query/v4/fact-names` | Fan out to all and serve the **union** of the flat name arrays, deduped and re-sorted, re-paged across backends, plus the `pdbmux_source` name while injection is on. `order_by` is only valid on `name`. | | `GET /pdb/query/v4/resources` | An `extract` query with a `function` column is fanned out and **combined**; 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. | +| `GET /pdb/query/v4/events` | Fan out to all and serve the **union**, deduped by record identity, re-ordered and re-paged. An `extract` query with a `function` column is **combined** instead. | | `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 none does. | @@ -89,14 +89,14 @@ 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 **combined**, not deduped. This covers `/event-counts`, `/aggregate-event-counts`, and any `/reports`, - `/nodes`, `/resources`, `/facts` or `/facts/[/]` query whose - `extract` carries a `["function", ...]` column. + `/events`, `/nodes`, `/resources`, `/facts` or `/facts/[/]` query + whose `extract` carries a `["function", ...]` column. - The grouping key is the row's full set of non-aggregate columns: for - `/reports`, `/nodes`, `/resources`, `/facts` and `/facts/` they come - from the query — the plain `extract` fields, the row-function columns and - 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. + `/reports`, `/events`, `/nodes`, `/resources`, `/facts` and `/facts/` + they come from the query — the plain `extract` fields, the row-function + columns and 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`, `/facts` 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 @@ -157,8 +157,9 @@ paths add two more headers `pdbmux` sets itself, `X-Cache` and `Age` — see fetched and the window cut after the fold, which an aggregate's row count — one per distinct group value — keeps affordable. `include_total` still reports the merged group count. - - A `/reports` query with no `function` column is a projection of real reports, - not an aggregate, and stays on the union path. + - A `/reports` or `/events` query with no `function` column is a projection of + real records, not an aggregate, and stays on the union path — so an event + stored identically in two backends is still served once. - `include_total=true` on a combined endpoint reports the **merged** row count, not the sum of the backends' `X-Records`, since shared keys collapse. diff --git a/e2e_query_test.go b/e2e_query_test.go index f95606e..3263204 100644 --- a/e2e_query_test.go +++ b/e2e_query_test.go @@ -649,6 +649,72 @@ func countsByName(t *testing.T, rows []map[string]any) map[string]int { return out } +// An events count row is an aggregate, not an event, so the union's +// verbatim-record dedupe would fold two backends' equal counts into one number. +func TestEventsAggregatesAreSummed(t *testing.T) { + ctx := context.Background() + const q = `["extract",[["function","count"]]]` + + wantA := backendCount(ctx, t, h.a, eventsPath, q) + wantB := backendCount(ctx, t, h.b, eventsPath, q) + if wantA == 0 || wantB == 0 { + t.Fatalf("the fixture gives %s %d and %s %d events, so a sum proves nothing", h.a.name, wantA, h.b.name, wantB) + } + + resp := get(t, eventsPath, query(q)) + if got := countOf(t, resp.rows(t)); got != wantA+wantB { + t.Fatalf("/events count = %d, want %d (%s=%d + %s=%d)", got, wantA+wantB, h.a.name, wantA, h.b.name, wantB) + } + if got := resp.header.Get(backendsHeader); got != "2/2" { + t.Errorf("%s = %q, want %q", backendsHeader, got, "2/2") + } + + t.Run("grouped counts are summed per key", func(t *testing.T) { + const grouped = `["extract",[["function","count"],"certname"],["group_by","certname"]]` + want := map[string]int{} + for _, b := range []*backend{h.a, h.b} { + for cn, n := range countsByCertname(t, b.query(ctx, t, eventsPath, query(grouped))) { + want[cn] += n + } + } + // shared reports to both backends, so its row is the one a dedupe would + // leave frozen at a single backend's number. + if want[nodeShared] < 2 { + t.Fatalf("%s has %d events across the estate, so its merged row cannot show a sum", nodeShared, want[nodeShared]) + } + got := countsByCertname(t, get(t, eventsPath, query(grouped)).rows(t)) + if !reflect.DeepEqual(got, want) { + t.Errorf("grouped /events counts = %v, want %v", got, want) + } + }) + + t.Run("a query with no function column stays on the union", func(t *testing.T) { + rows := get(t, eventsPath, nil).rows(t) + if len(rows) != wantA+wantB { + t.Fatalf("/events returned %d records, want %d: the union dropped or duplicated events", len(rows), wantA+wantB) + } + for _, row := range rows { + if _, ok := row["resource_title"].(string); !ok { + t.Fatalf("/events record is not an event: %v", row) + } + } + }) +} + +// countsByCertname reads a ["function","count"] + group_by "certname" result set. +func countsByCertname(t *testing.T, rows []map[string]any) map[string]int { + t.Helper() + out := map[string]int{} + for _, row := range rows { + certname, ok := row["certname"].(string) + if !ok { + t.Fatalf("grouped aggregate row has no certname: %v", row) + } + out[certname] = int(aggregateNumber(t, row, "count")) + } + return out +} + // The provenance fact must name the backend whose data won, and must be absent // from the query shapes it would corrupt. func TestSourceFactInjectionAndGating(t *testing.T) { diff --git a/routes_test.go b/routes_test.go index d5f6300..8a3ace1 100644 --- a/routes_test.go +++ b/routes_test.go @@ -105,7 +105,6 @@ func TestQueryRoutes_UnsummedRoutesAreTheKnownOnes(t *testing.T) { want := []string{ aggregateEventCountsPath, eventCountsPath, - eventsPath, factNamesPath, reportsPath + "//", } diff --git a/server.go b/server.go index 2d38958..15762c3 100644 --- a/server.go +++ b/server.go @@ -151,6 +151,7 @@ var queryRoutes = []route{ // /resources has no cross-backend record identity, so only its aggregates merge. {name: resourcesPath, matches: pathIs(resourcesPath), fanOut: resourcesPath, serve: (*Server).proxyUnmerged}, {name: reportsPath, matches: pathIs(reportsPath), fanOut: reportsPath, serve: (*Server).serveReports}, + {name: eventsPath, matches: pathIs(eventsPath), fanOut: eventsPath, serve: (*Server).serveEvents}, {name: factsPath + "/", matches: isFactsSubPath, serve: (*Server).serveFactsByName}, { name: factNamesPath, @@ -158,12 +159,6 @@ var queryRoutes = []route{ serve: (*Server).serveFactNames, unsummed: "the union dedupes names across backends, so a count of it is not the sum of the backends' counts", }, - { - name: eventsPath, - matches: pathIs(eventsPath), - serve: (*Server).serveEvents, - unsummed: "unioned on the verbatim record; summing aggregates here would change what the route answers, so it is a change of its own", - }, { name: eventCountsPath, matches: pathIs(eventCountsPath), @@ -452,6 +447,11 @@ func (s *Server) serveReports(w http.ResponseWriter, r *http.Request) { s.serveUnion(w, r, reportsPath, reportKey) } +// Only a plain query reaches here: openvoxdb serves events from the same generic +// query engine as every other entity, so an extract carrying a ["function", ...] +// column is an aggregate handleQuery has already diverted — +// src/puppetlabs/puppetdb/http/handlers.clj:178-189 and +// src/puppetlabs/puppetdb/query_eng/engine.clj:1120-1210,1889-1911,2719-2733. func (s *Server) serveEvents(w http.ResponseWriter, r *http.Request) { s.serveUnion(w, r, eventsPath, rawKey) } diff --git a/server_test.go b/server_test.go index 8768c59..dd53045 100644 --- a/server_test.go +++ b/server_test.go @@ -596,6 +596,44 @@ func TestHandler_EventsDedupedByIdentity(t *testing.T) { } } +// An events aggregate row is a count, not an event, so the union's +// verbatim-record key would collapse two backends' identical rows into one. +func TestHandler_EventsAggregatesAreCombined(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[eventsPath] = `[{"count":5}]` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[eventsPath] = `[{"count":5}]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), eventsPath, `["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{10}) { + t.Errorf("count = %v, want [10]; the identical rows were deduped instead of added", got) + } +} + +func TestHandler_EventsGroupedAggregatesCombinePerKey(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[eventsPath] = `[{"status":"success","count":5,"max":7},{"status":"failure","count":1,"max":2}]` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[eventsPath] = `[{"status":"success","count":5,"max":3}]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + const q = `["extract",[["function","count"],["function","max","line"],"status"],["group_by","status"]]` + rec := doGet(t, srv.Handler(), eventsPath, q) + 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{10, 1}) { + t.Errorf("counts = %v, want [10 1]", got) + } + if got := counts(t, rec.Body.Bytes(), "max"); !slices.Equal(got, []float64{7, 2}) { + t.Errorf("max = %v, want [7 2]; the column was combined by the wrong operation", got) + } +} + func TestHandler_ReportSubResourceFromHoldingBackend(t *testing.T) { // Only a holds report r1, so its logs come from a; an unmerged pass-through // to whichever backend answered first could have 404'd. -- 2.47.3 From 8baa511ef979d5f09f01599243a0e97a99786606 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Mon, 7 Sep 2026 18:23:48 +1000 Subject: [PATCH 2/2] Refuse an /events aggregate that also asks for distinct_resources distinct_resources sends /events to openvoxdb's legacy compiler, which has no function or group_by, so every backend failed and the client saw a generic 502 instead of the reason. - Refuse an aggregate carrying a truthy distinct_resources with 400, before any fan-out - Read the param the way openvoxdb does, so any capitalisation of "true" counts - Leave non-aggregate distinct_resources queries and every other route alone - Document the refusal --- README.md | 7 +++ aggregate.go | 21 +++++++++ e2e_query_test.go | 48 +++++++++++++++++++++ server.go | 11 ++++- server_test.go | 107 ++++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 193 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 5cf41b7..46005e5 100644 --- a/README.md +++ b/README.md @@ -160,6 +160,13 @@ paths add two more headers `pdbmux` sets itself, `X-Cache` and `Age` — see - A `/reports` or `/events` query with no `function` column is a projection of real records, not an aggregate, and stays on the union path — so an event stored identically in two backends is still served once. + - `distinct_resources=true` on an `/events` `extract` with a `function` column + is refused with **400** naming the incompatibility, before any fan-out. + openvoxdb answers the distinct-resources form of `/events` from its legacy + compiler, which supports neither `function` nor `group_by`, so every backend + fails and the mistake would otherwise surface as a `502`. A + `distinct_resources` query with no `function` column is untouched and still + fans out; no other endpoint `pdbmux` combines honours the parameter. - `include_total=true` on a combined endpoint reports the **merged** row count, not the sum of the backends' `X-Records`, since shared keys collapse. diff --git a/aggregate.go b/aggregate.go index 41da8b2..df23457 100644 --- a/aggregate.go +++ b/aggregate.go @@ -2,7 +2,9 @@ package main import ( "encoding/json" + "errors" "fmt" + "net/url" "sort" "strconv" "strings" @@ -167,6 +169,25 @@ func groupByClash(name string) error { return fmt.Errorf("group_by names the aggregate column %q: pdbmux folds that column across backends, so it cannot also be a grouping key", name) } +const distinctResourcesParam = "distinct_resources" + +// errDistinctResourcesAggregate refuses an aggregate asking for the +// distinct-resources form of /events. openvoxdb answers that form from its +// legacy events compiler, whose operator map carries neither function nor +// group_by (src/puppetlabs/puppetdb/query_eng.clj:189-190 and +// src/puppetlabs/puppetdb/query/events.clj:183-187), so every backend fails and +// a fan-out could only report a client mistake as an outage. +var errDistinctResourcesAggregate = errors.New( + "distinct_resources cannot be combined with an extract function column: openvoxdb answers a distinct_resources /events query from its legacy compiler, which supports neither function nor group_by") + +// distinctResources reports whether a request asks for the distinct-resources +// form. openvoxdb coerces the param with Boolean/parseBoolean +// (src/puppetlabs/puppetdb/http/query.clj:245-250, applied at query.clj:296), +// so any capitalisation of "true" turns it on and everything else reads false. +func distinctResources(params url.Values) bool { + return strings.EqualFold(params.Get(distinctResourcesParam), "true") +} + // rewriteAvg replaces the client's avg column with the sum and count of the same // expression, so the true weighted average can be computed from the shards: // Postgres avg(x) is sum(x)/count(x), and both of those do combine. diff --git a/e2e_query_test.go b/e2e_query_test.go index 3263204..6778f08 100644 --- a/e2e_query_test.go +++ b/e2e_query_test.go @@ -13,6 +13,7 @@ import ( "sort" "strings" "testing" + "time" ) var allNodes = []string{nodeAlpha, nodeBeta, nodeGamma, nodeShared} @@ -701,6 +702,53 @@ func TestEventsAggregatesAreSummed(t *testing.T) { }) } +// distinct_resources moves /events onto openvoxdb's legacy compiler, which has +// no function or group_by: both backends fail identically, so without a refusal +// a client mistake reaches the caller as an outage-shaped 502. +func TestEventsAggregateWithDistinctResourcesIsRefused(t *testing.T) { + ctx := context.Background() + distinct := func(q string) url.Values { + v := url.Values{ + "distinct_resources": {"true"}, + "distinct_start_time": {fixtureTime(-24 * time.Hour)}, + "distinct_end_time": {fixtureTime(time.Hour)}, + } + if q != "" { + v.Set("query", q) + } + return v + } + + params := distinct(`["extract",[["function","count"]]]`) + resp := rawGet(t, eventsPath, params) + if resp.status != http.StatusBadRequest { + t.Fatalf("status %d, want 400: %s", resp.status, resp.body) + } + if !strings.Contains(string(resp.body), "distinct_resources") { + t.Errorf("refusal %q does not name the incompatibility", resp.body) + } + if got := h.a.queryStatus(ctx, t, eventsPath, params); got == http.StatusOK { + t.Errorf("backend %s answered the aggregate with 200, so the refusal is unnecessary", h.a.name) + } + + t.Run("a plain distinct_resources query is still served", func(t *testing.T) { + params := distinct("") + want := len(h.a.query(ctx, t, eventsPath, params)) + len(h.b.query(ctx, t, eventsPath, params)) + if want == 0 { + t.Fatal("the fixture yields no distinct-resources events, so the union proves nothing") + } + rows := get(t, eventsPath, params).rows(t) + if len(rows) != want { + t.Fatalf("/events with distinct_resources returned %d records, want %d", len(rows), want) + } + for _, row := range rows { + if _, ok := row["resource_title"].(string); !ok { + t.Fatalf("/events record is not an event: %v", row) + } + } + }) +} + // countsByCertname reads a ["function","count"] + group_by "certname" result set. func countsByCertname(t *testing.T, rows []map[string]any) map[string]int { t.Helper() diff --git a/server.go b/server.go index 15762c3..9ad914c 100644 --- a/server.go +++ b/server.go @@ -204,12 +204,21 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) { if rt.unsummed == "" { // The one place an aggregate is read, so a query pdbmux cannot fold is // refused here rather than by whichever handler happens to notice. - spec, err := parseAggregate(r.URL.Query().Get("query")) + in := r.URL.Query() + spec, err := parseAggregate(in.Get("query")) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } if spec != nil { + // Of the routes that reach an aggregate, only /events honours + // distinct_resources; every other one rejects it as an unsupported + // query parameter (src/puppetlabs/puppetdb/http/handlers.clj:178-189, + // 475-496 and src/puppetlabs/puppetdb/http/query.clj:273-277). + if r.URL.Path == eventsPath && distinctResources(in) { + http.Error(w, errDistinctResourcesAggregate.Error(), http.StatusBadRequest) + return + } path := rt.fanOut if path == "" { path = r.URL.Path diff --git a/server_test.go b/server_test.go index dd53045..7f39cde 100644 --- a/server_test.go +++ b/server_test.go @@ -95,6 +95,19 @@ func (fb *fakeBackend) params(path string) (url.Values, bool) { return v, ok } +// hits reports how many requests the backend received for a path. +func (fb *fakeBackend) hits(path string) int { + fb.mu.Lock() + defer fb.mu.Unlock() + n := 0 + for _, p := range fb.rawPaths { + if p == path { + n++ + } + } + return n +} + // sawRawPath reports whether the backend was asked for a path with exactly that // escaping. func (fb *fakeBackend) sawRawPath(p string) bool { @@ -634,6 +647,100 @@ func TestHandler_EventsGroupedAggregatesCombinePerKey(t *testing.T) { } } +// distinct_resources sends /events to openvoxdb's legacy compiler, which has no +// function or group_by, so an aggregate asking for it is refused with the reason +// rather than fanned out into two identical failures and a 502. +func TestHandler_EventsAggregateWithDistinctResourcesIsRefused(t *testing.T) { + const agg = `["extract",[["function","count"]]]` + distinct := func(q, value string) url.Values { + v := url.Values{"distinct_start_time": {"2026-07-01T00:00:00Z"}, "distinct_end_time": {"2026-07-02T00:00:00Z"}} + if q != "" { + v.Set("query", q) + } + if value != "" { + v.Set(distinctResourcesParam, value) + } + return v + } + + t.Run("refused without any fan-out", func(t *testing.T) { + for _, value := range []string{"true", "TRUE", "True"} { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[eventsPath] = `[{"count":5}]` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[eventsPath] = `[{"count":5}]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGetParams(t, srv.Handler(), eventsPath, distinct(agg, value)) + if rec.Code != http.StatusBadRequest { + t.Fatalf("distinct_resources=%s: status %d, want 400: %s", value, rec.Code, rec.Body.String()) + } + if !strings.Contains(rec.Body.String(), distinctResourcesParam) { + t.Errorf("distinct_resources=%s: refusal %q does not name the incompatibility", value, rec.Body.String()) + } + if got := a.hits(eventsPath) + b.hits(eventsPath); got != 0 { + t.Errorf("distinct_resources=%s: backends saw %d requests, want 0", value, got) + } + } + }) + + // openvoxdb reads the param with Boolean/parseBoolean, so only "true" turns + // the distinct form on and any other spelling is an ordinary aggregate. + t.Run("a non-true value is still an ordinary aggregate", func(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[eventsPath] = `[{"count":5}]` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[eventsPath] = `[{"count":5}]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGetParams(t, srv.Handler(), eventsPath, distinct(agg, "yes")) + if rec.Code != http.StatusOK { + t.Fatalf("status %d, want 200: %s", rec.Code, rec.Body.String()) + } + if got := counts(t, rec.Body.Bytes(), "count"); !slices.Equal(got, []float64{10}) { + t.Errorf("count = %v, want [10]", got) + } + }) + + t.Run("a plain distinct_resources query still fans out", func(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[eventsPath] = `[` + event("h1", "r1", "Package[nginx]") + `]` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[eventsPath] = `[` + event("h1", "r2", "Service[nginx]") + `]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGetParams(t, srv.Handler(), eventsPath, distinct("", "true")) + if rec.Code != http.StatusOK { + t.Fatalf("status %d, want 200: %s", rec.Code, rec.Body.String()) + } + if a.hits(eventsPath) != 1 || b.hits(eventsPath) != 1 { + t.Fatalf("backends saw %d and %d requests, want 1 each", a.hits(eventsPath), b.hits(eventsPath)) + } + got, _ := a.params(eventsPath) + if got.Get(distinctResourcesParam) != "true" { + t.Errorf("backend a got distinct_resources=%q, want %q", got.Get(distinctResourcesParam), "true") + } + }) + + // Only /events honours the param; on any other route openvoxdb refuses it + // itself, so pdbmux must not shadow that with a refusal of its own. + t.Run("another route keeps combining", func(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[resourcesPath] = `[{"count":5}]` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[resourcesPath] = `[{"count":5}]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGetParams(t, srv.Handler(), resourcesPath, distinct(agg, "true")) + if rec.Code != http.StatusOK { + t.Fatalf("status %d, want 200: %s", rec.Code, rec.Body.String()) + } + if got := counts(t, rec.Body.Bytes(), "count"); !slices.Equal(got, []float64{10}) { + t.Errorf("count = %v, want [10]", got) + } + }) +} + func TestHandler_ReportSubResourceFromHoldingBackend(t *testing.T) { // Only a holds report r1, so its logs come from a; an unmerged pass-through // to whichever backend answered first could have 404'd. -- 2.47.3