merge main
This commit is contained in:
@@ -1,21 +1,23 @@
|
||||
# pdbmux — merging PuppetDB proxy
|
||||
|
||||
`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`, Puppetboard, or anything else at `pdbmux` instead of a raw
|
||||
PuppetDB and it sees one consistent view spanning both.
|
||||
`pdbmux` is a small HTTP daemon that fronts **several** PuppetDB backends and
|
||||
serves a single, merged PuppetDB v4 query surface on one address. Point
|
||||
Puppetboard, or any other PuppetDB API client, at `pdbmux` instead of a raw
|
||||
PuppetDB and it sees one consistent view spanning all of them.
|
||||
|
||||
## Why
|
||||
|
||||
During the VM→k8s Puppet migration there are two PuppetDBs:
|
||||
Running more than one PuppetDB — during a migration between two of them, or
|
||||
across regions — means a given node's current data lives in exactly one at any
|
||||
moment, and consumers have to know which, or query each in turn. Consider two
|
||||
backends being merged during a migration:
|
||||
|
||||
- **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)
|
||||
- **old** — the PuppetDB nodes are moving off, e.g. `http://puppetdb1.example.com:8080`
|
||||
- **new** — the PuppetDB nodes are moving on to, e.g. `http://puppetdb2.example.com: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
|
||||
Nodes move from old to new as they migrate. `pdbmux` merges both so consumers
|
||||
don't have to know (or query twice) which PuppetDB a node currently lives in.
|
||||
The backend names are arbitrary labels; there is no fixed number of backends.
|
||||
|
||||
## Endpoints
|
||||
|
||||
@@ -24,16 +26,16 @@ 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/reports` | Fan out to both and serve the **union**, deduped by report `hash`, re-ordered and re-paged across the two backends. |
|
||||
| `GET /pdb/query/v4/events` | Fan out to both and serve the **union**, deduped by record identity, re-ordered and re-paged. |
|
||||
| `GET /pdb/query/v4/reports/<hash>/{events,logs,metrics}` | Ask both; serve the answer from whichever backend actually holds that report. `404` when neither does. |
|
||||
| `GET /pdb/query/v4/nodes` | Fan out to all backends, dedupe by `certname`, keep the record with the newer `report_timestamp`. |
|
||||
| `GET /pdb/query/v4/facts` | Fan out to all, and per `certname` keep **all** facts from the backend that owns that node (see merge semantics). |
|
||||
| `GET /pdb/query/v4/reports` | Fan out to all and serve the **union**, deduped by report `hash`, re-ordered and re-paged across backends. |
|
||||
| `GET /pdb/query/v4/events` | Fan out to all and serve the **union**, deduped by record identity, re-ordered and re-paged. |
|
||||
| `GET /pdb/query/v4/reports/<hash>/{events,logs,metrics}` | Ask every backend; serve the answer from whichever backend actually holds that report. `404` when neither does. |
|
||||
| `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
|
||||
surviving backends' 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.
|
||||
|
||||
@@ -42,12 +44,12 @@ unknown fields survive untouched.
|
||||
- **`/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
|
||||
- **`/facts`** — node-level granularity. For a `certname` present in more than
|
||||
one backend, `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 map built by querying `/nodes` from every backend, 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.
|
||||
@@ -81,16 +83,16 @@ 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 Kubernetes there is no
|
||||
Config file: `$XDG_CONFIG_HOME/pdbmux/config.yaml`. In a container there is no
|
||||
config file — everything comes from `PDBMUX_*` env vars.
|
||||
|
||||
```yaml
|
||||
listen: ":8080"
|
||||
backends:
|
||||
- name: old
|
||||
url: http://puppetdbapi.service.consul:8080
|
||||
url: http://puppetdb1.example.com:8080
|
||||
- name: new
|
||||
url: https://puppetdb.k8s.syd1.au.unkin.net
|
||||
url: https://puppetdb2.example.com
|
||||
primary: new # backend used for non-merged /pdb/query/v4/* pass-through
|
||||
merge: freshness # freshness | static
|
||||
prefer: new # winner on ties / static merge / fallback
|
||||
@@ -116,11 +118,13 @@ Flags: `--listen`, `--primary`, `--merge`.
|
||||
## Running
|
||||
|
||||
Subcommands: `serve` (default), `config init`, `config show`, `version`. Run
|
||||
`pdbmux --help` for details.
|
||||
`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://puppetdbapi.service.consul:8080,new=https://puppetdb.k8s.syd1.au.unkin.net' pdbmux
|
||||
node-lookup --url http://localhost:8080/pdb/query/v4/facts -R
|
||||
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
|
||||
@@ -129,10 +133,11 @@ node-lookup --url http://localhost:8080/pdb/query/v4/facts -R
|
||||
|
||||
## Deployment
|
||||
|
||||
Kubernetes only — no RPM. Every `v*` tag builds and pushes
|
||||
`artifactapi.k8s.syd1.au.unkin.net/docker-internal/pdbmux:<tag>`
|
||||
(`.woodpecker/docker.yaml`); tag with `make patch` / `minor` / `major`.
|
||||
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`.
|
||||
|
||||
Manifests live in `argocd-apps` under `apps/base/pdbmux/` (namespace `pdbmux`,
|
||||
2 replicas). Use `/healthz` for liveness/readiness probes. Reachable from VMs and
|
||||
workstations at `https://pdbmux.k8s.syd1.au.unkin.net`.
|
||||
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,15 +16,17 @@ const (
|
||||
configFileName = "config.yaml"
|
||||
envPrefix = "PDBMUX_"
|
||||
|
||||
defaultListen = ":8080"
|
||||
defaultOldURL = "http://puppetdbapi.service.consul:8080"
|
||||
defaultNewURL = "https://puppetdb.k8s.syd1.au.unkin.net"
|
||||
defaultPrimary = "new"
|
||||
defaultListen = ":8080"
|
||||
|
||||
defaultTimeout = 10 * time.Second
|
||||
defaultFreshnessTTL = 30 * time.Second
|
||||
)
|
||||
|
||||
var exampleBackends = []Backend{
|
||||
{Name: "primary", URL: "http://puppetdb1.example.com:8080"},
|
||||
{Name: "secondary", URL: "http://puppetdb2.example.com:8080"},
|
||||
}
|
||||
|
||||
type Backend struct {
|
||||
Name string `yaml:"name"`
|
||||
URL string `yaml:"url"` // base URL only; the query path is appended per request
|
||||
@@ -47,19 +49,21 @@ const (
|
||||
|
||||
func DefaultConfig() Config {
|
||||
return Config{
|
||||
Listen: defaultListen,
|
||||
Backends: []Backend{
|
||||
{Name: "old", URL: defaultOldURL},
|
||||
{Name: "new", URL: defaultNewURL},
|
||||
},
|
||||
Primary: defaultPrimary,
|
||||
Listen: defaultListen,
|
||||
Merge: mergeFreshness,
|
||||
Prefer: defaultPrimary,
|
||||
Timeout: defaultTimeout,
|
||||
FreshnessTTL: defaultFreshnessTTL,
|
||||
}
|
||||
}
|
||||
|
||||
func ExampleConfig() Config {
|
||||
cfg := DefaultConfig()
|
||||
cfg.Backends = append([]Backend(nil), exampleBackends...)
|
||||
cfg.Primary = exampleBackends[0].Name
|
||||
cfg.Prefer = exampleBackends[0].Name
|
||||
return cfg
|
||||
}
|
||||
|
||||
func ConfigDir() string {
|
||||
base := os.Getenv("XDG_CONFIG_HOME")
|
||||
if base == "" {
|
||||
@@ -89,10 +93,7 @@ func Load() (Config, error) {
|
||||
}
|
||||
|
||||
applyEnv(&cfg, os.Getenv)
|
||||
|
||||
if err := cfg.Validate(); err != nil {
|
||||
return cfg, err
|
||||
}
|
||||
cfg.normalize()
|
||||
return cfg, nil
|
||||
}
|
||||
|
||||
@@ -144,9 +145,22 @@ func parseBackends(s string) []Backend {
|
||||
return out
|
||||
}
|
||||
|
||||
func (c *Config) normalize() {
|
||||
if len(c.Backends) == 0 {
|
||||
return
|
||||
}
|
||||
if c.Primary == "" {
|
||||
c.Primary = c.Backends[0].Name
|
||||
}
|
||||
if c.Prefer == "" {
|
||||
c.Prefer = c.Backends[0].Name
|
||||
}
|
||||
}
|
||||
|
||||
func (c Config) Validate() error {
|
||||
if len(c.Backends) == 0 {
|
||||
return fmt.Errorf("no backends configured")
|
||||
return fmt.Errorf("no backends configured: set %sBACKENDS to \"name=url,name=url\" or add a backends list to %s",
|
||||
envPrefix, ConfigPath())
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
for _, b := range c.Backends {
|
||||
@@ -193,9 +207,10 @@ func writeDefaultConfig() error {
|
||||
if _, err := os.Stat(path); err == nil {
|
||||
return fmt.Errorf("config already exists at %s", path)
|
||||
}
|
||||
data, _ := yaml.Marshal(DefaultConfig())
|
||||
data, _ := yaml.Marshal(ExampleConfig())
|
||||
header := []byte("# pdbmux configuration\n" +
|
||||
"# A merging proxy over two PuppetDBs (old Consul + new k8s) during migration.\n" +
|
||||
"# A merging proxy presenting one PuppetDB v4 query surface over several\n" +
|
||||
"# PuppetDB backends. The backend URLs below are placeholders — edit them.\n" +
|
||||
"# Env overrides: PDBMUX_LISTEN, PDBMUX_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 {
|
||||
|
||||
+71
-15
@@ -3,26 +3,69 @@ package main
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestLoad_Defaults(t *testing.T) {
|
||||
// testConfigValid returns a minimal valid config for Validate/merge tests, with
|
||||
// neutral placeholder backends.
|
||||
func testConfigValid() Config {
|
||||
cfg := DefaultConfig()
|
||||
cfg.Backends = []Backend{
|
||||
{Name: "old", URL: "http://localhost:18080"},
|
||||
{Name: "new", URL: "http://localhost:18081"},
|
||||
}
|
||||
cfg.Primary = "new"
|
||||
cfg.Prefer = "new"
|
||||
return cfg
|
||||
}
|
||||
|
||||
func TestDefaultConfig_NoBackends(t *testing.T) {
|
||||
cfg := DefaultConfig()
|
||||
if cfg.Listen != defaultListen {
|
||||
t.Errorf("listen = %q, want %q", cfg.Listen, defaultListen)
|
||||
}
|
||||
if len(cfg.Backends) != 0 {
|
||||
t.Errorf("defaults must not ship backends, got %+v", cfg.Backends)
|
||||
}
|
||||
if cfg.Merge != mergeFreshness {
|
||||
t.Errorf("merge = %q, want %q", cfg.Merge, mergeFreshness)
|
||||
}
|
||||
if err := cfg.Validate(); err == nil {
|
||||
t.Error("defaults alone must not validate: backends are required")
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoad_NoBackendsLoadsButFailsValidation(t *testing.T) {
|
||||
t.Setenv("XDG_CONFIG_HOME", t.TempDir())
|
||||
clearEnv(t)
|
||||
|
||||
// Load itself must succeed so `config init` / `version` work unconfigured.
|
||||
cfg, err := Load()
|
||||
if err != nil {
|
||||
t.Fatalf("load: %v", err)
|
||||
}
|
||||
err = cfg.Validate()
|
||||
if err == nil {
|
||||
t.Fatal("expected a validation error when no backends are configured")
|
||||
}
|
||||
if !strings.Contains(err.Error(), envPrefix+"BACKENDS") {
|
||||
t.Errorf("error should name the env var to set, got: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoad_PrimaryDefaultsToFirstBackend(t *testing.T) {
|
||||
t.Setenv("XDG_CONFIG_HOME", t.TempDir())
|
||||
clearEnv(t)
|
||||
t.Setenv(envPrefix+"BACKENDS", "a=http://localhost:18080,b=http://localhost:18081")
|
||||
|
||||
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)
|
||||
if cfg.Primary != "a" || cfg.Prefer != "a" {
|
||||
t.Errorf("primary/prefer should default to first backend, got %q/%q", cfg.Primary, cfg.Prefer)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -35,7 +78,8 @@ func TestLoad_FileAndEnvOverride(t *testing.T) {
|
||||
if err := os.MkdirAll(cfgDir, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
body := "listen: :9999\nmerge: static\nprimary: old\nprefer: old\n"
|
||||
body := "listen: :9999\nmerge: static\nprimary: old\nprefer: old\n" +
|
||||
"backends:\n - name: old\n url: http://localhost:18080\n - name: new\n url: http://localhost:18081\n"
|
||||
if err := os.WriteFile(filepath.Join(cfgDir, configFileName), []byte(body), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -56,7 +100,7 @@ func TestLoad_FileAndEnvOverride(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestApplyEnv_Backends(t *testing.T) {
|
||||
cfg := DefaultConfig()
|
||||
cfg := testConfigValid()
|
||||
env := map[string]string{
|
||||
envPrefix + "BACKENDS": "a=http://a:8080,b=http://b:8080",
|
||||
envPrefix + "PRIMARY": "a",
|
||||
@@ -91,7 +135,7 @@ func TestValidate(t *testing.T) {
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
cfg := DefaultConfig()
|
||||
cfg := testConfigValid()
|
||||
tc.mutate(&cfg)
|
||||
err := cfg.Validate()
|
||||
if (err != nil) != tc.wantErr {
|
||||
@@ -112,9 +156,21 @@ func TestParseBackends(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestPrimaryBackend(t *testing.T) {
|
||||
cfg := DefaultConfig()
|
||||
if cfg.PrimaryBackend().URL != defaultNewURL {
|
||||
t.Errorf("primary backend URL = %q, want %q", cfg.PrimaryBackend().URL, defaultNewURL)
|
||||
cfg := testConfigValid()
|
||||
if got, want := cfg.PrimaryBackend().URL, cfg.Backends[1].URL; got != want {
|
||||
t.Errorf("primary backend URL = %q, want %q", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExampleConfig_IsValidAndNeutral(t *testing.T) {
|
||||
cfg := ExampleConfig()
|
||||
if err := cfg.Validate(); err != nil {
|
||||
t.Fatalf("example config must validate: %v", err)
|
||||
}
|
||||
for _, b := range cfg.Backends {
|
||||
if !strings.Contains(b.URL, "example.com") {
|
||||
t.Errorf("example backend %q must use a placeholder host, got %q", b.Name, b.URL)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
// Command pdbmux serves one merged PuppetDB v4 query surface over two PuppetDBs.
|
||||
// Command pdbmux is a small merging HTTP proxy over several PuppetDB backends.
|
||||
package main
|
||||
|
||||
import (
|
||||
@@ -48,11 +48,11 @@ func main() {
|
||||
|
||||
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,\n" +
|
||||
"pblastreport and Puppetboard see one consistent view. Running pdbmux with\n" +
|
||||
"no subcommand (or `pdbmux serve`) starts the proxy.",
|
||||
Short: "Merging HTTP proxy over several PuppetDB backends.",
|
||||
Long: "pdbmux presents a single merged PuppetDB v4 query surface over several\n" +
|
||||
"PuppetDB backends, so clients see one consistent view of nodes, facts and\n" +
|
||||
"reports spanning all of them. Running pdbmux with no subcommand (or\n" +
|
||||
"`pdbmux serve`) starts the proxy.",
|
||||
SilenceUsage: true,
|
||||
RunE: func(cmd *cobra.Command, args []string) error { return serve(cmd) },
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user