From c8efa26383dc69b4b461a66c281204f0865ad95f Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 6 Sep 2026 15:18:29 +1000 Subject: [PATCH] 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 {