unkin/pdbmux
4
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Serve fact queries live and drop the cache headers #26

Merged
benvin merged 2 commits from benvin/drop-fact-caching into main 2026-09-13 15:04:26 +10:00
Conversation 1 Commits 2 Files Changed 6
+263 -323
6 changed files with 263 additions and 323 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+32 -34
View File
@@ -75,9 +75,10 @@ reporting their own fault, so it is replayed just as faithfully but still counts
as a failed round and still falls back to a stale cache entry where there is one.
Responses carry PuppetDB's `X-Records` when the query asked for a total, and on
the merged paths `X-Backends` (see [Backend health](#backend-health)). Cached
paths add two more headers `pdbmux` sets itself, `X-Cache` and `Age` — see
[Caching](#caching).
the merged paths `X-Backends` (see [Backend health](#backend-health)). Successful
query responses carry PuppetDB's own `Content-Type: application/json;charset=utf-8`.
Nothing tells a client whether a response came from the cache: PuppetDB sets no
`X-Cache` or `Age`, so neither does `pdbmux` — see [Caching](#caching).
## Merge semantics
@@ -258,7 +259,7 @@ record of that name, so the route does not serve its own fan-out: it takes the
records from the `/facts` merge that produces them, which makes the `certname`
set, the owner and the `environment` identical to the ones an unfiltered `/facts`
response reports, and lets the request's own `query` narrow the result upstream.
That costs one `/facts` fan-out per cache miss — the widest fan-out `pdbmux`
That costs one `/facts` fan-out per request — the widest fan-out `pdbmux`
makes — on a rare, user-initiated path.
`/facts/pdbmux_source/<value>` pins the backend name, so it answers with the
@@ -419,8 +420,8 @@ not answering.
asked, never what a merged answer means: a response built from a subset is
still served, as before. Every merged response carries `X-Backends:
<contributed>/<configured>` naming how many backends' records went into it, so
a client can tell a full answer from a partial one. On a cache hit the header
describes the stored body, not the current backend count.
a client can tell a full answer from a partial one. On a `/nodes` cache hit the
header describes the stored body, not the current backend count.
- **`/healthz`** gives each backend a `state` (`healthy`, `unhealthy`,
`probe_unsupported`, `unprobed`, or `unmonitored` when probing is off),
`consecutive_failures`,
@@ -440,21 +441,27 @@ not answering.
## Caching
`pdbmux` caches merged `/nodes`, `/facts`, `/facts/<name>[/<value>]` and
`/fact-names` record sets **in memory** so a busy Puppetboard does not re-fan-out
the same query every few seconds — its facts overview and fact drilldown are two
of the pages that hit hardest. Everything else runs uncached — including
`extract` aggregates on those paths, and
the `/pdb/meta/v1/*` and `/metrics/*` endpoints, which are served live on every
request. The cache is an interface, and `/reports` gets its own (S3-backed)
backend later without further handler changes.
`pdbmux` caches the merged `/nodes` record set **in memory** so a busy
Puppetboard does not re-fan-out the node list every few seconds. Its report
columns only move when a node finishes a run, so a `30s` answer is a `30s`-old
report timestamp and nothing else.
**No fact-serving path is cached.** `/facts`, `/facts/<name>[/<value>]`,
`/fact-names`, `/factsets*`, `/nodes/<certname>/facts`, `/fact-contents`,
`/fact-paths` and `/inventory` all fan out on every request, so a fact answer is
exactly as current as the backend's own. Cache fact data client-side if you want
it cached. Everything else runs uncached too — including `extract` aggregates on
`/nodes`, and the `/pdb/meta/v1/*` and `/metrics/*` endpoints. The cache is an
interface, and `/reports` gets its own (S3-backed) backend later without further
handler changes.
- **Key** — `<path>?<params>`, where the params are the ones that actually
determine the response, URL-encoded with keys sorted ascending and a repeated
param's values sorted ascending. Param order in the request is therefore
irrelevant: one canonical key per distinct request. A request with no params
keys on the bare path.
- **TTL** — `facts_ttl`, default `30s`, **hard cap `30s`**. A larger configured
- **TTL** — `facts_ttl` (the key predates the cache narrowing to `/nodes`),
default `30s`, **hard cap `30s`**. A larger configured
value is **clamped** down to the cap, not rejected, so a stray env var cannot
crash-loop a container; `pdbmux config show` prints
`facts_ttl : 30s (clamped from 600s, cap 30s)` when that happens. `facts_ttl: 0`
@@ -478,29 +485,20 @@ backend later without further handler changes.
client goes away leaves the flight running for the others. The flight is
cancelled once its last participant leaves, so a lone client disconnecting
releases the upstream connections straight away.
- **Response headers** — every response on a cached path carries `X-Cache`
(`hit` served from a fresh entry, `miss` built by this request, `stale` the
expired-entry fallback) and `Age` in whole seconds since the served copy was
stored (`0` on a `miss`). Uncached paths carry neither.
- **Invisible to clients** — a cached response is byte-for-byte a live one, with
the same headers. PuppetDB emits no `X-Cache` and no `Age`, so neither does
`pdbmux`; read the cache state off `/healthz` instead.
- **Visibility** — `/healthz` carries a `cache` object: `backend`
(`memory`/`none`), `ttl`, `entries`, `stale_entries`, `bytes`, `serving_stale`,
`stale_served` and `last_stale_served`. `serving_stale` is `true` from the
moment a stale fallback is served until the next response comes from a live
fan-out or a fresh entry.
- **Provenance is stored, not re-applied** — what a cache entry holds is the
fully merged body, `pdbmux_source` already injected and upstream records of
that name already dropped. Attribution names the backend that supplied the
data, which is a property of that fetch, so it stays correct for as long as the
body does and ages out with it — `X-Cache` and `Age` say how old both are. Two
requests can only share an entry when they share a key, and the key is path
plus query, which is exactly what decides whether injection applies; a
name-filtered `/facts` query and a plain one therefore cache separately and
neither is ever served the other's shape. The one path that keys on less than
it is asked is the `pdbmux_source` drilldown, whose `<value>` is dropped from
the key and applied to the shared entry instead. `source_fact` and
`source_fact_enabled` are read once at startup, and the cache lives for the
same process, so changing either cannot leave differently-shaped entries
behind.
- **Merged bodies are stored, not re-merged** — an entry holds the fully merged
`/nodes` body, `pdbmux_source` already stamped on, and the
`X-Records`/`X-Backends` it was built with, so a hit reproduces all three.
Attribution is a property of that fetch, so it stays correct for as long as the
body does and ages out with it. Two requests share an entry only when they
share a key, and the key is path plus query.
## Config
@@ -526,7 +524,7 @@ backends: # order is a tie-break only, not a ranking
merge: freshness # freshness | static
timeout: 10s # per-upstream request timeout
freshness_ttl: 30s # freshness-map cache TTL (freshness merge only)
facts_ttl: 30s # /facts + /nodes response cache TTL; 0 disables, capped at 30s
facts_ttl: 30s # merged /nodes response cache TTL; 0 disables, capped at 30s
facts_cache_bytes: 67108864 # byte budget for that cache (64 MiB), LRU-evicted
source_fact: pdbmux_source # name of the synthetic provenance fact
source_fact_enabled: true # false serves backends' records untouched
cache_test.go
+165 -204
View File
@@ -136,13 +136,13 @@ func (cb *countingBackend) setBlock(ch chan struct{}) {
cb.block = ch
}
// newCachedServer builds a server whose facts cache runs on a fake clock.
// newCachedServer builds a server whose /nodes cache runs on a fake clock.
func newCachedServer(t *testing.T, cfg Config) (*Server, *fakeClock) {
t.Helper()
srv := newTestServer(cfg)
mc, ok := srv.factsCache.(*memoryCache)
mc, ok := srv.nodeCache.(*memoryCache)
if !ok {
t.Fatalf("expected a memory cache, got %T", srv.factsCache)
t.Fatalf("expected a memory cache, got %T", srv.nodeCache)
}
clk := newFakeClock()
mc.now = clk.now
@@ -277,14 +277,14 @@ func TestNoopCache_AlwaysMisses(t *testing.T) {
}
}
func TestHandler_FactsCacheFreshHit(t *testing.T) {
body := `[` + fact("h1", "role", "web", "") + `]`
a := newCountingBackend(t, map[string]string{factsPath: body})
b := newCountingBackend(t, map[string]string{factsPath: `[]`})
func TestHandler_NodesCacheFreshHit(t *testing.T) {
body := `[` + node("h1", "2026-01-01T00:00:00.000Z") + `]`
a := newCountingBackend(t, map[string]string{nodesPath: body})
b := newCountingBackend(t, map[string]string{nodesPath: `[]`})
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
first := doGet(t, srv.Handler(), factsPath, `["=","name","role"]`)
second := doGet(t, srv.Handler(), factsPath, `["=","name","role"]`)
first := doGet(t, srv.Handler(), nodesPath, `["=","certname","h1"]`)
second := doGet(t, srv.Handler(), nodesPath, `["=","certname","h1"]`)
if first.Code != http.StatusOK || second.Code != http.StatusOK {
t.Fatalf("statuses %d/%d", first.Code, second.Code)
@@ -292,40 +292,40 @@ func TestHandler_FactsCacheFreshHit(t *testing.T) {
if first.Body.String() != second.Body.String() {
t.Errorf("cache hit changed the body:\n %s\n %s", first.Body.String(), second.Body.String())
}
if got := a.hitCount(factsPath); got != 1 {
if got := a.hitCount(nodesPath); got != 1 {
t.Errorf("backend a saw %d requests, want 1 (second served from cache)", got)
}
if got := b.hitCount(factsPath); got != 1 {
if got := b.hitCount(nodesPath); got != 1 {
t.Errorf("backend b saw %d requests, want 1", got)
}
// A different query is a different key and must go upstream.
doGet(t, srv.Handler(), factsPath, `["=","name","osfamily"]`)
if got := a.hitCount(factsPath); got != 2 {
doGet(t, srv.Handler(), nodesPath, `["=","certname","h2"]`)
if got := a.hitCount(nodesPath); got != 2 {
t.Errorf("a different query should refetch: %d requests, want 2", got)
}
}
func TestHandler_FactsCacheExpires(t *testing.T) {
a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "role", "web", "") + `]`})
b := newCountingBackend(t, map[string]string{factsPath: `[]`})
func TestHandler_NodesCacheExpires(t *testing.T) {
a := newCountingBackend(t, map[string]string{nodesPath: `[` + node("h1", "2026-01-01T00:00:00.000Z") + `]`})
b := newCountingBackend(t, map[string]string{nodesPath: `[]`})
srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
doGet(t, srv.Handler(), factsPath, "")
doGet(t, srv.Handler(), nodesPath, "")
clk.advance(31 * time.Second)
doGet(t, srv.Handler(), factsPath, "")
doGet(t, srv.Handler(), nodesPath, "")
if got := a.hitCount(factsPath); got != 2 {
if got := a.hitCount(nodesPath); got != 2 {
t.Errorf("an expired entry should refetch: %d requests, want 2", got)
}
}
func TestHandler_ServesStaleOnlyWhenBackendsFail(t *testing.T) {
a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "role", "old", "") + `]`})
b := newCountingBackend(t, map[string]string{factsPath: `[]`})
a := newCountingBackend(t, map[string]string{nodesPath: `[` + node("old", "2026-01-01T00:00:00.000Z") + `]`})
b := newCountingBackend(t, map[string]string{nodesPath: `[]`})
srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
warm := doGet(t, srv.Handler(), factsPath, "")
warm := doGet(t, srv.Handler(), nodesPath, "")
if warm.Code != http.StatusOK {
t.Fatalf("warm-up status %d", warm.Code)
}
@@ -335,8 +335,8 @@ func TestHandler_ServesStaleOnlyWhenBackendsFail(t *testing.T) {
// Backends healthy but the entry expired: fresh data wins, never the stale copy.
clk.advance(31 * time.Second)
a.setBody(factsPath, `[`+fact("h1", "role", "new", "")+`]`)
refetch := doGet(t, srv.Handler(), factsPath, "")
a.setBody(nodesPath, `[`+node("new", "2026-01-01T00:00:00.000Z")+`]`)
refetch := doGet(t, srv.Handler(), nodesPath, "")
if !strings.Contains(refetch.Body.String(), `"new"`) {
t.Errorf("a healthy backend must not be shadowed by the stale entry: %s", refetch.Body.String())
}
@@ -348,7 +348,7 @@ func TestHandler_ServesStaleOnlyWhenBackendsFail(t *testing.T) {
clk.advance(31 * time.Second)
a.setFail(true)
b.setFail(true)
stale := doGet(t, srv.Handler(), factsPath, "")
stale := doGet(t, srv.Handler(), nodesPath, "")
if stale.Code != http.StatusOK {
t.Fatalf("stale fallback status %d: %s", stale.Code, stale.Body.String())
}
@@ -362,51 +362,14 @@ func TestHandler_ServesStaleOnlyWhenBackendsFail(t *testing.T) {
}
}
// The stored entry is the whole estate's record set and the pinned value narrows
// it per request, so the fallback has to keep narrowing: a client asking for one
// backend's records must not be handed every backend's because the entry expired.
func TestHandler_StaleSourceFactDrilldownStaysFilteredByOwner(t *testing.T) {
a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "osfamily", "RedHat", "") + `]`})
b := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h2", "osfamily", "Debian", "") + `]`})
srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
drilldown := sourceFactURL + "/a"
want := map[string]string{"h1": "a"}
warm := doGet(t, srv.Handler(), drilldown, "")
if warm.Code != http.StatusOK {
t.Fatalf("warm-up status %d: %s", warm.Code, warm.Body.String())
}
if got, n := sourceValues(t, warm.Body.Bytes(), defaultSourceFact); n != len(want) || !reflect.DeepEqual(got, want) {
t.Fatalf("warm-up %s = %v (%d records), want %v", drilldown, got, n, want)
}
// Every backend down and the entry expired: the stale copy is served.
clk.advance(31 * time.Second)
a.setFail(true)
b.setFail(true)
rec := doGet(t, srv.Handler(), drilldown, "")
if rec.Code != http.StatusOK {
t.Fatalf("stale fallback status %d: %s", rec.Code, rec.Body.String())
}
if got := rec.Header().Get(cacheStatusHeader); got != "stale" {
t.Fatalf("%s = %q, want stale: the request did not take the fallback path", cacheStatusHeader, got)
}
got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact)
if n != len(want) || !reflect.DeepEqual(got, want) {
t.Errorf("stale %s = %v (%d records), want only backend a's %v", drilldown, got, n, want)
}
}
func TestHandler_NoCacheEntryMeansBackendFailureIs502(t *testing.T) {
a := newCountingBackend(t, map[string]string{factsPath: `[]`})
b := newCountingBackend(t, map[string]string{factsPath: `[]`})
a := newCountingBackend(t, map[string]string{nodesPath: `[]`})
b := newCountingBackend(t, map[string]string{nodesPath: `[]`})
a.setDead(true)
b.setDead(true)
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
rec := doGet(t, srv.Handler(), factsPath, "")
rec := doGet(t, srv.Handler(), nodesPath, "")
if rec.Code != http.StatusBadGateway {
t.Fatalf("status %d, want 502", rec.Code)
}
@@ -416,8 +379,8 @@ func TestHandler_NoCacheEntryMeansBackendFailureIs502(t *testing.T) {
}
func TestHandler_SingleFlightCollapsesConcurrentRequests(t *testing.T) {
a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "role", "web", "") + `]`})
b := newCountingBackend(t, map[string]string{factsPath: `[]`})
a := newCountingBackend(t, map[string]string{nodesPath: `[` + node("h1", "2026-01-01T00:00:00.000Z") + `]`})
b := newCountingBackend(t, map[string]string{nodesPath: `[]`})
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
h := srv.Handler()
@@ -432,13 +395,13 @@ func TestHandler_SingleFlightCollapsesConcurrentRequests(t *testing.T) {
wg.Add(1)
go func(i int) {
defer wg.Done()
codes[i] = doGet(t, h, factsPath, "").Code
codes[i] = doGet(t, h, nodesPath, "").Code
}(i)
}
// The leader is parked inside the blocked backend, so every caller that
// reaches the handler before the release joins its flight.
waitFor(t, func() bool { return a.hitCount(factsPath) >= 1 })
waitFor(t, func() bool { return a.hitCount(nodesPath) >= 1 })
time.Sleep(250 * time.Millisecond)
close(release)
wg.Wait()
@@ -448,10 +411,10 @@ func TestHandler_SingleFlightCollapsesConcurrentRequests(t *testing.T) {
t.Fatalf("caller %d got %d", i, code)
}
}
if got := a.hitCount(factsPath); got != 1 {
if got := a.hitCount(nodesPath); got != 1 {
t.Errorf("backend a saw %d requests, want 1 for %d concurrent callers", got, callers)
}
if got := b.hitCount(factsPath); got != 1 {
if got := b.hitCount(nodesPath); got != 1 {
t.Errorf("backend b saw %d requests, want 1 for %d concurrent callers", got, callers)
}
}
@@ -567,15 +530,15 @@ func TestFlightGroup_LeaderPanicFailsLeaderAndWaiters(t *testing.T) {
}
func TestServeCached_BuildPanicIs502ThenStale(t *testing.T) {
a := newCountingBackend(t, map[string]string{factsPath: `[]`})
b := newCountingBackend(t, map[string]string{factsPath: `[]`})
a := newCountingBackend(t, map[string]string{nodesPath: `[]`})
b := newCountingBackend(t, map[string]string{nodesPath: `[]`})
srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
panicBuild := func(context.Context) (cachedResponse, error) { panic("build exploded") }
serve := func(build func(context.Context) (cachedResponse, error)) *httptest.ResponseRecorder {
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, factsPath, nil)
srv.serveCached(rec, req, factsPath, nil, build)
req := httptest.NewRequest(http.MethodGet, nodesPath, nil)
srv.serveCached(rec, req, nodesPath, nil, build)
return rec
}
@@ -585,7 +548,7 @@ func TestServeCached_BuildPanicIs502ThenStale(t *testing.T) {
t.Fatalf("status %d (%s), want 502", rec.Code, rec.Body.String())
}
stored := `[` + fact("h1", "role", "web", "") + `]`
stored := `[` + node("h1", "2026-01-01T00:00:00.000Z") + `]`
rec = serve(func(context.Context) (cachedResponse, error) {
return cachedResponse{Body: json.RawMessage(stored), Records: 1}, nil
})
@@ -610,16 +573,16 @@ func TestServeCached_BuildPanicIs502ThenStale(t *testing.T) {
}
func TestServeCached_BuildPanicFailsConcurrentWaiters(t *testing.T) {
a := newCountingBackend(t, map[string]string{factsPath: `[]`})
b := newCountingBackend(t, map[string]string{factsPath: `[]`})
a := newCountingBackend(t, map[string]string{nodesPath: `[]`})
b := newCountingBackend(t, map[string]string{nodesPath: `[]`})
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
entered := make(chan struct{})
release := make(chan struct{})
serve := func(build func(context.Context) (cachedResponse, error)) *httptest.ResponseRecorder {
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, factsPath, nil)
srv.serveCached(rec, req, factsPath, nil, build)
req := httptest.NewRequest(http.MethodGet, nodesPath, nil)
srv.serveCached(rec, req, nodesPath, nil, build)
return rec
}
@@ -720,8 +683,8 @@ func TestHandler_LeaderDisconnectDoesNotFailFollowers(t *testing.T) {
// The flight the leader started still populated the cache.
warmed := doGet(t, h, nodesPath, "")
if got := warmed.Header().Get(cacheStatusHeader); got != "hit" {
t.Errorf("%s = %q, want hit: the abandoned leader's flight must still warm the cache", cacheStatusHeader, got)
if got := strings.TrimSpace(warmed.Body.String()); !sameJSON(t, got, want) {
t.Errorf("warmed body = %s, want %s: the abandoned leader's flight must still warm the cache", got, want)
}
if got := a.hitCount(nodesPath); got != 1 {
t.Errorf("backend a saw %d requests after the cached read, want 1", got)
@@ -760,7 +723,7 @@ func TestHandler_SoloDisconnectAbortsFanOutPromptly(t *testing.T) {
}()
start := time.Now()
req := httptest.NewRequest(http.MethodGet, factsPath, nil).WithContext(ctx)
req := httptest.NewRequest(http.MethodGet, nodesPath, nil).WithContext(ctx)
srv.Handler().ServeHTTP(httptest.NewRecorder(), req)
if elapsed := time.Since(start); elapsed >= cfg.Timeout/2 {
t.Errorf("handler returned after %s, want well under cfg.Timeout %s", elapsed, cfg.Timeout)
@@ -793,7 +756,7 @@ func TestHandler_DisconnectedRequestsDoNotPinBackends(t *testing.T) {
time.Sleep(disconnectAfter)
cancel()
}()
target := factsPath + "?query=" + url.QueryEscape(strconv.Itoa(i))
target := nodesPath + "?query=" + url.QueryEscape(strconv.Itoa(i))
req := httptest.NewRequest(http.MethodGet, target, nil).WithContext(ctx)
h.ServeHTTP(httptest.NewRecorder(), req)
}(i)
@@ -1126,47 +1089,39 @@ func assertGoroutinesSettle(t *testing.T, baseline int, frames ...string) {
}
}
func TestServeCached_CacheStatusHeaders(t *testing.T) {
stored := `[` + fact("h1", "role", "web", "") + `]`
a := newCountingBackend(t, map[string]string{factsPath: stored})
b := newCountingBackend(t, map[string]string{factsPath: `[]`})
// PuppetDB emits no X-Cache and no Age, so a client cannot tell a cached
// response from a live one. Covers the fresh hit, the miss and the stale
// fallback, which are the three ways a response can leave the cached path.
func TestHandler_CachedResponsesCarryNoCacheHeaders(t *testing.T) {
a := newCountingBackend(t, map[string]string{nodesPath: `[` + node("h1", "2026-01-01T00:00:00.000Z") + `]`})
b := newCountingBackend(t, map[string]string{nodesPath: `[]`})
srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
h := srv.Handler()
rec := doGet(t, h, factsPath, "")
if got := rec.Header().Get(cacheStatusHeader); got != "miss" {
t.Errorf("first request %s = %q, want miss", cacheStatusHeader, got)
}
if got := rec.Header().Get(ageHeader); got != "0" {
t.Errorf("first request %s = %q, want 0", ageHeader, got)
}
assertNoCacheHeaders(t, doGet(t, h, nodesPath, ""), "miss")
clk.advance(7 * time.Second)
rec = doGet(t, h, factsPath, "")
if got := rec.Header().Get(cacheStatusHeader); got != "hit" {
t.Errorf("cached request %s = %q, want hit", cacheStatusHeader, got)
}
if got := rec.Header().Get(ageHeader); got != "7" {
t.Errorf("cached request %s = %q, want 7", ageHeader, got)
}
assertNoCacheHeaders(t, doGet(t, h, nodesPath, ""), "fresh hit")
// Past the TTL with every backend down, the stale fallback must say so.
clk.advance(24 * time.Second)
a.setFail(true)
b.setFail(true)
rec = doGet(t, h, factsPath, "")
rec := doGet(t, h, nodesPath, "")
if rec.Code != http.StatusOK {
t.Fatalf("stale fallback status %d (%s)", rec.Code, rec.Body.String())
}
if got := rec.Header().Get(cacheStatusHeader); got != "stale" {
t.Errorf("stale fallback %s = %q, want stale", cacheStatusHeader, got)
if serving, _, _ := srv.stale.snapshot(); !serving {
t.Fatal("the third request did not take the stale fallback path")
}
if got := rec.Header().Get(ageHeader); got != "31" {
t.Errorf("stale fallback %s = %q, want 31 seconds since the entry was stored", ageHeader, got)
}
want := `[` + fact("h1", "role", "web", "") + `,` + factEnv("h1", defaultSourceFact, "a", "") + `]`
if got := strings.TrimSpace(rec.Body.String()); !sameJSON(t, got, want) {
t.Errorf("stale body = %s, want %s", got, want)
assertNoCacheHeaders(t, rec, "stale fallback")
}
func assertNoCacheHeaders(t *testing.T, rec *httptest.ResponseRecorder, what string) {
t.Helper()
for _, h := range []string{"X-Cache", "Age"} {
if got := rec.Header().Get(h); got != "" {
t.Errorf("%s response carries %s: %q, want no such header", what, h, got)
}
}
}
@@ -1203,18 +1158,18 @@ func (c *recordingCache) Stats() CacheStats { return CacheStats{Backend: "record
func TestServeCached_PutRunsOnDetachedContext(t *testing.T) {
srv := newTestServer(cacheTestConfig("http://backend.invalid", "http://backend.invalid"))
cache := newRecordingCache()
srv.factsCache = cache
srv.nodeCache = cache
entered := make(chan struct{})
release := make(chan struct{})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
req := httptest.NewRequest(http.MethodGet, factsPath, nil).WithContext(ctx)
req := httptest.NewRequest(http.MethodGet, nodesPath, nil).WithContext(ctx)
served := make(chan struct{})
go func() {
defer close(served)
srv.serveCached(httptest.NewRecorder(), req, factsPath, nil, func(context.Context) (cachedResponse, error) {
srv.serveCached(httptest.NewRecorder(), req, nodesPath, nil, func(context.Context) (cachedResponse, error) {
close(entered)
<-release
return cachedResponse{Body: json.RawMessage(`[]`), Records: -1}, nil
@@ -1246,8 +1201,8 @@ func TestServeCached_PutRunsOnDetachedContext(t *testing.T) {
}
func TestHandler_HealthzReportsCacheState(t *testing.T) {
a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "role", "web", "") + `]`})
b := newCountingBackend(t, map[string]string{factsPath: `[]`})
a := newCountingBackend(t, map[string]string{nodesPath: `[` + node("h1", "2026-01-01T00:00:00.000Z") + `]`})
b := newCountingBackend(t, map[string]string{nodesPath: `[]`})
srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
hr := health(t, srv)
@@ -1258,7 +1213,7 @@ func TestHandler_HealthzReportsCacheState(t *testing.T) {
t.Fatalf("a cold cache should be empty and not stale: %+v", hr.Cache)
}
doGet(t, srv.Handler(), factsPath, "")
doGet(t, srv.Handler(), nodesPath, "")
hr = health(t, srv)
if hr.Cache.Entries != 1 || hr.Cache.StaleEntries != 0 || hr.Cache.Bytes == 0 {
t.Fatalf("after one request: %+v", hr.Cache)
@@ -1274,7 +1229,7 @@ func TestHandler_HealthzReportsCacheState(t *testing.T) {
a.setFail(true)
b.setFail(true)
doGet(t, srv.Handler(), factsPath, "")
doGet(t, srv.Handler(), nodesPath, "")
hr = health(t, srv)
if !hr.Cache.ServingStale || hr.Cache.StaleServed != 1 || hr.Cache.LastStale == "" {
t.Errorf("staleness not surfaced in /healthz: %+v", hr.Cache)
@@ -1454,9 +1409,9 @@ func TestNewServer_ClampsFactsTTL(t *testing.T) {
cfg.CacheBytes = 1 << 20
srv := newTestServer(cfg)
mc, ok := srv.factsCache.(*memoryCache)
mc, ok := srv.nodeCache.(*memoryCache)
if !ok {
t.Fatalf("expected a memory cache, got %T", srv.factsCache)
t.Fatalf("expected a memory cache, got %T", srv.nodeCache)
}
if mc.ttl != maxFactsTTL {
t.Errorf("cache ttl = %s, want %s", mc.ttl, maxFactsTTL)
@@ -1530,35 +1485,6 @@ func stamped(t *testing.T, raw, field, value string) string {
// The injector is per-request but a cache entry is shared, so a second caller is
// served a body built for the first. Provenance names the backend that supplied
// the data, which is a property of that fetch, so the shared body stays correct.
func TestHandler_CachedFactsKeepSourceAttribution(t *testing.T) {
a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "role", "web", "") + `]`})
b := newCountingBackend(t, map[string]string{factsPath: `[]`})
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
h := srv.Handler()
first := doGet(t, h, factsPath, "")
if got := first.Header().Get(cacheStatusHeader); got != "miss" {
t.Fatalf("first %s = %q, want miss", cacheStatusHeader, got)
}
srcs, n := sourceValues(t, first.Body.Bytes(), defaultSourceFact)
if n != 1 || srcs["h1"] != "a" {
t.Fatalf("first request sources = %v (%d records), want h1 -> a", srcs, n)
}
second := doGet(t, h, factsPath, "")
if got := second.Header().Get(cacheStatusHeader); got != "hit" {
t.Fatalf("second %s = %q, want hit", cacheStatusHeader, got)
}
srcs, n = sourceValues(t, second.Body.Bytes(), defaultSourceFact)
if n != 1 || srcs["h1"] != "a" {
t.Errorf("cached request sources = %v (%d records), want exactly one h1 -> a", srcs, n)
}
if got := a.hitCount(factsPath); got != 1 {
t.Errorf("backend a saw %d requests, want 1: the second read must come from the cache", got)
}
}
// The same, for the /nodes stamp rather than the synthetic /facts record.
func TestHandler_CachedNodesKeepSourceStamp(t *testing.T) {
a := newCountingBackend(t, map[string]string{nodesPath: `[` + node("h1", "2026-01-02T00:00:00.000Z") + `]`})
b := newCountingBackend(t, map[string]string{nodesPath: `[` + node("h1", "2026-01-01T00:00:00.000Z") + `]`})
@@ -1571,9 +1497,6 @@ func TestHandler_CachedNodesKeepSourceStamp(t *testing.T) {
}
second := doGet(t, h, nodesPath, "")
if got := second.Header().Get(cacheStatusHeader); got != "hit" {
t.Fatalf("second %s = %q, want hit", cacheStatusHeader, got)
}
want := `[` + stamped(t, node("h1", "2026-01-02T00:00:00.000Z"), defaultSourceFact, "a") + `]`
if got := strings.TrimSpace(second.Body.String()); !sameJSON(t, got, want) {
t.Errorf("cached body = %s, want %s", got, want)
@@ -1587,77 +1510,115 @@ func TestHandler_CachedNodesKeepSourceStamp(t *testing.T) {
// while the entry is served the attribution is the one that fetch had, and the
// rebuild after the TTL picks up the move.
func TestHandler_CachedSourceAgesWithItsData(t *testing.T) {
a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "role", "web", "") + `]`})
b := newCountingBackend(t, map[string]string{factsPath: `[]`})
a := newCountingBackend(t, map[string]string{nodesPath: `[` + node("h1", "2026-01-01T00:00:00.000Z") + `]`})
b := newCountingBackend(t, map[string]string{nodesPath: `[]`})
srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
h := srv.Handler()
doGet(t, h, factsPath, "")
doGet(t, h, nodesPath, "")
// The node moves to b while the entry is still fresh.
a.setBody(factsPath, `[]`)
b.setBody(factsPath, `[`+fact("h1", "role", "web", "")+`]`)
a.setBody(nodesPath, `[]`)
b.setBody(nodesPath, `[`+node("h1", "2026-01-01T00:00:00.000Z")+`]`)
cached := doGet(t, h, factsPath, "")
if got := cached.Header().Get(cacheStatusHeader); got != "hit" {
t.Fatalf("%s = %q, want hit", cacheStatusHeader, got)
cached := doGet(t, h, nodesPath, "")
if got := nodeSources(t, cached.Body.Bytes(), defaultSourceFact); got["h1"] != "a" {
t.Errorf("cached stamp = %v, want h1 -> a: the body and its attribution come from the same fetch", got)
}
if srcs, _ := sourceValues(t, cached.Body.Bytes(), defaultSourceFact); srcs["h1"] != "a" {
t.Errorf("cached sources = %v, want h1 -> a: the body and its attribution come from the same fetch", srcs)
if got := a.hitCount(nodesPath); got != 1 {
t.Errorf("backend a saw %d requests, want 1 while the entry is fresh", got)
}
clk.advance(31 * time.Second)
rebuilt := doGet(t, h, factsPath, "")
if srcs, _ := sourceValues(t, rebuilt.Body.Bytes(), defaultSourceFact); srcs["h1"] != "b" {
t.Errorf("rebuilt sources = %v, want h1 -> b once the entry expired", srcs)
rebuilt := doGet(t, h, nodesPath, "")
if got := nodeSources(t, rebuilt.Body.Bytes(), defaultSourceFact); got["h1"] != "b" {
t.Errorf("rebuilt stamp = %v, want h1 -> b once the entry expired", got)
}
}
// The injection gate is a pure function of path and query, both of which are in
// the cache key, so a gated request can never be served an injected body cached
// for an ungated one.
func TestHandler_SourceGateIsPerCacheKey(t *testing.T) {
// Fact answers must be as current as a backend's own, so every fact-serving
// route refetches: the second request sees data that changed between the two.
func TestHandler_FactPathsAreNeverCached(t *testing.T) {
for _, path := range []string{
factsPath,
factsPath + "/role",
factsPath + "/role/web",
factNamesPath,
factsetsPath,
nodesPath + "/h1/facts",
"/pdb/query/v4/fact-contents",
"/pdb/query/v4/fact-paths",
"/pdb/query/v4/inventory",
} {
t.Run(path, func(t *testing.T) {
before := `[{"certname":"h1","name":"role","value":"before"}]`
after := `[{"certname":"h1","name":"role","value":"after"}]`
a := newCountingBackend(t, map[string]string{path: before})
b := newCountingBackend(t, map[string]string{path: `[]`})
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
h := srv.Handler()
if first := doGet(t, h, path, ""); !strings.Contains(first.Body.String(), "before") {
t.Fatalf("first %s = %s, want the backend's own records", path, first.Body.String())
}
a.setBody(path, after)
second := doGet(t, h, path, "")
if strings.Contains(second.Body.String(), "before") {
t.Errorf("second %s = %s, want the changed data rather than a cached copy", path, second.Body.String())
}
if !strings.Contains(second.Body.String(), "after") {
t.Errorf("second %s = %s, want the changed data", path, second.Body.String())
}
if got := a.hitCount(path); got != 2 {
t.Errorf("backend a saw %d requests for %s, want one per request", got, path)
}
})
}
}
// No fact response may carry the cache headers PuppetDB never sends, whatever
// else pdbmux is doing with the merged body.
func TestHandler_FactResponsesCarryNoCacheHeaders(t *testing.T) {
body := `[` + fact("h1", "role", "web", "") + `]`
a := newCountingBackend(t, map[string]string{factsPath: body})
b := newCountingBackend(t, map[string]string{factsPath: `[]`})
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
h := srv.Handler()
for _, path := range []string{factsPath, factsPath + "/role", factNamesPath, sourceFactURL} {
t.Run(path, func(t *testing.T) {
a := newCountingBackend(t, map[string]string{path: body, factsPath: body})
b := newCountingBackend(t, map[string]string{path: `[]`, factsPath: `[]`})
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
h := srv.Handler()
const nameFiltered = `["=","name","role"]`
for _, pass := range []string{"first", "cached"} {
open := doGet(t, h, factsPath, "")
if srcs, n := sourceValues(t, open.Body.Bytes(), defaultSourceFact); n != 1 || srcs["h1"] != "a" {
t.Errorf("%s unfiltered request sources = %v (%d records), want h1 -> a", pass, srcs, n)
}
gated := doGet(t, h, factsPath, nameFiltered)
if _, n := sourceValues(t, gated.Body.Bytes(), defaultSourceFact); n != 0 {
t.Errorf("%s name-filtered request carried %d synthetic record(s), want none", pass, n)
}
if got := strings.TrimSpace(gated.Body.String()); !sameJSON(t, got, body) {
t.Errorf("%s name-filtered body = %s, want %s", pass, got, body)
}
}
if got := a.hitCount(factsPath); got != 2 {
t.Errorf("backend a saw %d requests, want 2: one per distinct cache key", got)
assertNoCacheHeaders(t, doGet(t, h, path, ""), path)
assertNoCacheHeaders(t, doGet(t, h, path, ""), path+" repeated")
})
}
}
// Suppression of an upstream fact of the configured name is part of the merged
// body, so it survives into the cache rather than being re-applied per request.
func TestHandler_SuppressionSurvivesCacheHit(t *testing.T) {
upstream := `[` + fact("h1", defaultSourceFact, "somewhere-else", "") + `,` + fact("h1", "role", "web", "") + `]`
a := newCountingBackend(t, map[string]string{factsPath: upstream})
b := newCountingBackend(t, map[string]string{factsPath: `[]`})
// openvoxdb answers a successful query with application/json;charset=utf-8
// (src/puppetlabs/puppetdb/query_eng.clj:555 through http.clj:80), so a merged
// response has to as well or a client can tell the proxy from the real thing.
func TestHandler_SuccessContentTypeMatchesPuppetDB(t *testing.T) {
bodies := map[string]string{
nodesPath: `[` + node("h1", "2026-01-01T00:00:00.000Z") + `]`,
factsPath: `[` + fact("h1", "role", "web", "") + `]`,
factNamesPath: `["role"]`,
reportsPath: `[` + report("h1", "abc", "2026-01-01T00:00:00.000Z") + `]`,
}
a := newCountingBackend(t, bodies)
b := newCountingBackend(t, map[string]string{nodesPath: `[]`, factsPath: `[]`, factNamesPath: `[]`, reportsPath: `[]`})
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
h := srv.Handler()
doGet(t, h, factsPath, "")
cached := doGet(t, h, factsPath, "")
if got := cached.Header().Get(cacheStatusHeader); got != "hit" {
t.Fatalf("%s = %q, want hit", cacheStatusHeader, got)
}
srcs, n := sourceValues(t, cached.Body.Bytes(), defaultSourceFact)
if n != 1 || srcs["h1"] != "a" {
t.Errorf("cached sources = %v (%d records), want exactly one h1 -> a, the upstream value dropped", srcs, n)
for path := range bodies {
// Twice, so a cache hit is held to the same content type as a live build.
for _, pass := range []string{"live", "repeated"} {
rec := doGet(t, h, path, "")
if rec.Code != http.StatusOK {
t.Fatalf("%s %s status %d (%s)", pass, path, rec.Code, rec.Body.String())
}
if got := rec.Header().Get("Content-Type"); got != "application/json;charset=utf-8" {
t.Errorf("%s %s Content-Type = %q, want application/json;charset=utf-8", pass, path, got)
}
}
}
}
config.go
+5 -5
View File
@@ -59,7 +59,7 @@ type Config struct {
Merge string `yaml:"merge"`
Timeout time.Duration `yaml:"timeout"`
FreshnessTTL time.Duration `yaml:"freshness_ttl"`
FactsTTL time.Duration `yaml:"facts_ttl"` // 0 disables the /facts+/nodes cache
FactsTTL time.Duration `yaml:"facts_ttl"` // 0 disables the merged /nodes cache
CacheBytes int64 `yaml:"facts_cache_bytes"` // byte budget for that cache
SourceFact string `yaml:"source_fact"`
@@ -344,8 +344,8 @@ func (c Config) Validate() error {
return nil
}
// cacheEnabled reports whether a facts/nodes cache should be built: both a TTL
// and a byte budget are required.
// cacheEnabled reports whether the merged /nodes cache should be built: both a
// TTL and a byte budget are required.
func (c Config) cacheEnabled() bool { return c.FactsTTL > 0 && c.CacheBytes > 0 }
func writeDefaultConfig(path string) error {
@@ -366,8 +366,8 @@ func writeDefaultConfig(path string) error {
"# PDBMUX_HEALTH_PROBE_ENABLED, PDBMUX_HEALTH_PROBE_PATH,\n" +
"# PDBMUX_HEALTH_PROBE_INTERVAL, PDBMUX_HEALTH_PROBE_TIMEOUT,\n" +
"# PDBMUX_HEALTH_PROBE_FAILURES, PDBMUX_HEALTH_PROBE_SUCCESSES.\n" +
"# facts_ttl caches merged /facts and /nodes in memory; it is capped at 30s\n" +
"# (a larger value is clamped) and 0 disables the cache.\n" +
"# facts_ttl caches merged /nodes in memory; it is capped at 30s (a larger\n" +
"# value is clamped) and 0 disables the cache. Fact answers are never cached.\n" +
"# health_probe_* polls each backend's status endpoint so queries skip a\n" +
"# backend that is down; when every backend is down all are queried anyway.\n\n")
if err := os.WriteFile(path, append(header, data...), 0o644); err != nil {
factroutes_test.go
+20 -14
View File
@@ -370,9 +370,9 @@ func TestHandler_FactsBySourceNameUnknownValueSkipsFanOut(t *testing.T) {
}
}
// The record set is the estate's, not the pinned value's, so every value shares
// one entry: two valid values must not cost two whole-estate fan-outs.
func TestHandler_FactsBySourceNameValuesShareOneFetch(t *testing.T) {
// The record set is the estate's, not the pinned value's: the <value> segment is
// applied to the merged records rather than fanned out as a path of its own.
func TestHandler_FactsBySourceNameValuesFilterOneFetch(t *testing.T) {
a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "osfamily", "RedHat", "") + `]`})
b := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h2", "osfamily", "Debian", "") + `]`})
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
@@ -392,8 +392,8 @@ func TestHandler_FactsBySourceNameValuesShareOneFetch(t *testing.T) {
}
}
for _, cb := range []*countingBackend{a, b} {
if got := cb.hitCount(factsPath); got != 1 {
t.Errorf("%s fetched %d times, want 1 shared fetch", factsPath, got)
if got := cb.hitCount(factsPath); got != 3 {
t.Errorf("%s fetched %d times, want one uncached fetch per request", factsPath, got)
}
if got := cb.hitCount(sourceFactURL); got != 0 {
t.Errorf("%s was fanned out %d times, want 0", sourceFactURL, got)
@@ -678,21 +678,27 @@ func TestHandler_FactNamesSurvivesOneBackend(t *testing.T) {
}
}
// Both routes are merged record sets, so they use the same cache /facts does.
func TestHandler_FactRoutesAreCached(t *testing.T) {
// Every route that serves fact records fans out per request, even with the
// cache configured: a fact answer is only ever as old as the backend's own.
func TestHandler_FactRoutesAreNotCached(t *testing.T) {
for _, path := range []string{roleFactPath, roleFactPath + "/web", sourceFactURL, factNamesPath} {
t.Run(path, func(t *testing.T) {
// The source-fact drilldown is answered from the /facts merge, so
// that is the path its fan-out lands on.
fanOut := path
if path == sourceFactURL {
fanOut = factsPath
}
a := newFakeBackend(t, `[]`, `[]`)
a.bodies[path] = `[]`
a.bodies[fanOut] = `[]`
b := newFakeBackend(t, `[]`, `[]`)
b.bodies[path] = `[]`
b.bodies[fanOut] = `[]`
srv := newTestServer(cacheTestConfig(a.srv.URL, b.srv.URL))
if got := doGet(t, srv.Handler(), path, "").Header().Get(cacheStatusHeader); got != "miss" {
t.Errorf("first %s = %q, want miss", cacheStatusHeader, got)
}
if got := doGet(t, srv.Handler(), path, "").Header().Get(cacheStatusHeader); got != "hit" {
t.Errorf("second %s = %q, want hit", cacheStatusHeader, got)
doGet(t, srv.Handler(), path, "")
doGet(t, srv.Handler(), path, "")
if got := a.hits(fanOut); got != 2 {
t.Errorf("backend a saw %d requests for %s, want one per request", got, fanOut)
}
})
}
server.go
+27 -52
View File
@@ -32,11 +32,6 @@ const (
// PuppetDB only sends this when the request carries include_total=true.
recordsHeader = "X-Records"
// Set by pdbmux, not by PuppetDB: how a cache-backed response was answered
// and how old the served copy is.
cacheStatusHeader = "X-Cache"
ageHeader = "Age"
// Set by pdbmux: "<contributed>/<configured>" backends behind a merged response.
backendsHeader = "X-Backends"
)
@@ -61,16 +56,17 @@ type Server struct {
client *http.Client
log *log.Logger
// factsCache is nil when caching is disabled; cacheFor hands out a noop then.
factsCache Cache
flights flightGroup
stale staleTracker
// nodeCache is nil when caching is disabled; cacheFor hands out a noop then.
nodeCache Cache
flights flightGroup
stale staleTracker
// health is nil when probing is disabled, which makes every backend healthy.
health *prober
partial partialTracker
// now is shared with the cache's clock so Age matches the stored timestamp.
// now is shared with the cache's clock so a stale fallback is reported
// against the same timestamps the cache stamps entries with.
now func() time.Time
// freshness cache (freshness merge only).
@@ -88,7 +84,7 @@ func NewServer(cfg Config, logger *log.Logger) *Server {
now: time.Now,
}
if cfg.cacheEnabled() {
s.factsCache = newMemoryCache(cfg.FactsTTL, cfg.CacheBytes)
s.nodeCache = newMemoryCache(cfg.FactsTTL, cfg.CacheBytes)
}
s.health = newProber(cfg, logger)
return s
@@ -101,20 +97,21 @@ func (s *Server) StartProbes(ctx context.Context) { s.health.Start(ctx) }
// StopProbes stops the probing goroutines and waits for them to exit.
func (s *Server) StopProbes() { s.health.Stop() }
// cacheFor picks the cache backing a request. The merged /nodes and fact record
// sets — /facts, /facts/<name>[/<value>] and /fact-names — share the in-memory
// cache; every other path is uncached until the reports cache lands, and a new
// backend is a case here rather than a change to any handler.
// cacheFor picks the cache backing a request. Only the merged /nodes record set
// is cached: its report columns move once per node run, while fact answers must
// be as current as a backend's own, so no fact-serving path is cached here.
// Every other path is uncached until the reports cache lands, and a new backend
// is a case here rather than a change to any handler.
func (s *Server) cacheFor(path string, params url.Values) (Cache, bool) {
switch {
case path == factsPath, path == nodesPath, path == factNamesPath, isFactsSubPath(path):
switch path {
case nodesPath:
// An aggregate row is a combined count, not the merged record set the
// cache was built for, so it stays on the live path.
if spec, err := parseAggregate(params.Get("query")); spec != nil || err != nil {
return noopCache{}, false
}
if s.factsCache != nil {
return s.factsCache, true
if s.nodeCache != nil {
return s.nodeCache, true
}
}
return noopCache{}, false
@@ -595,10 +592,9 @@ type cachedResponse struct {
Configured int `json:"configured"` // backends configured at build time
}
// recordFilter narrows a response's records after it has been built or read back
// from the cache, so requests differing only in the filter share one stored entry
// and one fan-out. It leaves Records alone, so it only suits responses that set
// no X-Records.
// recordFilter narrows a response's records after it has been built, so a route
// that answers from another path's merge can pin part of it per request. It
// leaves Records alone, so it only suits responses that set no X-Records.
type recordFilter func([]json.RawMessage) []json.RawMessage
func (f recordFilter) apply(resp cachedResponse) cachedResponse {
@@ -644,7 +640,7 @@ func (s *Server) serveFiltered(w http.ResponseWriter, r *http.Request, path stri
s.log.Printf("warning: cache lookup for %s failed: %v", key, err)
case status == CacheFresh:
s.stale.markFresh()
s.writeStored(w, ent, CacheFresh, filter)
s.writeStored(w, ent, filter)
return
case status == CacheStale:
stale = &ent
@@ -687,14 +683,13 @@ func (s *Server) serveFiltered(w http.ResponseWriter, r *http.Request, path stri
s.stale.markStale(s.now())
s.log.Printf("warning: serving stale %s from cache (stored %s): %v",
path, stale.StoredAt.UTC().Format(time.RFC3339), err)
s.writeStored(w, *stale, CacheStale, filter)
s.writeStored(w, *stale, filter)
return
}
s.writeUpstreamError(w, err)
return
}
s.stale.markFresh()
s.setCacheHeaders(w, CacheMiss, time.Time{})
writeCached(w, filter.apply(resp))
}
@@ -707,39 +702,19 @@ func (s *Server) flightTimeout() time.Duration {
return defaultTimeout
}
func (s *Server) writeStored(w http.ResponseWriter, ent CacheEntry, status CacheStatus, filter recordFilter) {
// writeStored answers from a cache entry. Whether a response came from the cache
// is deliberately invisible to the client: PuppetDB sets no X-Cache or Age, so
// neither does pdbmux, and /healthz carries the cache state instead.
func (s *Server) writeStored(w http.ResponseWriter, ent CacheEntry, filter recordFilter) {
var resp cachedResponse
if err := json.Unmarshal(ent.Body, &resp); err != nil {
s.log.Printf("warning: unreadable cache entry: %v", err)
http.Error(w, "unreadable cache entry", http.StatusBadGateway)
return
}
s.setCacheHeaders(w, status, ent.StoredAt)
writeCached(w, filter.apply(resp))
}
// setCacheHeaders labels a response from a cache-backed path: X-Cache is
// hit/stale/miss and Age is whole seconds since the served copy was stored (0
// for a response built by this request). It reads the same clock the cache
// stamps entries with, so the two never disagree.
func (s *Server) setCacheHeaders(w http.ResponseWriter, status CacheStatus, storedAt time.Time) {
label := "miss"
switch status {
case CacheFresh:
label = "hit"
case CacheStale:
label = "stale"
}
age := 0
if !storedAt.IsZero() {
if secs := int(s.now().Sub(storedAt).Seconds()); secs > 0 {
age = secs
}
}
w.Header().Set(cacheStatusHeader, label)
w.Header().Set(ageHeader, strconv.Itoa(age))
}
func writeCached(w http.ResponseWriter, resp cachedResponse) {
w.Header().Set("Content-Type", jsonContentType)
if resp.Records >= 0 {
@@ -1033,8 +1008,8 @@ type cacheHealth struct {
func (s *Server) cacheHealth() cacheHealth {
stats := CacheStats{Backend: "none"}
ttl := time.Duration(0)
if s.factsCache != nil {
stats = s.factsCache.Stats()
if s.nodeCache != nil {
stats = s.nodeCache.Stats()
ttl = s.cfg.FactsTTL
}
serving, served, last := s.stale.snapshot()
upstream_test.go
+14 -14
View File
@@ -371,18 +371,18 @@ func TestHandler_RejectedQueryIsNotCached(t *testing.T) {
b.reject, b.rejectBody = http.StatusBadRequest, badOrderBy
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
rec := doGet(t, srv.Handler(), factsPath, "")
rec := doGet(t, srv.Handler(), nodesPath, "")
if rec.Code != http.StatusBadRequest {
t.Fatalf("status = %d, want the upstream 400 replayed: %s", rec.Code, rec.Body.String())
}
if n := srv.factsCache.Stats().Entries; n != 0 {
if n := srv.nodeCache.Stats().Entries; n != 0 {
t.Fatalf("cache holds %d entries after a rejected query", n)
}
// The rejection left nothing behind, so the next good query is served fresh.
a.reject, b.reject = 0, 0
a.factsBody = `[` + fact("h1", "role", "web", "") + `]`
ok := doGet(t, srv.Handler(), factsPath, "")
a.nodesBody = `[` + node("web", "2026-01-01T00:00:00.000Z") + `]`
ok := doGet(t, srv.Handler(), nodesPath, "")
if ok.Code != http.StatusOK || !strings.Contains(ok.Body.String(), "web") {
t.Fatalf("follow-up = %d %s", ok.Code, ok.Body.String())
}
@@ -394,10 +394,10 @@ func TestHandler_AllBackendsFailedIsNotCached(t *testing.T) {
a.dead, b.dead = true, true
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
if rec := doGet(t, srv.Handler(), factsPath, ""); rec.Code != http.StatusBadGateway {
if rec := doGet(t, srv.Handler(), nodesPath, ""); rec.Code != http.StatusBadGateway {
t.Fatalf("status = %d, want 502", rec.Code)
}
if n := srv.factsCache.Stats().Entries; n != 0 {
if n := srv.nodeCache.Stats().Entries; n != 0 {
t.Fatalf("cache holds %d entries after a failed fan-out", n)
}
}
@@ -405,18 +405,18 @@ func TestHandler_AllBackendsFailedIsNotCached(t *testing.T) {
// Stale records answer an outage. They do not answer a query the estate refused:
// the client has to be told why, not handed data for a question it did not ask.
func TestHandler_RejectedQueryIsNotAnsweredFromStale(t *testing.T) {
a := newFakeBackend(t, `[]`, `[`+fact("h1", "role", "old", "")+`]`)
a := newFakeBackend(t, `[`+node("old", "2026-01-01T00:00:00.000Z")+`]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
if warm := doGet(t, srv.Handler(), factsPath, ""); warm.Code != http.StatusOK {
if warm := doGet(t, srv.Handler(), nodesPath, ""); warm.Code != http.StatusOK {
t.Fatalf("warm-up status %d", warm.Code)
}
clk.advance(31 * time.Second)
a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
b.reject, b.rejectBody = http.StatusBadRequest, badOrderBy
rec := doGet(t, srv.Handler(), factsPath, "")
rec := doGet(t, srv.Handler(), nodesPath, "")
if rec.Code != http.StatusBadRequest {
t.Fatalf("status = %d, want the rejection rather than the stale entry: %s", rec.Code, rec.Body.String())
}
@@ -430,16 +430,16 @@ func TestHandler_RejectedQueryIsNotAnsweredFromStale(t *testing.T) {
// An outage still falls back to the stale copy, unchanged by the replay path.
func TestHandler_OutageStillFallsBackToStale(t *testing.T) {
a := newFakeBackend(t, `[]`, `[`+fact("h1", "role", "old", "")+`]`)
a := newFakeBackend(t, `[`+node("old", "2026-01-01T00:00:00.000Z")+`]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
if warm := doGet(t, srv.Handler(), factsPath, ""); warm.Code != http.StatusOK {
if warm := doGet(t, srv.Handler(), nodesPath, ""); warm.Code != http.StatusOK {
t.Fatalf("warm-up status %d", warm.Code)
}
clk.advance(31 * time.Second)
a.fail, b.fail = true, true
rec := doGet(t, srv.Handler(), factsPath, "")
rec := doGet(t, srv.Handler(), nodesPath, "")
if rec.Code != http.StatusOK || !strings.Contains(rec.Body.String(), "old") {
t.Fatalf("stale fallback = %d %s", rec.Code, rec.Body.String())
}
@@ -652,14 +652,14 @@ func TestHandler_SuccessContentTypeMatchesUpstream(t *testing.T) {
}
func TestHandler_CachedSuccessContentTypeMatchesUpstream(t *testing.T) {
a := newFakeBackend(t, `[]`, `[`+fact("h1", "role", "web", "")+`]`)
a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
h := srv.Handler()
// The miss builds the entry and the hit replays it; both are the client's view.
for _, label := range []string{"miss", "hit"} {
rec := doGet(t, h, factsPath, "")
rec := doGet(t, h, nodesPath, "")
if got := rec.Header().Get("Content-Type"); got != upstreamJSONContentType {
t.Errorf("%s Content-Type = %q, want %q", label, got, upstreamJSONContentType)
}