Merge pull request 'Sum /events aggregates instead of keeping one backend's row' (#21) from benvin/events-aggregates into main

Reviewed-on: #21

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/<hash>/{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/<name>[/<value>]` query whose
-  `extract` carries a `["function", ...]` column.
+  `/events`, `/nodes`, `/resources`, `/facts` or `/facts/<name>[/<value>]` 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/<name>` 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/<name>`
+    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/<name>[/<value>]` 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,16 @@ 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.
+  - `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.
 

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.

e2e_query_test.go

@@ -13,6 +13,7 @@ import (
 	"sort"
 	"strings"
 	"testing"
+	"time"
 )
 
 var allNodes = []string{nodeAlpha, nodeBeta, nodeGamma, nodeShared}
@@ -649,6 +650,119 @@ 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)
+			}
+		}
+	})
+}
+
+// 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()
+	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) {

routes_test.go

@@ -105,7 +105,6 @@ func TestQueryRoutes_UnsummedRoutesAreTheKnownOnes(t *testing.T) {
 	want := []string{
 		aggregateEventCountsPath,
 		eventCountsPath,
-		eventsPath,
 		factNamesPath,
 		reportsPath + "/<hash>/<sub>",
 	}

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 + "/<name>", 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),
@@ -209,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
@@ -452,6 +456,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)
 }

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 {
@@ -596,6 +609,138 @@ 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)
+	}
+}
+
+// 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.