From c8efa26383dc69b4b461a66c281204f0865ad95f Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 6 Sep 2026 15:18:29 +1000 Subject: [PATCH 1/6] Merge the /facts/ and /fact-names routes Both fell to the unmerged pass-through, so one backend's answer was served as if it were the estate's: Puppetboard's fact drilldown lost the other backend's nodes and its facts overview lost that backend's fact names. - Serve /facts/ and /facts// through the /facts merge. - Serve /fact-names as a deduped, re-sorted, re-paged union of name arrays. - Gate provenance on the path: only /facts/ may be injected. - Keep the owned fact name out of /fact-names while the feature is on. - Cache both alongside the merged /facts and /nodes record sets. - Turn the two recorded e2e gaps into positive assertions. --- README.md | 34 ++-- e2e_puppetboard_test.go | 36 ++-- e2e_query_test.go | 111 ++++++++++++ factroutes_test.go | 362 ++++++++++++++++++++++++++++++++++++++++ merge.go | 45 +++++ server.go | 104 +++++++++++- source.go | 8 + 7 files changed, 658 insertions(+), 42 deletions(-) create mode 100644 factroutes_test.go diff --git a/README.md b/README.md index cda7844..278ad93 100644 --- a/README.md +++ b/README.md @@ -26,6 +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/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. | @@ -67,6 +69,12 @@ paths add two more headers `pdbmux` sets itself, `X-Cache` and `Age` — see facts from the first backend in configured order that holds it. - 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. +- **`/fact-names`** — a flat array of strings, not records: **union**, deduped by + the name and re-sorted, ascending unless `order_by` says otherwise. + `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 immutable history, so a node's reports can legitimately exist in more than one backend and all of them belong in the merged view. Reports dedupe on `hash`; @@ -148,6 +156,10 @@ matters more. Only the outer query is inspected: a `name` filter inside an `in`/`select_facts` subquery narrows which *nodes* match, not which facts come back, so injection still happens; +- the path is `/facts/` for any other fact. The path segment is the same + outer `name` constraint, so only `/facts/pdbmux_source` may carry the record; + `/facts//` never does, since the pinned value need not equal the + backend name the record holds; - injection is turned off (see `source_fact_enabled`). **Not supported in v1: server-side filtering on the fact.** A query that selects @@ -156,8 +168,11 @@ the backends like any other, and they return nothing, because the fact does not exist upstream. `pdbmux` does not evaluate the AST itself, so it cannot answer 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. The same applies to the -`/pdb/query/v4/facts/` route, which is served unmerged pass-through. +`/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. **Not covered:** `/factsets` and `/inventory`. Both carry facts, but `pdbmux` does not merge either today — they take the unmerged pass-through path, where @@ -316,9 +331,11 @@ not answering. ## Caching -`pdbmux` caches merged `/nodes` and `/facts` record sets **in memory** so a busy -Puppetboard does not re-fan-out the same query every few seconds. Everything else -runs uncached — including `extract`/`count` aggregates on those two paths, and +`pdbmux` caches merged `/nodes`, `/facts`, `/facts/[/]` and +`/fact-names` record sets **in memory** so a busy Puppetboard does not re-fan-out +the same query every few seconds — its facts overview and fact drilldown are two +of the pages that hit hardest. Everything else runs uncached — including +`extract`/`count` aggregates on those paths, and the `/pdb/meta/v1/*` and `/metrics/*` endpoints, which are served live on every request. The cache is an interface, and `/reports` gets its own (S3-backed) backend later without further handler changes. @@ -474,11 +491,8 @@ 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). +One test skips rather than asserts, naming a known gap and failing if that gap +closes: `/facts` aggregates are not summed. ## Deployment diff --git a/e2e_puppetboard_test.go b/e2e_puppetboard_test.go index 00586c9..d168732 100644 --- a/e2e_puppetboard_test.go +++ b/e2e_puppetboard_test.go @@ -143,29 +143,20 @@ func TestPuppetboardNodeDetail(t *testing.T) { } } -// The facts overview lists fact names, which come from the unmerged /fact-names -// route, so only the first backend's names reach the page today. +// The facts overview lists fact names, which come from the merged /fact-names +// route, so both backends' names have to reach the page. func TestPuppetboardFactsOverview(t *testing.T) { body := pbPage(t, pbAllEnvs+"/facts") - for _, name := range []string{"osfamily", "kernel", "only_a"} { + // only_a lives in backend A alone; only_b and extra_b in backend B alone. + for _, name := range []string{"osfamily", "kernel", "only_a", "only_b", "extra_b"} { if !strings.Contains(body, name) { t.Errorf("Puppetboard facts overview does not list %s", name) } } - - t.Run("names from the second backend are missing", func(t *testing.T) { - // Known gap: /fact-names has no merge rule, so handleQuery falls through to - // proxyUnmerged and the first backend to answer supplies the whole list. - // Recorded rather than asserted; fails once the gap closes. - if strings.Contains(body, "only_b") { - t.Fatal("the facts overview now lists a fact name only the second backend holds: /fact-names is merged, so assert this properly and drop the skip") - } - t.Skip("known gap: /fact-names is an unmerged pass-through, so only_b and extra_b never reach the facts overview") - }) } -// The single-fact drilldown is the page that exercises the /facts/ path -// route. +// The single-fact drilldown is the page that exercises the merged /facts/ +// path route, so every backend's nodes have to appear on it. func TestPuppetboardFactDrilldown(t *testing.T) { if body := pbPage(t, pbAllEnvs+"/fact/osfamily"); !strings.Contains(body, "osfamily") { t.Errorf("Puppetboard fact page for osfamily does not name it") @@ -189,24 +180,17 @@ func TestPuppetboardFactDrilldown(t *testing.T) { } } - // Known gap: Puppetboard fetches a single fact through - // GET /pdb/query/v4/facts/, a path route with no merge rule, so - // proxyUnmerged streams back the first backend's answer alone. The other - // backend's nodes go missing with no error. Tracked separately; this test - // records the gap and fails once it closes. + // Puppetboard fetches a single fact through GET /pdb/query/v4/facts/, + // so the page is only whole if that path route merges every backend. missing := []string{} for _, cn := range allNodes { if !listed[cn] { missing = append(missing, cn) } } - if len(missing) == 0 { - t.Fatal("the fact drilldown now lists every node: the /facts/ path route is merged, so assert this properly and drop the skip") + if len(missing) > 0 { + t.Fatalf("the fact drilldown omits %v; /pdb/query/v4/facts/ must serve every backend's nodes", missing) } - if !listed[nodeAlpha] { - t.Fatalf("the fact drilldown lists neither backend's nodes (%v missing), which is not the pass-through behaviour under test", missing) - } - t.Skipf("known gap: /pdb/query/v4/facts/ is served unmerged, so the drilldown silently omits %v", missing) } // Reports are the least exercised merged path, and Puppetboard reads them diff --git a/e2e_query_test.go b/e2e_query_test.go index 11968e7..2c2dfac 100644 --- a/e2e_query_test.go +++ b/e2e_query_test.go @@ -5,6 +5,7 @@ package main import ( "context" "encoding/json" + "fmt" "net/url" "sort" "testing" @@ -58,6 +59,116 @@ func TestFactsUnionAcrossBackends(t *testing.T) { } } +// The /facts/ path route carries the same records /facts does, so it takes +// the same merge: every node in the estate, each resolved to one backend. +func TestFactsByNamePathIsMerged(t *testing.T) { + resp := get(t, factsPath+"/osfamily", nil) + rows := resp.rows(t) + + if got := e2eCertnames(rows); !equalStrings(got, allNodes) { + t.Fatalf("merged /facts/osfamily certnames = %v, want %v", got, allNodes) + } + if len(rows) != len(allNodes) { + t.Fatalf("/facts/osfamily returned %d records for %d nodes, so a shared node was not deduped", len(rows), len(allNodes)) + } + if got, _ := factValue(rows, nodeShared, "osfamily"); got != "Debian" { + t.Errorf("osfamily of %s = %v, want Debian from the fresher backend", nodeShared, got) + } + if got := resp.header.Get(backendsHeader); got != "2/2" { + t.Errorf("%s = %q, want %q", backendsHeader, got, "2/2") + } + + t.Run("the name constraint gates injection", func(t *testing.T) { + for _, row := range rows { + if row["name"] == defaultSourceFact { + t.Fatalf("%s was injected into a /facts/ response: %v", defaultSourceFact, row) + } + } + }) + + t.Run("the value sub-route merges too", func(t *testing.T) { + rows := get(t, factsPath+"/osfamily/Debian", nil).rows(t) + want := []string{nodeBeta, nodeGamma, nodeShared} + if got := e2eCertnames(rows); !equalStrings(got, want) { + t.Errorf("/facts/osfamily/Debian certnames = %v, want %v", got, want) + } + }) + + t.Run("a fact only one backend holds survives", func(t *testing.T) { + rows := get(t, factsPath+"/only_b", nil).rows(t) + if got, ok := factValue(rows, nodeBeta, "only_b"); !ok || got != "yes" { + t.Errorf("/facts/only_b for %s = %v (present %v), want yes", nodeBeta, got, ok) + } + }) +} + +// /fact-names is a flat array of strings, so it needs its own union: the merged +// list is every backend's names, deduped and sorted. +func TestFactNamesAreMerged(t *testing.T) { + resp := get(t, factNamesPath, nil) + var got []string + if err := json.Unmarshal(resp.body, &got); err != nil { + t.Fatalf("/fact-names is not a flat string array: %v: %s", err, resp.body) + } + + for _, name := range []string{"osfamily", "kernel", "role", "owner", "only_a", "only_b", "extra_b"} { + if !containsString(got, name) { + t.Errorf("merged /fact-names is missing %s: %v", name, got) + } + } + if !sort.StringsAreSorted(got) { + t.Errorf("merged /fact-names is not sorted ascending: %v", got) + } + seen := map[string]bool{} + for _, name := range got { + if seen[name] { + t.Errorf("merged /fact-names lists %s twice: %v", name, got) + } + 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) + } + if got := resp.header.Get(backendsHeader); got != "2/2" { + t.Errorf("%s = %q, want %q", backendsHeader, got, "2/2") + } + + t.Run("paging and ordering are redone across the union", func(t *testing.T) { + params := url.Values{ + "order_by": {`[{"field":"name","order":"desc"}]`}, + "limit": {"3"}, + "include_total": {"true"}, + } + resp := get(t, factNamesPath, params) + var page []string + if err := json.Unmarshal(resp.body, &page); err != nil { + t.Fatalf("decoding the paged /fact-names response: %v: %s", err, resp.body) + } + want := append([]string{}, got...) + sort.Sort(sort.Reverse(sort.StringSlice(want))) + if len(want) > 3 { + want = want[:3] + } + if !equalStrings(page, want) { + t.Errorf("descending /fact-names page = %v, want %v", page, want) + } + if n := resp.header.Get(recordsHeader); n != fmt.Sprint(len(got)) { + t.Errorf("%s = %q, want the merged count %d", recordsHeader, n, len(got)) + } + }) +} + +func containsString(s []string, v string) bool { + for _, x := range s { + if x == v { + return true + } + } + return false +} + // The shared node's two copies disagree on every fact value; the backend holding // its newer report_timestamp has to win both endpoints. func TestSharedNodeResolvesToTheFresherBackend(t *testing.T) { diff --git a/factroutes_test.go b/factroutes_test.go new file mode 100644 index 0000000..772b864 --- /dev/null +++ b/factroutes_test.go @@ -0,0 +1,362 @@ +package main + +import ( + "encoding/json" + "net/http" + "net/url" + "reflect" + "testing" +) + +const ( + roleFactPath = factsPath + "/role" + osFactPath = factsPath + "/osfamily" + sourceFactURL = factsPath + "/" + defaultSourceFact +) + +func names(t *testing.T, body []byte) []string { + t.Helper() + var out []string + if err := json.Unmarshal(body, &out); err != nil { + t.Fatalf("unmarshal %s: %v", body, err) + } + return out +} + +func factNamesBody(names ...string) string { + b, err := json.Marshal(names) + if err != nil { + panic(err) + } + return string(b) +} + +func TestFactsSubPath(t *testing.T) { + for _, tc := range []struct { + path string + wantName string + wantValue bool + }{ + {factsPath, "", false}, + {factsPath + "/", "", false}, + {roleFactPath, "role", false}, + {roleFactPath + "/web", "role", true}, + {roleFactPath + "/", "", false}, + {roleFactPath + "/web/extra", "", false}, + {nodesPath + "/h1/facts/role", "", false}, + {factNamesPath, "", false}, + } { + name, valued := factsSubPath(tc.path) + if name != tc.wantName || valued != tc.wantValue { + t.Errorf("factsSubPath(%q) = (%q, %v), want (%q, %v)", tc.path, name, valued, tc.wantName, tc.wantValue) + } + } +} + +// The path route carries the same records /facts does, so it merges the same +// way: every backend's nodes survive and a shared node resolves to one owner. +func TestHandler_FactsByNameMerged(t *testing.T) { + 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, "") + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + got := map[string]string{} + var rows []struct { + Certname string `json:"certname"` + Value string `json:"value"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &rows); err != nil { + t.Fatalf("unmarshal %s: %v", rec.Body.String(), err) + } + for _, row := range rows { + got[row.Certname] = row.Value + } + want := map[string]string{"h1": "web-a", "h2": "db-a", "h3": "db-b"} + if !reflect.DeepEqual(got, want) { + t.Errorf("merged %s = %v, want %v", roleFactPath, got, want) + } + if h := rec.Header().Get(backendsHeader); h != "2/2" { + t.Errorf("%s = %q, want 2/2", backendsHeader, h) + } +} + +// The // sub-route is the same entity with one more constraint, so +// it takes the same merge rather than falling through to the pass-through path. +func TestHandler_FactsByNameAndValueMerged(t *testing.T) { + path := roleFactPath + "/web" + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[path] = `[` + fact("h1", "role", "web", "") + `]` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[path] = `[` + fact("h3", "role", "web", "") + `]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), path, "") + var rows []recordMeta + if err := json.Unmarshal(rec.Body.Bytes(), &rows); err != nil { + t.Fatalf("unmarshal %s: %v", rec.Body.String(), err) + } + if len(rows) != 2 { + t.Fatalf("merged %s returned %d records, want both backends': %s", path, len(rows), rec.Body.String()) + } + if h := rec.Header().Get(backendsHeader); h != "2/2" { + t.Errorf("%s = %q, want 2/2", backendsHeader, h) + } +} + +func TestHandler_FactsByNameForwardsPathAndQuery(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[roleFactPath] = `[]` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[roleFactPath] = `[]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + q := `["=","certname","h1"]` + doGet(t, srv.Handler(), roleFactPath, q) + for _, fb := range []*fakeBackend{a, b} { + if got := fb.gotQuery(roleFactPath); got != q { + t.Errorf("backend got query %q on %s, want %q", got, roleFactPath, q) + } + } +} + +// The path is a name constraint, the shape the injection gate already excludes. +func TestHandler_FactsByNameNotInjected(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[osFactPath] = `[` + fact("h1", "osfamily", "RedHat", "") + `]` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[osFactPath] = `[` + fact("h2", "osfamily", "Debian", "") + `]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), osFactPath, "") + if _, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact); n != 0 { + t.Errorf("%s was injected into %s: %s", defaultSourceFact, osFactPath, rec.Body.String()) + } +} + +// Only the source fact's own path may carry it; the upstream record of that name +// is still dropped, as it is on every other query shape. +func TestHandler_FactsBySourceNamePathInjectsAndSuppresses(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[sourceFactURL] = `[` + fact("h1", defaultSourceFact, "stale", "") + `]` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[sourceFactURL] = `[]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), sourceFactURL, "") + got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact) + if n != 1 { + t.Fatalf("%s carried %d %s records, want exactly the synthetic one: %s", sourceFactURL, n, defaultSourceFact, rec.Body.String()) + } + if got["h1"] != "a" { + t.Errorf("%s for h1 = %q, want the owning backend %q", defaultSourceFact, got["h1"], "a") + } +} + +// The // form also pins a value the synthetic record's own value +// need not equal, so it never carries one. +func TestHandler_FactsBySourceNameAndValueNotInjected(t *testing.T) { + path := sourceFactURL + "/a" + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[path] = `[` + fact("h1", defaultSourceFact, "a", "") + `]` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[path] = `[]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), path, "") + if _, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact); n != 0 { + t.Errorf("%s appeared on %s: %s", defaultSourceFact, path, rec.Body.String()) + } +} + +// A name-constrained query on the source fact's own path is still gated by the +// query, not just the path. +func TestHandler_FactsBySourceNamePathRespectsQueryGate(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[sourceFactURL] = `[` + fact("h1", defaultSourceFact, "stale", "") + `]` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[sourceFactURL] = `[]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), sourceFactURL, `["extract",["certname","value"]]`) + if _, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact); n != 0 { + t.Errorf("%s survived an extract projection: %s", defaultSourceFact, rec.Body.String()) + } +} + +func TestHandler_FactNamesUnionDedupedAndSorted(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[factNamesPath] = factNamesBody("kernel", "only_a", "osfamily") + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[factNamesPath] = factNamesBody("extra_b", "kernel", "only_b", "osfamily") + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), factNamesPath, "") + 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"} + if got := names(t, rec.Body.Bytes()); !reflect.DeepEqual(got, want) { + t.Errorf("merged %s = %v, want %v", factNamesPath, got, want) + } + if h := rec.Header().Get(backendsHeader); h != "2/2" { + t.Errorf("%s = %q, want 2/2", backendsHeader, h) + } +} + +func TestHandler_FactNamesOrderByDescending(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[factNamesPath] = factNamesBody("kernel", "osfamily") + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[factNamesPath] = factNamesBody("zone") + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGetParams(t, srv.Handler(), factNamesPath, url.Values{ + "order_by": {`[{"field":"name","order":"desc"}]`}, + }) + want := []string{"zone", "osfamily", "kernel"} + if got := names(t, rec.Body.Bytes()); !reflect.DeepEqual(got, want) { + t.Errorf("descending %s = %v, want %v", factNamesPath, got, want) + } +} + +// Paging is applied to the merged list: each backend can only page its own. +func TestHandler_FactNamesPagedAcrossBackends(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[factNamesPath] = factNamesBody("a1", "a2", "a3") + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[factNamesPath] = factNamesBody("b1", "b2") + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGetParams(t, srv.Handler(), factNamesPath, url.Values{ + "limit": {"2"}, + "offset": {"1"}, + }) + want := []string{"a2", "a3"} + if got := names(t, rec.Body.Bytes()); !reflect.DeepEqual(got, want) { + t.Errorf("paged %s = %v, want %v", factNamesPath, got, want) + } + // Backends are asked for limit+offset with no offset, so the union has enough + // records to page locally. + if got, _ := a.params(factNamesPath); got.Get("limit") != "3" || got.Get("offset") != "" { + t.Errorf("backend a got limit=%q offset=%q, want 3 and none", got.Get("limit"), got.Get("offset")) + } +} + +// A merged total counts the union, so the limit cannot be pushed upstream. +func TestHandler_FactNamesTotalCountsTheWholeUnion(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[factNamesPath] = factNamesBody("a1", "a2", "a3") + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[factNamesPath] = factNamesBody("a3", "b1") + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGetParams(t, srv.Handler(), factNamesPath, url.Values{ + "limit": {"2"}, + "include_total": {"true"}, + }) + 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, _ := 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")) + } +} + +func TestHandler_FactNamesRejectsBadPaging(t *testing.T) { + a := newFakeBackend(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()) + } +} + +// 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") + for _, tc := range []struct { + name string + enabled bool + want []string + }{ + {"enabled", true, []string{"osfamily"}}, + {"disabled", false, []string{"osfamily", defaultSourceFact}}, + } { + t.Run(tc.name, func(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[factNamesPath] = body + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[factNamesPath] = `[]` + cfg := testConfig(a.srv.URL, b.srv.URL, mergeStatic) + cfg.SourceFactEnabled = tc.enabled + srv := newTestServer(cfg) + + rec := doGet(t, srv.Handler(), factNamesPath, "") + if got := names(t, rec.Body.Bytes()); !reflect.DeepEqual(got, tc.want) { + t.Errorf("%s = %v, want %v", factNamesPath, got, tc.want) + } + }) + } +} + +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)) + + rec := doGet(t, srv.Handler(), factNamesPath, "") + if got := rec.Body.String(); got != "[]\n" { + t.Errorf("empty %s body = %q, want %q", factNamesPath, got, "[]\n") + } +} + +// A backend that is down must not black-hole the other's names. +func TestHandler_FactNamesSurvivesOneBackend(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.fail = true + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[factNamesPath] = factNamesBody("kernel") + 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"}) { + t.Errorf("%s = %v, want the surviving backend's names", factNamesPath, got) + } + if h := rec.Header().Get(backendsHeader); h != "1/2" { + t.Errorf("%s = %q, want 1/2", backendsHeader, h) + } +} + +// Both routes are merged record sets, so they use the same cache /facts does. +func TestHandler_FactRoutesAreCached(t *testing.T) { + for _, path := range []string{roleFactPath, roleFactPath + "/web", factNamesPath} { + t.Run(path, func(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[path] = `[]` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[path] = `[]` + srv := newTestServer(cacheTestConfig(a.srv.URL, b.srv.URL)) + + if got := doGet(t, srv.Handler(), path, "").Header().Get(cacheStatusHeader); got != "miss" { + t.Errorf("first %s = %q, want miss", cacheStatusHeader, got) + } + if got := doGet(t, srv.Handler(), path, "").Header().Get(cacheStatusHeader); got != "hit" { + t.Errorf("second %s = %q, want hit", cacheStatusHeader, got) + } + }) + } +} diff --git a/merge.go b/merge.go index fcb530e..74c455b 100644 --- a/merge.go +++ b/merge.go @@ -2,6 +2,7 @@ package main import ( "encoding/json" + "sort" "time" ) @@ -170,6 +171,50 @@ func mergeFacts(results []backendResult, owner func(certname string) string, inj return out } +// 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 { + seen := map[string]bool{} + raws := []json.RawMessage{} + values := []any{} + 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) + } + } + idx := make([]int, len(raws)) + for i := range idx { + idx[i] = i + } + sort.SliceStable(idx, func(a, b int) bool { + c := compareValues(values[idx[a]], values[idx[b]]) + if desc { + return c > 0 + } + return c < 0 + }) + out := make([]json.RawMessage, 0, len(raws)) + for _, i := range idx { + out = append(out, raws[i]) + } + return out +} + func environmentOf(recs []record) string { for _, rec := range recs { if rec.Environment != "" { diff --git a/server.go b/server.go index a16b96a..75f3d4c 100644 --- a/server.go +++ b/server.go @@ -17,6 +17,7 @@ import ( const ( factsPath = "/pdb/query/v4/facts" + factNamesPath = "/pdb/query/v4/fact-names" nodesPath = "/pdb/query/v4/nodes" resourcesPath = "/pdb/query/v4/resources" reportsPath = "/pdb/query/v4/reports" @@ -91,13 +92,13 @@ func (s *Server) StartProbes(ctx context.Context) { s.health.Start(ctx) } // StopProbes stops the probing goroutines and waits for them to exit. func (s *Server) StopProbes() { s.health.Stop() } -// cacheFor picks the cache backing a request. Merged /facts and /nodes record -// sets share the in-memory cache; every other path is uncached until the reports -// cache lands, and a new backend is a case here rather than a change to any -// handler. +// cacheFor picks the cache backing a request. The merged /nodes and fact record +// sets — /facts, /facts/[/] and /fact-names — share the in-memory +// cache; every other path is uncached until the reports cache lands, and a new +// backend is a case here rather than a change to any handler. func (s *Server) cacheFor(path string, params url.Values) (Cache, bool) { - switch path { - case factsPath, nodesPath: + switch { + case path == factsPath, path == nodesPath, path == factNamesPath, isFactsSubPath(path): // An aggregate row is a summed count, not the merged record set the // cache was built for, so it stays on the live path. if parseAggregate(params.Get("query")) != nil { @@ -132,6 +133,8 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) { s.serveResources(w, r) case factsPath: s.serveMerged(w, r, factsPath, s.mergeFactsResponse(r)) + case factNamesPath: + s.serveFactNames(w, r) case reportsPath: s.serveReports(w, r) case eventsPath: @@ -139,6 +142,10 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) { case eventCountsPath, aggregateEventCountsPath: 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)) + return + } if isReportSubResource(r.URL.Path) { s.serveFirstHolder(w, r) return @@ -147,6 +154,76 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) { } } +// factsSubPath matches /pdb/query/v4/facts/ and /pdb/query/v4/facts//, +// returning the fact name the path constrains and whether it also pins a value. +// openvoxdb serves both from the facts entity, ANDing ["=","name",] (and +// ["=","value",]) onto the request's own query, so the record shape is +// exactly /facts' — src/puppetlabs/puppetdb/http/handlers.clj:283-302 and +// src/puppetlabs/puppetdb/http/query.clj:136-143,193-209. +func factsSubPath(path string) (name string, valued bool) { + rest, ok := strings.CutPrefix(path, factsPath+"/") + if !ok { + return "", false + } + name, value, hasValue := strings.Cut(rest, "/") + if name == "" { + return "", false + } + if hasValue && (value == "" || strings.Contains(value, "/")) { + return "", false + } + return name, hasValue +} + +func isFactsSubPath(path string) bool { + name, _ := factsSubPath(path) + return name != "" +} + +// /fact-names answers with a flat array of fact-name strings rather than +// certname-keyed records, so it gets its own union: dedupe by name and re-sort, +// 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. +func (s *Server) serveFactNames(w http.ResponseWriter, r *http.Request) { + in := r.URL.Query() + page, err := parsePaging(in) + if err != nil { + http.Error(w, err.Error(), 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() + + upstream := page.upstreamParams(in) + if page.wantTotal { + // A merged total counts the whole union, so the window cannot be pushed + // upstream; the full name list is small enough to fetch. + upstream.Del("limit") + } + + s.serveCached(w, r, factNamesPath, in, func(ctx context.Context) (cachedResponse, error) { + alive, err := s.aliveResults(ctx, factNamesPath, upstream) + if err != nil { + return cachedResponse{}, err + } + merged := mergeFactNames(alive, suppress, desc) + suppress.logSuppressed(s.log) + resp := cachedResponse{Body: encodeRecords(page.apply(merged)), Records: -1} + s.countBackends(&resp, alive) + if page.wantTotal { + resp.Records = len(merged) + } + return resp, nil + }) +} + // Matches /pdb/query/v4/reports//{events,logs,metrics}, whose data lives in exactly one backend. func isReportSubResource(path string) bool { rest, ok := strings.CutPrefix(path, reportsPath+"/") @@ -489,7 +566,22 @@ func (s *Server) mergeNodesResponse(r *http.Request) func([]backendResult) []jso } func (s *Server) mergeFactsResponse(r *http.Request) func([]backendResult) []json.RawMessage { + return s.mergeFactsWith(s.newSourceInjector(r.URL.Query().Get("query"), true)) +} + +// The path segment of /facts/ is the same outer `name` constraint the +// query gate already rules injection out on, so the synthetic record survives +// only on the source fact's own path — and not on the // form, +// whose pinned value the record's own value need not equal. +func (s *Server) mergeFactsByNameResponse(r *http.Request, name string, valued bool) func([]backendResult) []json.RawMessage { inject := s.newSourceInjector(r.URL.Query().Get("query"), true) + if valued || !inject.claims(name) { + inject.disableInject() + } + return s.mergeFactsWith(inject) +} + +func (s *Server) mergeFactsWith(inject *sourceInjector) func([]backendResult) []json.RawMessage { return func(results []backendResult) []json.RawMessage { var merged []json.RawMessage if s.cfg.Merge == mergeStatic { diff --git a/source.go b/source.go index feec363..2aeeed1 100644 --- a/source.go +++ b/source.go @@ -39,6 +39,14 @@ func (si *sourceInjector) injects() bool { return si != nil && si.inject } +// disableInject rules the synthetic record out for a request shape the query +// gate cannot see, leaving suppression on. +func (si *sourceInjector) disableInject() { + if si != nil { + si.inject = false + } +} + // logSuppressed reports, once per request, that upstream records were dropped. func (si *sourceInjector) logSuppressed(l *log.Logger) { if si == nil || si.suppressed == 0 || l == nil { -- 2.47.3 From cc71902a0d84610c7a30f9ed37d5f3dcc70e1930 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 6 Sep 2026 15:50:49 +1000 Subject: [PATCH 2/6] 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 { -- 2.47.3 From 148be4fe0f208dbc8b20c53fa74ac5cb29aa3c70 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 6 Sep 2026 16:09:55 +1000 Subject: [PATCH 3/6] Synthesise the /facts/ drilldown /fact-names advertises the fact, so its drilldown must not be a dead link. - Serve the source fact's own path from the /facts merge that produces the records, so certname set, owner and environment match /facts. - Filter /facts// by the owning backend. - Keep the aggregate, query-gate and disabled paths answering as before. --- README.md | 33 ++++-- e2e_puppetboard_test.go | 55 +++++++--- e2e_query_test.go | 77 +++++++++++++ factroutes_test.go | 238 ++++++++++++++++++++++++++++++++-------- merge.go | 22 ++++ server.go | 73 ++++++++---- 6 files changed, 399 insertions(+), 99 deletions(-) diff --git a/README.md b/README.md index 1b2626e..bc2eb3c 100644 --- a/README.md +++ b/README.md @@ -26,7 +26,7 @@ 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`, 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/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 — except on the fact's own path, which is **synthesised** from the `/facts` merge (see provenance). | | `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. | @@ -71,7 +71,9 @@ paths add two more headers `pdbmux` sets itself, `X-Cache` and `Age` — see 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 — and the same - aggregate branch, since a count row has no `certname` there either. + aggregate branch, since a count row has no `certname` there either. The + `pdbmux_source` path is the exception: no backend holds that name, so it is + synthesised from the `/facts` merge (see provenance). - **`/fact-names`** — a flat array of strings, not records: **union**, deduped by 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 @@ -162,24 +164,31 @@ matters more. subquery narrows which *nodes* match, not which facts come back, so injection still happens; - the path is `/facts/` for any other fact. The path segment is the same - outer `name` constraint, so only `/facts/pdbmux_source` may carry the record; - `/facts//` never does, since the pinned value need not equal the - backend name the record holds; + outer `name` constraint, so only the fact's own path carries the record; - injection is turned off (see `source_fact_enabled`). +**The fact's own path is synthesised.** `/pdb/query/v4/fact-names` lists the name +while injection is on, so a client that discovers names there can click through +to it, and `/pdb/query/v4/facts/pdbmux_source` has to answer. No backend holds a +record of that name, so the route does not serve its own fan-out: it takes the +records from the `/facts` merge that produces them, which makes the `certname` +set, the owner and the `environment` identical to the ones an unfiltered `/facts` +response reports, and lets the request's own `query` narrow the result upstream. +That costs one `/facts` fan-out per cache miss — the widest fan-out `pdbmux` +makes — on a rare, user-initiated path. `/facts/pdbmux_source/` pins the +backend name, so it answers with the nodes that backend owns, and with `[]` for a +value naming no backend. An `extract`/`count` query still takes the summing +branch, and the gated query shapes above still answer `[]`, as does every form +while injection is off — with the name kept out of `/fact-names`, since nothing +then produces it. + **Not supported in v1: server-side filtering on the fact.** A query that selects it — `["=","name","pdbmux_source"]`, or an `extract` naming it — is forwarded to the backends like any other, and they return nothing, because the fact does not exist upstream. `pdbmux` does not evaluate the AST itself, so it cannot answer 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, 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. +`/facts` response, from the path route above, and filter client-side. **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_puppetboard_test.go b/e2e_puppetboard_test.go index d168732..68b483f 100644 --- a/e2e_puppetboard_test.go +++ b/e2e_puppetboard_test.go @@ -153,6 +153,42 @@ func TestPuppetboardFactsOverview(t *testing.T) { t.Errorf("Puppetboard facts overview does not list %s", name) } } + + // A name the overview lists is a link a user can click, so the one pdbmux + // owns has to lead to a page with every node on it rather than an empty one. + t.Run("the owned fact is listed and its link resolves", func(t *testing.T) { + if !strings.Contains(body, defaultSourceFact) { + t.Fatalf("Puppetboard facts overview does not list %s", defaultSourceFact) + } + listed := pbFactRows(t, defaultSourceFact) + for _, cn := range allNodes { + if !listed[cn] { + t.Errorf("the %s drilldown omits %s, so the overview links to a dead page", defaultSourceFact, cn) + } + } + }) +} + +// pbFactRows reads the JSON table a Puppetboard fact page renders from, and +// returns the certnames it lists. +func pbFactRows(t *testing.T, name string) map[string]bool { + t.Helper() + var payload struct { + Data [][]string `json:"data"` + } + body := pbPage(t, pbAllEnvs+"/fact/"+name+"/json") + if err := json.Unmarshal([]byte(body), &payload); err != nil { + t.Fatalf("decoding the %s drilldown table: %v: %s", name, err, body) + } + listed := map[string]bool{} + for _, row := range payload.Data { + for _, cn := range allNodes { + if len(row) > 0 && strings.Contains(row[0], cn) { + listed[cn] = true + } + } + } + return listed } // The single-fact drilldown is the page that exercises the merged /facts/ @@ -162,26 +198,9 @@ func TestPuppetboardFactDrilldown(t *testing.T) { t.Errorf("Puppetboard fact page for osfamily does not name it") } - // The page's table is filled from this endpoint, so it is what a user sees. - var payload struct { - Data [][]string `json:"data"` - } - body := pbPage(t, pbAllEnvs+"/fact/osfamily/json") - if err := json.Unmarshal([]byte(body), &payload); err != nil { - t.Fatalf("decoding the fact drilldown table: %v: %s", err, body) - } - - listed := map[string]bool{} - for _, row := range payload.Data { - for _, cn := range allNodes { - if len(row) > 0 && strings.Contains(row[0], cn) { - listed[cn] = true - } - } - } - // Puppetboard fetches a single fact through GET /pdb/query/v4/facts/, // so the page is only whole if that path route merges every backend. + listed := pbFactRows(t, "osfamily") missing := []string{} for _, cn := range allNodes { if !listed[cn] { diff --git a/e2e_query_test.go b/e2e_query_test.go index 4f7bb4e..17a6861 100644 --- a/e2e_query_test.go +++ b/e2e_query_test.go @@ -456,6 +456,83 @@ func TestSourceFactInjectionAndGating(t *testing.T) { }) } +// /fact-names advertises the fact, so its drilldown has to answer with the same +// records /facts carries rather than the empty set the backends hold. +func TestSourceFactDrilldown(t *testing.T) { + want := map[string]string{ + nodeAlpha: backendAName, + nodeBeta: backendBName, + nodeGamma: backendBName, + nodeShared: backendBName, // won on freshness, not on configured order + } + path := factsPath + "/" + defaultSourceFact + + t.Run("one record per node, attributed as /facts attributes it", func(t *testing.T) { + resp := get(t, path, nil) + rows := resp.rows(t) + if got := e2eCertnames(rows); !equalStrings(got, allNodes) { + t.Fatalf("%s certnames = %v, want %v", path, got, allNodes) + } + if len(rows) != len(allNodes) { + t.Fatalf("%s returned %d records for %d nodes", path, len(rows), len(allNodes)) + } + unfiltered := get(t, factsPath, nil).rows(t) + for _, row := range rows { + cn, _ := row["certname"].(string) + if row["name"] != defaultSourceFact { + t.Errorf("%s returned a record named %v", path, row["name"]) + } + if row["value"] != want[cn] { + t.Errorf("%s for %s = %v, want %q", path, cn, row["value"], want[cn]) + } + // The drilldown must not disagree with the response it stands for. + if got, ok := factValue(unfiltered, cn, defaultSourceFact); !ok || got != row["value"] { + t.Errorf("%s for %s = %v, want the %s value %v", path, cn, row["value"], factsPath, got) + } + if _, ok := row["environment"]; !ok { + t.Errorf("%s record for %s has no environment key", path, cn) + } + } + if got := resp.header.Get(backendsHeader); got != "2/2" { + t.Errorf("%s = %q, want %q", backendsHeader, got, "2/2") + } + }) + + t.Run("the value sub-route filters by owning backend", func(t *testing.T) { + for _, tc := range []struct { + value string + want []string + }{ + {backendAName, []string{nodeAlpha}}, + {backendBName, []string{nodeBeta, nodeGamma, nodeShared}}, + {"nosuchbackend", nil}, + } { + rows := get(t, path+"/"+tc.value, nil).rows(t) + if got := e2eCertnames(rows); !equalStrings(got, tc.want) { + t.Errorf("%s/%s certnames = %v, want %v", path, tc.value, got, tc.want) + } + } + }) + + t.Run("a certname query narrows the drilldown", func(t *testing.T) { + rows := get(t, path, query(`["=","certname","`+nodeAlpha+`"]`)).rows(t) + if got := e2eCertnames(rows); !equalStrings(got, []string{nodeAlpha}) { + t.Errorf("certname-filtered %s = %v, want only %s", path, got, nodeAlpha) + } + }) + + // An aggregate carries no certname to attribute, so it stays summed. + t.Run("an aggregate is still summed", func(t *testing.T) { + ctx := context.Background() + const q = `["extract",[["function","count"]]]` + wantA := backendCount(ctx, t, h.a, path, q) + wantB := backendCount(ctx, t, h.b, path, q) + if got := countOf(t, get(t, path, query(q)).rows(t)); got != wantA+wantB { + t.Errorf("%s count = %d, want %d (%s=%d + %s=%d)", path, got, wantA+wantB, h.a.name, wantA, h.b.name, wantB) + } + }) +} + // X-Backends has to report what the response was actually built from, not what // is configured. func TestBackendsHeaderReportsContributors(t *testing.T) { diff --git a/factroutes_test.go b/factroutes_test.go index ced7655..722dd21 100644 --- a/factroutes_test.go +++ b/factroutes_test.go @@ -34,22 +34,24 @@ func factNamesBody(names ...string) string { func TestFactsSubPath(t *testing.T) { for _, tc := range []struct { - path string - wantName string - wantValue bool + path string + wantName string + wantValue string + wantValued bool }{ - {factsPath, "", false}, - {factsPath + "/", "", false}, - {roleFactPath, "role", false}, - {roleFactPath + "/web", "role", true}, - {roleFactPath + "/", "", false}, - {roleFactPath + "/web/extra", "", false}, - {nodesPath + "/h1/facts/role", "", false}, - {factNamesPath, "", false}, + {factsPath, "", "", false}, + {factsPath + "/", "", "", false}, + {roleFactPath, "role", "", false}, + {roleFactPath + "/web", "role", "web", true}, + {roleFactPath + "/", "", "", false}, + {roleFactPath + "/web/extra", "", "", false}, + {nodesPath + "/h1/facts/role", "", "", false}, + {factNamesPath, "", "", false}, } { - name, valued := factsSubPath(tc.path) - if name != tc.wantName || valued != tc.wantValue { - t.Errorf("factsSubPath(%q) = (%q, %v), want (%q, %v)", tc.path, name, valued, tc.wantName, tc.wantValue) + name, value, valued := factsSubPath(tc.path) + if name != tc.wantName || value != tc.wantValue || valued != tc.wantValued { + t.Errorf("factsSubPath(%q) = (%q, %q, %v), want (%q, %q, %v)", + tc.path, name, value, valued, tc.wantName, tc.wantValue, tc.wantValued) } } } @@ -187,7 +189,7 @@ func TestHandler_FactsByNameNonAggregateStillMergedByCertname(t *testing.T) { // 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"} { + for _, path := range []string{roleFactPath, roleFactPath + "/web", sourceFactURL} { 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}]`}) @@ -236,48 +238,172 @@ func TestHandler_FactsByNameNotInjected(t *testing.T) { } } -// Only the source fact's own path may carry it; the upstream record of that name -// is still dropped, as it is on every other query shape. -func TestHandler_FactsBySourceNamePathInjectsAndSuppresses(t *testing.T) { - a := newFakeBackend(t, `[]`, `[]`) - a.bodies[sourceFactURL] = `[` + fact("h1", defaultSourceFact, "stale", "") + `]` - b := newFakeBackend(t, `[]`, `[]`) - b.bodies[sourceFactURL] = `[]` - srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) +// sourceFactBackends is a pair of fake backends whose /facts bodies give h1 to a +// and h2/h3 to b, the shape every drilldown assertion below rests on. +func sourceFactBackends(t *testing.T) (*fakeBackend, *fakeBackend) { + t.Helper() + a := newFakeBackend(t, + `[`+node("h1", "2026-01-01T00:00:00Z")+`,`+node("h3", "2026-01-01T00:00:00Z")+`]`, + `[`+factEnv("h1", "osfamily", "RedHat", "prod")+`,`+factEnv("h3", "osfamily", "RedHat", "prod")+`]`) + b := newFakeBackend(t, + `[`+node("h2", "2026-01-01T00:00:00Z")+`,`+node("h3", "2026-06-01T00:00:00Z")+`]`, + `[`+factEnv("h2", "osfamily", "Debian", "dev")+`,`+factEnv("h3", "osfamily", "Debian", "dev")+`]`) + return a, b +} - rec := doGet(t, srv.Handler(), sourceFactURL, "") - got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact) - if n != 1 { - t.Fatalf("%s carried %d %s records, want exactly the synthetic one: %s", sourceFactURL, n, defaultSourceFact, rec.Body.String()) - } - if got["h1"] != "a" { - t.Errorf("%s for h1 = %q, want the owning backend %q", defaultSourceFact, got["h1"], "a") +// The drilldown has to answer with the records /facts carries, which no backend +// holds: the certname set and the per-node owner come from the same merge. +func TestHandler_FactsBySourceNamePathSynthesisesOneRecordPerNode(t *testing.T) { + for _, merge := range []string{mergeStatic, mergeFreshness} { + t.Run(merge, func(t *testing.T) { + a, b := sourceFactBackends(t) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, merge)) + + // h3 lives in both: static keeps the first configured backend, freshness + // the one holding its newer report. + wantH3 := "a" + if merge == mergeFreshness { + wantH3 = "b" + } + want := map[string]string{"h1": "a", "h2": "b", "h3": wantH3} + + rec := doGet(t, srv.Handler(), sourceFactURL, "") + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact) + if n != len(want) || !reflect.DeepEqual(got, want) { + t.Fatalf("%s = %v (%d records), want %v", sourceFactURL, got, n, want) + } + + // The drilldown must agree with the records the unfiltered route reports. + unfiltered, _ := sourceValues(t, doGet(t, srv.Handler(), factsPath, "").Body.Bytes(), defaultSourceFact) + if !reflect.DeepEqual(got, unfiltered) { + t.Errorf("%s = %v, want the same attribution %s reports: %v", sourceFactURL, got, factsPath, unfiltered) + } + if h := rec.Header().Get(backendsHeader); h != "2/2" { + t.Errorf("%s = %q, want 2/2", backendsHeader, h) + } + }) } } -// The // form also pins a value the synthetic record's own value -// need not equal, so it never carries one. -func TestHandler_FactsBySourceNameAndValueNotInjected(t *testing.T) { - path := sourceFactURL + "/a" - a := newFakeBackend(t, `[]`, `[]`) - a.bodies[path] = `[` + fact("h1", defaultSourceFact, "a", "") + `]` - b := newFakeBackend(t, `[]`, `[]`) - b.bodies[path] = `[]` - srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) +// A fact record's four keys are always present, and environment is the owning +// backend's, since that is the record the merge attributed the node to. +func TestHandler_FactsBySourceNamePathCarriesTheOwnersEnvironment(t *testing.T) { + a, b := sourceFactBackends(t) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness)) - rec := doGet(t, srv.Handler(), path, "") - if _, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact); n != 0 { - t.Errorf("%s appeared on %s: %s", defaultSourceFact, path, rec.Body.String()) + rec := doGet(t, srv.Handler(), sourceFactURL, "") + var rows []map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &rows); err != nil { + t.Fatalf("unmarshal %s: %v", rec.Body.String(), err) + } + want := map[string]string{"h1": "prod", "h2": "dev", "h3": "dev"} + for _, row := range rows { + for _, key := range []string{"certname", "name", "value", "environment"} { + if _, ok := row[key]; !ok { + t.Fatalf("synthetic record %v is missing %s", row, key) + } + } + cn, _ := row["certname"].(string) + if got := row["environment"]; got != want[cn] { + t.Errorf("environment of %s = %v, want the owner's %q", cn, got, want[cn]) + } + } +} + +// The pinned value names a backend, so the sub-route is the drilldown filtered +// by owner; a value naming no backend matches nothing. +func TestHandler_FactsBySourceNameAndValueFiltersByOwner(t *testing.T) { + for _, tc := range []struct { + value string + want map[string]string + }{ + {"a", map[string]string{"h1": "a"}}, + {"b", map[string]string{"h2": "b", "h3": "b"}}, + {"nosuchbackend", map[string]string{}}, + } { + t.Run(tc.value, func(t *testing.T) { + a, b := sourceFactBackends(t) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness)) + + rec := doGet(t, srv.Handler(), sourceFactURL+"/"+tc.value, "") + got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact) + if n != len(tc.want) || !reflect.DeepEqual(got, tc.want) { + t.Errorf("%s/%s = %v (%d records), want %v", sourceFactURL, tc.value, got, n, tc.want) + } + }) + } +} + +// Nothing produces the fact while injection is off, so the drilldown is empty +// and the name is not advertised for a client to click through to. +func TestHandler_FactsBySourceNamePathEmptyWhenDisabled(t *testing.T) { + a, b := sourceFactBackends(t) + a.bodies[factNamesPath] = factNamesBody("osfamily") + b.bodies[factNamesPath] = `[]` + cfg := testConfig(a.srv.URL, b.srv.URL, mergeFreshness) + cfg.SourceFactEnabled = false + srv := newTestServer(cfg) + + // No backend holds the fact, so the route falls through to the plain merge. + for _, fb := range []*fakeBackend{a, b} { + fb.bodies[sourceFactURL] = `[]` + fb.bodies[sourceFactURL+"/a"] = `[]` + } + for _, path := range []string{sourceFactURL, sourceFactURL + "/a"} { + if got := doGet(t, srv.Handler(), path, "").Body.String(); got != "[]\n" { + t.Errorf("disabled %s = %q, want %q", path, got, "[]\n") + } + } + if got := names(t, doGet(t, srv.Handler(), factNamesPath, "").Body.Bytes()); !reflect.DeepEqual(got, []string{"osfamily"}) { + t.Errorf("disabled %s = %v, want no %s entry", factNamesPath, got, defaultSourceFact) + } +} + +// The query narrows the certnames upstream, so the drilldown only covers what it +// selects. +func TestHandler_FactsBySourceNamePathRespectsQuery(t *testing.T) { + a, b := sourceFactBackends(t) + a.bodies[factsPath] = `[` + factEnv("h1", "osfamily", "RedHat", "prod") + `]` + b.bodies[factsPath] = `[]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness)) + + const q = `["=","certname","h1"]` + rec := doGet(t, srv.Handler(), sourceFactURL, q) + got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact) + if n != 1 || got["h1"] != "a" { + t.Fatalf("%s?query=%s = %v (%d records), want only h1: %s", sourceFactURL, q, got, n, rec.Body.String()) + } + // The query has to reach the backends for them to narrow anything. + for _, fb := range []*fakeBackend{a, b} { + if seen := fb.gotQuery(factsPath); seen != q { + t.Errorf("backend got query %q on %s, want %q", seen, factsPath, q) + } + } +} + +// pdbmux owns the name, so an upstream record of it is dropped rather than +// listed beside the synthetic one. +func TestHandler_FactsBySourceNamePathSuppressesUpstream(t *testing.T) { + a, b := sourceFactBackends(t) + a.bodies[factsPath] = `[` + factEnv("h1", "osfamily", "RedHat", "prod") + `,` + + fact("h1", defaultSourceFact, "stale", "") + `]` + b.bodies[factsPath] = `[]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness)) + + rec := doGet(t, srv.Handler(), sourceFactURL, "") + got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact) + if n != 1 || got["h1"] != "a" { + t.Fatalf("%s = %v (%d records), want exactly the synthetic one: %s", sourceFactURL, got, n, rec.Body.String()) } } // A name-constrained query on the source fact's own path is still gated by the // query, not just the path. func TestHandler_FactsBySourceNamePathRespectsQueryGate(t *testing.T) { - a := newFakeBackend(t, `[]`, `[]`) - a.bodies[sourceFactURL] = `[` + fact("h1", defaultSourceFact, "stale", "") + `]` - b := newFakeBackend(t, `[]`, `[]`) - b.bodies[sourceFactURL] = `[]` + a, b := sourceFactBackends(t) srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) rec := doGet(t, srv.Handler(), sourceFactURL, `["extract",["certname","value"]]`) @@ -286,6 +412,24 @@ func TestHandler_FactsBySourceNamePathRespectsQueryGate(t *testing.T) { } } +// An aggregate carries no certname to attribute, so the route stays on the +// summing path rather than synthesising records. +func TestHandler_FactsBySourceNameAggregateStillSummed(t *testing.T) { + for _, path := range []string{sourceFactURL, sourceFactURL + "/a"} { + t.Run(path, func(t *testing.T) { + a, b := sourceFactBackends(t) + a.bodies[path] = `[{"count":7}]` + b.bodies[path] = `[{"count":3}]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness)) + + rec := doGet(t, srv.Handler(), path, `["extract",[["function","count"]]]`) + if got := counts(t, rec.Body.Bytes(), "count"); !slices.Equal(got, []float64{10}) { + t.Errorf("count = %v, want [10]", got) + } + }) + } +} + func TestHandler_FactNamesUnionDedupedAndSorted(t *testing.T) { a := newFakeBackend(t, `[]`, `[]`) a.bodies[factNamesPath] = factNamesBody("kernel", "only_a", "osfamily") @@ -472,7 +616,7 @@ func TestHandler_FactNamesSurvivesOneBackend(t *testing.T) { // Both routes are merged record sets, so they use the same cache /facts does. func TestHandler_FactRoutesAreCached(t *testing.T) { - for _, path := range []string{roleFactPath, roleFactPath + "/web", factNamesPath} { + for _, path := range []string{roleFactPath, roleFactPath + "/web", sourceFactURL, factNamesPath} { t.Run(path, func(t *testing.T) { a := newFakeBackend(t, `[]`, `[]`) a.bodies[path] = `[]` diff --git a/merge.go b/merge.go index ece3b96..672b1f9 100644 --- a/merge.go +++ b/merge.go @@ -171,6 +171,28 @@ func mergeFacts(results []backendResult, owner func(certname string) string, inj return out } +// sourceFactRecords keeps the merged /facts records naming the fact pdbmux +// owns, and, when the path pins a value, only those naming that backend. The +// merge drops every upstream record of that name, so what survives is exactly +// the synthetic set — one record per certname the merge attributed. +func sourceFactRecords(merged []json.RawMessage, name, value string, valued bool) []json.RawMessage { + out := []json.RawMessage{} + for _, raw := range merged { + var m struct { + Name string `json:"name"` + Value string `json:"value"` + } + if json.Unmarshal(raw, &m) != nil || m.Name != name { + continue + } + if valued && m.Value != value { + continue + } + out = append(out, raw) + } + return out +} + // mergeFactNames unions the backends' /fact-names arrays, dedupes by the name // 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 diff --git a/server.go b/server.go index 4f23639..689ed38 100644 --- a/server.go +++ b/server.go @@ -145,8 +145,8 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) { case eventCountsPath, aggregateEventCountsPath: s.serveSummed(w, r, r.URL.Path, inferredColumns) default: - if name, valued := factsSubPath(r.URL.Path); name != "" { - s.serveFactsByName(w, r, name, valued) + if name, value, valued := factsSubPath(r.URL.Path); name != "" { + s.serveFactsByName(w, r, name, value, valued) return } if isReportSubResource(r.URL.Path) { @@ -158,28 +158,32 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) { } // factsSubPath matches /pdb/query/v4/facts/ and /pdb/query/v4/facts//, -// returning the fact name the path constrains and whether it also pins a value. -// openvoxdb serves both from the facts entity, ANDing ["=","name",] (and -// ["=","value",]) onto the request's own query, so the record shape is -// exactly /facts' — src/puppetlabs/puppetdb/http/handlers.clj:283-302 and +// returning the fact name the path constrains, the value it pins and whether it +// pins one at all. openvoxdb serves both from the facts entity, ANDing +// ["=","name",] (and ["=","value",]) onto the request's own query, +// so the record shape is exactly /facts' — +// src/puppetlabs/puppetdb/http/handlers.clj:283-302 and // src/puppetlabs/puppetdb/http/query.clj:136-143,193-209. -func factsSubPath(path string) (name string, valued bool) { +func factsSubPath(path string) (name, value string, valued bool) { rest, ok := strings.CutPrefix(path, factsPath+"/") if !ok { - return "", false + return "", "", false } name, value, hasValue := strings.Cut(rest, "/") if name == "" { - return "", false + return "", "", false } if hasValue && (value == "" || strings.Contains(value, "/")) { - return "", false + return "", "", false } - return name, hasValue + if !hasValue { + value = "" + } + return name, value, hasValue } func isFactsSubPath(path string) bool { - name, _ := factsSubPath(path) + name, _, _ := factsSubPath(path) return name != "" } @@ -302,13 +306,40 @@ 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) { +// to key on, and is summed instead — the same split /nodes makes. A plain query +// on the source fact's own path is the one name no backend can answer for, so it +// is synthesised instead of fanned out as-is. +func (s *Server) serveFactsByName(w http.ResponseWriter, r *http.Request, name, value 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)) + if inject := s.newSourceInjector(r.URL.Query().Get("query"), true); inject.claims(name) && inject.injects() { + s.serveSourceFact(w, r, inject, value, valued) + return + } + s.serveMerged(w, r, r.URL.Path, s.mergeFactsByNameResponse(r)) +} + +// serveSourceFact answers the drilldown on pdbmux's own fact. No backend holds a +// record of that name, so the route's own fan-out would return nothing; the +// records are taken from the /facts merge that produces them instead, which +// makes the certname set, the owner and the environment identical to the ones an +// unfiltered /facts response reports, and lets the request's query narrow the +// result upstream. That costs one extra fan-out on a rare, user-initiated path. +func (s *Server) serveSourceFact(w http.ResponseWriter, r *http.Request, inject *sourceInjector, value string, valued bool) { + params := queryParams(r.URL.Query().Get("query")) + merge := s.mergeFactsWith(inject) + s.serveCached(w, r, r.URL.Path, params, func(ctx context.Context) (cachedResponse, error) { + alive, err := s.aliveResults(ctx, factsPath, params) + if err != nil { + return cachedResponse{}, err + } + recs := sourceFactRecords(merge(alive), inject.name, value, valued) + resp := cachedResponse{Body: encodeRecords(recs), Records: -1} + s.countBackends(&resp, alive) + return resp, nil + }) } // 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. @@ -592,14 +623,12 @@ func (s *Server) mergeFactsResponse(r *http.Request) func([]backendResult) []jso } // The path segment of /facts/ is the same outer `name` constraint the -// query gate already rules injection out on, so the synthetic record survives -// only on the source fact's own path — and not on the // form, -// whose pinned value the record's own value need not equal. -func (s *Server) mergeFactsByNameResponse(r *http.Request, name string, valued bool) func([]backendResult) []json.RawMessage { +// query gate already rules injection out on, so nothing is synthesised here: +// the source fact's own path is diverted to serveSourceFact before this. +// Suppression of an upstream record of the owned name stays on. +func (s *Server) mergeFactsByNameResponse(r *http.Request) func([]backendResult) []json.RawMessage { inject := s.newSourceInjector(r.URL.Query().Get("query"), true) - if valued || !inject.claims(name) { - inject.disableInject() - } + inject.disableInject() return s.mergeFactsWith(inject) } -- 2.47.3 From abf565b0f68bab13647f4d42e8999a4ed696da51 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 6 Sep 2026 16:51:00 +1000 Subject: [PATCH 4/6] Stop the source-fact drilldown fanning out per pinned value The path segment is client-supplied and reached a full unfiltered /facts fan-out, keyed per value, so every distinct value was a fresh whole-estate query and a fresh cache entry. - validate against the configured backend names, answering [] with no fan-out when it names none - key the drilldown's fetch on the fact name alone and apply to the shared record set, so all values share one entry and one fan-out - report every configured backend on the no-fan-out empty response, which is complete rather than partial --- README.md | 22 +++++++++----- cache_test.go | 12 ++++++++ factroutes_test.go | 60 +++++++++++++++++++++++++++++++++++++ server.go | 75 ++++++++++++++++++++++++++++++++++++++++------ 4 files changed, 153 insertions(+), 16 deletions(-) diff --git a/README.md b/README.md index a5597aa..a45d752 100644 --- a/README.md +++ b/README.md @@ -182,12 +182,18 @@ records from the `/facts` merge that produces them, which makes the `certname` set, the owner and the `environment` identical to the ones an unfiltered `/facts` response reports, and lets the request's own `query` narrow the result upstream. That costs one `/facts` fan-out per cache miss — the widest fan-out `pdbmux` -makes — on a rare, user-initiated path. `/facts/pdbmux_source/` pins the -backend name, so it answers with the nodes that backend owns, and with `[]` for a -value naming no backend. An `extract`/`count` query still takes the summing -branch, and the gated query shapes above still answer `[]`, as does every form -while injection is off — with the name kept out of `/fact-names`, since nothing -then produces it. +makes — on a rare, user-initiated path. + +`/facts/pdbmux_source/` pins the backend name, so it answers with the +nodes that backend owns. The `` segment never reaches that fan-out: the +synthetic record's value is always a backend name, so a value naming none is +answered `[]` from the configured names alone, with no fan-out at all, and a +value naming one filters a record set fetched under a key the value is not part +of. The record set is a property of the estate rather than of the filter, so +every value of it — and the unfiltered path — share one entry and one fetch. +An `extract`/`count` query still takes the summing branch, and the gated query +shapes above still answer `[]`, as does every form while injection is off — with +the name kept out of `/fact-names`, since nothing then produces it. **Not supported in v1: server-side filtering on the fact.** A query that selects it — `["=","name","pdbmux_source"]`, or an `extract` naming it — is forwarded to @@ -407,7 +413,9 @@ backend later without further handler changes. requests can only share an entry when they share a key, and the key is path plus query, which is exactly what decides whether injection applies; a name-filtered `/facts` query and a plain one therefore cache separately and - neither is ever served the other's shape. `source_fact` and + neither is ever served the other's shape. The one path that keys on less than + it is asked is the `pdbmux_source` drilldown, whose `` is dropped from + the key and applied to the shared entry instead. `source_fact` and `source_fact_enabled` are read once at startup, and the cache lives for the same process, so changing either cannot leave differently-shaped entries behind. diff --git a/cache_test.go b/cache_test.go index 9f29901..33c8618 100644 --- a/cache_test.go +++ b/cache_test.go @@ -96,6 +96,18 @@ func (cb *countingBackend) hitCount(path string) int { return cb.hits[path] } +// totalHits counts requests across every path, so a test can assert a route +// reached no backend at all. +func (cb *countingBackend) totalHits() int { + cb.mu.Lock() + defer cb.mu.Unlock() + n := 0 + for _, c := range cb.hits { + n += c + } + return n +} + func (cb *countingBackend) setFail(v bool) { cb.mu.Lock() defer cb.mu.Unlock() diff --git a/factroutes_test.go b/factroutes_test.go index 722dd21..9f76369 100644 --- a/factroutes_test.go +++ b/factroutes_test.go @@ -337,6 +337,66 @@ func TestHandler_FactsBySourceNameAndValueFiltersByOwner(t *testing.T) { } } +// The pinned value is client-supplied and the drilldown's fetch is the widest +// query pdbmux makes, so a value naming no backend is answered from the +// configured names alone: no fan-out, and so no per-value cache key either. +func TestHandler_FactsBySourceNameUnknownValueSkipsFanOut(t *testing.T) { + facts := map[string]string{factsPath: `[` + fact("h1", "osfamily", "RedHat", "") + `]`} + a := newCountingBackend(t, facts) + b := newCountingBackend(t, facts) + srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + + for _, value := range []string{"nosuchbackend", "nonce-1", "nonce-2"} { + rec := doGet(t, srv.Handler(), sourceFactURL+"/"+value, "") + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + if got := rec.Body.String(); got != "[]\n" { + t.Errorf("%s/%s = %q, want %q", sourceFactURL, value, got, "[]\n") + } + // The answer is complete, not built from a subset of backends. + if h := rec.Header().Get(backendsHeader); h != "2/2" { + t.Errorf("%s/%s %s = %q, want 2/2", sourceFactURL, value, backendsHeader, h) + } + } + for _, cb := range []*countingBackend{a, b} { + if got := cb.totalHits(); got != 0 { + t.Errorf("unknown value reached a backend %d times, want 0", got) + } + } +} + +// The record set is the estate's, not the pinned value's, so every value shares +// one entry: two valid values must not cost two whole-estate fan-outs. +func TestHandler_FactsBySourceNameValuesShareOneFetch(t *testing.T) { + a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "osfamily", "RedHat", "") + `]`}) + b := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h2", "osfamily", "Debian", "") + `]`}) + srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + + for _, tc := range []struct { + path string + want map[string]string + }{ + {sourceFactURL + "/a", map[string]string{"h1": "a"}}, + {sourceFactURL + "/b", map[string]string{"h2": "b"}}, + {sourceFactURL, map[string]string{"h1": "a", "h2": "b"}}, + } { + rec := doGet(t, srv.Handler(), tc.path, "") + got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact) + if n != len(tc.want) || !reflect.DeepEqual(got, tc.want) { + t.Fatalf("%s = %v (%d records), want %v", tc.path, got, n, tc.want) + } + } + for _, cb := range []*countingBackend{a, b} { + if got := cb.hitCount(factsPath); got != 1 { + t.Errorf("%s fetched %d times, want 1 shared fetch", factsPath, got) + } + if got := cb.hitCount(sourceFactURL); got != 0 { + t.Errorf("%s was fanned out %d times, want 0", sourceFactURL, got) + } + } +} + // Nothing produces the fact while injection is off, so the drilldown is empty // and the name is not advertised for a client to click through to. func TestHandler_FactsBySourceNamePathEmptyWhenDisabled(t *testing.T) { diff --git a/server.go b/server.go index 19914dc..0a4fa5d 100644 --- a/server.go +++ b/server.go @@ -335,22 +335,55 @@ func (s *Server) serveFactsByName(w http.ResponseWriter, r *http.Request, name, // records are taken from the /facts merge that produces them instead, which // makes the certname set, the owner and the environment identical to the ones an // unfiltered /facts response reports, and lets the request's query narrow the -// result upstream. That costs one extra fan-out on a rare, user-initiated path. +// result upstream. +// +// That merge is the widest query pdbmux makes and the pinned value is +// client-supplied, so the value never reaches it: a value naming no configured +// backend is answered empty without any fan-out, and a value naming one filters +// a record set fetched under a value-independent key. Otherwise each distinct +// value would be a fresh cache key, a fresh flight and a fresh whole-estate +// fan-out. func (s *Server) serveSourceFact(w http.ResponseWriter, r *http.Request, inject *sourceInjector, value string, valued bool) { + // The synthetic record's value is always a backend name, so any other value + // matches zero records. The response is complete rather than degraded, so it + // reports every configured backend. + if valued && !s.hasBackend(value) { + n := len(s.cfg.Backends) + writeCached(w, cachedResponse{Records: -1, Backends: n, Configured: n}) + return + } + params := queryParams(r.URL.Query().Get("query")) merge := s.mergeFactsWith(inject) - s.serveCached(w, r, r.URL.Path, params, func(ctx context.Context) (cachedResponse, error) { + var filter recordFilter + if valued { + filter = func(recs []json.RawMessage) []json.RawMessage { + return sourceFactRecords(recs, inject.name, value, true) + } + } + // The stored set is the whole owned-fact record set, so every value of it + // keys, and waits on, the same fetch. + s.serveFiltered(w, r, factsPath+"/"+inject.name, params, filter, func(ctx context.Context) (cachedResponse, error) { alive, err := s.aliveResults(ctx, factsPath, params) if err != nil { return cachedResponse{}, err } - recs := sourceFactRecords(merge(alive), inject.name, value, valued) + recs := sourceFactRecords(merge(alive), inject.name, "", false) resp := cachedResponse{Body: encodeRecords(recs), Records: -1} s.countBackends(&resp, alive) return resp, nil }) } +func (s *Server) hasBackend(name string) bool { + for _, b := range s.cfg.Backends { + if b.Name == name { + return true + } + } + return false +} + // 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 { @@ -457,12 +490,36 @@ type cachedResponse struct { Configured int `json:"configured"` // backends configured at build time } +// recordFilter narrows a response's records after it has been built or read back +// from the cache, so requests differing only in the filter share one stored entry +// and one fan-out. It leaves Records alone, so it only suits responses that set +// no X-Records. +type recordFilter func([]json.RawMessage) []json.RawMessage + +func (f recordFilter) apply(resp cachedResponse) cachedResponse { + if f == nil { + return resp + } + var recs []json.RawMessage + if json.Unmarshal(resp.Body, &recs) != nil { + return resp + } + resp.Body = encodeRecords(f(recs)) + return resp +} + // serveCached answers from the cache when the entry is fresh, otherwise runs // build — single-flighted, so N concurrent identical requests cause one upstream // fan-out — and stores the result. A build failure falls back to a stale entry // when one exists; that is the only path on which stale data is served. Paths // with no cache configured run build directly, unchanged. func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string, params url.Values, build func(context.Context) (cachedResponse, error)) { + s.serveFiltered(w, r, path, params, nil, build) +} + +// serveFiltered is serveCached with a per-request narrowing applied to whatever +// the shared entry holds. +func (s *Server) serveFiltered(w http.ResponseWriter, r *http.Request, path string, params url.Values, filter recordFilter, build func(context.Context) (cachedResponse, error)) { cache, enabled := s.cacheFor(path, params) if !enabled { resp, err := build(r.Context()) @@ -470,7 +527,7 @@ func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string http.Error(w, err.Error(), http.StatusBadGateway) return } - writeCached(w, resp) + writeCached(w, filter.apply(resp)) return } @@ -482,7 +539,7 @@ func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string s.log.Printf("warning: cache lookup for %s failed: %v", key, err) case status == CacheFresh: s.stale.markFresh() - s.writeStored(w, ent, CacheFresh) + s.writeStored(w, ent, CacheFresh, filter) return case status == CacheStale: stale = &ent @@ -523,7 +580,7 @@ func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string s.stale.markStale(s.now()) s.log.Printf("warning: serving stale %s from cache (stored %s): %v", path, stale.StoredAt.UTC().Format(time.RFC3339), err) - s.writeStored(w, *stale, CacheStale) + s.writeStored(w, *stale, CacheStale, filter) return } http.Error(w, err.Error(), http.StatusBadGateway) @@ -531,7 +588,7 @@ func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string } s.stale.markFresh() s.setCacheHeaders(w, CacheMiss, time.Time{}) - writeCached(w, resp) + writeCached(w, filter.apply(resp)) } // http.Client reads a zero Timeout as "no deadline", but it would expire a @@ -543,7 +600,7 @@ func (s *Server) flightTimeout() time.Duration { return defaultTimeout } -func (s *Server) writeStored(w http.ResponseWriter, ent CacheEntry, status CacheStatus) { +func (s *Server) writeStored(w http.ResponseWriter, ent CacheEntry, status CacheStatus, filter recordFilter) { var resp cachedResponse if err := json.Unmarshal(ent.Body, &resp); err != nil { s.log.Printf("warning: unreadable cache entry: %v", err) @@ -551,7 +608,7 @@ func (s *Server) writeStored(w http.ResponseWriter, ent CacheEntry, status Cache return } s.setCacheHeaders(w, status, ent.StoredAt) - writeCached(w, resp) + writeCached(w, filter.apply(resp)) } // setCacheHeaders labels a response from a cache-backed path: X-Cache is -- 2.47.3 From e299f64b07e3c51c6d1551c4854896c1c0c3e3c6 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 6 Sep 2026 16:53:11 +1000 Subject: [PATCH 5/6] Assert the drilldown's no-fan-out path marks no partial round The empty answer for an unknown value never reaches a backend, so it must leave /healthz reporting a whole estate. - assert partial_rounds stays 0 after an unknown-value drilldown --- factroutes_test.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/factroutes_test.go b/factroutes_test.go index 9f76369..23de0bc 100644 --- a/factroutes_test.go +++ b/factroutes_test.go @@ -364,6 +364,10 @@ func TestHandler_FactsBySourceNameUnknownValueSkipsFanOut(t *testing.T) { t.Errorf("unknown value reached a backend %d times, want 0", got) } } + // No fan-out happened, so none was heard from partially either. + if got := health(t, srv).Query.PartialRounds; got != 0 { + t.Errorf("partial_rounds = %d, want 0", got) + } } // The record set is the estate's, not the pinned value's, so every value shares -- 2.47.3 From 6bf6a8024cd8361e265f8d42243e9641cf8efd53 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 6 Sep 2026 17:10:34 +1000 Subject: [PATCH 6/6] Assert the stale drilldown keeps its owner filter --- cache_test.go | 37 +++++++++++++++++++++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/cache_test.go b/cache_test.go index 33c8618..e199561 100644 --- a/cache_test.go +++ b/cache_test.go @@ -352,6 +352,43 @@ func TestHandler_ServesStaleOnlyWhenBackendsFail(t *testing.T) { } } +// The stored entry is the whole estate's record set and the pinned value narrows +// it per request, so the fallback has to keep narrowing: a client asking for one +// backend's records must not be handed every backend's because the entry expired. +func TestHandler_StaleSourceFactDrilldownStaysFilteredByOwner(t *testing.T) { + a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "osfamily", "RedHat", "") + `]`}) + b := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h2", "osfamily", "Debian", "") + `]`}) + srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + + drilldown := sourceFactURL + "/a" + want := map[string]string{"h1": "a"} + + warm := doGet(t, srv.Handler(), drilldown, "") + if warm.Code != http.StatusOK { + t.Fatalf("warm-up status %d: %s", warm.Code, warm.Body.String()) + } + if got, n := sourceValues(t, warm.Body.Bytes(), defaultSourceFact); n != len(want) || !reflect.DeepEqual(got, want) { + t.Fatalf("warm-up %s = %v (%d records), want %v", drilldown, got, n, want) + } + + // Every backend down and the entry expired: the stale copy is served. + clk.advance(31 * time.Second) + a.setFail(true) + b.setFail(true) + + rec := doGet(t, srv.Handler(), drilldown, "") + if rec.Code != http.StatusOK { + t.Fatalf("stale fallback status %d: %s", rec.Code, rec.Body.String()) + } + if got := rec.Header().Get(cacheStatusHeader); got != "stale" { + t.Fatalf("%s = %q, want stale: the request did not take the fallback path", cacheStatusHeader, got) + } + got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact) + if n != len(want) || !reflect.DeepEqual(got, want) { + t.Errorf("stale %s = %v (%d records), want only backend a's %v", drilldown, got, n, want) + } +} + func TestHandler_NoCacheEntryMeansBackendFailureIs502(t *testing.T) { a := newCountingBackend(t, map[string]string{factsPath: `[]`}) b := newCountingBackend(t, map[string]string{factsPath: `[]`}) -- 2.47.3