From a4a29866e136386d2a85d6a3a98cad9cb917fdd6 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sat, 5 Sep 2026 21:31:50 +1000 Subject: [PATCH] Override the source fact on every query shape - drop upstream facts of the configured name whenever the feature is enabled, independent of the per-query injection gate, and log the drop once per request - walk the whole AST for a nested extract and skip injection when one is found outside an in subquery - skip the environment scan on /facts when nothing is injected - document the override rule and that PQL-syntax queries never get the fact --- README.md | 26 ++++++--- merge.go | 9 ++- server.go | 10 +++- source.go | 91 +++++++++++++++++++++++------ source_test.go | 153 +++++++++++++++++++++++++++++++++++++++++++++++-- 5 files changed, 252 insertions(+), 37 deletions(-) diff --git a/README.md b/README.md index 95e8712..9671dd3 100644 --- a/README.md +++ b/README.md @@ -114,17 +114,25 @@ attributes a shared node to the first backend in configured order, while `/nodes` always attributes it to the backend holding the newer `report_timestamp`. Each answer describes the record it is attached to. -If a backend genuinely reports a fact of the configured name, `pdbmux` -**overrides** it — the upstream record is dropped and replaced, never duplicated, -so the fact means exactly one thing and a node never carries two of it. Rename -the synthetic fact via `source_fact` if the real one matters more. +While the fact is enabled the configured name is `pdbmux`'s alone. A `/facts` +record of that name coming from a backend is **always dropped**, on every query +shape — including the shapes below, where nothing is injected in its place — so +the fact means exactly one thing and a node never carries two of it. Each request +that drops one logs it once. Only `source_fact_enabled: false` restores upstream +records of that name; rename the synthetic fact via `source_fact` if the real one +matters more. -**Injection is skipped**, and the response passes through untouched, when: +**Injection is skipped**, and no synthetic record is added, when: -- the query has a top-level `extract` — it projects a column subset, and with a - `["function", ...]` column it aggregates. Injecting there would break the row - shape or silently inflate a `count()`, so **aggregate results are never - changed**; +- the query contains an `extract` anywhere outside an `in` subquery — it projects + a column subset, and with a `["function", ...]` column it aggregates. Injecting + there would break the row shape or silently inflate a `count()`, so **aggregate + results are never changed**. The whole query is walked, so an `extract` nested + under `and`/`or`/`not`/`from` skips injection too; an `extract` inside an `in` + operand projects the subquery rather than the response, so it does not; +- the query is not an AST array — every **PQL-syntax** query (`facts { certname + = "web1" }`) lands here. `pdbmux` cannot tell what such a query projects, so it + never injects into a PQL response. Use the AST form to get the fact; - (`/facts` only) the query constrains `name` — `["=","name","osfamily"]` and friends ask for specific facts, and the synthetic record is not one of them. Only the outer query is inspected: a `name` filter inside an `in`/`select_facts` diff --git a/merge.go b/merge.go index a5b61de..fcb530e 100644 --- a/merge.go +++ b/merge.go @@ -114,7 +114,7 @@ func buildFreshness(results []backendResult) freshness { } // owner names the winning backend per certname; a nil owner (static merge), or one holding no facts for that certname, falls back to configured order. -// A non-nil inject appends the synthetic source fact after each certname's block, naming the backend that won. +// inject appends the synthetic source fact after each certname's block, naming the backend that won, and always drops upstream facts of that name. func mergeFacts(results []backendResult, owner func(certname string) string, inject *sourceInjector) []json.RawMessage { present := map[string][]string{} // certname -> backend names, in configured order byKey := map[string][]record{} @@ -152,12 +152,17 @@ func mergeFacts(results []backendResult, owner func(certname string) string, inj } recs := byKey[cn+"\x00"+chosen] for _, rec := range recs { - // An upstream fact of the same name is dropped: pdbmux's own value is authoritative. + // An upstream fact of the configured name is dropped on every query shape, + // injected or not: while the feature is on the name is pdbmux's alone. if inject.claims(rec.Name) { + inject.suppressed++ continue } out = append(out, rec.Raw) } + if !inject.injects() { + continue + } if synth := inject.factRecord(cn, chosen, environmentOf(recs)); synth != nil { out = append(out, synth) } diff --git a/server.go b/server.go index 8231b14..4e046e6 100644 --- a/server.go +++ b/server.go @@ -258,11 +258,15 @@ func (s *Server) mergeNodesResponse(r *http.Request) func([]backendResult) []jso func (s *Server) mergeFactsResponse(r *http.Request) func([]backendResult) []json.RawMessage { inject := s.newSourceInjector(r.URL.Query().Get("query"), true) return func(results []backendResult) []json.RawMessage { + var merged []json.RawMessage if s.cfg.Merge == mergeStatic { - return mergeFacts(results, nil, inject) + merged = mergeFacts(results, nil, inject) + } else { + fresh := s.freshnessMap(context.Background(), results) + merged = mergeFacts(results, func(cn string) string { return fresh[cn] }, inject) } - fresh := s.freshnessMap(context.Background(), results) - return mergeFacts(results, func(cn string) string { return fresh[cn] }, inject) + inject.logSuppressed(s.log) + return merged } } diff --git a/source.go b/source.go index a04e737..2f74449 100644 --- a/source.go +++ b/source.go @@ -1,36 +1,56 @@ package main -import "encoding/json" +import ( + "encoding/json" + "log" +) -// sourceInjector synthesises the provenance fact naming the backend whose data -// won the merge for a given certname. A nil *sourceInjector is the disabled -// case, so every method is nil-safe and callers need no branch. +// sourceInjector owns the configured fact name for one request. A nil +// *sourceInjector is the feature-disabled case, so every method is nil-safe and +// callers need no branch. type sourceInjector struct { name string + // inject is false when the query shape rules synthesis out. Suppression of an + // upstream fact of the same name does not depend on it. + inject bool + suppressed int } -// newSourceInjector returns nil when injection is off for this request. +// newSourceInjector returns nil only when the feature is off; a gated query +// yields an injector that suppresses but does not synthesise. func (s *Server) newSourceInjector(query string, factEntity bool) *sourceInjector { if !s.cfg.SourceFactEnabled || s.cfg.SourceFact == "" { return nil } - if !injectable(query, factEntity) { - return nil - } - return &sourceInjector{name: s.cfg.SourceFact} + return &sourceInjector{name: s.cfg.SourceFact, inject: injectable(query, factEntity)} } -// claims reports whether an upstream record is the one the injector replaces. +// claims reports whether an upstream record carries the name pdbmux owns. While +// the feature is enabled the name means one thing on every query shape, so such +// a record is dropped even when the query gate has ruled synthesis out. func (si *sourceInjector) claims(factName string) bool { return si != nil && factName != "" && factName == si.name } +// injects reports whether this response may carry the synthetic record. +func (si *sourceInjector) injects() bool { + return si != nil && si.inject +} + +// 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 { + return + } + l.Printf("info: dropped %d upstream %q fact record(s); pdbmux owns that fact name", si.suppressed, si.name) +} + // factRecord builds the synthetic /facts record, or nil when disabled. // environment is copied from the node's real facts. All four keys of a fact // record are always emitted, empty environment included: pypuppetdb indexes them // directly (types.py Fact.create_from_dict), so an omitted key is a KeyError. func (si *sourceInjector) factRecord(certname, backend, environment string) json.RawMessage { - if si == nil { + if !si.injects() { return nil } raw, err := json.Marshal(struct { @@ -48,7 +68,7 @@ func (si *sourceInjector) factRecord(certname, backend, environment string) json // stamp adds the provenance key to a /nodes record, overwriting any existing // key of that name. A record that is not a JSON object passes through untouched. func (si *sourceInjector) stamp(raw json.RawMessage, backend string) json.RawMessage { - if si == nil { + if !si.injects() { return raw } var obj map[string]json.RawMessage @@ -68,12 +88,14 @@ func (si *sourceInjector) stamp(raw json.RawMessage, backend string) json.RawMes } // injectable reports whether a response to this query may carry the synthetic -// record. Two shapes are excluded, both because the client asked for something +// record. Three shapes are excluded, each because the client asked for something // the synthetic record is not part of: // -// - a top-level `extract`, which projects a column subset and, with a -// `["function", ...]` column, aggregates — injecting there would corrupt the -// row shape or silently inflate a count(); +// - a query that is not an AST array, which includes every PQL-syntax query: +// pdbmux cannot tell what it projects, so it changes nothing; +// - an `extract` anywhere in the query's own projection scope, which projects a +// column subset and, with a `["function", ...]` column, aggregates — injecting +// there would corrupt the row shape or silently inflate a count(); // - on the facts entity, any outer constraint on `name`, which selects // specific facts. Subquery operands are not descended into: they choose which // nodes match, not which facts come back. @@ -90,7 +112,7 @@ func injectable(query string, factEntity bool) bool { if json.Unmarshal(ast[0], &op) != nil { return false } - if op == "extract" { + if hasExtract(ast) { return false } if !factEntity { @@ -99,6 +121,41 @@ func injectable(query string, factEntity bool) bool { return !constrainsField(ast, "name") } +// hasExtract reports whether an extract appears anywhere in the query's own +// projection scope. openvoxdb accepts an extract as an operand of a boolean +// operator — engine.clj's user-node->plan-node sends every and/or/not operand +// back through itself (src/puppetlabs/puppetdb/query_eng/engine.clj:2697-2733) +// and valid-operator? lists "extract" (:2780-2784) — so the row shape can be +// rewritten below the top level, and the whole tree is walked to fail closed. +// `in` is the one operator not descended into: its operand becomes +// InExpression's :subquery (:2705-2712), projecting the subquery rather than +// the response. +func hasExtract(parts []json.RawMessage) bool { + if len(parts) == 0 { + return false + } + var op string + if json.Unmarshal(parts[0], &op) != nil { + return false + } + switch op { + case "extract": + return true + case "in": + return false + } + for _, p := range parts[1:] { + var sub []json.RawMessage + if json.Unmarshal(p, &sub) != nil { + continue + } + if hasExtract(sub) { + return true + } + } + return false +} + // constrainsField walks the boolean skeleton of an AST node looking for a // comparison whose field operand is field. Only and/or/not are descended into; // anything else, including the subquery operand of `in`, is left alone. diff --git a/source_test.go b/source_test.go index 89aca22..fe63ad5 100644 --- a/source_test.go +++ b/source_test.go @@ -1,7 +1,10 @@ package main import ( + "bytes" "encoding/json" + "io" + "log" "net/http" "slices" "strings" @@ -220,9 +223,90 @@ func TestHandler_UpstreamSourceFactOverridden(t *testing.T) { } } -// A count() must report the backends' real fact count, not one inflated by a -// record pdbmux invented. -func TestHandler_SourceNotInjectedOnAggregate(t *testing.T) { +// The configured name means one thing on every query shape: an upstream fact of +// that name is dropped whether or not the query gate allows synthesis. +func TestHandler_UpstreamSourceFactSuppressedOnEveryGateState(t *testing.T) { + const upstream = "REAL-UPSTREAM-VALUE" + tests := []struct { + name string + query string + want string // synthetic value, or "" when the gate blocks injection + }{ + {"injection on", "", "a"}, + {"gated by extract", `["extract",["certname","name","value"],["=","certname","h1"]]`, ""}, + {"gated by name filter", `["=","name","` + defaultSourceFact + `"]`, ""}, + {"gated by nested extract", `["and",["=","certname","h1"],["extract",["certname"]]]`, ""}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, + `[`+fact("h1", "role", "web", "")+`,`+fact("h1", defaultSourceFact, upstream, "")+`]`) + b := newFakeBackend(t, `[]`, `[]`) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), factsPath, tc.query) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + if strings.Contains(rec.Body.String(), upstream) { + t.Fatalf("upstream value survived: %s", rec.Body.String()) + } + got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact) + if tc.want == "" { + if n != 0 { + t.Errorf("gated query produced %d %s records: %s", n, defaultSourceFact, rec.Body.String()) + } + } else if n != 1 || got["h1"] != tc.want { + t.Errorf("provenance = %v (%d records), want h1=%s: %s", got, n, tc.want, rec.Body.String()) + } + if !strings.Contains(rec.Body.String(), `"role"`) { + t.Errorf("real facts were dropped: %s", rec.Body.String()) + } + }) + } +} + +// Dropping an upstream record is invisible in the response, so it is logged — +// once per request, not once per record. +func TestHandler_SuppressedUpstreamFactLoggedOncePerRequest(t *testing.T) { + a := newFakeBackend(t, + `[`+node("h1", "2026-07-20T00:00:00Z")+`,`+node("h2", "2026-07-20T00:00:00Z")+`]`, + `[`+fact("h1", defaultSourceFact, "old-h1", "")+`,`+fact("h2", defaultSourceFact, "old-h2", "")+`]`) + b := newFakeBackend(t, `[]`, `[]`) + var buf bytes.Buffer + srv := NewServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic), log.New(&buf, "", 0)) + + doGet(t, srv.Handler(), factsPath, "") + if n := strings.Count(buf.String(), defaultSourceFact); n != 1 { + t.Fatalf("expected 1 log line naming the fact, got %d: %s", n, buf.String()) + } + if !strings.Contains(buf.String(), "dropped 2") { + t.Errorf("log does not report the number dropped: %s", buf.String()) + } +} + +// Disabled means untouched: an upstream fact of the configured name is served as +// the backend reported it. +func TestHandler_SourceDisabledKeepsUpstreamFact(t *testing.T) { + a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, + `[`+fact("h1", defaultSourceFact, "upstream-value", "")+`]`) + b := newFakeBackend(t, `[]`, `[]`) + cfg := testConfig(a.srv.URL, b.srv.URL, mergeStatic) + cfg.SourceFactEnabled = false + srv := newTestServer(cfg) + + rec := doGet(t, srv.Handler(), factsPath, "") + got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact) + if n != 1 || got["h1"] != "upstream-value" { + t.Errorf("disabled injection altered the upstream fact: %s", rec.Body.String()) + } +} + +// Covers only that an aggregate /facts response gains no synthetic record and no +// rewritten row. It does not cover whether the aggregate rows are correct: +// /facts has no parseAggregate branch, so its rows take the certname merge +// instead of serveSummed and are not summed across backends. +func TestHandler_SourceNotInjectedOnFactsAggregate(t *testing.T) { a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`) b := newFakeBackend(t, `[]`, `[]`) a.bodies[factsPath] = `[{"count":3}]` @@ -333,6 +417,16 @@ func TestInjectable(t *testing.T) { {"name only inside subquery", `["in","certname",["extract",["certname"],["select_facts",["=","name","os"]]]]`, true, true}, {"top-level extract", `["extract",["certname","value"],["=","certname","h1"]]`, true, false}, {"aggregate extract", `["extract",[["function","count"]]]`, true, false}, + // openvoxdb hands each boolean operand back to user-node->plan-node, which + // builds an extract node from it, so a nested extract can reshape the rows. + {"extract under and", `["and",["=","certname","h1"],["extract",["certname"]]]`, true, false}, + {"extract under or", `["or",["extract",["certname"]],["=","certname","h1"]]`, true, false}, + {"extract under not", `["not",["extract",["certname"]]]`, true, false}, + {"extract nested two deep", `["and",["or",["extract",[["function","count"]]]]]`, true, false}, + {"extract under from", `["from","facts",["extract",["certname"]]]`, true, false}, + {"nested extract on nodes", `["and",["extract",["certname"]]]`, false, false}, + // An extract inside an `in` operand projects the subquery, not the response. + {"extract under in stays injectable", `["in","certname",["extract",["certname"],["select_facts",["=","name","os"]]]]`, true, true}, {"nodes name filter is not a fact filter", `["=","name","os"]`, false, true}, {"nodes extract", `["extract",["certname"]]`, false, false}, {"unparseable query", `not json`, true, false}, @@ -350,7 +444,7 @@ func TestInjectable(t *testing.T) { // pypuppetdb reads certname/name/value/environment by direct index, so a // missing key is a KeyError there — all four are always present. func TestSourceInjector_FactRecordHasEveryFactKey(t *testing.T) { - si := &sourceInjector{name: defaultSourceFact} + si := &sourceInjector{name: defaultSourceFact, inject: true} for _, env := range []string{"production", ""} { var obj map[string]json.RawMessage if err := json.Unmarshal(si.factRecord("h1", "a", env), &obj); err != nil { @@ -373,6 +467,10 @@ func TestSourceInjector_NilIsInert(t *testing.T) { if si.claims(defaultSourceFact) { t.Error("nil injector claims a fact name") } + if si.injects() { + t.Error("nil injector injects") + } + si.logSuppressed(log.New(io.Discard, "", 0)) if si.factRecord("h1", "a", "production") != nil { t.Error("nil injector produced a record") } @@ -385,7 +483,7 @@ func TestSourceInjector_NilIsInert(t *testing.T) { // A response element that is not a JSON object cannot be stamped, and must be // passed through rather than dropped or mangled. func TestSourceInjector_StampNonObject(t *testing.T) { - si := &sourceInjector{name: defaultSourceFact} + si := &sourceInjector{name: defaultSourceFact, inject: true} for _, raw := range []string{`"scalar"`, `[1,2]`, `null`} { if got := si.stamp(json.RawMessage(raw), "a"); string(got) != raw { t.Errorf("stamp(%s) = %s, want unchanged", raw, got) @@ -393,10 +491,53 @@ func TestSourceInjector_StampNonObject(t *testing.T) { } } +// A gated injector suppresses without synthesising anything, on either endpoint. +func TestSourceInjector_GatedSuppressesButDoesNotInject(t *testing.T) { + si := &sourceInjector{name: defaultSourceFact} + if !si.claims(defaultSourceFact) { + t.Error("gated injector does not claim its own fact name") + } + if si.factRecord("h1", "a", "production") != nil { + t.Error("gated injector produced a record") + } + raw := json.RawMessage(`{"certname":"h1"}`) + if got := si.stamp(raw, "a"); string(got) != string(raw) { + t.Errorf("gated injector rewrote %s to %s", raw, got) + } +} + +// The disabled path must return the backends' records verbatim, source fact +// included. +func TestMergeFacts_DisabledIsUntouched(t *testing.T) { + a := recs(t, "a", fact("h1", "role", "web-a", ""), fact("h1", defaultSourceFact, "upstream", "")) + merged := mergeFacts([]backendResult{a}, nil, nil) + + got := factValues(t, merged) + want := []string{"h1:role=web-a", "h1:" + defaultSourceFact + "=upstream"} + if !slices.Equal(got, want) { + t.Errorf("merged = %v, want %v", got, want) + } +} + +// A gated query still gets the upstream record removed, and nothing added. +func TestMergeFacts_GatedSuppressesUpstream(t *testing.T) { + a := recs(t, "a", fact("h1", "role", "web-a", ""), fact("h1", defaultSourceFact, "upstream", "")) + si := &sourceInjector{name: defaultSourceFact} + merged := mergeFacts([]backendResult{a}, nil, si) + + got := factValues(t, merged) + if !slices.Equal(got, []string{"h1:role=web-a"}) { + t.Errorf("merged = %v, want only the real fact", got) + } + if si.suppressed != 1 { + t.Errorf("suppressed = %d, want 1", si.suppressed) + } +} + func TestMergeFacts_SourceOrderedAfterOwnersFacts(t *testing.T) { a := recs(t, "a", fact("h1", "role", "web-a", ""), fact("h1", "kernel", "Linux", "")) b := recs(t, "b", fact("h1", "role", "web-b", "")) - merged := mergeFacts([]backendResult{a, b}, nil, &sourceInjector{name: defaultSourceFact}) + merged := mergeFacts([]backendResult{a, b}, nil, &sourceInjector{name: defaultSourceFact, inject: true}) got := factValues(t, merged) want := []string{"h1:role=web-a", "h1:kernel=Linux", "h1:" + defaultSourceFact + "=a"}