diff --git a/README.md b/README.md index a5747d2..07fdba0 100644 --- a/README.md +++ b/README.md @@ -24,8 +24,8 @@ not PQL) is forwarded verbatim. | Path | Behaviour | |---|---| -| `GET /pdb/query/v4/nodes` | Fan out to all backends, dedupe by `certname`, keep the record with the newer `report_timestamp`. 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). | +| `GET /pdb/query/v4/nodes` | Fan out to all backends, dedupe by `certname`, keep the record with the newer `report_timestamp`, stamped with the winning backend's name (see provenance). An `extract`/`count` query is **summed** instead. | +| `GET /pdb/query/v4/facts` | Fan out to all, and per `certname` keep **all** facts from the backend that owns that node (see merge semantics), plus a synthetic `pdbmux_source` fact naming it. | | `GET /pdb/query/v4/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. | @@ -91,6 +91,74 @@ unknown fields survive untouched. - `include_total=true` on a summed endpoint reports the **merged** row count, not the sum of the backends' `X-Records`, since shared keys collapse. +### Provenance: the `pdbmux_source` fact + +Once several PuppetDBs sit behind one endpoint, a consumer can no longer tell +which backend a node's data came from. `pdbmux` makes that visible in the +response itself, so nothing has to query each backend to find out: + +- **`/facts`** gains one extra fact record per `certname`, alongside the node's + real facts, in the shape of a real fact record — `certname`, `name`, `value`, + `environment` — with `value` set to the **backend name** from `backends` / + `PDBMUX_BACKENDS`. `environment` is copied from that node's own facts (all + four keys are always present, since clients index them directly). +- **`/nodes`** gains a `pdbmux_source` **key** on each merged node record. A + node record carries no facts, so this is a synthetic field, not a fact — the + one key outside PuppetDB's documented node schema. Clients read node fields by + name, so an extra key is ignored by anything that doesn't want it. + +The value always names the backend **whose data won that endpoint's merge**, not +a backend that merely holds the node. The two endpoints resolve their winner +separately, so under `merge: static` they can legitimately disagree: `/facts` +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. + +While the fact is enabled, any `/facts` record whose own `name` field equals the +configured name is dropped, on every query shape — including the shapes below, +where nothing is injected in its place. The rule reads the record, not the query, +so a projection that filters on `name` without returning it — say +`["extract",["certname","value"],["=","name","pdbmux_source"]]` — produces rows +that no longer identify themselves, and an upstream value of that name comes +through. Ask for the `name` column and the guarantee holds. Each request that +drops a record 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 no synthetic record is added, when: + +- the query contains an `extract` outside a 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` under `in`, + `subquery`, or `select_` projects that 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` + subquery narrows which *nodes* match, not which facts come back, so injection + still happens; +- injection is turned off (see `source_fact_enabled`). + +**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. + +**Not covered:** `/factsets` and `/inventory`. Both carry facts, but `pdbmux` +does not merge either today — they take the unmerged pass-through path, where +the answer comes from whichever backend replied first rather than from a merge +winner, so there is no owner to attribute. Injecting there would state a +provenance that isn't true. + ### Metadata and metrics - **`/pdb/meta/v1/version`** — when the backends agree, that version is served. @@ -164,6 +232,8 @@ backends: # order is a tie-break only, not a ranking merge: freshness # freshness | static timeout: 10s # per-upstream request timeout freshness_ttl: 30s # freshness-map cache TTL (freshness merge only) +source_fact: pdbmux_source # name of the synthetic provenance fact +source_fact_enabled: true # false serves backends' records untouched ``` `backends[*].url` is a **base** URL (`scheme://host[:port]`); `pdbmux` appends @@ -177,6 +247,8 @@ the `/pdb/query/v4/...` path per request. | `PDBMUX_TIMEOUT` | `timeout` (Go duration, e.g. `10s`) | | `PDBMUX_FRESHNESS_TTL` | `freshness_ttl` | | `PDBMUX_BACKENDS` | whole backend list, as `name=url,name=url` | +| `PDBMUX_SOURCE_FACT` | `source_fact` (default `pdbmux_source`) | +| `PDBMUX_SOURCE_FACT_ENABLED` | `source_fact_enabled` (default `true`); `false` disables injection | Flags: `--config`, `--listen`, `--merge`. diff --git a/config.go b/config.go index 45102ec..6b5cddc 100644 --- a/config.go +++ b/config.go @@ -25,6 +25,8 @@ const ( defaultTimeout = 10 * time.Second defaultFreshnessTTL = 30 * time.Second + + defaultSourceFact = "pdbmux_source" ) var exampleBackends = []Backend{ @@ -44,6 +46,9 @@ type Config struct { Timeout time.Duration `yaml:"timeout"` FreshnessTTL time.Duration `yaml:"freshness_ttl"` + SourceFact string `yaml:"source_fact"` + SourceFactEnabled bool `yaml:"source_fact_enabled"` + sourcePath string // file this config was read from, empty if none was found } @@ -57,10 +62,12 @@ const ( func DefaultConfig() Config { return Config{ - Listen: defaultListen, - Merge: mergeFreshness, - Timeout: defaultTimeout, - FreshnessTTL: defaultFreshnessTTL, + Listen: defaultListen, + Merge: mergeFreshness, + Timeout: defaultTimeout, + FreshnessTTL: defaultFreshnessTTL, + SourceFact: defaultSourceFact, + SourceFactEnabled: true, } } @@ -159,6 +166,14 @@ func applyEnv(cfg *Config, getenv func(string) string) { cfg.FreshnessTTL = d } } + if v := getenv(envPrefix + "SOURCE_FACT"); v != "" { + cfg.SourceFact = v + } + if v := getenv(envPrefix + "SOURCE_FACT_ENABLED"); v != "" { + if b, err := strconv.ParseBool(v); err == nil { + cfg.SourceFactEnabled = b + } + } if v := getenv(envPrefix + "BACKENDS"); v != "" { if bs := parseBackends(v); len(bs) > 0 { cfg.Backends = bs @@ -216,6 +231,9 @@ func (c Config) Validate() error { if c.Timeout <= 0 { return fmt.Errorf("timeout must be positive") } + if c.SourceFactEnabled && c.SourceFact == "" { + return fmt.Errorf("source_fact must be non-empty, or set source_fact_enabled to false") + } return nil } @@ -231,7 +249,8 @@ func writeDefaultConfig(path string) error { "# A merging proxy presenting one PuppetDB v4 query surface over several\n" + "# PuppetDB backends. The backend URLs below are placeholders — edit them.\n" + "# Env overrides: PDBMUX_LISTEN, PDBMUX_MERGE, PDBMUX_TIMEOUT,\n" + - "# PDBMUX_FRESHNESS_TTL, PDBMUX_BACKENDS (name=url,name=url).\n\n") + "# PDBMUX_FRESHNESS_TTL, PDBMUX_BACKENDS (name=url,name=url),\n" + + "# PDBMUX_SOURCE_FACT, PDBMUX_SOURCE_FACT_ENABLED.\n\n") if err := os.WriteFile(path, append(header, data...), 0o644); err != nil { return fmt.Errorf("writing config: %w", err) } diff --git a/config_test.go b/config_test.go index 9cc4096..bfdbf36 100644 --- a/config_test.go +++ b/config_test.go @@ -118,6 +118,69 @@ func TestApplyEnv_Backends(t *testing.T) { } } +func TestDefaultConfig_SourceFact(t *testing.T) { + cfg := DefaultConfig() + if cfg.SourceFact != defaultSourceFact || !cfg.SourceFactEnabled { + t.Errorf("source fact defaults to %q enabled=%v, want %q enabled=true", + cfg.SourceFact, cfg.SourceFactEnabled, defaultSourceFact) + } +} + +func TestApplyEnv_SourceFact(t *testing.T) { + cfg := testConfigValid() + env := map[string]string{envPrefix + "SOURCE_FACT": "origin_pdb"} + applyEnv(&cfg, func(k string) string { return env[k] }) + if cfg.SourceFact != "origin_pdb" || !cfg.SourceFactEnabled { + t.Errorf("name override failed: %q enabled=%v", cfg.SourceFact, cfg.SourceFactEnabled) + } + + cfg = testConfigValid() + env = map[string]string{envPrefix + "SOURCE_FACT_ENABLED": "false"} + applyEnv(&cfg, func(k string) string { return env[k] }) + if cfg.SourceFactEnabled { + t.Error("PDBMUX_SOURCE_FACT_ENABLED=false must disable injection") + } + + // A junk boolean leaves the default alone rather than disabling silently. + cfg = testConfigValid() + env = map[string]string{envPrefix + "SOURCE_FACT_ENABLED": "maybe"} + applyEnv(&cfg, func(k string) string { return env[k] }) + if !cfg.SourceFactEnabled { + t.Error("unparseable bool must not change the setting") + } +} + +// A config file omitting the key keeps the default; setting it false wins. +func TestLoad_SourceFactFileOverride(t *testing.T) { + dir := t.TempDir() + t.Setenv("XDG_CONFIG_HOME", dir) + path := filepath.Join(dir, appName, configFileName) + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + t.Fatal(err) + } + write := func(body string) Config { + t.Helper() + if err := os.WriteFile(path, []byte(body), 0o644); err != nil { + t.Fatal(err) + } + cfg, err := Load("") + if err != nil { + t.Fatal(err) + } + return cfg + } + + cfg := write("backends:\n - name: a\n url: http://a:8080\n") + if cfg.SourceFact != defaultSourceFact || !cfg.SourceFactEnabled { + t.Errorf("omitted keys must keep defaults: %q enabled=%v", cfg.SourceFact, cfg.SourceFactEnabled) + } + + cfg = write("backends:\n - name: a\n url: http://a:8080\nsource_fact: origin_pdb\nsource_fact_enabled: false\n") + if cfg.SourceFact != "origin_pdb" || cfg.SourceFactEnabled { + t.Errorf("file override failed: %q enabled=%v", cfg.SourceFact, cfg.SourceFactEnabled) + } +} + func TestValidate(t *testing.T) { cases := []struct { name string @@ -130,6 +193,8 @@ func TestValidate(t *testing.T) { {"missing url", func(c *Config) { c.Backends[0].URL = "" }, true}, {"bad merge", func(c *Config) { c.Merge = "wrong" }, true}, {"zero timeout", func(c *Config) { c.Timeout = 0 }, true}, + {"empty source fact while enabled", func(c *Config) { c.SourceFact = "" }, true}, + {"empty source fact while disabled", func(c *Config) { c.SourceFact = ""; c.SourceFactEnabled = false }, false}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { diff --git a/main.go b/main.go index 413d99b..de1cd43 100644 --- a/main.go +++ b/main.go @@ -161,6 +161,11 @@ func printConfig(cfg Config) { fmt.Printf("merge : %s\n", cfg.Merge) fmt.Printf("timeout : %s\n", durationString(cfg.Timeout)) fmt.Printf("freshness_ttl: %s\n", durationString(cfg.FreshnessTTL)) + if cfg.SourceFactEnabled { + fmt.Printf("source_fact : %s\n", cfg.SourceFact) + } else { + fmt.Printf("source_fact : disabled\n") + } fmt.Println("backends:") for _, b := range cfg.Backends { fmt.Printf(" - %-8s %s\n", b.Name, b.URL) diff --git a/merge.go b/merge.go index d9cce56..fcb530e 100644 --- a/merge.go +++ b/merge.go @@ -11,12 +11,16 @@ type record struct { Certname string ReportTimestamp string // only populated for /nodes records Hash string // only populated for /reports records + Name string // only populated for /facts records + Environment string } type recordMeta struct { Certname string `json:"certname"` ReportTimestamp string `json:"report_timestamp"` Hash string `json:"hash"` + Name string `json:"name"` + Environment string `json:"environment"` } func decodeRecords(body []byte) ([]record, error) { @@ -33,6 +37,8 @@ func decodeRecords(body []byte) ([]record, error) { Certname: m.Certname, ReportTimestamp: m.ReportTimestamp, Hash: m.Hash, + Name: m.Name, + Environment: m.Environment, }) } return out, nil @@ -50,10 +56,12 @@ func parseTimestamp(s string) time.Time { } // Ties keep the earlier backend's record — a deterministic tie-break, not a preference. -func mergeNodes(results []backendResult) []json.RawMessage { +// A non-nil inject stamps each surviving record with the backend that supplied it. +func mergeNodes(results []backendResult, inject *sourceInjector) []json.RawMessage { type pick struct { - raw json.RawMessage - ts time.Time + raw json.RawMessage + ts time.Time + backend string } best := map[string]pick{} var order []string @@ -62,18 +70,19 @@ func mergeNodes(results []backendResult) []json.RawMessage { ts := parseTimestamp(rec.ReportTimestamp) cur, ok := best[rec.Certname] if !ok { - best[rec.Certname] = pick{raw: rec.Raw, ts: ts} + best[rec.Certname] = pick{raw: rec.Raw, ts: ts, backend: res.name} order = append(order, rec.Certname) continue } if ts.After(cur.ts) { - best[rec.Certname] = pick{raw: rec.Raw, ts: ts} + best[rec.Certname] = pick{raw: rec.Raw, ts: ts, backend: res.name} } } } out := make([]json.RawMessage, 0, len(order)) for _, cn := range order { - out = append(out, best[cn].raw) + p := best[cn] + out = append(out, inject.stamp(p.raw, p.backend)) } return out } @@ -105,16 +114,17 @@ 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. -func mergeFacts(results []backendResult, owner func(certname string) string) []json.RawMessage { +// 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][]json.RawMessage{} + byKey := map[string][]record{} for _, res := range results { for _, rec := range res.records { key := rec.Certname + "\x00" + res.name if _, ok := byKey[key]; !ok { present[rec.Certname] = append(present[rec.Certname], res.name) } - byKey[key] = append(byKey[key], rec.Raw) + byKey[key] = append(byKey[key], rec) } } @@ -140,11 +150,35 @@ func mergeFacts(results []backendResult, owner func(certname string) string) []j if !contains(backends, chosen) { chosen = backends[0] } - out = append(out, byKey[cn+"\x00"+chosen]...) + recs := byKey[cn+"\x00"+chosen] + for _, rec := range recs { + // 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) + } } return out } +func environmentOf(recs []record) string { + for _, rec := range recs { + if rec.Environment != "" { + return rec.Environment + } + } + return "" +} + func contains(s []string, v string) bool { for _, x := range s { if x == v { diff --git a/merge_test.go b/merge_test.go index 8c111d8..8e6027d 100644 --- a/merge_test.go +++ b/merge_test.go @@ -92,7 +92,7 @@ func TestMergeNodes_NewerWins(t *testing.T) { a := recs(t, "a", node("h1", "2026-07-01T00:00:00Z"), node("h2", "2026-07-10T00:00:00Z")) b := recs(t, "b", node("h1", "2026-07-20T00:00:00Z"), node("h3", "2026-07-05T00:00:00Z")) - merged := mergeNodes([]backendResult{a, b}) + merged := mergeNodes([]backendResult{a, b}, nil) got := map[string]string{} for _, r := range merged { var m recordMeta @@ -117,7 +117,7 @@ func TestMergeNodes_OneBackendOnly(t *testing.T) { a := recs(t, "a", node("h1", "2026-07-01T00:00:00Z")) // b returned nothing (e.g. empty result). b := backendResult{name: "b"} - merged := mergeNodes([]backendResult{a, b}) + merged := mergeNodes([]backendResult{a, b}, nil) if len(merged) != 1 || certnames(t, merged)[0] != "h1" { t.Fatalf("expected only h1, got %v", certnames(t, merged)) } @@ -127,7 +127,7 @@ func TestMergeNodes_TieKeepsEarlierBackend(t *testing.T) { // Equal timestamps: the backend listed first wins, as a tie-break. first := recs(t, "b", node("h1", "2026-07-01T00:00:00Z")) second := recs(t, "a", node("h1", "2026-07-01T00:00:00Z")) - merged := mergeNodes([]backendResult{first, second}) + merged := mergeNodes([]backendResult{first, second}, nil) if len(merged) != 1 { t.Fatalf("expected 1 record, got %d", len(merged)) } @@ -139,7 +139,7 @@ func TestMergeNodes_TieKeepsEarlierBackend(t *testing.T) { func TestMergeNodes_PreservesUnknownFields(t *testing.T) { a := recs(t, "a", `{"certname":"h1","report_timestamp":"2026-07-01T00:00:00Z","extra":{"deep":42}}`) - merged := mergeNodes([]backendResult{a}) + merged := mergeNodes([]backendResult{a}, nil) if len(merged) != 1 { t.Fatalf("expected 1 record") } @@ -156,7 +156,7 @@ func TestMergeFacts_NilOwnerUsesConfiguredOrder(t *testing.T) { a := recs(t, "a", fact("h1", "role", "web-a", ""), fact("h2", "role", "db-a", "")) b := recs(t, "b", fact("h1", "role", "web-b", "")) - merged := mergeFacts([]backendResult{b, a}, nil) + merged := mergeFacts([]backendResult{b, a}, nil, nil) got := factValues(t, merged) assertContains(t, got, "h1:role=web-b") assertNotContains(t, got, "h1:role=web-a") @@ -180,7 +180,7 @@ func TestMergeFacts_Freshness_NewerBackendWins(t *testing.T) { } return "b" } - merged := mergeFacts([]backendResult{b, a}, owner) + merged := mergeFacts([]backendResult{b, a}, owner, nil) got := factValues(t, merged) // h1 -> all a facts, no b facts. assertContains(t, got, "h1:role=web-a") @@ -198,7 +198,7 @@ func TestMergeFacts_OwnerMissingFallsBackToConfiguredOrder(t *testing.T) { // backend in the slice that has some. first := recs(t, "b", fact("h1", "role", "web-b", "")) second := recs(t, "a", fact("h1", "role", "web-a", "")) - merged := mergeFacts([]backendResult{first, second}, func(string) string { return "ghost" }) + merged := mergeFacts([]backendResult{first, second}, func(string) string { return "ghost" }, nil) got := factValues(t, merged) assertContains(t, got, "h1:role=web-b") // b is first in slice assertNotContains(t, got, "h1:role=web-a") diff --git a/server.go b/server.go index 7381ca0..4e046e6 100644 --- a/server.go +++ b/server.go @@ -75,7 +75,7 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) { case resourcesPath: s.serveResources(w, r) case factsPath: - s.serveMerged(w, r, factsPath, s.mergeFactsResponse) + s.serveMerged(w, r, factsPath, s.mergeFactsResponse(r)) case reportsPath: s.serveReports(w, r) case eventsPath: @@ -146,7 +146,7 @@ func (s *Server) serveNodes(w http.ResponseWriter, r *http.Request) { s.serveSummed(w, r, nodesPath, spec.columns) return } - s.serveMerged(w, r, nodesPath, s.mergeNodesResponse) + s.serveMerged(w, r, nodesPath, s.mergeNodesResponse(r)) } // Only aggregates merge: a resource record has no cross-backend identity to dedupe on, so a plain query stays on the pass-through path. @@ -248,16 +248,26 @@ func rawRecords(recs []record) []json.RawMessage { return out } -func (s *Server) mergeNodesResponse(results []backendResult) []json.RawMessage { - return mergeNodes(results) +func (s *Server) mergeNodesResponse(r *http.Request) func([]backendResult) []json.RawMessage { + inject := s.newSourceInjector(r.URL.Query().Get("query"), false) + return func(results []backendResult) []json.RawMessage { + return mergeNodes(results, inject) + } } -func (s *Server) mergeFactsResponse(results []backendResult) []json.RawMessage { - if s.cfg.Merge == mergeStatic { - return mergeFacts(results, nil) +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 { + merged = mergeFacts(results, nil, inject) + } else { + fresh := s.freshnessMap(context.Background(), results) + merged = mergeFacts(results, func(cn string) string { return fresh[cn] }, inject) + } + inject.logSuppressed(s.log) + return merged } - fresh := s.freshnessMap(context.Background(), results) - return mergeFacts(results, func(cn string) string { return fresh[cn] }) } // Queries /nodes unfiltered rather than reusing the request's results, because a /facts query's certname set can differ. diff --git a/server_test.go b/server_test.go index 1b0dfb6..a426837 100644 --- a/server_test.go +++ b/server_test.go @@ -133,11 +133,13 @@ func truncate(t *testing.T, body, limit string) string { func testConfig(aURL, bURL, merge string) Config { return Config{ - Listen: ":0", - Backends: []Backend{{Name: "a", URL: aURL}, {Name: "b", URL: bURL}}, - Merge: merge, - Timeout: 2 * time.Second, - FreshnessTTL: 30 * time.Second, + Listen: ":0", + Backends: []Backend{{Name: "a", URL: aURL}, {Name: "b", URL: bURL}}, + Merge: merge, + Timeout: 2 * time.Second, + FreshnessTTL: 30 * time.Second, + SourceFact: defaultSourceFact, + SourceFactEnabled: true, } } diff --git a/source.go b/source.go new file mode 100644 index 0000000..feec363 --- /dev/null +++ b/source.go @@ -0,0 +1,198 @@ +package main + +import ( + "encoding/json" + "log" + "strings" +) + +// 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 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 + } + return &sourceInjector{name: s.cfg.SourceFact, inject: injectable(query, factEntity)} +} + +// claims reports whether an upstream record carries the name pdbmux owns, and is +// keyed on the record's own name field: a projection that omits the name column +// yields records that cannot be identified, so they pass through. Suppression +// does not depend on the query gate. +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.injects() { + return nil + } + raw, err := json.Marshal(struct { + Certname string `json:"certname"` + Environment string `json:"environment"` + Name string `json:"name"` + Value string `json:"value"` + }{Certname: certname, Environment: environment, Name: si.name, Value: backend}) + if err != nil { + return nil + } + return raw +} + +// 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.injects() { + return raw + } + var obj map[string]json.RawMessage + if json.Unmarshal(raw, &obj) != nil || obj == nil { + return raw + } + value, err := json.Marshal(backend) + if err != nil { + return raw + } + obj[si.name] = value + out, err := json.Marshal(obj) + if err != nil { + return raw + } + return out +} + +// injectable reports whether a response to this query may carry the synthetic +// record. Three shapes are excluded, each because the client asked for something +// the synthetic record is not part of: +// +// - 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(). Operands +// scoped to a subquery are excluded; see hasExtract; +// - 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. +func injectable(query string, factEntity bool) bool { + if query == "" { + return true + } + var ast []json.RawMessage + if json.Unmarshal([]byte(query), &ast) != nil || len(ast) == 0 { + // Not an AST array pdbmux can reason about; leave the response alone. + return false + } + var op string + if json.Unmarshal(ast[0], &op) != nil { + return false + } + if hasExtract(ast) { + return false + } + if !factEntity { + return true + } + 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" (:2779-2784) — so the row shape can be +// rewritten below the top level, and the whole tree is walked to fail closed. +// +// Operators whose operand is scoped to a subquery are not descended into, +// because an extract there projects the subquery rather than the response: +// +// - `in`, whose operand becomes InExpression's :subquery (:2705-2712); +// - `subquery`, which the AST-rewrite stage expands into +// ["in" cols ["extract" cols ["select_" expr]]] (:2111-2123) +// before any plan node is built; +// - `select_`, the explicit subquery form (:1889-1911), reachable +// only under one of the two above in a query openvoxdb accepts. +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 { + case op == "extract": + return true + case op == "in", op == "subquery", strings.HasPrefix(op, "select_"): + 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. +func constrainsField(parts []json.RawMessage, field string) bool { + if len(parts) == 0 { + return false + } + var op string + if json.Unmarshal(parts[0], &op) != nil { + return false + } + switch op { + case "and", "or", "not": + for _, p := range parts[1:] { + var sub []json.RawMessage + if json.Unmarshal(p, &sub) != nil { + continue + } + if constrainsField(sub, field) { + return true + } + } + return false + } + if len(parts) < 2 { + return false + } + var name string + return json.Unmarshal(parts[1], &name) == nil && name == field +} diff --git a/source_test.go b/source_test.go new file mode 100644 index 0000000..ae63bf0 --- /dev/null +++ b/source_test.go @@ -0,0 +1,578 @@ +package main + +import ( + "bytes" + "encoding/json" + "io" + "log" + "net/http" + "slices" + "strings" + "testing" +) + +// sourceValues returns certname -> value of the synthetic fact record, and the +// number of records carrying that fact name. +func sourceValues(t *testing.T, body []byte, factName string) (map[string]string, int) { + t.Helper() + var raws []json.RawMessage + if err := json.Unmarshal(body, &raws); err != nil { + t.Fatalf("unmarshal %s: %v", body, err) + } + out := map[string]string{} + n := 0 + for _, raw := range raws { + var m struct { + Certname string `json:"certname"` + Name string `json:"name"` + Value string `json:"value"` + } + if json.Unmarshal(raw, &m) != nil || m.Name != factName { + continue + } + out[m.Certname] = m.Value + n++ + } + return out, n +} + +// nodeSources returns certname -> the stamped provenance field on /nodes records. +func nodeSources(t *testing.T, body []byte, field string) map[string]string { + t.Helper() + var raws []json.RawMessage + if err := json.Unmarshal(body, &raws); err != nil { + t.Fatalf("unmarshal %s: %v", body, err) + } + out := map[string]string{} + for _, raw := range raws { + var obj map[string]json.RawMessage + if err := json.Unmarshal(raw, &obj); err != nil { + t.Fatalf("unmarshal record %s: %v", raw, err) + } + var m recordMeta + _ = json.Unmarshal(raw, &m) + v, ok := obj[field] + if !ok { + continue + } + var s string + if err := json.Unmarshal(v, &s); err != nil { + t.Fatalf("provenance field of %s is not a string: %v", raw, err) + } + out[m.Certname] = s + } + return out +} + +func factEnv(cn, name, val, env string) string { + return `{"certname":"` + cn + `","name":"` + name + `","value":"` + val + `","environment":"` + env + `"}` +} + +// Both backends hold h1; a holds its newer report, so h1's facts and its +// provenance fact must both come from a. +func TestHandler_FactsSourceFollowsMergeOwner(t *testing.T) { + a := newFakeBackend(t, + `[`+node("h1", "2026-07-20T00:00:00Z")+`,`+node("h2", "2026-07-01T00:00:00Z")+`]`, + `[`+factEnv("h1", "role", "web-a", "production")+`,`+factEnv("h2", "role", "db-a", "production")+`]`) + b := newFakeBackend(t, + `[`+node("h1", "2026-07-01T00:00:00Z")+`,`+node("h2", "2026-07-20T00:00:00Z")+`]`, + `[`+factEnv("h1", "role", "web-b", "staging")+`,`+factEnv("h2", "role", "db-b", "staging")+`]`) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness)) + + rec := doGet(t, srv.Handler(), factsPath, "") + 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 != 2 { + t.Fatalf("expected one %s record per certname, got %d: %s", defaultSourceFact, n, rec.Body.String()) + } + if got["h1"] != "a" || got["h2"] != "b" { + t.Errorf("provenance must name the backend that won the merge, got %v", got) + } +} + +// The synthetic record carries the node's own environment so it groups with the +// real facts rather than landing in an unrelated environment. +func TestHandler_FactsSourceCopiesEnvironment(t *testing.T) { + a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, + `[`+factEnv("h1", "role", "web", "staging")+`]`) + b := newFakeBackend(t, `[]`, `[]`) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), factsPath, "") + var raws []json.RawMessage + if err := json.Unmarshal(rec.Body.Bytes(), &raws); err != nil { + t.Fatal(err) + } + var found bool + for _, raw := range raws { + var m recordMeta + if json.Unmarshal(raw, &m) != nil || m.Name != defaultSourceFact { + continue + } + found = true + if m.Environment != "staging" { + t.Errorf("environment = %q, want staging: %s", m.Environment, raw) + } + } + if !found { + t.Fatalf("no %s record: %s", defaultSourceFact, rec.Body.String()) + } +} + +// A node record's provenance names the backend whose node record won /nodes' +// own report_timestamp merge. +func TestHandler_NodesSourceStamped(t *testing.T) { + a := newFakeBackend(t, + `[`+node("h1", "2026-07-01T00:00:00Z")+`,`+node("h2", "2026-07-20T00:00:00Z")+`]`, `[]`) + b := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), nodesPath, "") + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + got := nodeSources(t, rec.Body.Bytes(), defaultSourceFact) + if got["h1"] != "b" || got["h2"] != "a" { + t.Errorf("node provenance = %v, want h1=b h2=a", got) + } +} + +// Stamping must not drop unknown upstream fields. +func TestHandler_NodesSourceKeepsUpstreamFields(t *testing.T) { + a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`) + b := newFakeBackend(t, `[]`, `[]`) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), nodesPath, "") + var raws []json.RawMessage + if err := json.Unmarshal(rec.Body.Bytes(), &raws); err != nil { + t.Fatal(err) + } + if len(raws) != 1 { + t.Fatalf("expected 1 node, got %d", len(raws)) + } + var obj map[string]json.RawMessage + if err := json.Unmarshal(raws[0], &obj); err != nil { + t.Fatal(err) + } + for _, k := range []string{"certname", "report_timestamp", "latest_report_status", defaultSourceFact} { + if _, ok := obj[k]; !ok { + t.Errorf("field %q missing from stamped record: %s", k, raws[0]) + } + } +} + +func TestHandler_SourceDisabled(t *testing.T) { + a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, + `[`+fact("h1", "role", "web", "")+`]`) + b := newFakeBackend(t, `[]`, `[]`) + cfg := testConfig(a.srv.URL, b.srv.URL, mergeStatic) + cfg.SourceFactEnabled = false + srv := newTestServer(cfg) + + facts := doGet(t, srv.Handler(), factsPath, "") + if _, n := sourceValues(t, facts.Body.Bytes(), defaultSourceFact); n != 0 { + t.Errorf("disabled injection still produced %d records: %s", n, facts.Body.String()) + } + nodes := doGet(t, srv.Handler(), nodesPath, "") + if got := nodeSources(t, nodes.Body.Bytes(), defaultSourceFact); len(got) != 0 { + t.Errorf("disabled injection still stamped nodes: %v", got) + } +} + +func TestHandler_SourceFactNameOverride(t *testing.T) { + a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, + `[`+fact("h1", "role", "web", "")+`]`) + b := newFakeBackend(t, `[]`, `[]`) + cfg := testConfig(a.srv.URL, b.srv.URL, mergeStatic) + cfg.SourceFact = "origin_pdb" + srv := newTestServer(cfg) + + facts := doGet(t, srv.Handler(), factsPath, "") + got, n := sourceValues(t, facts.Body.Bytes(), "origin_pdb") + if n != 1 || got["h1"] != "a" { + t.Errorf("override name not honoured: %s", facts.Body.String()) + } + if _, n := sourceValues(t, facts.Body.Bytes(), defaultSourceFact); n != 0 { + t.Errorf("default name still emitted alongside the override: %s", facts.Body.String()) + } + + nodes := doGet(t, srv.Handler(), nodesPath, "") + if got := nodeSources(t, nodes.Body.Bytes(), "origin_pdb"); got["h1"] != "a" { + t.Errorf("override name not honoured on /nodes: %v", got) + } +} + +// An upstream fact of the configured name is replaced, not duplicated: pdbmux's +// own attribution is authoritative. +func TestHandler_UpstreamSourceFactOverridden(t *testing.T) { + a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, + `[`+fact("h1", "role", "web", "")+`,`+fact("h1", defaultSourceFact, "stale-value", "")+`]`) + b := newFakeBackend(t, `[]`, `[]`) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), factsPath, "") + got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact) + if n != 1 { + t.Fatalf("expected exactly 1 %s record, got %d: %s", defaultSourceFact, n, rec.Body.String()) + } + if got["h1"] != "a" { + t.Errorf("upstream value survived: %v", got) + } +} + +// 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()) + } +} + +// Suppression matches a record's own name field, so rows from a projection that +// filters on name without returning it are not self-identifying and pass +// through. Pinned as a known limit of the guarantee, and documented as one. +func TestHandler_ProjectionWithoutNameColumnCarriesUpstreamValue(t *testing.T) { + a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`) + b := newFakeBackend(t, `[]`, `[]`) + a.bodies[factsPath] = `[{"certname":"h1","value":"upstream-value"}]` + var buf bytes.Buffer + srv := NewServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic), log.New(&buf, "", 0)) + + rec := doGet(t, srv.Handler(), factsPath, + `["extract",["certname","value"],["=","name","`+defaultSourceFact+`"]]`) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + if got := rec.Body.String(); !strings.Contains(got, "upstream-value") { + t.Errorf("unidentifiable row was dropped: %s", got) + } + if strings.Contains(buf.String(), "dropped") { + t.Errorf("a row with no name field was counted as suppressed: %s", buf.String()) + } + if _, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact); n != 0 { + t.Errorf("gated projection gained %d synthetic records: %s", n, 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}]` + b.bodies[factsPath] = `[{"count":2}]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), factsPath, `["extract",[["function","count"]]]`) + if _, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact); n != 0 { + t.Errorf("aggregate response gained %d synthetic records: %s", n, rec.Body.String()) + } + // Every row must be one an upstream actually returned: an aggregate row pdbmux + // invented or rewrote would change the count the client sees. + var raws []json.RawMessage + if err := json.Unmarshal(rec.Body.Bytes(), &raws); err != nil { + t.Fatal(err) + } + upstream := []string{`{"count":3}`, `{"count":2}`} + for _, raw := range raws { + if !slices.Contains(upstream, string(raw)) { + t.Errorf("aggregate row %s is not an upstream row", raw) + } + } +} + +// /nodes aggregates are summed rather than merged, so nothing may stamp them. +func TestHandler_NodesAggregateNotStamped(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + b := newFakeBackend(t, `[]`, `[]`) + a.bodies[nodesPath] = `[{"count":3}]` + b.bodies[nodesPath] = `[{"count":2}]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), nodesPath, `["extract",[["function","count"]]]`) + if strings.Contains(rec.Body.String(), defaultSourceFact) { + t.Errorf("aggregate rows were stamped: %s", rec.Body.String()) + } + if got := rec.Body.String(); !strings.Contains(got, `"count":5`) { + t.Errorf("count = %s, want the summed 5", got) + } +} + +// A plain extract projects columns and skips the aggregate path, so the stamp +// must not add a key the client did not ask for. +func TestHandler_NodesProjectionNotStamped(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + b := newFakeBackend(t, `[]`, `[]`) + a.bodies[nodesPath] = `[{"certname":"h1"}]` + b.bodies[nodesPath] = `[]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), nodesPath, `["extract",["certname"]]`) + if strings.Contains(rec.Body.String(), defaultSourceFact) { + t.Errorf("projection gained a stamp: %s", rec.Body.String()) + } +} + +// A query naming a specific fact asked for that fact only. +func TestHandler_SourceNotInjectedWhenNameFiltered(t *testing.T) { + a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, + `[`+fact("h1", "role", "web", "")+`]`) + b := newFakeBackend(t, `[]`, `[]`) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + for _, q := range []string{ + `["=","name","role"]`, + `["and",["=","certname","h1"],["=","name","role"]]`, + `["=","name","` + defaultSourceFact + `"]`, + } { + rec := doGet(t, srv.Handler(), factsPath, q) + if _, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact); n != 0 { + t.Errorf("query %s gained %d synthetic records: %s", q, n, rec.Body.String()) + } + } +} + +// A certname filter selects nodes, not facts, so the full fact set — synthetic +// record included — is still the right answer. +func TestHandler_SourceInjectedWhenOnlyCertnameFiltered(t *testing.T) { + a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, + `[`+fact("h1", "role", "web", "")+`]`) + b := newFakeBackend(t, `[]`, `[]`) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), factsPath, `["=","certname","h1"]`) + if _, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact); n != 1 { + t.Errorf("expected 1 synthetic record, got %d: %s", n, rec.Body.String()) + } +} + +func TestInjectable(t *testing.T) { + tests := []struct { + name string + query string + factEntity bool + want bool + }{ + {"empty query", "", true, true}, + {"certname filter", `["=","certname","h1"]`, true, true}, + {"regex certname filter", `["~","certname","^web"]`, true, true}, + {"name filter", `["=","name","os"]`, true, false}, + {"name regex filter", `["~","name","^net"]`, true, false}, + {"name under and", `["and",["=","certname","h1"],["=","name","os"]]`, true, false}, + {"name under or", `["or",["=","name","os"],["=","name","kernel"]]`, true, false}, + {"name under not", `["not",["=","name","os"]]`, true, false}, + {"name in list", `["in","name",["array",["os"]]]`, true, false}, + // A select_facts subquery narrows which nodes match; the outer response is + // still whole fact sets, so the synthetic record belongs in it. + {"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}, + // `subquery` is rewritten to ["in" ... ["extract" ... ["select_x" ...]]] + // before any plan node is built, so its operand is subquery-scoped too. + {"subquery operand stays injectable", `["and",["=","certname","h1"],["subquery","facts",["extract",["certname"],["=","name","os"]]]]`, true, true}, + {"bare subquery stays injectable", `["subquery","facts",["extract",["certname"]]]`, true, true}, + {"extract under select_facts stays injectable", `["and",["select_facts",["extract",["certname"]]]]`, 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}, + {"non-array query", `{"a":1}`, true, false}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := injectable(tc.query, tc.factEntity); got != tc.want { + t.Errorf("injectable(%s, %v) = %v, want %v", tc.query, tc.factEntity, got, tc.want) + } + }) + } +} + +// 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, 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 { + t.Fatal(err) + } + for _, k := range []string{"certname", "name", "value", "environment"} { + if _, ok := obj[k]; !ok { + t.Errorf("environment=%q: key %q missing from synthetic fact", env, k) + } + } + if len(obj) != 4 { + t.Errorf("synthetic fact has %d keys, want the 4 of a real fact record: %v", len(obj), obj) + } + } +} + +// A nil injector is the disabled path and must leave every input untouched. +func TestSourceInjector_NilIsInert(t *testing.T) { + var si *sourceInjector + 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") + } + raw := json.RawMessage(`{"certname":"h1"}`) + if got := si.stamp(raw, "a"); string(got) != string(raw) { + t.Errorf("nil injector rewrote %s to %s", raw, got) + } +} + +// 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, 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) + } + } +} + +// 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, inject: true}) + + got := factValues(t, merged) + want := []string{"h1:role=web-a", "h1:kernel=Linux", "h1:" + defaultSourceFact + "=a"} + if !slices.Equal(got, want) { + t.Errorf("merged = %v, want %v", got, want) + } +}