From cc71902a0d84610c7a30f9ed37d5f3dcc70e1930 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 6 Sep 2026 15:50:49 +1000 Subject: [PATCH] Sum aggregates on the /facts/ routes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An aggregate row carries no certname, so the per-certname merge collapsed every backend's row into one bucket and served a single backend's count as the estate's — no error, no warning, X-Backends still 2/2. - Route an aggregate query on /facts/[/] to the summing path. - List the injected fact's name in /fact-names instead of hiding it. - Reject an order_by on any field but name, as the backends do. --- README.md | 41 ++++++----- e2e_backend_test.go | 20 ++++++ e2e_query_test.go | 101 +++++++++++++++++++++++++-- factroutes_test.go | 167 +++++++++++++++++++++++++++++++++++++++----- merge.go | 50 ++++++++----- server.go | 40 ++++++++--- 6 files changed, 352 insertions(+), 67 deletions(-) diff --git a/README.md b/README.md index 278ad93..1b2626e 100644 --- a/README.md +++ b/README.md @@ -26,8 +26,8 @@ not PQL) is forwarded verbatim. |---|---| | `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/[/]` | Same fan-out and merge as `/facts`. The path segment is a `name` constraint, so no synthetic `pdbmux_source` record is added unless the path names it. | -| `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. | +| `GET /pdb/query/v4/facts/[/]` | Same fan-out and merge as `/facts`, and an `extract`/`count` query is **summed** the same way. The path segment is a `name` constraint, so no synthetic `pdbmux_source` record is added unless the path names it. | +| `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`/`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. | @@ -70,9 +70,13 @@ paths add two more headers `pdbmux` sets itself, `X-Cache` and `Age` — see - A node present in only one backend always appears (falls back to whichever backend actually returned facts for it). - `/facts/` and `/facts//` are the same records with one - more constraint applied upstream, so they take the same rule. + more constraint applied upstream, so they take the same rule — and the same + aggregate branch, since a count row has no `certname` there either. - **`/fact-names`** — a flat array of strings, not records: **union**, deduped by - the name and re-sorted, ascending unless `order_by` says otherwise. + the name and re-sorted, ascending unless `order_by` says otherwise. `name` is + the only column the entity projects, so an `order_by` on any other field is + rejected with `400`, as the backends reject it. While injection is on the + `pdbmux_source` name is listed too, exactly once (see provenance). `include_total=true` reports the deduped union's size, so the `limit` is applied to the merged list rather than pushed upstream. - **`/reports`, `/events`** — **union**, not a per-node winner. Reports are @@ -83,17 +87,18 @@ 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`, `/resources` or `/facts/[/]` 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`, `/resources` and `/facts/` 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: a count row has no `certname`, so deduping would collapse + every backend's count into one backend's number. A query with no `function` + column — including a plain `extract` projection — still merges by + `certname`. - `/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 @@ -170,9 +175,11 @@ such a query correctly for every operator (`not`, `or`, subqueries) and does not pretend to for some. Read the fact from an unfiltered (or `certname`-filtered) `/facts` response and filter client-side. `/pdb/query/v4/facts/pdbmux_source` is the same story: the route is merged and injection is allowed there, but the -backends hold no record to attach it to. `/pdb/query/v4/fact-names` therefore -leaves the name out of its list — advertising a name no `/facts` response serves -would offer a drilldown with nothing behind it. +backends hold no record to attach it to, so it answers empty. +`/pdb/query/v4/fact-names` still lists the name while injection is on, because +every merged `/facts` response does carry records of it — leaving it out hides a +fact every node has from any client that discovers names there. Turning injection +off removes it, since nothing then produces it. **Not covered:** `/factsets` and `/inventory`. Both carry facts, but `pdbmux` does not merge either today — they take the unmerged pass-through path, where diff --git a/e2e_backend_test.go b/e2e_backend_test.go index d85f483..cfa2523 100644 --- a/e2e_backend_test.go +++ b/e2e_backend_test.go @@ -340,6 +340,26 @@ func (b *backend) query(ctx context.Context, t fatalf, path string, params url.V return rows } +// queryStatus is query without the body, for asserting what a backend rejects. +func (b *backend) queryStatus(ctx context.Context, t fatalf, path string, params url.Values) int { + t.Helper() + target := b.url + path + if len(params) > 0 { + target += "?" + params.Encode() + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil) + if err != nil { + t.Fatalf("building query for %s: %v", b.name, err) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("querying %s%s: %v", b.name, path, err) + } + defer func() { _ = resp.Body.Close() }() + _, _ = io.Copy(io.Discard, resp.Body) + return resp.StatusCode +} + func newNetwork(ctx context.Context, t fatalf) (string, func()) { t.Helper() nw, err := tcnet.New(ctx) diff --git a/e2e_query_test.go b/e2e_query_test.go index 2c2dfac..4f7bb4e 100644 --- a/e2e_query_test.go +++ b/e2e_query_test.go @@ -6,6 +6,7 @@ import ( "context" "encoding/json" "fmt" + "net/http" "net/url" "sort" "testing" @@ -126,10 +127,10 @@ func TestFactNamesAreMerged(t *testing.T) { } seen[name] = true } - // No merged /facts response can serve the name pdbmux owns, so the list must - // not advertise it. - if seen[defaultSourceFact] { - t.Errorf("merged /fact-names advertises %s, which no /facts response carries upstream", defaultSourceFact) + // The merged /facts response carries a record of the name pdbmux owns, so a + // client discovering names here has to see it. + if !seen[defaultSourceFact] { + t.Errorf("merged /fact-names omits %s, which every merged /facts response carries: %v", defaultSourceFact, got) } if got := resp.header.Get(backendsHeader); got != "2/2" { t.Errorf("%s = %q, want %q", backendsHeader, got, "2/2") @@ -158,6 +159,98 @@ func TestFactNamesAreMerged(t *testing.T) { t.Errorf("%s = %q, want the merged count %d", recordsHeader, n, len(got)) } }) + + // name is the only column the entity projects, so pdbmux rejects any other + // order_by field the way the backends do. + t.Run("an order_by on another field is rejected", func(t *testing.T) { + params := url.Values{"order_by": {`[{"field":"bogus","order":"desc"}]`}} + if r := rawGet(t, factNamesPath, params); r.status != http.StatusBadRequest { + t.Errorf("merged /fact-names order_by bogus = HTTP %d, want 400: %s", r.status, r.body) + } + if r := h.a.queryStatus(context.Background(), t, factNamesPath, params); r != http.StatusBadRequest { + t.Errorf("backend %s answered HTTP %d for the same order_by, so 400 is not what it does", h.a.name, r) + } + }) +} + +// A fact aggregate row carries no certname, so the per-certname merge would keep +// one backend's row and drop the other's; only adding the numbers is right. +func TestFactsByNamePathAggregatesAreSummed(t *testing.T) { + ctx := context.Background() + const q = `["extract",[["function","count"]]]` + + for _, path := range []string{factsPath + "/kernel", factsPath + "/kernel/Linux"} { + t.Run(path, func(t *testing.T) { + wantA := backendCount(ctx, t, h.a, path, q) + wantB := backendCount(ctx, t, h.b, path, q) + if wantA == wantB { + t.Fatalf("the fixture gives both backends %d records on %s, so a sum is indistinguishable from one backend's number", wantA, path) + } + + resp := get(t, path, query(q)) + if got := countOf(t, resp.rows(t)); got != wantA+wantB { + t.Fatalf("%s count = %d, want %d (%s=%d + %s=%d)", path, 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"],"value"],["group_by","value"]]` + path := factsPath + "/osfamily" + want := map[string]int{} + for _, b := range []*backend{h.a, h.b} { + for value, n := range countsByValue(t, b.query(ctx, t, path, query(grouped))) { + want[value] += n + } + } + got := countsByValue(t, get(t, path, query(grouped)).rows(t)) + if !equalCounts(got, want) { + t.Errorf("grouped %s counts = %v, want %v", path, got, want) + } + }) + + t.Run("a query with no function column still merges by certname", func(t *testing.T) { + rows := get(t, factsPath+"/kernel", nil).rows(t) + if got := e2eCertnames(rows); !equalStrings(got, allNodes) { + t.Errorf("/facts/kernel certnames = %v, want %v", got, allNodes) + } + if len(rows) != len(allNodes) { + t.Errorf("/facts/kernel returned %d records for %d nodes, so a shared node was not deduped", len(rows), len(allNodes)) + } + }) +} + +// countsByValue reads a ["function","count"] + group_by "value" result set. +func countsByValue(t *testing.T, rows []map[string]any) map[string]int { + t.Helper() + out := map[string]int{} + for _, row := range rows { + value, ok := row["value"].(string) + if !ok { + t.Fatalf("grouped aggregate row has no value: %v", row) + } + n, ok := row["count"].(float64) + if !ok { + t.Fatalf("grouped aggregate row has no numeric count: %v", row) + } + out[value] = int(n) + } + return out +} + +func equalCounts(a, b map[string]int) bool { + if len(a) != len(b) { + return false + } + for k, v := range a { + if b[k] != v { + return false + } + } + return true } func containsString(s []string, v string) bool { diff --git a/factroutes_test.go b/factroutes_test.go index 772b864..ced7655 100644 --- a/factroutes_test.go +++ b/factroutes_test.go @@ -5,6 +5,7 @@ import ( "net/http" "net/url" "reflect" + "slices" "testing" ) @@ -109,6 +110,102 @@ func TestHandler_FactsByNameAndValueMerged(t *testing.T) { } } +// An aggregate row carries no certname, so the per-certname merge would keep one +// backend's row and drop the other's without any error or partial-backend signal. +func TestHandler_FactsByNameAggregateSummed(t *testing.T) { + for _, path := range []string{roleFactPath, roleFactPath + "/web"} { + t.Run(path, func(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[path] = `[{"count":7}]` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[path] = `[{"count":3}]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), path, `["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]", got) + } + if h := rec.Header().Get(backendsHeader); h != "2/2" { + t.Errorf("%s = %q, want 2/2", backendsHeader, h) + } + }) + } +} + +func TestHandler_FactsByNameAggregateGroupedSummed(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[roleFactPath] = `[{"count":2,"value":"web"},{"count":1,"value":"db"}]` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[roleFactPath] = `[{"count":5,"value":"web"}]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), roleFactPath, + `["extract",[["function","count"],"value"],["group_by","value"]]`) + var rows []map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &rows); err != nil { + t.Fatalf("unmarshal %s: %v", rec.Body.String(), err) + } + got := map[string]float64{} + for _, row := range rows { + v, _ := row["value"].(string) + n, _ := row["count"].(float64) + got[v] = n + } + want := map[string]float64{"web": 7, "db": 1} + if !reflect.DeepEqual(got, want) { + t.Errorf("grouped counts = %v, want %v", got, want) + } +} + +// The summing branch must not divert a plain query, including an extract +// projection that carries no function column. +func TestHandler_FactsByNameNonAggregateStillMergedByCertname(t *testing.T) { + for _, q := range []string{"", `["extract",["certname","value"],["~","certname",".*"]]`} { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[roleFactPath] = `[` + fact("h1", "role", "web-a", "") + `,` + fact("h2", "role", "db-a", "") + `]` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[roleFactPath] = `[` + fact("h1", "role", "web-b", "") + `,` + fact("h3", "role", "db-b", "") + `]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), roleFactPath, q) + var rows []recordMeta + if err := json.Unmarshal(rec.Body.Bytes(), &rows); err != nil { + t.Fatalf("query %q: unmarshal %s: %v", q, rec.Body.String(), err) + } + got := map[string]bool{} + for _, row := range rows { + got[row.Certname] = true + } + if len(rows) != 3 || !got["h1"] || !got["h2"] || !got["h3"] { + t.Errorf("query %q: merged %s = %s, want one record per certname", q, roleFactPath, rec.Body.String()) + } + } +} + +// A summed row is not the merged record set the cache stores, so it stays live. +func TestHandler_FactsByNameAggregateNotCached(t *testing.T) { + for _, path := range []string{roleFactPath, roleFactPath + "/web"} { + t.Run(path, func(t *testing.T) { + a := newCountingBackend(t, map[string]string{path: `[{"count":7}]`}) + b := newCountingBackend(t, map[string]string{path: `[{"count":3}]`}) + srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + + const q = `["extract",[["function","count"]]]` + doGet(t, srv.Handler(), path, q) + second := doGet(t, srv.Handler(), path, q) + if got := a.hitCount(path); got != 2 { + t.Errorf("%s aggregates are uncached: %d requests, want 2", path, got) + } + if got := counts(t, second.Body.Bytes(), "count"); !slices.Equal(got, []float64{10}) { + t.Errorf("count = %v, want [10]", got) + } + }) + } +} + func TestHandler_FactsByNameForwardsPathAndQuery(t *testing.T) { a := newFakeBackend(t, `[]`, `[]`) a.bodies[roleFactPath] = `[]` @@ -200,7 +297,7 @@ func TestHandler_FactNamesUnionDedupedAndSorted(t *testing.T) { if rec.Code != http.StatusOK { t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) } - want := []string{"extra_b", "kernel", "only_a", "only_b", "osfamily"} + want := []string{"extra_b", "kernel", "only_a", "only_b", "osfamily", defaultSourceFact} if got := names(t, rec.Body.Bytes()); !reflect.DeepEqual(got, want) { t.Errorf("merged %s = %v, want %v", factNamesPath, got, want) } @@ -219,7 +316,7 @@ func TestHandler_FactNamesOrderByDescending(t *testing.T) { rec := doGetParams(t, srv.Handler(), factNamesPath, url.Values{ "order_by": {`[{"field":"name","order":"desc"}]`}, }) - want := []string{"zone", "osfamily", "kernel"} + want := []string{"zone", defaultSourceFact, "osfamily", "kernel"} if got := names(t, rec.Body.Bytes()); !reflect.DeepEqual(got, want) { t.Errorf("descending %s = %v, want %v", factNamesPath, got, want) } @@ -263,8 +360,8 @@ func TestHandler_FactNamesTotalCountsTheWholeUnion(t *testing.T) { if got := names(t, rec.Body.Bytes()); !reflect.DeepEqual(got, []string{"a1", "a2"}) { t.Errorf("paged %s = %v, want [a1 a2]", factNamesPath, got) } - if got := rec.Header().Get(recordsHeader); got != "4" { - t.Errorf("%s = %q, want the deduped union size 4", recordsHeader, got) + if got := rec.Header().Get(recordsHeader); got != "5" { + t.Errorf("%s = %q, want the deduped union size 5", recordsHeader, got) } if got, _ := a.params(factNamesPath); got.Get("limit") != "" { t.Errorf("backend a got limit=%q, want none: a total cannot be counted from a window", got.Get("limit")) @@ -276,27 +373,36 @@ func TestHandler_FactNamesRejectsBadPaging(t *testing.T) { b := newFakeBackend(t, `[]`, `[]`) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) - rec := doGetParams(t, srv.Handler(), factNamesPath, url.Values{"limit": {"-1"}}) - if rec.Code != http.StatusBadRequest { - t.Errorf("status %d, want 400: %s", rec.Code, rec.Body.String()) + for _, params := range []url.Values{ + {"limit": {"-1"}}, + // name is the only column the entity projects, so anything else is a 400 + // upstream and must be one here. + {"order_by": {`[{"field":"bogus","order":"desc"}]`}}, + } { + rec := doGetParams(t, srv.Handler(), factNamesPath, params) + if rec.Code != http.StatusBadRequest { + t.Errorf("%v: status %d, want 400: %s", params, rec.Code, rec.Body.String()) + } } } -// No merged /facts response can serve the name pdbmux owns, so the overview must -// not advertise it — and must keep it once the feature is off. -func TestHandler_FactNamesDropsTheOwnedName(t *testing.T) { - body := factNamesBody(defaultSourceFact, "osfamily") +// The merged /facts response carries records of the name pdbmux owns, so the +// overview has to list it — and must not once nothing produces it. +func TestHandler_FactNamesListsTheOwnedName(t *testing.T) { for _, tc := range []struct { - name string - enabled bool - want []string + name string + enabled bool + upstream string + want []string }{ - {"enabled", true, []string{"osfamily"}}, - {"disabled", false, []string{"osfamily", defaultSourceFact}}, + {"enabled", true, factNamesBody("osfamily"), []string{"osfamily", defaultSourceFact}}, + {"disabled", false, factNamesBody("osfamily"), []string{"osfamily"}}, + {"listed once when a backend reports it too", true, factNamesBody(defaultSourceFact, "osfamily"), []string{"osfamily", defaultSourceFact}}, + {"kept when a backend reports it and the feature is off", false, factNamesBody(defaultSourceFact, "osfamily"), []string{"osfamily", defaultSourceFact}}, } { t.Run(tc.name, func(t *testing.T) { a := newFakeBackend(t, `[]`, `[]`) - a.bodies[factNamesPath] = body + a.bodies[factNamesPath] = tc.upstream b := newFakeBackend(t, `[]`, `[]`) b.bodies[factNamesPath] = `[]` cfg := testConfig(a.srv.URL, b.srv.URL, mergeStatic) @@ -311,12 +417,35 @@ func TestHandler_FactNamesDropsTheOwnedName(t *testing.T) { } } +// The owned name is part of the union, so it is counted and ordered with it. +func TestHandler_FactNamesOwnedNameIsCountedAndOrdered(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[factNamesPath] = factNamesBody("osfamily", "zone") + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[factNamesPath] = `[]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGetParams(t, srv.Handler(), factNamesPath, url.Values{ + "order_by": {`[{"field":"name","order":"desc"}]`}, + "include_total": {"true"}, + }) + want := []string{"zone", defaultSourceFact, "osfamily"} + if got := names(t, rec.Body.Bytes()); !reflect.DeepEqual(got, want) { + t.Errorf("descending %s = %v, want %v", factNamesPath, got, want) + } + if got := rec.Header().Get(recordsHeader); got != "3" { + t.Errorf("%s = %q, want 3", recordsHeader, got) + } +} + func TestHandler_FactNamesServesEmptyArray(t *testing.T) { a := newFakeBackend(t, `[]`, `[]`) a.bodies[factNamesPath] = `[]` b := newFakeBackend(t, `[]`, `[]`) b.bodies[factNamesPath] = `[]` - srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + cfg := testConfig(a.srv.URL, b.srv.URL, mergeStatic) + cfg.SourceFactEnabled = false + srv := newTestServer(cfg) rec := doGet(t, srv.Handler(), factNamesPath, "") if got := rec.Body.String(); got != "[]\n" { @@ -333,7 +462,7 @@ func TestHandler_FactNamesSurvivesOneBackend(t *testing.T) { srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) rec := doGet(t, srv.Handler(), factNamesPath, "") - if got := names(t, rec.Body.Bytes()); !reflect.DeepEqual(got, []string{"kernel"}) { + if got := names(t, rec.Body.Bytes()); !reflect.DeepEqual(got, []string{"kernel", defaultSourceFact}) { t.Errorf("%s = %v, want the surviving backend's names", factNamesPath, got) } if h := rec.Header().Get(backendsHeader); h != "1/2" { diff --git a/merge.go b/merge.go index 74c455b..ece3b96 100644 --- a/merge.go +++ b/merge.go @@ -172,29 +172,33 @@ func mergeFacts(results []backendResult, owner func(certname string) string, inj } // mergeFactNames unions the backends' /fact-names arrays, dedupes by the name -// itself and re-sorts, since each backend only ordered its own slice. The name -// pdbmux owns is dropped for the same reason its fact records are: while the -// feature is on, no merged /facts response can serve it. -func mergeFactNames(results []backendResult, suppress *sourceInjector, desc bool) []json.RawMessage { +// itself and re-sorts, since each backend only ordered its own slice. owned is +// the fact name pdbmux injects, or "" while injection is off: it is listed +// whether or not a backend reports it, because the merged /facts response +// carries records of that name. +func mergeFactNames(results []backendResult, owned string, desc bool) []json.RawMessage { seen := map[string]bool{} raws := []json.RawMessage{} values := []any{} + add := func(raw json.RawMessage) { + key := factNameKey(raw) + if seen[key] { + return + } + seen[key] = true + var v any + _ = json.Unmarshal(raw, &v) // an undecodable element sorts as null + raws = append(raws, raw) + values = append(values, v) + } for _, res := range results { for _, rec := range res.records { - var name string - if json.Unmarshal(rec.Raw, &name) == nil && suppress.claims(name) { - suppress.suppressed++ - continue - } - key := string(rec.Raw) - if seen[key] { - continue - } - seen[key] = true - var v any - _ = json.Unmarshal(rec.Raw, &v) // an undecodable element sorts as null - raws = append(raws, rec.Raw) - values = append(values, v) + add(rec.Raw) + } + } + if owned != "" { + if raw, err := json.Marshal(owned); err == nil { + add(raw) } } idx := make([]int, len(raws)) @@ -215,6 +219,16 @@ func mergeFactNames(results []backendResult, suppress *sourceInjector, desc bool return out } +// Two encoders can spell one name differently — Go escapes <, > and & — so a +// name is deduped on its decoded value, not on its bytes. +func factNameKey(raw json.RawMessage) string { + var name string + if json.Unmarshal(raw, &name) == nil { + return "s\x00" + name + } + return "r\x00" + string(raw) +} + func environmentOf(recs []record) string { for _, rec := range recs { if rec.Environment != "" { diff --git a/server.go b/server.go index 75f3d4c..4f23639 100644 --- a/server.go +++ b/server.go @@ -26,6 +26,9 @@ const ( aggregateEventCountsPath = "/pdb/query/v4/aggregate-event-counts" queryV4 = "/pdb/query/v4/" + // The only column /fact-names projects, so the only one it can be ordered on. + factNamesColumn = "name" + // PuppetDB only sends this when the request carries include_total=true. recordsHeader = "X-Records" @@ -143,7 +146,7 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) { s.serveSummed(w, r, r.URL.Path, inferredColumns) default: if name, valued := factsSubPath(r.URL.Path); name != "" { - s.serveMerged(w, r, r.URL.Path, s.mergeFactsByNameResponse(r, name, valued)) + s.serveFactsByName(w, r, name, valued) return } if isReportSubResource(r.URL.Path) { @@ -185,8 +188,9 @@ func isFactsSubPath(path string) bool { // because each backend only ordered its own slice. openvoxdb projects a single // DISTINCT `name` column and defaults to name-ascending // (src/puppetlabs/puppetdb/query_eng/engine.clj:488-498, -// src/puppetlabs/puppetdb/http/handlers.clj:349-362), so an order_by can only be -// on that column and its direction is all this route reads. +// src/puppetlabs/puppetdb/http/handlers.clj:349-362), so an order_by on any +// other field is rejected as openvoxdb rejects it, and the direction is all this +// route reads. func (s *Server) serveFactNames(w http.ResponseWriter, r *http.Request) { in := r.URL.Query() page, err := parsePaging(in) @@ -194,12 +198,20 @@ func (s *Server) serveFactNames(w http.ResponseWriter, r *http.Request) { http.Error(w, err.Error(), http.StatusBadRequest) return } + for _, f := range page.order { + if f.Field != factNamesColumn { + http.Error(w, fmt.Sprintf("order_by field must be %q, got %q", factNamesColumn, f.Field), http.StatusBadRequest) + return + } + } desc := len(page.order) > 0 && page.order[0].Desc - // Built only for the name pdbmux owns: a string list has no query shape for - // the injection gate to read, and nothing is ever injected here. - suppress := s.newSourceInjector("", true) - suppress.disableInject() + // The merged /facts response carries the injected fact, so the list of names + // has to carry its name; nothing produces it while injection is off. + owned := "" + if s.cfg.SourceFactEnabled { + owned = s.cfg.SourceFact + } upstream := page.upstreamParams(in) if page.wantTotal { @@ -213,8 +225,7 @@ func (s *Server) serveFactNames(w http.ResponseWriter, r *http.Request) { if err != nil { return cachedResponse{}, err } - merged := mergeFactNames(alive, suppress, desc) - suppress.logSuppressed(s.log) + merged := mergeFactNames(alive, owned, desc) resp := cachedResponse{Body: encodeRecords(page.apply(merged)), Records: -1} s.countBackends(&resp, alive) if page.wantTotal { @@ -289,6 +300,17 @@ func (s *Server) serveUnion(w http.ResponseWriter, r *http.Request, path string, }) } +// Both path forms are the facts entity with a name (and value) constraint ANDed +// on, so an aggregate over them carries no certname for the per-certname merge +// to key on, and is summed instead — the same split /nodes makes. +func (s *Server) serveFactsByName(w http.ResponseWriter, r *http.Request, name string, valued bool) { + if spec := parseAggregate(r.URL.Query().Get("query")); spec != nil { + s.serveSummed(w, r, r.URL.Path, spec.columns) + return + } + s.serveMerged(w, r, r.URL.Path, s.mergeFactsByNameResponse(r, name, valued)) +} + // 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 {