diff --git a/README.md b/README.md index 3fa9cf0..a45d752 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. An `extract`/`count` query is **summed** instead. | +| `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. | | `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,18 @@ 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 — and the same + 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 + 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 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`; @@ -75,17 +89,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`, `/facts` or `/resources` query whose `extract` carries a - `["function", ...]` column. + `/nodes`, `/resources`, `/facts` or `/facts/[/]` query whose + `extract` carries a `["function", ...]` column. - The grouping key is the row's non-aggregate fields: for `/reports`, - `/nodes`, `/facts` 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 + `/nodes`, `/resources`, `/facts` 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: an - aggregate row has no `certname`, so deduping would collapse every backend's - rows into one backend's numbers. A query with no `function` column — - including a plain `extract` projection — still merges by `certname`. + - On `/nodes`, `/facts` and `/facts/[/]` this takes precedence + over the `certname` merge: an aggregate row has no `certname`, so deduping + would collapse every backend's rows into one backend's numbers. A query with + no `function` column — including a plain `extract` projection — still merges + by `certname`. - PuppetDB accepts `count`, `sum`, `avg`, `min`, `max`, `to_string` and `jsonb_typeof` as `extract` functions. Only `count` and `sum` are additive, so only those two merge correctly. `avg`, `min` and `max` are folded like any @@ -155,16 +170,38 @@ 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 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. 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 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, 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 @@ -323,9 +360,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. @@ -374,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. @@ -481,11 +522,7 @@ 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) | -Two tests skip rather than assert, each naming a known gap and failing if that -gap closes: `/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). +Every test asserts; the suite records no known gaps. ## Deployment diff --git a/cache_test.go b/cache_test.go index 9f29901..e199561 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() @@ -340,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: `[]`}) 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_puppetboard_test.go b/e2e_puppetboard_test.go index 00586c9..68b483f 100644 --- a/e2e_puppetboard_test.go +++ b/e2e_puppetboard_test.go @@ -143,43 +143,43 @@ 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") + // 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) + } } - 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. -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") - } - - // The page's table is filled from this endpoint, so it is what a user sees. +// 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/osfamily/json") + body := pbPage(t, pbAllEnvs+"/fact/"+name+"/json") if err := json.Unmarshal([]byte(body), &payload); err != nil { - t.Fatalf("decoding the fact drilldown table: %v: %s", err, body) + 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 { @@ -188,25 +188,28 @@ func TestPuppetboardFactDrilldown(t *testing.T) { } } } + return listed +} - // 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. +// 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") + } + + // 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] { 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 dffde52..dc59eb3 100644 --- a/e2e_query_test.go +++ b/e2e_query_test.go @@ -5,6 +5,8 @@ package main import ( "context" "encoding/json" + "fmt" + "net/http" "net/url" "reflect" "sort" @@ -59,6 +61,208 @@ 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 + } + // 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") + } + + 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)) + } + }) + + // 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 { + 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) { @@ -284,6 +488,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 new file mode 100644 index 0000000..23de0bc --- /dev/null +++ b/factroutes_test.go @@ -0,0 +1,699 @@ +package main + +import ( + "encoding/json" + "net/http" + "net/url" + "reflect" + "slices" + "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 string + wantValued bool + }{ + {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, 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) + } + } +} + +// 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) + } +} + +// 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", 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}]`}) + 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] = `[]` + 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()) + } +} + +// 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 +} + +// 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) + } + }) + } +} + +// 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(), 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) + } + }) + } +} + +// 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) + } + } + // 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 +// 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) { + 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, b := sourceFactBackends(t) + 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()) + } +} + +// 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") + 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", defaultSourceFact} + 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", defaultSourceFact, "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 != "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")) + } +} + +func TestHandler_FactNamesRejectsBadPaging(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + b := newFakeBackend(t, `[]`, `[]`) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + 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()) + } + } +} + +// 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 + upstream string + want []string + }{ + {"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] = tc.upstream + 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) + } + }) + } +} + +// 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] = `[]` + 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" { + 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", defaultSourceFact}) { + 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", sourceFactURL, 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..672b1f9 100644 --- a/merge.go +++ b/merge.go @@ -2,6 +2,7 @@ package main import ( "encoding/json" + "sort" "time" ) @@ -170,6 +171,86 @@ 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 +// 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 { + add(rec.Raw) + } + } + if owned != "" { + if raw, err := json.Marshal(owned); err == nil { + add(raw) + } + } + 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 +} + +// 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 da39330..0a4fa5d 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" @@ -25,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" @@ -91,13 +95,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 +136,8 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) { s.serveResources(w, r) case factsPath: s.serveFacts(w, r) + case factNamesPath: + s.serveFactNames(w, r) case reportsPath: s.serveReports(w, r) case eventsPath: @@ -139,6 +145,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, value, valued := factsSubPath(r.URL.Path); name != "" { + s.serveFactsByName(w, r, name, value, valued) + return + } if isReportSubResource(r.URL.Path) { s.serveFirstHolder(w, r) return @@ -147,6 +157,88 @@ 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, 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, value 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 + } + if !hasValue { + value = "" + } + return name, value, 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 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) + if err != nil { + 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 + + // 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 { + // 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, owned, desc) + 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+"/") @@ -221,6 +313,77 @@ func (s *Server) serveFacts(w http.ResponseWriter, r *http.Request) { s.serveMerged(w, r, factsPath, s.mergeFactsResponse(r)) } +// 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. 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 + } + 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 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) + 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, "", 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 { @@ -327,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()) @@ -340,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 } @@ -352,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 @@ -393,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) @@ -401,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 @@ -413,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) @@ -421,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 @@ -498,7 +685,20 @@ 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 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) + 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 {