docs: strip over-commenting from README and source #6
@@ -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*
|
||||
|
||||
@@ -1,5 +1,3 @@
|
||||
# Container image for pdbmux, the merging PuppetDB proxy daemon. It ships only
|
||||
# as a distroless static image; there is no OS package.
|
||||
FROM golang:1.25-alpine AS builder
|
||||
|
||||
RUN apk add --no-cache git
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -83,16 +83,10 @@ Each backend applies `order_by`/`limit`/`offset` to its own slice only, so
|
||||
|
||||
Precedence (lowest → highest): **defaults < config file < env vars (`PDBMUX_*`) < flags**.
|
||||
|
||||
Config file: `$XDG_CONFIG_HOME/pdbmux/config.yaml`. In a container, configuration
|
||||
is supplied entirely via `PDBMUX_*` env vars (no config file) — see
|
||||
[Deployment](#deployment).
|
||||
|
||||
**`backends` has no default.** `pdbmux` refuses to start until at least one
|
||||
backend is configured, via the config file or `PDBMUX_BACKENDS`. `primary` and
|
||||
`prefer` default to the first configured backend.
|
||||
Config file: `$XDG_CONFIG_HOME/pdbmux/config.yaml`. In a container there is no
|
||||
config file — everything comes from `PDBMUX_*` env vars.
|
||||
|
||||
```yaml
|
||||
# ~/.config/pdbmux/config.yaml (local dev; in a container use PDBMUX_* env instead)
|
||||
listen: ":8080"
|
||||
backends:
|
||||
- name: old
|
||||
@@ -106,8 +100,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 |
|
||||
|---|---|
|
||||
@@ -123,69 +117,27 @@ Flags: `--listen`, `--primary`, `--merge`.
|
||||
|
||||
## Running
|
||||
|
||||
```bash
|
||||
pdbmux # start the proxy (serve is the default action)
|
||||
pdbmux serve # explicit
|
||||
pdbmux config init # write an example config file to edit
|
||||
pdbmux config show # print active config after all overrides
|
||||
pdbmux version
|
||||
```
|
||||
|
||||
Point a consumer at it — any PuppetDB v4 client works, it just needs the
|
||||
`pdbmux` base URL in place of a PuppetDB one. For Puppetboard, that is its
|
||||
PuppetDB host/port setting; for a raw query:
|
||||
Subcommands: `serve` (default), `config init`, `config show`, `version`. Run
|
||||
`pdbmux --help` for details. Any PuppetDB v4 client works against the `pdbmux`
|
||||
base URL in place of a PuppetDB one.
|
||||
|
||||
```bash
|
||||
PDBMUX_BACKENDS='old=http://puppetdb1.example.com:8080,new=http://puppetdb2.example.com:8080' pdbmux
|
||||
curl -s --get http://localhost:8080/pdb/query/v4/nodes \
|
||||
--data-urlencode 'query=["=","certname","host1.example.com"]'
|
||||
```
|
||||
|
||||
## 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` is distributed as a container image only — there is no OS package. The
|
||||
image is built and pushed on every `v*` tag (`.woodpecker/docker.yaml`); the
|
||||
registry and repository are pipeline settings, so point them at your own.
|
||||
Container image only — no OS package. Every `v*` tag builds and pushes the image
|
||||
(`.woodpecker/docker.yaml`); registry and repository are pipeline settings. Tag
|
||||
with `make patch` / `minor` / `major`.
|
||||
|
||||
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. It is stateless, so run
|
||||
as many replicas as you like behind an ordinary Service/Ingress.
|
||||
|
||||
A container needs at minimum `PDBMUX_BACKENDS`; everything else has a default:
|
||||
|
||||
```yaml
|
||||
env:
|
||||
- name: PDBMUX_BACKENDS
|
||||
value: "old=http://puppetdb1.example.com:8080,new=http://puppetdb2.example.com:8080"
|
||||
- name: PDBMUX_PRIMARY
|
||||
value: "new"
|
||||
- name: PDBMUX_PREFER
|
||||
value: "new"
|
||||
```
|
||||
|
||||
Locally you can run the binary directly for development:
|
||||
|
||||
```bash
|
||||
PDBMUX_BACKENDS='old=http://puppetdb1.example.com:8080,new=http://puppetdb2.example.com: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
|
||||
```
|
||||
A static (`CGO_ENABLED=0`) binary on a distroless base, configured entirely via
|
||||
`PDBMUX_*` env vars; a container needs at minimum `PDBMUX_BACKENDS`. Stateless,
|
||||
so run as many replicas as you like; use `/healthz` for liveness/readiness
|
||||
probes.
|
||||
|
||||
@@ -16,48 +16,29 @@ const (
|
||||
configFileName = "config.yaml"
|
||||
envPrefix = "PDBMUX_"
|
||||
|
||||
// defaultListen is the default HTTP listen address.
|
||||
defaultListen = ":8080"
|
||||
|
||||
defaultTimeout = 10 * time.Second
|
||||
defaultFreshnessTTL = 30 * time.Second
|
||||
)
|
||||
|
||||
// exampleBackends are the placeholder backends written by `config init`. They
|
||||
// are a scaffold to edit, not a working configuration.
|
||||
var exampleBackends = []Backend{
|
||||
{Name: "primary", URL: "http://puppetdb1.example.com:8080"},
|
||||
{Name: "secondary", URL: "http://puppetdb2.example.com:8080"},
|
||||
}
|
||||
|
||||
// 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"`
|
||||
}
|
||||
|
||||
@@ -66,9 +47,6 @@ const (
|
||||
mergeStatic = "static"
|
||||
)
|
||||
|
||||
// DefaultConfig returns the built-in defaults. Backends have no default: they
|
||||
// must come from the config file or PDBMUX_BACKENDS. Primary and Prefer default
|
||||
// to the first configured backend (see normalize).
|
||||
func DefaultConfig() Config {
|
||||
return Config{
|
||||
Listen: defaultListen,
|
||||
@@ -78,8 +56,6 @@ func DefaultConfig() Config {
|
||||
}
|
||||
}
|
||||
|
||||
// ExampleConfig returns DefaultConfig with placeholder backends filled in, as
|
||||
// written by `config init`.
|
||||
func ExampleConfig() Config {
|
||||
cfg := DefaultConfig()
|
||||
cfg.Backends = append([]Backend(nil), exampleBackends...)
|
||||
@@ -88,7 +64,6 @@ func ExampleConfig() Config {
|
||||
return cfg
|
||||
}
|
||||
|
||||
// ConfigDir returns the XDG_CONFIG_HOME/pdbmux directory.
|
||||
func ConfigDir() string {
|
||||
base := os.Getenv("XDG_CONFIG_HOME")
|
||||
if base == "" {
|
||||
@@ -98,19 +73,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").
|
||||
//
|
||||
// Load reports only read/parse errors; the result is not validated, so commands
|
||||
// that do not serve (config init, version) still work on an unconfigured host.
|
||||
// Callers that serve must call Validate after applying flags.
|
||||
// Precedence: defaults < config file < env vars < flags, and flags are applied by the caller.
|
||||
func Load() (Config, error) {
|
||||
cfg := DefaultConfig()
|
||||
|
||||
@@ -130,7 +97,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
|
||||
@@ -161,8 +127,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, ",") {
|
||||
@@ -180,8 +145,6 @@ func parseBackends(s string) []Backend {
|
||||
return out
|
||||
}
|
||||
|
||||
// normalize fills in values that default to something derived rather than
|
||||
// constant: Primary and Prefer both fall back to the first configured backend.
|
||||
func (c *Config) normalize() {
|
||||
if len(c.Backends) == 0 {
|
||||
return
|
||||
@@ -194,7 +157,6 @@ func (c *Config) normalize() {
|
||||
}
|
||||
}
|
||||
|
||||
// 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: set %sBACKENDS to \"name=url,name=url\" or add a backends list to %s",
|
||||
@@ -227,8 +189,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 {
|
||||
@@ -238,7 +198,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 {
|
||||
@@ -261,8 +220,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"
|
||||
|
||||
@@ -1,27 +1,4 @@
|
||||
// Command pdbmux is a small merging HTTP proxy over several PuppetDB backends.
|
||||
//
|
||||
// Running more than one PuppetDB — during a migration between them, or across
|
||||
// regions — means a given node's current data lives in exactly one of them at
|
||||
// any moment. pdbmux presents a single merged PuppetDB v4 query surface so
|
||||
// clients see one consistent view:
|
||||
//
|
||||
// - GET /pdb/query/v4/nodes — fan out to all backends, dedupe by certname,
|
||||
// keep the record with the newer report_timestamp.
|
||||
// - GET /pdb/query/v4/facts — fan out to all, and for a certname present in
|
||||
// more than one keep ALL facts from the backend holding that node's newer
|
||||
// report (freshness merge) or a static preferred backend (static merge).
|
||||
// - GET /pdb/query/v4/reports and /events — fan out to all and serve the
|
||||
// deduped union, re-ordered and re-paged across backends, because reports
|
||||
// are immutable history and a moved node has some in each.
|
||||
// - GET /pdb/query/v4/reports/<hash>/{events,logs,metrics} — served by
|
||||
// whichever backend actually holds that report.
|
||||
// - 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); order_by, limit,
|
||||
// offset and include_total are re-applied over the merged result set. If one
|
||||
// backend errors/times out, the others' results are served; only if every
|
||||
// backend fails does a merged endpoint return 502.
|
||||
package main
|
||||
|
||||
import (
|
||||
@@ -125,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)
|
||||
@@ -161,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)
|
||||
|
||||
@@ -5,9 +5,7 @@ import (
|
||||
"time"
|
||||
)
|
||||
|
||||
// record is a single PuppetDB result element kept as raw JSON so unknown fields
|
||||
// survive the merge untouched. certname/report_timestamp/hash 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
|
||||
@@ -15,16 +13,12 @@ type record struct {
|
||||
Hash string // only populated for /reports records
|
||||
}
|
||||
|
||||
// recordMeta is the subset we decode from any /nodes, /facts or /reports element
|
||||
// to drive merge decisions.
|
||||
type recordMeta struct {
|
||||
Certname string `json:"certname"`
|
||||
ReportTimestamp string `json:"report_timestamp"`
|
||||
Hash string `json:"hash"`
|
||||
}
|
||||
|
||||
// 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 {
|
||||
@@ -44,8 +38,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{}
|
||||
@@ -56,10 +49,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
|
||||
@@ -76,7 +66,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}
|
||||
}
|
||||
@@ -89,12 +78,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
|
||||
@@ -117,19 +104,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 {
|
||||
@@ -157,7 +134,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]
|
||||
}
|
||||
|
||||
+8
-33
@@ -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 {
|
||||
|
||||
@@ -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/<hash>/{events,logs,metrics} — whose data lives in
|
||||
// exactly one backend.
|
||||
// Matches /pdb/query/v4/reports/<hash>/{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 {
|
||||
|
||||
Reference in New Issue
Block a user