From 31ad4ae457cdab551a3a8539dc0837e227f0d22c Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sat, 5 Sep 2026 11:38:24 +1000 Subject: [PATCH 1/2] docs: strip over-commenting from README and source --- .woodpecker/docker.yaml | 5 +-- Dockerfile | 3 -- Makefile | 3 -- README.md | 72 ++++++++--------------------------------- config.go | 61 +++++++--------------------------- main.go | 22 +------------ merge.go | 38 ++++------------------ 7 files changed, 35 insertions(+), 169 deletions(-) diff --git a/.woodpecker/docker.yaml b/.woodpecker/docker.yaml index 1ebc697..6c74c13 100644 --- a/.woodpecker/docker.yaml +++ b/.woodpecker/docker.yaml @@ -1,7 +1,4 @@ -# Build and push the pdbmux container image on a v* tag. pdbmux is a k8s-only -# daemon (deployed via argocd-apps), so it ships as an image. Mirrors the estate -# convention: the CA-baked plugin-docker-buildx image pushes to the -# artifactapi local docker registry (unauthenticated in-cluster push). +# plugin-docker-buildx is the CA-baked variant; artifactapi's cert is not in the default trust store. when: - event: tag ref: refs/tags/v* diff --git a/Dockerfile b/Dockerfile index 404c768..e600e52 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,6 +1,3 @@ -# Container image for pdbmux, the merging PuppetDB proxy daemon. pdbmux is a -# k8s-only service (deployed via argocd-apps), so it ships as a distroless -# static image rather than an RPM. FROM golang:1.25-alpine AS builder RUN apk add --no-cache git diff --git a/Makefile b/Makefile index 380edf7..bb8a1e5 100644 --- a/Makefile +++ b/Makefile @@ -9,7 +9,6 @@ ARCH ?= $(shell go env GOARCH) all: build -# Build the single static binary into dist/. build: CGO_ENABLED=0 GOOS=$(OS) GOARCH=$(ARCH) go build $(GOFLAGS) -o $(DIST)/$(BINARY) . @@ -28,8 +27,6 @@ clean: install: go install $(GOFLAGS) . -# Bump helpers — read the latest semver tag and create the next one. -# If no tag exists yet, start from v0.0.0. _LATEST := $(shell git tag --sort=-v:refname | grep -E '^v[0-9]+\.[0-9]+\.[0-9]+$$' | head -1) _BASE := $(if $(_LATEST),$(_LATEST),v0.0.0) _MAJ := $(shell echo $(_BASE) | sed 's/^v//' | cut -d. -f1) diff --git a/README.md b/README.md index 32b904d..333a07b 100644 --- a/README.md +++ b/README.md @@ -55,12 +55,10 @@ unknown fields survive untouched. Precedence (lowest → highest): **defaults < config file < env vars (`PDBMUX_*`) < flags**. -Config file: `$XDG_CONFIG_HOME/pdbmux/config.yaml`. In Kubernetes, configuration -is supplied entirely via `PDBMUX_*` env vars (no config file), which is the -supported deployment path — see [Deployment](#deployment). +Config file: `$XDG_CONFIG_HOME/pdbmux/config.yaml`. In Kubernetes there is no +config file — everything comes from `PDBMUX_*` env vars. ```yaml -# ~/.config/pdbmux/config.yaml (local dev; in k8s use PDBMUX_* env instead) listen: ":8080" backends: - name: old @@ -74,8 +72,8 @@ timeout: 10s # per-upstream request timeout freshness_ttl: 30s # freshness-map cache TTL (freshness merge only) ``` -`backends[*].url` is a **base** URL (`scheme://host[:port]`), without the -`/pdb/query/v4/...` path — `pdbmux` appends the path per request. +`backends[*].url` is a **base** URL (`scheme://host[:port]`); `pdbmux` appends +the `/pdb/query/v4/...` path per request. | Env var | Overrides | |---|---| @@ -91,66 +89,24 @@ Flags: `--listen`, `--primary`, `--merge`. ## Running -```bash -pdbmux # start the proxy (serve is the default action) -pdbmux serve # explicit -pdbmux config init # write a default config file -pdbmux config show # print active config after all overrides -pdbmux version -``` - -Point a consumer at it: +Subcommands: `serve` (default), `config init`, `config show`, `version`. Run +`pdbmux --help` for details. ```bash +PDBMUX_BACKENDS='old=http://puppetdbapi.service.consul:8080,new=https://puppetdb.k8s.syd1.au.unkin.net' pdbmux node-lookup --url http://localhost:8080/pdb/query/v4/facts -R -NODE_LOOKUP_URL=http://localhost:8080/pdb/query/v4/facts pblastreport somehost ``` ## Build -```bash -make build # -> dist/pdbmux (CGO disabled, static) -make test # go test -race ./... -make lint # golangci-lint -``` - -Requires Go 1.25+. Dependencies: `github.com/spf13/cobra` (CLI), -`gopkg.in/yaml.v3` (config file). +`make build` (static binary into `dist/`), `make test`, `make lint`. Requires Go 1.25+. ## Deployment -`pdbmux` runs **in Kubernetes** as a container, in line with the all-in-k8s -estate direction — it is not shipped as a per-VM RPM/systemd service. The image -is built and pushed on every `v*` tag (`.woodpecker/docker.yaml`) to: +Kubernetes only — no RPM. Every `v*` tag builds and pushes +`artifactapi.k8s.syd1.au.unkin.net/docker-internal/pdbmux:` +(`.woodpecker/docker.yaml`); tag with `make patch` / `minor` / `major`. -``` -artifactapi.k8s.syd1.au.unkin.net/docker-internal/pdbmux: -``` - -It is a minimal static (`CGO_ENABLED=0`) binary on a distroless base -(`Dockerfile`), configured entirely via `PDBMUX_*` env vars, with a single HTTP -listener and `/healthz` for liveness/readiness probes. - -The Deployment/Service/Gateway manifests live in the estate's `argocd-apps` repo -under `apps/base/pdbmux/` (namespace `pdbmux`, 2 replicas), and it is exposed to -VM/workstation `node-lookup` consumers over HTTPS at: - -``` -https://pdbmux.k8s.syd1.au.unkin.net -``` - -Locally you can still run the binary directly for development: - -```bash -PDBMUX_BACKENDS='old=http://puppetdbapi.service.consul:8080,new=http://puppetdb.puppet.svc.cluster.local:8080' \ - pdbmux serve -curl -s localhost:8080/healthz -``` - -## Version bumps - -```bash -make patch # tag vX.Y.(Z+1) and push (triggers the docker release) -make minor # tag vX.(Y+1).0 -make major # tag v(X+1).0.0 -``` +Manifests live in `argocd-apps` under `apps/base/pdbmux/` (namespace `pdbmux`, +2 replicas). Use `/healthz` for liveness/readiness probes. Reachable from VMs and +workstations at `https://pdbmux.k8s.syd1.au.unkin.net`. diff --git a/config.go b/config.go index 3e8714b..8265cca 100644 --- a/config.go +++ b/config.go @@ -16,49 +16,27 @@ const ( configFileName = "config.yaml" envPrefix = "PDBMUX_" - // defaultListen is the default HTTP listen address. - defaultListen = ":8080" - // defaultOldURL / defaultNewURL are the two PuppetDBs merged during the - // VM -> k8s migration. old = legacy Consul-registered puppetdbapi; new = - // the k8s PuppetDB behind the gateway (TLS terminated there). - defaultOldURL = "http://puppetdbapi.service.consul:8080" - defaultNewURL = "https://puppetdb.k8s.syd1.au.unkin.net" - // defaultPrimary is the backend name used for pass-through (non-merged) - // /pdb/query/v4/* paths and as static precedence for merge fallback. + defaultListen = ":8080" + defaultOldURL = "http://puppetdbapi.service.consul:8080" + defaultNewURL = "https://puppetdb.k8s.syd1.au.unkin.net" defaultPrimary = "new" defaultTimeout = 10 * time.Second defaultFreshnessTTL = 30 * time.Second ) -// Backend is one upstream PuppetDB. URL is the base URL (scheme://host[:port]), -// without the /pdb/query/v4/... path — that is appended per request. type Backend struct { Name string `yaml:"name"` - URL string `yaml:"url"` + URL string `yaml:"url"` // base URL only; the query path is appended per request } -// Config holds every configurable value. Fields map 1:1 to config-file keys and -// env vars (PDBMUX_*). See Load for precedence. type Config struct { - // Listen is the HTTP listen address (host:port). - Listen string `yaml:"listen"` - // Backends is the ordered list of upstream PuppetDBs to fan out to. - Backends []Backend `yaml:"backends"` - // Primary is the backend Name used for transparent pass-through of - // non-merged /pdb/query/v4/* paths. - Primary string `yaml:"primary"` - // Merge selects how /facts records are attributed to a backend when a - // certname appears in both: "freshness" (query /nodes report_timestamp, - // newer wins) or "static" (always prefer the Prefer backend). - Merge string `yaml:"merge"` - // Prefer names the backend that wins under static merge and as the - // tie-breaker/fallback under freshness merge. - Prefer string `yaml:"prefer"` - // Timeout bounds each upstream request. - Timeout time.Duration `yaml:"timeout"` - // FreshnessTTL is how long a per-certname freshness map (from /nodes) is - // cached under the "freshness" merge strategy. + Listen string `yaml:"listen"` + Backends []Backend `yaml:"backends"` + Primary string `yaml:"primary"` + Merge string `yaml:"merge"` + Prefer string `yaml:"prefer"` // wins under static merge, and breaks ties under freshness merge + Timeout time.Duration `yaml:"timeout"` FreshnessTTL time.Duration `yaml:"freshness_ttl"` } @@ -67,8 +45,6 @@ const ( mergeStatic = "static" ) -// DefaultConfig returns the built-in defaults: both migration PuppetDBs, -// freshness merge, "new" primary/preferred. func DefaultConfig() Config { return Config{ Listen: defaultListen, @@ -84,7 +60,6 @@ func DefaultConfig() Config { } } -// ConfigDir returns the XDG_CONFIG_HOME/pdbmux directory. func ConfigDir() string { base := os.Getenv("XDG_CONFIG_HOME") if base == "" { @@ -94,15 +69,11 @@ func ConfigDir() string { return filepath.Join(base, appName) } -// ConfigPath returns the full path to the config file. func ConfigPath() string { return filepath.Join(ConfigDir(), configFileName) } -// Load reads the config file (if present), then applies env var overrides. -// Precedence (lowest -> highest): defaults < config file < env vars < flags -// (flags are applied by the caller). Backends can be overridden wholesale via -// PDBMUX_BACKENDS ("name=url,name=url"). +// Precedence: defaults < config file < env vars < flags, and flags are applied by the caller. func Load() (Config, error) { cfg := DefaultConfig() @@ -125,7 +96,6 @@ func Load() (Config, error) { return cfg, nil } -// applyEnv overlays PDBMUX_* env vars onto cfg. getenv is injected for testing. func applyEnv(cfg *Config, getenv func(string) string) { if v := getenv(envPrefix + "LISTEN"); v != "" { cfg.Listen = v @@ -156,8 +126,7 @@ func applyEnv(cfg *Config, getenv func(string) string) { } } -// parseBackends parses "name=url,name=url" into Backends. Entries without an -// "=" are skipped. Used for the PDBMUX_BACKENDS env override. +// Parses the PDBMUX_BACKENDS form "name=url,name=url"; entries without an "=" are skipped. func parseBackends(s string) []Backend { var out []Backend for _, part := range strings.Split(s, ",") { @@ -175,7 +144,6 @@ func parseBackends(s string) []Backend { return out } -// Validate checks the config is internally consistent and usable. func (c Config) Validate() error { if len(c.Backends) == 0 { return fmt.Errorf("no backends configured") @@ -207,8 +175,6 @@ func (c Config) Validate() error { return nil } -// PrimaryBackend returns the backend named by Primary (guaranteed present after -// Validate). func (c Config) PrimaryBackend() Backend { for _, b := range c.Backends { if b.Name == c.Primary { @@ -218,7 +184,6 @@ func (c Config) PrimaryBackend() Backend { return c.Backends[0] } -// writeDefaultConfig creates the config dir and writes a default config file. func writeDefaultConfig() error { dir := ConfigDir() if err := os.MkdirAll(dir, 0o755); err != nil { @@ -240,8 +205,6 @@ func writeDefaultConfig() error { return nil } -// durationString renders a duration for `config show` (falls back to a plain -// seconds count for zero to avoid "0s" ambiguity in logs). func durationString(d time.Duration) string { if d == 0 { return "0" diff --git a/main.go b/main.go index 52beabc..b5c101d 100644 --- a/main.go +++ b/main.go @@ -1,21 +1,4 @@ -// Command pdbmux is a small merging HTTP proxy over two PuppetDB backends. -// -// During the VM -> k8s Puppet migration there are two PuppetDBs — the legacy -// Consul-registered one and the new k8s one — and nodes move between them as -// they migrate. pdbmux presents a single merged PuppetDB v4 query surface so -// node-lookup and pblastreport (and anything else) see one consistent view: -// -// - GET /pdb/query/v4/nodes — fan out to both backends, dedupe by certname, -// keep the record with the newer report_timestamp. -// - GET /pdb/query/v4/facts — fan out to both, and for a certname present in -// both keep ALL facts from the backend holding that node's newer report -// (freshness merge) or a static preferred backend (static merge). -// - any other GET /pdb/query/v4/* — transparently proxied to the primary. -// - GET /healthz — per-backend reachability. -// -// The query param is forwarded verbatim (PuppetDB AST JSON). If one backend -// errors/times out, the other's results are served; only if both fail does a -// merged endpoint return 502. +// Command pdbmux serves one merged PuppetDB v4 query surface over two PuppetDBs. package main import ( @@ -119,8 +102,6 @@ func main() { } } -// runServer starts the HTTP server and blocks until SIGINT/SIGTERM, then -// gracefully shuts down. func runServer(cfg Config) error { logger := log.New(os.Stderr, "pdbmux: ", log.LstdFlags) srv := NewServer(cfg, logger) @@ -155,7 +136,6 @@ func runServer(cfg Config) error { } } -// printConfig renders the active config for `config show`. func printConfig(cfg Config) { fmt.Printf("config file : %s\n", ConfigPath()) fmt.Printf("listen : %s\n", cfg.Listen) diff --git a/merge.go b/merge.go index 9240733..c98287f 100644 --- a/merge.go +++ b/merge.go @@ -5,24 +5,18 @@ import ( "time" ) -// record is a single PuppetDB result element kept as raw JSON so unknown fields -// survive the merge untouched. certname/report_timestamp are decoded only for -// merge decisions. +// Raw is kept verbatim so unknown PuppetDB fields survive the merge. type record struct { Raw json.RawMessage Certname string ReportTimestamp string // only populated for /nodes records } -// recordMeta is the subset we decode from any /nodes or /facts element to drive -// merge decisions. type recordMeta struct { Certname string `json:"certname"` ReportTimestamp string `json:"report_timestamp"` } -// decodeRecords turns a raw PuppetDB JSON array into records, preserving each -// element verbatim in Raw. A body that is not a JSON array yields (nil, err). func decodeRecords(body []byte) ([]record, error) { var raws []json.RawMessage if err := json.Unmarshal(body, &raws); err != nil { @@ -41,8 +35,7 @@ func decodeRecords(body []byte) ([]record, error) { return out, nil } -// parseTimestamp parses a PuppetDB RFC3339(nano) timestamp. Zero time on -// failure sorts oldest, so a backend with a well-formed newer timestamp wins. +// An unparseable timestamp yields the zero time, which sorts oldest. func parseTimestamp(s string) time.Time { if s == "" { return time.Time{} @@ -53,10 +46,7 @@ func parseTimestamp(s string) time.Time { return time.Time{} } -// mergeNodes dedupes /nodes records by certname, keeping the one with the newer -// report_timestamp. backends is the ordered list of (name, records) results; -// when timestamps tie (or both are zero), the earlier backend in the slice -// wins, so callers should order by precedence. +// results must be ordered by precedence: ties keep the earlier backend's record. func mergeNodes(results []backendResult) []json.RawMessage { type pick struct { raw json.RawMessage @@ -73,7 +63,6 @@ func mergeNodes(results []backendResult) []json.RawMessage { order = append(order, rec.Certname) continue } - // Strictly-newer wins; ties keep the existing (earlier-backend) pick. if ts.After(cur.ts) { best[rec.Certname] = pick{raw: rec.Raw, ts: ts} } @@ -86,12 +75,10 @@ func mergeNodes(results []backendResult) []json.RawMessage { return out } -// freshness maps certname -> backend name that holds that node's newest report. +// certname -> name of the backend holding that node's newest report. type freshness map[string]string -// buildFreshness computes, per certname, which backend has the newer -// report_timestamp. results must be ordered by precedence; on a tie the -// earlier backend wins. +// results must be ordered by precedence: ties keep the earlier backend. func buildFreshness(results []backendResult) freshness { type pick struct { backend string @@ -114,19 +101,9 @@ func buildFreshness(results []backendResult) freshness { return f } -// mergeFacts merges /facts records at node granularity: for each certname, all -// facts from the winning backend are kept and the other backend's facts for -// that certname are dropped. -// -// The winner is chosen per certname by `owner(certname)`. Callers supply owner -// from either a freshness map (freshness merge) or a constant preferred backend -// (static merge). When owner returns a backend that has no facts for a certname -// (or a name not in results), records fall back to precedence order so a node -// present in only one backend still appears. +// owner names the winning backend per certname; when it holds no facts for that certname, precedence order wins. func mergeFacts(results []backendResult, owner func(certname string) string) []json.RawMessage { - // Which backends actually returned facts for each certname, in precedence - // order, so we can fall back if the chosen owner has none. - present := map[string][]string{} // certname -> ordered backend names + present := map[string][]string{} // certname -> backend names, in precedence order byKey := map[string][]json.RawMessage{} for _, res := range results { for _, rec := range res.records { @@ -154,7 +131,6 @@ func mergeFacts(results []backendResult, owner func(certname string) string) []j for _, cn := range order { backends := present[cn] chosen := owner(cn) - // Fall back to precedence order if the chosen backend has no facts here. if !contains(backends, chosen) { chosen = backends[0] } -- 2.47.3 From 7b9082de08a54acbdad7dbf1dcf898578f886a32 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sat, 5 Sep 2026 11:42:43 +1000 Subject: [PATCH 2/2] docs: strip over-commenting from server.go and reports.go --- reports.go | 41 +++++++------------------------- server.go | 69 +++++++----------------------------------------------- 2 files changed, 17 insertions(+), 93 deletions(-) diff --git a/reports.go b/reports.go index 24c8d87..843f2c9 100644 --- a/reports.go +++ b/reports.go @@ -9,12 +9,7 @@ import ( "strings" ) -// mergeUnion concatenates every backend's records and drops duplicates by key. -// Reports and events are immutable history, so a certname that migrated between -// PuppetDBs legitimately has records in both and the union — not a per-node -// winner — is the correct merged view. results must be ordered by precedence; -// the first backend holding a key supplies the record. A key func returning -// ok=false means the record has no dedupe identity and is always kept. +// results must be ordered by precedence; a key func returning ok=false means the record has no identity and is always kept. func mergeUnion(results []backendResult, key func(record) (string, bool)) []json.RawMessage { seen := make(map[string]bool) out := []json.RawMessage{} @@ -32,10 +27,7 @@ func mergeUnion(results []backendResult, key func(record) (string, bool)) []json return out } -// reportKey identifies a report by its content hash, which PuppetDB guarantees -// is unique per report. An `extract`/`group_by` query returns synthetic rows -// with no hash and no identity — two backends can emit byte-identical aggregate -// rows that both count — so those are never deduped. +// extract/group_by rows are synthetic and carry no hash, so two backends can legitimately emit identical ones. func reportKey(rec record) (string, bool) { if rec.Hash == "" { return "", false @@ -43,19 +35,15 @@ func reportKey(rec record) (string, bool) { return "hash\x00" + rec.Hash, true } -// rawKey identifies a record by its verbatim JSON. Events carry no unique id, -// but two byte-identical events from the same PuppetDB serialiser describe the -// same resource change, so raw equality is a safe dedupe key. +// Events carry no id, but byte-identical events from the same PuppetDB serialiser are the same change. func rawKey(rec record) (string, bool) { return "raw\x00" + string(rec.Raw), true } -// orderField is one entry of PuppetDB's order_by param. type orderField struct { Field string Desc bool } -// parseOrderBy decodes PuppetDB's order_by param, a JSON array of -// {"field":..., "order":"asc"|"desc"} objects. An empty param yields no fields. +// order_by is a JSON array of {"field": ..., "order": "asc"|"desc"} objects. func parseOrderBy(s string) ([]orderField, error) { if strings.TrimSpace(s) == "" { return nil, nil @@ -77,9 +65,7 @@ func parseOrderBy(s string) ([]orderField, error) { return out, nil } -// sortRecords re-sorts a merged record set by order. Each backend only ordered -// its own slice, so the union has to be ordered again here. The sort is stable, -// so ties keep backend precedence order. +// Each backend ordered only its own slice, so the union is re-sorted here; stable, so ties keep backend precedence. func sortRecords(recs []json.RawMessage, order []orderField) { if len(order) == 0 || len(recs) < 2 { return @@ -113,8 +99,7 @@ func sortRecords(recs []json.RawMessage, order []orderField) { copy(recs, sorted) } -// compareValues orders two decoded JSON values. Unlike types are ordered by -// kind (null < bool < number < string) so a missing field always sorts first. +// Unlike types order by kind (null < bool < number < string), so a missing field sorts first. func compareValues(a, b any) int { ra, rb := valueRank(a), valueRank(b) if ra != rb { @@ -165,9 +150,6 @@ func valueRank(v any) int { } } -// paging holds the PuppetDB paging params a merged endpoint has to re-apply -// itself: each backend applies limit/offset to its own result set only, so the -// proxy must page the union instead. type paging struct { limit int // -1 when unset offset int @@ -175,8 +157,6 @@ type paging struct { wantTotal bool } -// parsePaging reads limit, offset, order_by and include_total from a request's -// query params. func parsePaging(v url.Values) (paging, error) { p := paging{limit: -1} if s := v.Get("limit"); s != "" { @@ -202,9 +182,7 @@ func parsePaging(v url.Values) (paging, error) { return p, nil } -// upstreamParams rewrites the client's params for the fan-out. A backend must -// return everything that could land in the merged page, so it is asked for the -// first offset+limit records and the offset is applied locally instead. +// Backends are asked for the first offset+limit records with no offset; the offset is applied to the union instead. func (p paging) upstreamParams(in url.Values) url.Values { out := url.Values{} for k, vs := range in { @@ -217,7 +195,6 @@ func (p paging) upstreamParams(in url.Values) url.Values { return out } -// apply slices the merged, ordered record set down to the requested page. func (p paging) apply(recs []json.RawMessage) []json.RawMessage { if p.offset >= len(recs) { return []json.RawMessage{} @@ -229,9 +206,7 @@ func (p paging) apply(recs []json.RawMessage) []json.RawMessage { return recs } -// sumTotals adds up the X-Records counts the backends reported, ignoring any -// backend that did not send one. It returns -1 when no backend reported a count. -// Deduped records are counted once per backend, so the total is an upper bound. +// Returns -1 when no backend reported a count; duplicates count once per backend, so the sum is an upper bound. func sumTotals(results []backendResult) int { total := -1 for _, res := range results { diff --git a/server.go b/server.go index 9efeb66..d0c61e6 100644 --- a/server.go +++ b/server.go @@ -22,14 +22,10 @@ const ( eventsPath = "/pdb/query/v4/events" queryV4 = "/pdb/query/v4/" - // recordsHeader is PuppetDB's total-result-count header, returned when a - // request carries include_total=true. + // PuppetDB only sends this when the request carries include_total=true. recordsHeader = "X-Records" ) -// backendResult is one backend's decoded response for a query. err is non-nil -// when the backend failed (network/timeout/non-2xx); such results carry no -// records and are excluded from the merge but logged. type backendResult struct { name string records []record @@ -37,7 +33,6 @@ type backendResult struct { err error } -// Server proxies and merges PuppetDB queries across the configured backends. type Server struct { cfg Config client *http.Client @@ -49,7 +44,6 @@ type Server struct { freshAt time.Time } -// NewServer builds a Server with an HTTP client bounded by cfg.Timeout. func NewServer(cfg Config, logger *log.Logger) *Server { return &Server{ cfg: cfg, @@ -58,7 +52,6 @@ func NewServer(cfg Config, logger *log.Logger) *Server { } } -// Handler returns the HTTP mux for the proxy. func (s *Server) Handler() http.Handler { mux := http.NewServeMux() mux.HandleFunc("/healthz", s.handleHealth) @@ -66,10 +59,6 @@ func (s *Server) Handler() http.Handler { return mux } -// handleQuery dispatches /pdb/query/v4/* requests: /facts and /nodes are merged -// per node, /reports and /events are unioned across backends, a report's -// sub-resources resolve to whichever backend stores that report, and every other -// v4 path is transparently proxied to the primary. func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "only GET is supported", http.StatusMethodNotAllowed) @@ -93,9 +82,7 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) { } } -// isReportSubResource reports whether path is a per-report child endpoint — -// /pdb/query/v4/reports//{events,logs,metrics} — whose data lives in -// exactly one backend. +// Matches /pdb/query/v4/reports//{events,logs,metrics}, whose data lives in exactly one backend. func isReportSubResource(path string) bool { rest, ok := strings.CutPrefix(path, reportsPath+"/") if !ok { @@ -112,9 +99,6 @@ func isReportSubResource(path string) bool { return false } -// serveMerged fans out the request to all backends, then hands the per-backend -// results to merge to produce the response body. If every backend fails it -// returns 502; if some fail it serves the survivors and logs a warning. func (s *Server) serveMerged(w http.ResponseWriter, r *http.Request, path string, merge func([]backendResult) []json.RawMessage) { alive, ok := s.aliveResults(w, r, path, queryParams(r.URL.Query().Get("query"))) if !ok { @@ -123,10 +107,7 @@ func (s *Server) serveMerged(w http.ResponseWriter, r *http.Request, path string writeJSON(w, merge(alive)) } -// serveUnion fans out a request whose records are immutable history — reports -// and events — and serves the deduped union of every backend. Because each -// backend ordered and paged only its own slice, the union is re-ordered and -// re-paged here from the client's order_by/limit/offset. +// Reports and events are immutable history, so both backends' records belong in the merged view. func (s *Server) serveUnion(w http.ResponseWriter, r *http.Request, path string, key func(record) (string, bool)) { in := r.URL.Query() page, err := parsePaging(in) @@ -150,11 +131,7 @@ func (s *Server) serveUnion(w http.ResponseWriter, r *http.Request, path string, writeJSON(w, page.apply(merged)) } -// serveFirstHolder answers a per-report sub-resource request. The report lives -// in exactly one backend, so all are asked concurrently and the first one (in -// precedence order) that actually holds it wins. Backends that do not have the -// report answer 404, which is indistinguishable here from any other failure, so -// an empty result is only served once every backend has been consulted. +// A backend without the report answers 404, indistinguishable from a failure, so every backend is consulted before serving empty. func (s *Server) serveFirstHolder(w http.ResponseWriter, r *http.Request) { results := s.fanOut(r.Context(), r.URL.Path, r.URL.Query()) @@ -179,8 +156,7 @@ func (s *Server) serveFirstHolder(w http.ResponseWriter, r *http.Request) { writeJSON(w, nil) } -// aliveResults fans out to every backend and returns the successful results. -// It writes a 502 and returns ok=false when every backend failed. +// Writes a 502 and returns ok=false only when every backend failed. func (s *Server) aliveResults(w http.ResponseWriter, r *http.Request, path string, params url.Values) ([]backendResult, bool) { results := s.fanOut(r.Context(), path, params) @@ -199,8 +175,6 @@ func (s *Server) aliveResults(w http.ResponseWriter, r *http.Request, path strin return alive, true } -// queryParams builds the upstream param set for a merged endpoint that only -// forwards the PuppetDB query. func queryParams(query string) url.Values { if query == "" { return nil @@ -208,7 +182,6 @@ func queryParams(query string) url.Values { return url.Values{"query": []string{query}} } -// rawRecords strips decoded metadata back down to the verbatim JSON elements. func rawRecords(recs []record) []json.RawMessage { out := make([]json.RawMessage, 0, len(recs)) for _, rec := range recs { @@ -217,21 +190,16 @@ func rawRecords(recs []record) []json.RawMessage { return out } -// mergeNodesResponse merges /nodes results (dedupe by certname, newer wins). func (s *Server) mergeNodesResponse(results []backendResult) []json.RawMessage { return mergeNodes(s.byPrecedence(results)) } -// mergeFactsResponse merges /facts results at node granularity, choosing each -// certname's owner via the configured merge strategy. func (s *Server) mergeFactsResponse(results []backendResult) []json.RawMessage { ordered := s.byPrecedence(results) if s.cfg.Merge == mergeStatic { prefer := s.cfg.Prefer return mergeFacts(ordered, func(string) string { return prefer }) } - // freshness merge: attribute each certname to the backend with the newer - // report_timestamp, taken from a short-TTL /nodes freshness map. fresh := s.freshnessMap(context.Background(), ordered) prefer := s.cfg.Prefer return mergeFacts(ordered, func(cn string) string { @@ -242,8 +210,7 @@ func (s *Server) mergeFactsResponse(results []backendResult) []json.RawMessage { }) } -// byPrecedence orders results so the Prefer backend comes first, giving it the -// tie-break on equal timestamps. Remaining backends keep config order. +// Puts Prefer first so it wins ties; the rest keep config order. func (s *Server) byPrecedence(results []backendResult) []backendResult { ordered := make([]backendResult, len(results)) copy(ordered, results) @@ -253,15 +220,7 @@ func (s *Server) byPrecedence(results []backendResult) []backendResult { return ordered } -// freshnessMap returns a per-certname owner map derived from each backend's -// /nodes report_timestamp, cached for cfg.FreshnessTTL. On cache miss it queries -// /nodes from all backends; a backend that fails is simply absent from the map, -// so its certnames fall back to precedence/Prefer. -// -// When the incoming request already carries /nodes data (results has records), -// we still query /nodes broadly here because a /facts query's certname set can -// differ from what the request's query filter returned. The cache keeps this -// cheap under load. +// Queries /nodes unfiltered rather than reusing the request's results, because a /facts query's certname set can differ. func (s *Server) freshnessMap(ctx context.Context, _ []backendResult) freshness { s.mu.Lock() if s.freshData != nil && time.Since(s.freshAt) < s.cfg.FreshnessTTL { @@ -271,7 +230,6 @@ func (s *Server) freshnessMap(ctx context.Context, _ []backendResult) freshness } s.mu.Unlock() - // Empty query = all nodes; cheap enough for a short-TTL cache. nodeResults := s.fanOut(ctx, nodesPath, nil) var alive []backendResult for _, res := range nodeResults { @@ -290,8 +248,7 @@ func (s *Server) freshnessMap(ctx context.Context, _ []backendResult) freshness return f } -// fanOut queries every backend concurrently for path with the given params and -// returns one backendResult per backend, in config order. +// Returns one result per backend, in config order. func (s *Server) fanOut(ctx context.Context, path string, params url.Values) []backendResult { results := make([]backendResult, len(s.cfg.Backends)) var wg sync.WaitGroup @@ -307,8 +264,7 @@ func (s *Server) fanOut(ctx context.Context, path string, params url.Values) []b return results } -// queryBackend performs one GET b.URL+path?params and decodes the JSON array. -// It also returns the upstream X-Records count, or -1 when the backend sent none. +// The returned count is the upstream X-Records value, or -1 when the backend sent none. func (s *Server) queryBackend(ctx context.Context, b Backend, path string, params url.Values) ([]record, int, error) { target := strings.TrimRight(b.URL, "/") + path req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil) @@ -338,8 +294,6 @@ func (s *Server) queryBackend(ctx context.Context, b Backend, path string, param return recs, total, err } -// proxyPrimary transparently forwards a non-merged /pdb/query/v4/* request to -// the primary backend and streams the response back verbatim. func (s *Server) proxyPrimary(w http.ResponseWriter, r *http.Request) { b := s.cfg.PrimaryBackend() target := strings.TrimRight(b.URL, "/") + r.URL.Path @@ -365,15 +319,11 @@ func (s *Server) proxyPrimary(w http.ResponseWriter, r *http.Request) { _, _ = io.Copy(w, resp.Body) } -// healthReport is the /healthz JSON body. type healthReport struct { Status string `json:"status"` Backends map[string]string `json:"backends"` // name -> "ok" | error text } -// handleHealth probes every backend's /nodes endpoint with a trivial query and -// reports per-backend reachability. Overall status is "ok" if any backend is -// reachable, "degraded" if some fail, "down" if all fail (503 in that case). func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { probe := `["=","certname","pdbmux-healthz-probe"]` results := s.fanOut(r.Context(), nodesPath, queryParams(probe)) @@ -406,7 +356,6 @@ func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { _ = enc.Encode(report) } -// writeJSON writes a JSON array of raw records as a PuppetDB-style response. func writeJSON(w http.ResponseWriter, recs []json.RawMessage) { w.Header().Set("Content-Type", "application/json") if recs == nil { -- 2.47.3