From dda6b8c8c8b984413b4b8dbb69f03fdcc9fc0011 Mon Sep 17 00:00:00 2001 From: benvin Date: Fri, 24 Jul 2026 23:24:35 +1000 Subject: [PATCH] Add pdbmux: merging PuppetDB proxy daemon Split out from node-lookup PR #17 into its own repo. pdbmux presents a single merged PuppetDB v4 query surface over the old (Consul) and new (k8s) PuppetDBs during the VM to k8s migration, and is deployed in-cluster via argocd-apps as a container image. --- .gitignore | 5 + .pre-commit-config.yaml | 17 +++ .woodpecker/build.yaml | 18 +++ .woodpecker/docker.yaml | 33 ++++ .woodpecker/pre-commit.yaml | 18 +++ .woodpecker/test.yaml | 33 ++++ Dockerfile | 24 +++ Makefile | 52 +++++++ README.md | 157 ++++++++++++++++++- config.go | 250 ++++++++++++++++++++++++++++++ config_test.go | 126 ++++++++++++++++ go.mod | 13 ++ go.sum | 13 ++ main.go | 171 +++++++++++++++++++++ merge.go | 173 +++++++++++++++++++++ merge_test.go | 228 ++++++++++++++++++++++++++++ server.go | 294 ++++++++++++++++++++++++++++++++++++ server_test.go | 286 +++++++++++++++++++++++++++++++++++ 18 files changed, 1909 insertions(+), 2 deletions(-) create mode 100644 .gitignore create mode 100644 .pre-commit-config.yaml create mode 100644 .woodpecker/build.yaml create mode 100644 .woodpecker/docker.yaml create mode 100644 .woodpecker/pre-commit.yaml create mode 100644 .woodpecker/test.yaml create mode 100644 Dockerfile create mode 100644 Makefile create mode 100644 config.go create mode 100644 config_test.go create mode 100644 go.mod create mode 100644 go.sum create mode 100644 main.go create mode 100644 merge.go create mode 100644 merge_test.go create mode 100644 server.go create mode 100644 server_test.go diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..0969cbb --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +# built binary (repo root) +/pdbmux +# cross-compiled artifacts (e.g. pdbmux-linux-amd64) +/pdbmux-* +dist/ diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml new file mode 100644 index 0000000..2e63b82 --- /dev/null +++ b/.pre-commit-config.yaml @@ -0,0 +1,17 @@ +repos: + - repo: https://github.com/pre-commit/pre-commit-hooks + rev: v4.6.0 + hooks: + - id: trailing-whitespace + - id: end-of-file-fixer + - id: check-yaml + - id: check-merge-conflict + - id: mixed-line-ending + args: [--fix=lf] + + - repo: https://github.com/dnephin/pre-commit-golang + rev: v0.5.1 + hooks: + - id: go-fmt + - id: go-vet + - id: go-unit-tests diff --git a/.woodpecker/build.yaml b/.woodpecker/build.yaml new file mode 100644 index 0000000..74123b5 --- /dev/null +++ b/.woodpecker/build.yaml @@ -0,0 +1,18 @@ +when: + - event: pull_request + +steps: + - name: build + image: golang:1.25 + commands: + - make build + backend_options: + kubernetes: + serviceAccountName: default + resources: + requests: + memory: 512Mi + cpu: 1 + limits: + memory: 2Gi + cpu: 2 diff --git a/.woodpecker/docker.yaml b/.woodpecker/docker.yaml new file mode 100644 index 0000000..9d938b1 --- /dev/null +++ b/.woodpecker/docker.yaml @@ -0,0 +1,33 @@ +# 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 woodpeckerci/plugin-docker-buildx plugin pushes to the Gitea +# registry using the droneci / DRONECI_PASSWORD credentials. +when: + - event: tag + ref: refs/tags/v* + +steps: + - name: docker + image: woodpeckerci/plugin-docker-buildx + settings: + registry: git.unkin.net + repo: git.unkin.net/unkin/pdbmux + dockerfile: Dockerfile + build_args: + VERSION: ${CI_COMMIT_TAG} + username: droneci + password: + from_secret: DRONECI_PASSWORD + tags: + - ${CI_COMMIT_TAG} + - latest + backend_options: + kubernetes: + serviceAccountName: default + resources: + requests: + memory: 1Gi + cpu: 1 + limits: + memory: 2Gi + cpu: 2 diff --git a/.woodpecker/pre-commit.yaml b/.woodpecker/pre-commit.yaml new file mode 100644 index 0000000..d57b508 --- /dev/null +++ b/.woodpecker/pre-commit.yaml @@ -0,0 +1,18 @@ +when: + - event: pull_request + +steps: + - name: pre-commit + image: git.unkin.net/unkin/almalinux9-gobuilder:20260606 + commands: + - uvx pre-commit run --all-files + backend_options: + kubernetes: + serviceAccountName: default + resources: + requests: + memory: 512Mi + cpu: 1 + limits: + memory: 2Gi + cpu: 2 diff --git a/.woodpecker/test.yaml b/.woodpecker/test.yaml new file mode 100644 index 0000000..5e179a7 --- /dev/null +++ b/.woodpecker/test.yaml @@ -0,0 +1,33 @@ +when: + - event: pull_request + +steps: + - name: lint + image: golangci/golangci-lint:latest + commands: + - golangci-lint run ./... + backend_options: + kubernetes: + serviceAccountName: default + resources: + requests: + memory: 512Mi + cpu: 1 + limits: + memory: 2Gi + cpu: 2 + + - name: test + image: golang:1.25 + commands: + - go test -v -race ./... + backend_options: + kubernetes: + serviceAccountName: default + resources: + requests: + memory: 512Mi + cpu: 1 + limits: + memory: 2Gi + cpu: 2 diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..404c768 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,24 @@ +# 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 + +WORKDIR /build + +COPY go.mod go.sum ./ +RUN go mod download + +COPY . . + +ARG VERSION=dev +RUN CGO_ENABLED=0 go build -ldflags="-s -w -X main.version=${VERSION}" -o pdbmux . + +FROM gcr.io/distroless/static-debian12:nonroot + +COPY --from=builder /build/pdbmux /usr/local/bin/pdbmux + +EXPOSE 8080 + +ENTRYPOINT ["pdbmux", "serve"] diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..380edf7 --- /dev/null +++ b/Makefile @@ -0,0 +1,52 @@ +BINARY := pdbmux +DIST := dist +VERSION := $(shell git describe --tags --always --dirty 2>/dev/null || echo dev) +GOFLAGS := -ldflags="-s -w -X main.version=$(VERSION)" +OS ?= $(shell go env GOOS) +ARCH ?= $(shell go env GOARCH) + +.PHONY: all build test lint fmt clean install patch minor major _tag + +all: build + +# Build the single static binary into dist/. +build: + CGO_ENABLED=0 GOOS=$(OS) GOARCH=$(ARCH) go build $(GOFLAGS) -o $(DIST)/$(BINARY) . + +test: + go test -v -race ./... + +lint: + golangci-lint run ./... + +fmt: + gofmt -w . + +clean: + rm -rf $(DIST) $(BINARY) + +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) +_MIN := $(shell echo $(_BASE) | sed 's/^v//' | cut -d. -f2) +_PAT := $(shell echo $(_BASE) | sed 's/^v//' | cut -d. -f3) + +patch: + @NEW=v$(_MAJ).$(_MIN).$(shell expr $(_PAT) + 1); \ + git tag $$NEW && echo "Tagged $$NEW" && $(MAKE) _tag TAG=$$NEW + +minor: + @NEW=v$(_MAJ).$(shell expr $(_MIN) + 1).0; \ + git tag $$NEW && echo "Tagged $$NEW" && $(MAKE) _tag TAG=$$NEW + +major: + @NEW=v$(shell expr $(_MAJ) + 1).0.0; \ + git tag $$NEW && echo "Tagged $$NEW" && $(MAKE) _tag TAG=$$NEW + +_tag: + git push origin $(TAG) diff --git a/README.md b/README.md index d36a4e5..6b4389d 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,156 @@ -# pdbmux +# pdbmux — merging PuppetDB proxy -Merging HTTP proxy over two PuppetDB backends, presenting a single merged PuppetDB v4 query surface during the VM to k8s Puppet migration. Deployed in-cluster via argocd-apps. \ No newline at end of file +`pdbmux` is a small HTTP daemon that fronts **two** PuppetDB backends and serves +a single, merged PuppetDB v4 query surface on one address. Point `node-lookup`, +`pblastreport`, or anything else at `pdbmux` instead of a raw PuppetDB and it +sees one consistent view spanning both. + +## Why + +During the VM→k8s Puppet migration there are two PuppetDBs: + +- **old** — the legacy Consul-registered `http://puppetdbapi.service.consul:8080` +- **new** — the k8s `https://puppetdb.k8s.syd1.au.unkin.net` (TLS terminated at + the gateway; backends are plain PuppetDB on 8080) + +Nodes move from old to new as they migrate, so at any moment a given node's +current data lives in exactly one of them. `pdbmux` merges both so consumers +don't have to know (or query twice) which PuppetDB a node currently lives in. + +## Endpoints + +`pdbmux` proxies **GET** requests only. The `query` param (PuppetDB AST JSON, +not PQL) is forwarded verbatim. + +| Path | Behaviour | +|---|---| +| `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 per `certname` keep **all** facts from the backend that owns that node (see merge semantics). | +| `GET /pdb/query/v4/*` (any other) | Transparently proxied to the **primary** backend, unmerged, streamed verbatim. | +| `GET /healthz` | Per-backend reachability. `200 {"status":"ok"}` if all reachable, `200 degraded` if some fail, `503 down` if all fail. | + +Fan-out is concurrent. If one backend errors or times out, `pdbmux` serves the +survivor's results and logs a warning; a merged endpoint only returns `502` when +**every** backend fails. Response records are passed through as raw JSON so +unknown fields survive untouched. + +## Merge semantics + +- **`/nodes`** — dedupe by `certname`; the record with the strictly-newer + `report_timestamp` wins. On a tie (or when a node exists in only one backend), + the **preferred** backend's record is kept. +- **`/facts`** — node-level granularity. For a `certname` present in both + backends, `pdbmux` keeps **all** of that node's facts from **one** backend and + drops the other's, chosen by the merge strategy: + - **`freshness`** (default) — attribute each `certname` to whichever backend + holds its newer `report_timestamp`. `pdbmux` derives this from a per-certname + freshness map built by querying `/nodes` from both backends, cached for + `freshness_ttl` (default 30s). Ties/fallbacks use `prefer`. + - **`static`** — always keep the `prefer` backend's facts for shared nodes. + No extra `/nodes` query. + - A node present in only one backend always appears (falls back to whichever + backend actually returned facts for it). + +## Config + +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). + +```yaml +# ~/.config/pdbmux/config.yaml (local dev; in k8s use PDBMUX_* env instead) +listen: ":8080" +backends: + - name: old + url: http://puppetdbapi.service.consul:8080 + - name: new + url: https://puppetdb.k8s.syd1.au.unkin.net +primary: new # backend used for non-merged /pdb/query/v4/* pass-through +merge: freshness # freshness | static +prefer: new # winner on ties / static merge / fallback +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. + +| Env var | Overrides | +|---|---| +| `PDBMUX_LISTEN` | `listen` | +| `PDBMUX_PRIMARY` | `primary` | +| `PDBMUX_MERGE` | `merge` | +| `PDBMUX_PREFER` | `prefer` | +| `PDBMUX_TIMEOUT` | `timeout` (Go duration, e.g. `10s`) | +| `PDBMUX_FRESHNESS_TTL` | `freshness_ttl` | +| `PDBMUX_BACKENDS` | whole backend list, as `name=url,name=url` | + +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: + +```bash +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). + +## 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: + +``` +git.unkin.net/unkin/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 +``` diff --git a/config.go b/config.go new file mode 100644 index 0000000..3e8714b --- /dev/null +++ b/config.go @@ -0,0 +1,250 @@ +package main + +import ( + "fmt" + "os" + "path/filepath" + "strconv" + "strings" + "time" + + "gopkg.in/yaml.v3" +) + +const ( + appName = "pdbmux" + 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. + 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"` +} + +// 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. + FreshnessTTL time.Duration `yaml:"freshness_ttl"` +} + +const ( + mergeFreshness = "freshness" + mergeStatic = "static" +) + +// DefaultConfig returns the built-in defaults: both migration PuppetDBs, +// freshness merge, "new" primary/preferred. +func DefaultConfig() Config { + return Config{ + Listen: defaultListen, + Backends: []Backend{ + {Name: "old", URL: defaultOldURL}, + {Name: "new", URL: defaultNewURL}, + }, + Primary: defaultPrimary, + Merge: mergeFreshness, + Prefer: defaultPrimary, + Timeout: defaultTimeout, + FreshnessTTL: defaultFreshnessTTL, + } +} + +// ConfigDir returns the XDG_CONFIG_HOME/pdbmux directory. +func ConfigDir() string { + base := os.Getenv("XDG_CONFIG_HOME") + if base == "" { + home, _ := os.UserHomeDir() + base = filepath.Join(home, ".config") + } + 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"). +func Load() (Config, error) { + cfg := DefaultConfig() + + path := ConfigPath() + data, err := os.ReadFile(path) + if err != nil && !os.IsNotExist(err) { + return cfg, fmt.Errorf("reading config %s: %w", path, err) + } + if err == nil { + if err := yaml.Unmarshal(data, &cfg); err != nil { + return cfg, fmt.Errorf("parsing config %s: %w", path, err) + } + } + + applyEnv(&cfg, os.Getenv) + + if err := cfg.Validate(); err != nil { + return cfg, err + } + 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 + } + if v := getenv(envPrefix + "PRIMARY"); v != "" { + cfg.Primary = v + } + if v := getenv(envPrefix + "MERGE"); v != "" { + cfg.Merge = v + } + if v := getenv(envPrefix + "PREFER"); v != "" { + cfg.Prefer = v + } + if v := getenv(envPrefix + "TIMEOUT"); v != "" { + if d, err := time.ParseDuration(v); err == nil { + cfg.Timeout = d + } + } + if v := getenv(envPrefix + "FRESHNESS_TTL"); v != "" { + if d, err := time.ParseDuration(v); err == nil { + cfg.FreshnessTTL = d + } + } + if v := getenv(envPrefix + "BACKENDS"); v != "" { + if bs := parseBackends(v); len(bs) > 0 { + cfg.Backends = bs + } + } +} + +// parseBackends parses "name=url,name=url" into Backends. Entries without an +// "=" are skipped. Used for the PDBMUX_BACKENDS env override. +func parseBackends(s string) []Backend { + var out []Backend + for _, part := range strings.Split(s, ",") { + part = strings.TrimSpace(part) + if part == "" { + continue + } + name, url, ok := strings.Cut(part, "=") + name, url = strings.TrimSpace(name), strings.TrimSpace(url) + if !ok || name == "" || url == "" { + continue + } + out = append(out, Backend{Name: name, URL: url}) + } + 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") + } + seen := map[string]bool{} + for _, b := range c.Backends { + if b.Name == "" || b.URL == "" { + return fmt.Errorf("backend requires both name and url: %+v", b) + } + if seen[b.Name] { + return fmt.Errorf("duplicate backend name %q", b.Name) + } + seen[b.Name] = true + } + if !seen[c.Primary] { + return fmt.Errorf("primary %q is not a configured backend", c.Primary) + } + switch c.Merge { + case mergeFreshness, mergeStatic: + default: + return fmt.Errorf("merge must be %q or %q, got %q", mergeFreshness, mergeStatic, c.Merge) + } + if !seen[c.Prefer] { + return fmt.Errorf("prefer %q is not a configured backend", c.Prefer) + } + if c.Timeout <= 0 { + return fmt.Errorf("timeout must be positive") + } + 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 { + return b + } + } + 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 { + return fmt.Errorf("creating config dir: %w", err) + } + path := ConfigPath() + if _, err := os.Stat(path); err == nil { + return fmt.Errorf("config already exists at %s", path) + } + data, _ := yaml.Marshal(DefaultConfig()) + header := []byte("# pdbmux configuration\n" + + "# A merging proxy over two PuppetDBs (old Consul + new k8s) during migration.\n" + + "# Env overrides: PDBMUX_LISTEN, PDBMUX_PRIMARY, PDBMUX_MERGE, PDBMUX_PREFER,\n" + + "# PDBMUX_TIMEOUT, PDBMUX_FRESHNESS_TTL, PDBMUX_BACKENDS (name=url,name=url).\n\n") + if err := os.WriteFile(path, append(header, data...), 0o644); err != nil { + return fmt.Errorf("writing config: %w", err) + } + fmt.Println("Config written to", path) + 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" + } + return strconv.FormatFloat(d.Seconds(), 'f', -1, 64) + "s" +} diff --git a/config_test.go b/config_test.go new file mode 100644 index 0000000..647e0c5 --- /dev/null +++ b/config_test.go @@ -0,0 +1,126 @@ +package main + +import ( + "os" + "path/filepath" + "testing" + "time" +) + +func TestLoad_Defaults(t *testing.T) { + t.Setenv("XDG_CONFIG_HOME", t.TempDir()) + clearEnv(t) + + cfg, err := Load() + if err != nil { + t.Fatal(err) + } + if cfg.Listen != defaultListen { + t.Errorf("listen = %q, want %q", cfg.Listen, defaultListen) + } + if len(cfg.Backends) != 2 || cfg.Backends[0].Name != "old" || cfg.Backends[1].Name != "new" { + t.Errorf("unexpected default backends: %+v", cfg.Backends) + } + if cfg.Merge != mergeFreshness || cfg.Primary != "new" { + t.Errorf("unexpected defaults merge=%s primary=%s", cfg.Merge, cfg.Primary) + } +} + +func TestLoad_FileAndEnvOverride(t *testing.T) { + dir := t.TempDir() + t.Setenv("XDG_CONFIG_HOME", dir) + clearEnv(t) + + cfgDir := filepath.Join(dir, appName) + if err := os.MkdirAll(cfgDir, 0o755); err != nil { + t.Fatal(err) + } + body := "listen: :9999\nmerge: static\nprimary: old\nprefer: old\n" + if err := os.WriteFile(filepath.Join(cfgDir, configFileName), []byte(body), 0o644); err != nil { + t.Fatal(err) + } + + // env beats file for listen. + t.Setenv(envPrefix+"LISTEN", "127.0.0.1:1234") + + cfg, err := Load() + if err != nil { + t.Fatalf("load: %v", err) + } + if cfg.Listen != "127.0.0.1:1234" { + t.Errorf("env should beat file for listen, got %q", cfg.Listen) + } + if cfg.Merge != mergeStatic || cfg.Primary != "old" { + t.Errorf("file override failed: merge=%s primary=%s", cfg.Merge, cfg.Primary) + } +} + +func TestApplyEnv_Backends(t *testing.T) { + cfg := DefaultConfig() + env := map[string]string{ + envPrefix + "BACKENDS": "a=http://a:8080,b=http://b:8080", + envPrefix + "PRIMARY": "a", + envPrefix + "PREFER": "a", + envPrefix + "TIMEOUT": "3s", + envPrefix + "FRESHNESS_TTL": "45s", + } + applyEnv(&cfg, func(k string) string { return env[k] }) + + if len(cfg.Backends) != 2 || cfg.Backends[0].Name != "a" || cfg.Backends[1].URL != "http://b:8080" { + t.Fatalf("PDBMUX_BACKENDS parse failed: %+v", cfg.Backends) + } + if cfg.Timeout != 3*time.Second || cfg.FreshnessTTL != 45*time.Second { + t.Errorf("duration envs failed: %v %v", cfg.Timeout, cfg.FreshnessTTL) + } +} + +func TestValidate(t *testing.T) { + cases := []struct { + name string + mutate func(*Config) + wantErr bool + }{ + {"ok", func(*Config) {}, false}, + {"no backends", func(c *Config) { c.Backends = nil }, true}, + {"dup name", func(c *Config) { c.Backends = append(c.Backends, Backend{Name: "old", URL: "x"}) }, true}, + {"missing url", func(c *Config) { c.Backends[0].URL = "" }, true}, + {"primary not a backend", func(c *Config) { c.Primary = "ghost" }, true}, + {"prefer not a backend", func(c *Config) { c.Prefer = "ghost" }, true}, + {"bad merge", func(c *Config) { c.Merge = "wrong" }, true}, + {"zero timeout", func(c *Config) { c.Timeout = 0 }, true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + cfg := DefaultConfig() + tc.mutate(&cfg) + err := cfg.Validate() + if (err != nil) != tc.wantErr { + t.Fatalf("Validate() err=%v, wantErr=%v", err, tc.wantErr) + } + }) + } +} + +func TestParseBackends(t *testing.T) { + got := parseBackends("a=http://a, b=http://b ,,broken,c=http://c") + if len(got) != 3 { + t.Fatalf("expected 3 valid backends, got %d: %+v", len(got), got) + } + if got[0].Name != "a" || got[2].URL != "http://c" { + t.Errorf("unexpected parse: %+v", got) + } +} + +func TestPrimaryBackend(t *testing.T) { + cfg := DefaultConfig() + if cfg.PrimaryBackend().URL != defaultNewURL { + t.Errorf("primary backend URL = %q, want %q", cfg.PrimaryBackend().URL, defaultNewURL) + } +} + +func clearEnv(t *testing.T) { + t.Helper() + for _, k := range []string{"LISTEN", "PRIMARY", "MERGE", "PREFER", "TIMEOUT", "FRESHNESS_TTL", "BACKENDS"} { + t.Setenv(envPrefix+k, "") + } +} diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..6fa53e0 --- /dev/null +++ b/go.mod @@ -0,0 +1,13 @@ +module pdbmux + +go 1.25.7 + +require ( + github.com/spf13/cobra v1.10.2 + gopkg.in/yaml.v3 v3.0.1 +) + +require ( + github.com/inconshreveable/mousetrap v1.1.0 // indirect + github.com/spf13/pflag v1.0.9 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..47edb24 --- /dev/null +++ b/go.sum @@ -0,0 +1,13 @@ +github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= +github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= +github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= +github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= +github.com/spf13/cobra v1.10.2 h1:DMTTonx5m65Ic0GOoRY2c16WCbHxOOw6xxezuLaBpcU= +github.com/spf13/cobra v1.10.2/go.mod h1:7C1pvHqHw5A4vrJfjNwvOdzYu0Gml16OCs2GRiTUUS4= +github.com/spf13/pflag v1.0.9 h1:9exaQaMOCwffKiiiYk6/BndUBv+iRViNW+4lEMi0PvY= +github.com/spf13/pflag v1.0.9/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= +go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/main.go b/main.go new file mode 100644 index 0000000..52beabc --- /dev/null +++ b/main.go @@ -0,0 +1,171 @@ +// 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. +package main + +import ( + "context" + "errors" + "fmt" + "log" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + "github.com/spf13/cobra" +) + +var version = "dev" + +func main() { + cfg, err := Load() + if err != nil { + fmt.Fprintln(os.Stderr, "config error:", err) + os.Exit(1) + } + + var ( + listen string + primary string + merge string + ) + + serve := func(cmd *cobra.Command) error { + if cmd.Flags().Changed("listen") { + cfg.Listen = listen + } + if cmd.Flags().Changed("primary") { + cfg.Primary = primary + } + if cmd.Flags().Changed("merge") { + cfg.Merge = merge + } + if err := cfg.Validate(); err != nil { + return err + } + return runServer(cfg) + } + + root := &cobra.Command{ + Use: appName, + Short: "Merging HTTP proxy over two PuppetDB backends.", + Long: "pdbmux presents a single merged PuppetDB v4 query surface over the old\n" + + "(Consul) and new (k8s) PuppetDBs during the migration, so node-lookup and\n" + + "pblastreport see one consistent view. Running pdbmux with no subcommand\n" + + "(or `pdbmux serve`) starts the proxy.", + SilenceUsage: true, + RunE: func(cmd *cobra.Command, args []string) error { return serve(cmd) }, + } + + pf := root.PersistentFlags() + pf.StringVar(&listen, "listen", cfg.Listen, "HTTP listen address (overrides config and PDBMUX_LISTEN)") + pf.StringVar(&primary, "primary", cfg.Primary, "Primary backend name for non-merged pass-through") + pf.StringVar(&merge, "merge", cfg.Merge, "Facts merge strategy: freshness or static") + + serveCmd := &cobra.Command{ + Use: "serve", + Short: "Start the proxy (default action)", + SilenceUsage: true, + RunE: func(cmd *cobra.Command, args []string) error { return serve(cmd) }, + } + + configCmd := &cobra.Command{Use: "config", Short: "Manage configuration"} + configCmd.AddCommand( + &cobra.Command{ + Use: "init", + Short: "Write a default config file to " + ConfigPath(), + SilenceUsage: true, + RunE: func(cmd *cobra.Command, args []string) error { return writeDefaultConfig() }, + }, + &cobra.Command{ + Use: "show", + Short: "Print the active configuration", + SilenceUsage: true, + RunE: func(cmd *cobra.Command, args []string) error { + printConfig(cfg) + return nil + }, + }, + ) + + versionCmd := &cobra.Command{ + Use: "version", + Short: "Print the version", + Run: func(cmd *cobra.Command, args []string) { fmt.Println(version) }, + SilenceUsage: true, + } + + root.AddCommand(serveCmd, configCmd, versionCmd) + + if err := root.Execute(); err != nil { + os.Exit(1) + } +} + +// 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) + + httpSrv := &http.Server{ + Addr: cfg.Listen, + Handler: srv.Handler(), + ReadHeaderTimeout: 10 * time.Second, + } + + logger.Printf("listening on %s (merge=%s primary=%s backends=%d)", + cfg.Listen, cfg.Merge, cfg.Primary, len(cfg.Backends)) + + errCh := make(chan error, 1) + go func() { + if err := httpSrv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + errCh <- err + } + }() + + stop := make(chan os.Signal, 1) + signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM) + + select { + case err := <-errCh: + return err + case <-stop: + logger.Println("shutting down") + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + return httpSrv.Shutdown(ctx) + } +} + +// 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) + fmt.Printf("primary : %s\n", cfg.Primary) + fmt.Printf("merge : %s\n", cfg.Merge) + fmt.Printf("prefer : %s\n", cfg.Prefer) + fmt.Printf("timeout : %s\n", durationString(cfg.Timeout)) + fmt.Printf("freshness_ttl: %s\n", durationString(cfg.FreshnessTTL)) + 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 new file mode 100644 index 0000000..9240733 --- /dev/null +++ b/merge.go @@ -0,0 +1,173 @@ +package main + +import ( + "encoding/json" + "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. +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 { + return nil, err + } + out := make([]record, 0, len(raws)) + for _, raw := range raws { + var m recordMeta + _ = json.Unmarshal(raw, &m) // best-effort; missing fields stay zero + out = append(out, record{ + Raw: raw, + Certname: m.Certname, + ReportTimestamp: m.ReportTimestamp, + }) + } + 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. +func parseTimestamp(s string) time.Time { + if s == "" { + return time.Time{} + } + if t, err := time.Parse(time.RFC3339Nano, s); err == nil { + return t + } + 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. +func mergeNodes(results []backendResult) []json.RawMessage { + type pick struct { + raw json.RawMessage + ts time.Time + } + best := map[string]pick{} + var order []string + for _, res := range results { + for _, rec := range res.records { + ts := parseTimestamp(rec.ReportTimestamp) + cur, ok := best[rec.Certname] + if !ok { + best[rec.Certname] = pick{raw: rec.Raw, ts: ts} + 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} + } + } + } + out := make([]json.RawMessage, 0, len(order)) + for _, cn := range order { + out = append(out, best[cn].raw) + } + return out +} + +// freshness maps certname -> backend name that holds 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. +func buildFreshness(results []backendResult) freshness { + type pick struct { + backend string + ts time.Time + } + best := map[string]pick{} + for _, res := range results { + for _, rec := range res.records { + ts := parseTimestamp(rec.ReportTimestamp) + cur, ok := best[rec.Certname] + if !ok || ts.After(cur.ts) { + best[rec.Certname] = pick{backend: res.name, ts: ts} + } + } + } + f := make(freshness, len(best)) + for cn, p := range best { + f[cn] = p.backend + } + 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. +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 + byKey := map[string][]json.RawMessage{} + 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) + } + } + + // Emit in first-seen certname order for stable output. + var order []string + seen := map[string]bool{} + for _, res := range results { + for _, rec := range res.records { + if !seen[rec.Certname] { + seen[rec.Certname] = true + order = append(order, rec.Certname) + } + } + } + + out := []json.RawMessage{} + 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] + } + out = append(out, byKey[cn+"\x00"+chosen]...) + } + return out +} + +func contains(s []string, v string) bool { + for _, x := range s { + if x == v { + return true + } + } + return false +} diff --git a/merge_test.go b/merge_test.go new file mode 100644 index 0000000..59c4a55 --- /dev/null +++ b/merge_test.go @@ -0,0 +1,228 @@ +package main + +import ( + "encoding/json" + "testing" +) + +// recs builds a backendResult from name + literal JSON element strings. +func recs(t *testing.T, name string, elems ...string) backendResult { + t.Helper() + body := "[" + join(elems) + "]" + r, err := decodeRecords([]byte(body)) + if err != nil { + t.Fatalf("decodeRecords(%s): %v", body, err) + } + return backendResult{name: name, records: r} +} + +func join(elems []string) string { + out := "" + for i, e := range elems { + if i > 0 { + out += "," + } + out += e + } + return out +} + +// certnames extracts the certname field from a merged result set. +func certnames(t *testing.T, raws []json.RawMessage) []string { + t.Helper() + var out []string + for _, r := range raws { + var m recordMeta + if err := json.Unmarshal(r, &m); err != nil { + t.Fatalf("unmarshal %s: %v", r, err) + } + out = append(out, m.Certname) + } + return out +} + +// factValues extracts "certname:name=value" for /facts records to assert which +// backend's facts survived. +func factValues(t *testing.T, raws []json.RawMessage) []string { + t.Helper() + var out []string + for _, r := range raws { + var m struct { + Certname string `json:"certname"` + Name string `json:"name"` + Value string `json:"value"` + } + if err := json.Unmarshal(r, &m); err != nil { + t.Fatalf("unmarshal %s: %v", r, err) + } + out = append(out, m.Certname+":"+m.Name+"="+m.Value) + } + return out +} + +func node(cn, ts string) string { + return `{"certname":"` + cn + `","report_timestamp":"` + ts + `","latest_report_status":"changed"}` +} + +func fact(cn, name, val, ts string) string { + // facts records don't carry report_timestamp in real PuppetDB, but including + // it is harmless and lets a couple of tests reuse the same helper. Merge + // attribution for facts comes from the owner func, not the record. + if ts == "" { + return `{"certname":"` + cn + `","name":"` + name + `","value":"` + val + `"}` + } + return `{"certname":"` + cn + `","name":"` + name + `","value":"` + val + `","report_timestamp":"` + ts + `"}` +} + +func TestMergeNodes_NewerWins(t *testing.T) { + old := recs(t, "old", node("h1", "2026-07-01T00:00:00Z"), node("h2", "2026-07-10T00:00:00Z")) + nw := recs(t, "new", node("h1", "2026-07-20T00:00:00Z"), node("h3", "2026-07-05T00:00:00Z")) + + merged := mergeNodes([]backendResult{old, nw}) + got := map[string]string{} + for _, r := range merged { + var m recordMeta + _ = json.Unmarshal(r, &m) + got[m.Certname] = m.ReportTimestamp + } + if got["h1"] != "2026-07-20T00:00:00Z" { + t.Errorf("h1: newer (new) should win, got %s", got["h1"]) + } + if got["h2"] != "2026-07-10T00:00:00Z" { + t.Errorf("h2: only in old, got %s", got["h2"]) + } + if got["h3"] != "2026-07-05T00:00:00Z" { + t.Errorf("h3: only in new, got %s", got["h3"]) + } + if len(merged) != 3 { + t.Errorf("expected 3 deduped nodes, got %d", len(merged)) + } +} + +func TestMergeNodes_OneBackendOnly(t *testing.T) { + old := recs(t, "old", node("h1", "2026-07-01T00:00:00Z")) + // new returned nothing (e.g. empty result). + nw := backendResult{name: "new"} + merged := mergeNodes([]backendResult{old, nw}) + if len(merged) != 1 || certnames(t, merged)[0] != "h1" { + t.Fatalf("expected only h1, got %v", certnames(t, merged)) + } +} + +func TestMergeNodes_TieKeepsEarlierBackend(t *testing.T) { + // Equal timestamps: the backend listed first (precedence) wins. + prefer := recs(t, "new", node("h1", "2026-07-01T00:00:00Z")) + other := recs(t, "old", node("h1", "2026-07-01T00:00:00Z")) + merged := mergeNodes([]backendResult{prefer, other}) + if len(merged) != 1 { + t.Fatalf("expected 1 record, got %d", len(merged)) + } + // Ensure the kept record is the first backend's (identical here, but assert count/dedupe). + if certnames(t, merged)[0] != "h1" { + t.Fatalf("expected h1") + } +} + +func TestMergeNodes_PreservesUnknownFields(t *testing.T) { + old := recs(t, "old", `{"certname":"h1","report_timestamp":"2026-07-01T00:00:00Z","extra":{"deep":42}}`) + merged := mergeNodes([]backendResult{old}) + if len(merged) != 1 { + t.Fatalf("expected 1 record") + } + var m map[string]json.RawMessage + _ = json.Unmarshal(merged[0], &m) + if _, ok := m["extra"]; !ok { + t.Fatalf("unknown field 'extra' was dropped: %s", merged[0]) + } +} + +func TestMergeFacts_Static_PreferWins(t *testing.T) { + // h1 in both; static prefer=new -> new's facts kept, old's dropped. + old := recs(t, "old", fact("h1", "role", "web-old", ""), fact("h2", "role", "db-old", "")) + nw := recs(t, "new", fact("h1", "role", "web-new", "")) + + merged := mergeFacts([]backendResult{nw, old}, func(string) string { return "new" }) + got := factValues(t, merged) + assertContains(t, got, "h1:role=web-new") + assertNotContains(t, got, "h1:role=web-old") + // h2 only in old -> falls back to old. + assertContains(t, got, "h2:role=db-old") +} + +func TestMergeFacts_Freshness_NewerBackendWins(t *testing.T) { + // owner map says h1 belongs to old (older backend has the newer report), + // h2 belongs to new. Multiple facts per node must all come from the winner. + old := recs(t, "old", + fact("h1", "role", "web-old", ""), fact("h1", "ip", "10.0.0.1", ""), + fact("h2", "role", "db-old", "")) + nw := recs(t, "new", + fact("h1", "role", "web-new", ""), fact("h1", "ip", "10.9.9.9", ""), + fact("h2", "role", "db-new", ""), fact("h2", "ip", "10.0.0.2", "")) + + owner := func(cn string) string { + if cn == "h1" { + return "old" + } + return "new" + } + merged := mergeFacts([]backendResult{nw, old}, owner) + got := factValues(t, merged) + // h1 -> all old facts, no new facts. + assertContains(t, got, "h1:role=web-old") + assertContains(t, got, "h1:ip=10.0.0.1") + assertNotContains(t, got, "h1:role=web-new") + assertNotContains(t, got, "h1:ip=10.9.9.9") + // h2 -> all new facts. + assertContains(t, got, "h2:role=db-new") + assertContains(t, got, "h2:ip=10.0.0.2") + assertNotContains(t, got, "h2:role=db-old") +} + +func TestMergeFacts_OwnerMissingFallsBackToPrecedence(t *testing.T) { + // owner returns a backend with no facts for h1 -> fall back to first + // backend present (precedence order of the slice). + prefer := recs(t, "new", fact("h1", "role", "web-new", "")) + other := recs(t, "old", fact("h1", "role", "web-old", "")) + merged := mergeFacts([]backendResult{prefer, other}, func(string) string { return "ghost" }) + got := factValues(t, merged) + assertContains(t, got, "h1:role=web-new") // new is first in slice + assertNotContains(t, got, "h1:role=web-old") +} + +func TestBuildFreshness(t *testing.T) { + // old has newer report for h1; new has newer for h2. + old := recs(t, "old", node("h1", "2026-07-20T00:00:00Z"), node("h2", "2026-07-01T00:00:00Z")) + nw := recs(t, "new", node("h1", "2026-07-01T00:00:00Z"), node("h2", "2026-07-20T00:00:00Z")) + f := buildFreshness([]backendResult{old, nw}) + if f["h1"] != "old" { + t.Errorf("h1 should belong to old, got %q", f["h1"]) + } + if f["h2"] != "new" { + t.Errorf("h2 should belong to new, got %q", f["h2"]) + } +} + +func TestDecodeRecords_NotArray(t *testing.T) { + if _, err := decodeRecords([]byte(`{"not":"array"}`)); err == nil { + t.Fatal("expected error decoding non-array body") + } +} + +func assertContains(t *testing.T, hay []string, needle string) { + t.Helper() + for _, h := range hay { + if h == needle { + return + } + } + t.Errorf("expected %q in %v", needle, hay) +} + +func assertNotContains(t *testing.T, hay []string, needle string) { + t.Helper() + for _, h := range hay { + if h == needle { + t.Errorf("did not expect %q in %v", needle, hay) + } + } +} diff --git a/server.go b/server.go new file mode 100644 index 0000000..1763a1a --- /dev/null +++ b/server.go @@ -0,0 +1,294 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "io" + "log" + "net/http" + "net/url" + "sort" + "strings" + "sync" + "time" +) + +const ( + factsPath = "/pdb/query/v4/facts" + nodesPath = "/pdb/query/v4/nodes" + queryV4 = "/pdb/query/v4/" +) + +// 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 + err error +} + +// Server proxies and merges PuppetDB queries across the configured backends. +type Server struct { + cfg Config + client *http.Client + log *log.Logger + + // freshness cache (freshness merge only). + mu sync.Mutex + freshData freshness + 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, + client: &http.Client{Timeout: cfg.Timeout}, + log: logger, + } +} + +// Handler returns the HTTP mux for the proxy. +func (s *Server) Handler() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/healthz", s.handleHealth) + mux.HandleFunc("/pdb/query/v4/", s.handleQuery) + return mux +} + +// handleQuery dispatches /pdb/query/v4/* requests: /facts and /nodes are merged +// across backends; 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) + return + } + switch r.URL.Path { + case nodesPath: + s.serveMerged(w, r, nodesPath, s.mergeNodesResponse) + case factsPath: + s.serveMerged(w, r, factsPath, s.mergeFactsResponse) + default: + s.proxyPrimary(w, r) + } +} + +// 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) { + query := r.URL.Query().Get("query") + results := s.fanOut(r.Context(), path, query) + + var alive []backendResult + for _, res := range results { + if res.err != nil { + s.log.Printf("warning: backend %q failed for %s: %v", res.name, path, res.err) + continue + } + alive = append(alive, res) + } + if len(alive) == 0 { + http.Error(w, "all backends failed", http.StatusBadGateway) + return + } + + merged := merge(alive) + writeJSON(w, merged) +} + +// 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 { + if b, ok := fresh[cn]; ok { + return b + } + return prefer + }) +} + +// byPrecedence orders results so the Prefer backend comes first, giving it the +// tie-break on equal timestamps. Remaining backends keep config order. +func (s *Server) byPrecedence(results []backendResult) []backendResult { + ordered := make([]backendResult, len(results)) + copy(ordered, results) + sort.SliceStable(ordered, func(i, j int) bool { + return ordered[i].name == s.cfg.Prefer && ordered[j].name != s.cfg.Prefer + }) + 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. +func (s *Server) freshnessMap(ctx context.Context, _ []backendResult) freshness { + s.mu.Lock() + if s.freshData != nil && time.Since(s.freshAt) < s.cfg.FreshnessTTL { + f := s.freshData + s.mu.Unlock() + return f + } + s.mu.Unlock() + + // Empty query = all nodes; cheap enough for a short-TTL cache. + nodeResults := s.fanOut(ctx, nodesPath, "") + var alive []backendResult + for _, res := range nodeResults { + if res.err != nil { + s.log.Printf("warning: freshness /nodes query to %q failed: %v", res.name, res.err) + continue + } + alive = append(alive, res) + } + f := buildFreshness(s.byPrecedence(alive)) + + s.mu.Lock() + s.freshData = f + s.freshAt = time.Now() + s.mu.Unlock() + return f +} + +// fanOut queries every backend concurrently for path?query=... and returns one +// backendResult per backend, in config order. +func (s *Server) fanOut(ctx context.Context, path, query string) []backendResult { + results := make([]backendResult, len(s.cfg.Backends)) + var wg sync.WaitGroup + for i, b := range s.cfg.Backends { + wg.Add(1) + go func(i int, b Backend) { + defer wg.Done() + recs, err := s.queryBackend(ctx, b, path, query) + results[i] = backendResult{name: b.Name, records: recs, err: err} + }(i, b) + } + wg.Wait() + return results +} + +// queryBackend performs one GET b.URL+path?query=... and decodes the JSON array. +func (s *Server) queryBackend(ctx context.Context, b Backend, path, query string) ([]record, error) { + target := strings.TrimRight(b.URL, "/") + path + req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil) + if err != nil { + return nil, err + } + if query != "" { + q := url.Values{} + q.Set("query", query) + req.URL.RawQuery = q.Encode() + } + resp, err := s.client.Do(req) + if err != nil { + return nil, err + } + defer func() { _ = resp.Body.Close() }() + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, err + } + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) + } + return decodeRecords(body) +} + +// 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 + if r.URL.RawQuery != "" { + target += "?" + r.URL.RawQuery + } + req, err := http.NewRequestWithContext(r.Context(), http.MethodGet, target, nil) + if err != nil { + http.Error(w, err.Error(), http.StatusBadGateway) + return + } + resp, err := s.client.Do(req) + if err != nil { + s.log.Printf("warning: primary %q pass-through failed for %s: %v", b.Name, r.URL.Path, err) + http.Error(w, "primary backend failed", http.StatusBadGateway) + return + } + defer func() { _ = resp.Body.Close() }() + if ct := resp.Header.Get("Content-Type"); ct != "" { + w.Header().Set("Content-Type", ct) + } + w.WriteHeader(resp.StatusCode) + _, _ = 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, probe) + + report := healthReport{Backends: map[string]string{}} + healthy := 0 + for _, res := range results { + if res.err != nil { + report.Backends[res.name] = res.err.Error() + continue + } + report.Backends[res.name] = "ok" + healthy++ + } + switch { + case healthy == len(results): + report.Status = "ok" + case healthy > 0: + report.Status = "degraded" + default: + report.Status = "down" + } + + w.Header().Set("Content-Type", "application/json") + if healthy == 0 { + w.WriteHeader(http.StatusServiceUnavailable) + } + enc := json.NewEncoder(w) + enc.SetIndent("", " ") + _ = 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 { + recs = []json.RawMessage{} + } + _ = json.NewEncoder(w).Encode(recs) +} diff --git a/server_test.go b/server_test.go new file mode 100644 index 0000000..18243e2 --- /dev/null +++ b/server_test.go @@ -0,0 +1,286 @@ +package main + +import ( + "encoding/json" + "io" + "log" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "testing" + "time" +) + +// fakeBackend is an httptest PuppetDB that returns canned bodies per path and +// records the query params it received. +type fakeBackend struct { + srv *httptest.Server + nodesBody string + factsBody string + fail bool // return 500 for everything + delay time.Duration // artificial latency + gotQueries map[string]string +} + +func newFakeBackend(t *testing.T, nodesBody, factsBody string) *fakeBackend { + t.Helper() + fb := &fakeBackend{nodesBody: nodesBody, factsBody: factsBody, gotQueries: map[string]string{}} + fb.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if fb.delay > 0 { + time.Sleep(fb.delay) + } + fb.gotQueries[r.URL.Path] = r.URL.Query().Get("query") + if fb.fail { + http.Error(w, "boom", http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/json") + switch r.URL.Path { + case nodesPath: + _, _ = io.WriteString(w, fb.nodesBody) + case factsPath: + _, _ = io.WriteString(w, fb.factsBody) + default: + _, _ = io.WriteString(w, `[{"path":"`+r.URL.Path+`"}]`) + } + })) + t.Cleanup(fb.srv.Close) + return fb +} + +func testConfig(oldURL, newURL, merge string) Config { + return Config{ + Listen: ":0", + Backends: []Backend{{Name: "old", URL: oldURL}, {Name: "new", URL: newURL}}, + Primary: "new", + Merge: merge, + Prefer: "new", + Timeout: 2 * time.Second, + FreshnessTTL: 30 * time.Second, + } +} + +func newTestServer(cfg Config) *Server { + return NewServer(cfg, log.New(io.Discard, "", 0)) +} + +func doGet(t *testing.T, h http.Handler, path, query string) *httptest.ResponseRecorder { + t.Helper() + target := path + if query != "" { + target += "?query=" + url.QueryEscape(query) + } + req := httptest.NewRequest(http.MethodGet, target, nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + return rec +} + +func TestHandler_NodesMerged(t *testing.T) { + old := newFakeBackend(t, + `[`+node("h1", "2026-07-01T00:00:00Z")+`,`+node("h2", "2026-07-10T00:00:00Z")+`]`, `[]`) + nw := newFakeBackend(t, + `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`) + srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), nodesPath, `["=","certname","h1"]`) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + var got []recordMeta + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + if len(got) != 2 { + t.Fatalf("expected 2 deduped nodes, got %d: %s", len(got), rec.Body.String()) + } + for _, m := range got { + if m.Certname == "h1" && m.ReportTimestamp != "2026-07-20T00:00:00Z" { + t.Errorf("h1 should be new's newer record, got %s", m.ReportTimestamp) + } + } +} + +func TestHandler_QueryPassthrough(t *testing.T) { + old := newFakeBackend(t, `[]`, `[]`) + nw := newFakeBackend(t, `[]`, `[]`) + srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) + + q := `["=","certname","abc.example.net"]` + doGet(t, srv.Handler(), factsPath, q) + if old.gotQueries[factsPath] != q { + t.Errorf("old backend got query %q, want %q", old.gotQueries[factsPath], q) + } + if nw.gotQueries[factsPath] != q { + t.Errorf("new backend got query %q, want %q", nw.gotQueries[factsPath], q) + } +} + +func TestHandler_FactsStaticMerge(t *testing.T) { + old := newFakeBackend(t, `[]`, + `[`+fact("h1", "role", "web-old", "")+`,`+fact("h2", "role", "db-old", "")+`]`) + nw := newFakeBackend(t, `[]`, + `[`+fact("h1", "role", "web-new", "")+`]`) + srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), factsPath, `["=","name","role"]`) + if rec.Code != http.StatusOK { + t.Fatalf("status %d", rec.Code) + } + body := rec.Body.String() + if !strings.Contains(body, "web-new") || strings.Contains(body, "web-old") { + t.Errorf("static prefer=new should keep web-new, drop web-old: %s", body) + } + if !strings.Contains(body, "db-old") { + t.Errorf("h2 only in old should survive: %s", body) + } +} + +func TestHandler_FactsFreshnessMerge(t *testing.T) { + // Freshness: old holds h1's newer report; new holds h2's newer report. + old := newFakeBackend(t, + `[`+node("h1", "2026-07-20T00:00:00Z")+`,`+node("h2", "2026-07-01T00:00:00Z")+`]`, + `[`+fact("h1", "role", "web-old", "")+`,`+fact("h2", "role", "db-old", "")+`]`) + nw := newFakeBackend(t, + `[`+node("h1", "2026-07-01T00:00:00Z")+`,`+node("h2", "2026-07-20T00:00:00Z")+`]`, + `[`+fact("h1", "role", "web-new", "")+`,`+fact("h2", "role", "db-new", "")+`]`) + srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeFreshness)) + + rec := doGet(t, srv.Handler(), factsPath, `["=","name","role"]`) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + body := rec.Body.String() + // h1 -> old (newer report there); h2 -> new. + if !strings.Contains(body, "web-old") || strings.Contains(body, "web-new") { + t.Errorf("h1 should resolve to old: %s", body) + } + if !strings.Contains(body, "db-new") || strings.Contains(body, "db-old") { + t.Errorf("h2 should resolve to new: %s", body) + } +} + +func TestHandler_OneBackendDown(t *testing.T) { + old := newFakeBackend(t, `[]`, `[]`) + old.fail = true + nw := newFakeBackend(t, + `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`) + srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), nodesPath, "") + if rec.Code != http.StatusOK { + t.Fatalf("expected 200 serving survivor, got %d", rec.Code) + } + if !strings.Contains(rec.Body.String(), "h1") { + t.Errorf("expected survivor's h1: %s", rec.Body.String()) + } +} + +func TestHandler_BothBackendsDown(t *testing.T) { + old := newFakeBackend(t, `[]`, `[]`) + nw := newFakeBackend(t, `[]`, `[]`) + old.fail, nw.fail = true, true + srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), nodesPath, "") + if rec.Code != http.StatusBadGateway { + t.Fatalf("expected 502 when all backends fail, got %d", rec.Code) + } +} + +func TestHandler_PassThroughToPrimary(t *testing.T) { + // A non-merged v4 path (e.g. /reports) goes only to the primary (new). + old := newFakeBackend(t, `[]`, `[]`) + nw := newFakeBackend(t, `[]`, `[]`) + srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), "/pdb/query/v4/reports", `["=","certname","h1"]`) + if rec.Code != http.StatusOK { + t.Fatalf("status %d", rec.Code) + } + if !strings.Contains(rec.Body.String(), "/pdb/query/v4/reports") { + t.Errorf("expected pass-through body, got %s", rec.Body.String()) + } + // Only primary (new) should have been queried. + if _, hit := old.gotQueries["/pdb/query/v4/reports"]; hit { + t.Errorf("non-primary backend should not be queried for pass-through") + } + if _, hit := nw.gotQueries["/pdb/query/v4/reports"]; !hit { + t.Errorf("primary backend should be queried for pass-through") + } +} + +func TestHandler_PostRejected(t *testing.T) { + old := newFakeBackend(t, `[]`, `[]`) + nw := newFakeBackend(t, `[]`, `[]`) + srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) + req := httptest.NewRequest(http.MethodPost, factsPath, nil) + rec := httptest.NewRecorder() + srv.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusMethodNotAllowed { + t.Fatalf("expected 405 for POST, got %d", rec.Code) + } +} + +func TestHandler_Health(t *testing.T) { + old := newFakeBackend(t, `[]`, `[]`) + nw := newFakeBackend(t, `[]`, `[]`) + srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), "/healthz", "") + if rec.Code != http.StatusOK { + t.Fatalf("expected 200 healthy, got %d: %s", rec.Code, rec.Body.String()) + } + var hr healthReport + if err := json.Unmarshal(rec.Body.Bytes(), &hr); err != nil { + t.Fatal(err) + } + if hr.Status != "ok" || hr.Backends["old"] != "ok" || hr.Backends["new"] != "ok" { + t.Fatalf("unexpected health: %+v", hr) + } +} + +func TestHandler_HealthDegradedAndDown(t *testing.T) { + old := newFakeBackend(t, `[]`, `[]`) + nw := newFakeBackend(t, `[]`, `[]`) + old.fail = true + srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), "/healthz", "") + var hr healthReport + _ = json.Unmarshal(rec.Body.Bytes(), &hr) + if hr.Status != "degraded" { + t.Errorf("expected degraded, got %s", hr.Status) + } + if rec.Code != http.StatusOK { + t.Errorf("degraded should still be 200, got %d", rec.Code) + } + + nw.fail = true + rec = doGet(t, srv.Handler(), "/healthz", "") + _ = json.Unmarshal(rec.Body.Bytes(), &hr) + if hr.Status != "down" || rec.Code != http.StatusServiceUnavailable { + t.Errorf("expected down/503, got %s/%d", hr.Status, rec.Code) + } +} + +func TestFreshnessCache_Reused(t *testing.T) { + old := newFakeBackend(t, + `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, + `[`+fact("h1", "role", "web-old", "")+`]`) + nw := newFakeBackend(t, + `[`+node("h1", "2026-07-01T00:00:00Z")+`]`, + `[`+fact("h1", "role", "web-new", "")+`]`) + srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeFreshness)) + + // Two facts queries; the freshness /nodes probe should be cached after the + // first, so query recording only reflects the last observed nodes query but + // results stay consistent (h1 -> old). + for i := 0; i < 2; i++ { + rec := doGet(t, srv.Handler(), factsPath, `["=","name","role"]`) + if !strings.Contains(rec.Body.String(), "web-old") { + t.Fatalf("iteration %d: expected h1->old, got %s", i, rec.Body.String()) + } + } +}