diff --git a/README.md b/README.md index 510ba51..d9f05f5 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`. | -| `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). | +| `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/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. | | `GET /pdb/query/v4/event-counts` | Fan out to all and **sum** each subject's counts into one row per subject. | @@ -77,6 +77,62 @@ 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. + +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. + +**Injection is skipped**, and the response passes through untouched, 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**; +- (`/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. + ### Paging and ordering on the merged endpoints Each backend applies `order_by`/`limit`/`offset` to its own slice only, so @@ -117,6 +173,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 @@ -130,6 +188,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..a5b61de 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 { +// A non-nil inject appends the synthetic source fact after each certname's block, naming the backend that won. +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,30 @@ 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 same name is dropped: pdbmux's own value is authoritative. + if inject.claims(rec.Name) { + continue + } + out = append(out, rec.Raw) + } + 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 1428478..e3ab0b2 100644 --- a/server.go +++ b/server.go @@ -67,9 +67,9 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) { } switch r.URL.Path { case nodesPath: - s.serveMerged(w, r, nodesPath, s.mergeNodesResponse) + s.serveMerged(w, r, nodesPath, s.mergeNodesResponse(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: @@ -224,16 +224,22 @@ 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 { + if s.cfg.Merge == mergeStatic { + return mergeFacts(results, nil, inject) + } + fresh := s.freshnessMap(context.Background(), results) + return 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] }) } // 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 73f9b07..253ef0b 100644 --- a/server_test.go +++ b/server_test.go @@ -121,11 +121,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..a04e737 --- /dev/null +++ b/source.go @@ -0,0 +1,131 @@ +package main + +import "encoding/json" + +// 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. +type sourceInjector struct { + name string +} + +// newSourceInjector returns nil when injection is off for this request. +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} +} + +// claims reports whether an upstream record is the one the injector replaces. +func (si *sourceInjector) claims(factName string) bool { + return si != nil && factName != "" && factName == 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 { + 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 == nil { + 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. Two shapes are excluded, both 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(); +// - 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 op == "extract" { + return false + } + if !factEntity { + return true + } + return !constrainsField(ast, "name") +} + +// 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..6eb7e7b --- /dev/null +++ b/source_test.go @@ -0,0 +1,373 @@ +package main + +import ( + "encoding/json" + "net/http" + "slices" + "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) + } +} + +// A count() must report the backends' real fact count, not one inflated by a +// record pdbmux invented. +func TestHandler_SourceNotInjectedOnAggregate(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) + } + } +} + +// 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}, + {"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} + 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.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} + 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) + } + } +} + +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}) + + 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) + } +}