feat: cache merged /facts and /nodes in memory, stale on backend failure
A busy Puppetboard re-fans-out the same /facts query every few seconds, and a 502 is worse than 30-second-old facts when every PuppetDB is unreachable. - Add a `Cache` interface (get reports fresh/stale/miss, put, stats) keyed on `<path>?<params>` with keys and repeated values sorted, plus a no-op default so uncached paths behave exactly as before. - Route serveMerged/serveUnion/serveSummed through `serveCached`, so the reports cache drops in at `cacheFor` without touching a handler. - Back /facts and /nodes with a byte-bounded LRU: `facts_ttl` (default 30s, clamped to a 30s cap) and `facts_cache_bytes` (default 64 MiB); expired entries are kept and served only when every backend fails. - Single-flight identical keys so N concurrent requests cause one fan-out. - Surface `cache` state and `serving_stale` in /healthz and the cache settings in `config show`.
This commit is contained in:
@@ -38,7 +38,7 @@ not PQL) is forwarded verbatim.
|
||||
| `GET /metrics/v2/read/<mbean>` | Fan out to all and merge the Jolokia response; numeric attributes are **summed** by default (see merge semantics). |
|
||||
| `GET /metrics/v2/list` | Fan out to all and serve the **union** of the backends' MBean trees. |
|
||||
| `GET /metrics/v1/mbeans[/<mbean>]` | Same merge, applied to the legacy envelope-less body. |
|
||||
| `GET /healthz` | Per-backend reachability. `200 {"status":"ok"}` if all reachable, `200 degraded` if some fail, `503 down` if all fail. |
|
||||
| `GET /healthz` | Per-backend reachability plus cache state. `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
|
||||
surviving backends' results and logs a warning; a merged endpoint only returns `502` when
|
||||
@@ -208,6 +208,41 @@ Each backend applies `order_by`/`limit`/`offset` to its own slice only, so
|
||||
- A malformed `limit`, `offset` or `order_by` gets a `400` rather than being
|
||||
forwarded.
|
||||
|
||||
## Caching
|
||||
|
||||
`pdbmux` caches merged `/nodes` and `/facts` record sets **in memory** so a busy
|
||||
Puppetboard does not re-fan-out the same query every few seconds. Everything else
|
||||
runs uncached — including `extract`/`count` aggregates on those two 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.
|
||||
|
||||
- **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
|
||||
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`
|
||||
disables the cache entirely and the merged endpoints behave exactly as before.
|
||||
- **Stale on failure only** — an expired entry is kept, not dropped. When the TTL
|
||||
has passed `pdbmux` always re-queries the backends; the expired copy is served
|
||||
**only** if every backend fails, which turns a `502` into slightly-old data. A
|
||||
healthy backend is never shadowed by a stale entry.
|
||||
- **Bounded** — `facts_cache_bytes` (default 64 MiB) is a byte budget, evicted
|
||||
least-recently-used; reads count as use, so a stale entry that is still being
|
||||
asked for survives. A single response larger than the whole budget is not
|
||||
cached at all.
|
||||
- **Single-flight** — concurrent requests for the same key collapse into one
|
||||
upstream fan-out; the rest wait for it and share the result.
|
||||
- **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.
|
||||
|
||||
## Config
|
||||
|
||||
Precedence (lowest → highest): **defaults < config file < env vars (`PDBMUX_*`) < flags**.
|
||||
@@ -229,9 +264,11 @@ backends: # order is a tie-break only, not a ranking
|
||||
url: http://puppetdb1.example.com:8080
|
||||
- name: pdb-b
|
||||
url: https://puppetdb2.example.com
|
||||
merge: freshness # freshness | static
|
||||
timeout: 10s # per-upstream request timeout
|
||||
freshness_ttl: 30s # freshness-map cache TTL (freshness merge only)
|
||||
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_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
|
||||
```
|
||||
@@ -246,6 +283,8 @@ the `/pdb/query/v4/...` path per request.
|
||||
| `PDBMUX_MERGE` | `merge` |
|
||||
| `PDBMUX_TIMEOUT` | `timeout` (Go duration, e.g. `10s`) |
|
||||
| `PDBMUX_FRESHNESS_TTL` | `freshness_ttl` |
|
||||
| `PDBMUX_FACTS_TTL` | `facts_ttl` (clamped to 30s) |
|
||||
| `PDBMUX_FACTS_CACHE_BYTES` | `facts_cache_bytes` (plain integer bytes) |
|
||||
| `PDBMUX_BACKENDS` | whole backend list, as `name=url,name=url` |
|
||||
| `PDBMUX_SOURCE_FACT` | `source_fact` (default `pdbmux_source`) |
|
||||
| `PDBMUX_SOURCE_FACT_ENABLED` | `source_fact_enabled` (default `true`); `false` disables injection |
|
||||
@@ -281,5 +320,6 @@ A static (`CGO_ENABLED=0`) binary on a distroless base. Configure it with
|
||||
`PDBMUX_*` env vars (at minimum `PDBMUX_BACKENDS`), or mount a config file — a
|
||||
configmap at `/etc/pdbmux/config.yaml` is picked up with no env var at all, and
|
||||
any other mount path works via `PDBMUX_CONFIG`. Env vars still override file
|
||||
values, so the two mix. Stateless, so run as many replicas as you like; use
|
||||
`/healthz` for liveness/readiness probes.
|
||||
values, so the two mix. Run as many replicas as you like — the only state is the
|
||||
in-memory cache, which is per-replica and bounded by `facts_cache_bytes`, so size
|
||||
the memory limit above it. Use `/healthz` for liveness/readiness probes.
|
||||
|
||||
@@ -0,0 +1,248 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"container/list"
|
||||
"context"
|
||||
"net/url"
|
||||
"sort"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// CacheStatus distinguishes the three outcomes of a lookup: nothing stored, a
|
||||
// stored entry within its TTL, and a stored entry past it.
|
||||
type CacheStatus int
|
||||
|
||||
const (
|
||||
CacheMiss CacheStatus = iota
|
||||
CacheFresh
|
||||
CacheStale
|
||||
)
|
||||
|
||||
func (s CacheStatus) String() string {
|
||||
switch s {
|
||||
case CacheFresh:
|
||||
return "fresh"
|
||||
case CacheStale:
|
||||
return "stale"
|
||||
default:
|
||||
return "miss"
|
||||
}
|
||||
}
|
||||
|
||||
// CacheEntry is a stored response body and the time it was stored.
|
||||
type CacheEntry struct {
|
||||
Body []byte
|
||||
StoredAt time.Time
|
||||
}
|
||||
|
||||
// CacheStats is the cache state reported by /healthz.
|
||||
type CacheStats struct {
|
||||
Backend string `json:"backend"`
|
||||
Entries int `json:"entries"`
|
||||
StaleEntries int `json:"stale_entries"`
|
||||
Bytes int64 `json:"bytes"`
|
||||
}
|
||||
|
||||
// Cache stores merged responses keyed by cacheKey. Get reports freshness rather
|
||||
// than hiding expired entries so a caller can fall back to a stale body when the
|
||||
// upstream fetch fails. The context and error exist for out-of-process backends
|
||||
// (the reports cache lands on S3); an in-process backend ignores both.
|
||||
//
|
||||
// A Body handed back by Get aliases the cache's copy and must not be mutated.
|
||||
type Cache interface {
|
||||
Get(ctx context.Context, key string) (CacheEntry, CacheStatus, error)
|
||||
Put(ctx context.Context, key string, body []byte) error
|
||||
Stats() CacheStats
|
||||
}
|
||||
|
||||
// cacheKey is the request path when there are no params, else "<path>?<params>"
|
||||
// where params is url.Values.Encode() over a copy whose repeated values have
|
||||
// been sorted. Encode() already sorts keys ascending, so both the order params
|
||||
// arrive in and the order of a repeated param's values are irrelevant to the
|
||||
// key: one canonical string per distinct request.
|
||||
func cacheKey(path string, params url.Values) string {
|
||||
if len(params) == 0 {
|
||||
return path
|
||||
}
|
||||
norm := make(url.Values, len(params))
|
||||
for k, vs := range params {
|
||||
sorted := append([]string(nil), vs...)
|
||||
sort.Strings(sorted)
|
||||
norm[k] = sorted
|
||||
}
|
||||
encoded := norm.Encode()
|
||||
if encoded == "" {
|
||||
return path
|
||||
}
|
||||
return path + "?" + encoded
|
||||
}
|
||||
|
||||
// noopCache is the default for every path with no cache configured, so wiring a
|
||||
// handler through the cache leaves its behaviour unchanged.
|
||||
type noopCache struct{}
|
||||
|
||||
func (noopCache) Get(context.Context, string) (CacheEntry, CacheStatus, error) {
|
||||
return CacheEntry{}, CacheMiss, nil
|
||||
}
|
||||
|
||||
func (noopCache) Put(context.Context, string, []byte) error { return nil }
|
||||
|
||||
func (noopCache) Stats() CacheStats { return CacheStats{Backend: "none"} }
|
||||
|
||||
type memoryEntry struct {
|
||||
key string
|
||||
body []byte
|
||||
storedAt time.Time
|
||||
}
|
||||
|
||||
// memoryCache is a byte-bounded LRU. Expired entries are kept, not dropped, so
|
||||
// they remain available as a stale fallback; only the byte budget evicts.
|
||||
type memoryCache struct {
|
||||
ttl time.Duration
|
||||
maxBytes int64
|
||||
now func() time.Time
|
||||
|
||||
mu sync.Mutex
|
||||
ll *list.List // front = most recently used
|
||||
items map[string]*list.Element
|
||||
bytes int64
|
||||
}
|
||||
|
||||
func newMemoryCache(ttl time.Duration, maxBytes int64) *memoryCache {
|
||||
return &memoryCache{
|
||||
ttl: ttl,
|
||||
maxBytes: maxBytes,
|
||||
now: time.Now,
|
||||
ll: list.New(),
|
||||
items: make(map[string]*list.Element),
|
||||
}
|
||||
}
|
||||
|
||||
func (c *memoryCache) Get(_ context.Context, key string) (CacheEntry, CacheStatus, error) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
el, ok := c.items[key]
|
||||
if !ok {
|
||||
return CacheEntry{}, CacheMiss, nil
|
||||
}
|
||||
c.ll.MoveToFront(el)
|
||||
e := el.Value.(*memoryEntry)
|
||||
status := CacheFresh
|
||||
if c.now().Sub(e.storedAt) >= c.ttl {
|
||||
status = CacheStale
|
||||
}
|
||||
return CacheEntry{Body: e.body, StoredAt: e.storedAt}, status, nil
|
||||
}
|
||||
|
||||
func (c *memoryCache) Put(_ context.Context, key string, body []byte) error {
|
||||
// A response larger than the whole budget would evict everything else.
|
||||
if int64(len(body)) > c.maxBytes {
|
||||
return nil
|
||||
}
|
||||
stored := append([]byte(nil), body...)
|
||||
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
if el, ok := c.items[key]; ok {
|
||||
e := el.Value.(*memoryEntry)
|
||||
c.bytes += int64(len(stored)) - int64(len(e.body))
|
||||
e.body, e.storedAt = stored, c.now()
|
||||
c.ll.MoveToFront(el)
|
||||
} else {
|
||||
c.items[key] = c.ll.PushFront(&memoryEntry{key: key, body: stored, storedAt: c.now()})
|
||||
c.bytes += int64(len(stored))
|
||||
}
|
||||
for c.bytes > c.maxBytes {
|
||||
back := c.ll.Back()
|
||||
if back == nil {
|
||||
break
|
||||
}
|
||||
e := c.ll.Remove(back).(*memoryEntry)
|
||||
delete(c.items, e.key)
|
||||
c.bytes -= int64(len(e.body))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *memoryCache) Stats() CacheStats {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
st := CacheStats{Backend: "memory", Entries: len(c.items), Bytes: c.bytes}
|
||||
now := c.now()
|
||||
for el := c.ll.Front(); el != nil; el = el.Next() {
|
||||
if now.Sub(el.Value.(*memoryEntry).storedAt) >= c.ttl {
|
||||
st.StaleEntries++
|
||||
}
|
||||
}
|
||||
return st
|
||||
}
|
||||
|
||||
// flightGroup collapses concurrent identical builds so N simultaneous requests
|
||||
// for one key cause one upstream fan-out.
|
||||
type flightGroup struct {
|
||||
mu sync.Mutex
|
||||
calls map[string]*flightCall
|
||||
}
|
||||
|
||||
type flightCall struct {
|
||||
wg sync.WaitGroup
|
||||
resp cachedResponse
|
||||
err error
|
||||
}
|
||||
|
||||
// Do returns fn's result and whether this caller shared another's in-flight run.
|
||||
func (g *flightGroup) Do(key string, fn func() (cachedResponse, error)) (cachedResponse, error, bool) {
|
||||
g.mu.Lock()
|
||||
if g.calls == nil {
|
||||
g.calls = make(map[string]*flightCall)
|
||||
}
|
||||
if c, ok := g.calls[key]; ok {
|
||||
g.mu.Unlock()
|
||||
c.wg.Wait()
|
||||
return c.resp, c.err, true
|
||||
}
|
||||
c := &flightCall{}
|
||||
c.wg.Add(1)
|
||||
g.calls[key] = c
|
||||
g.mu.Unlock()
|
||||
|
||||
defer func() {
|
||||
c.wg.Done()
|
||||
g.mu.Lock()
|
||||
delete(g.calls, key)
|
||||
g.mu.Unlock()
|
||||
}()
|
||||
|
||||
c.resp, c.err = fn()
|
||||
return c.resp, c.err, false
|
||||
}
|
||||
|
||||
// staleTracker records stale fallbacks for /healthz. serving flips back to false
|
||||
// as soon as a response is served from a live fan-out or a fresh entry.
|
||||
type staleTracker struct {
|
||||
mu sync.Mutex
|
||||
serving bool
|
||||
served uint64
|
||||
last time.Time
|
||||
}
|
||||
|
||||
func (t *staleTracker) markStale(now time.Time) {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
t.serving = true
|
||||
t.served++
|
||||
t.last = now
|
||||
}
|
||||
|
||||
func (t *staleTracker) markFresh() {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
t.serving = false
|
||||
}
|
||||
|
||||
func (t *staleTracker) snapshot() (serving bool, served uint64, last time.Time) {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
return t.serving, t.served, t.last
|
||||
}
|
||||
+657
@@ -0,0 +1,657 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// fakeClock drives the cache's TTL without sleeping.
|
||||
type fakeClock struct {
|
||||
mu sync.Mutex
|
||||
t time.Time
|
||||
}
|
||||
|
||||
func newFakeClock() *fakeClock { return &fakeClock{t: time.Unix(1_800_000_000, 0)} }
|
||||
|
||||
func (c *fakeClock) now() time.Time {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
return c.t
|
||||
}
|
||||
|
||||
func (c *fakeClock) advance(d time.Duration) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.t = c.t.Add(d)
|
||||
}
|
||||
|
||||
// countingBackend is a PuppetDB stand-in that counts requests per path and can
|
||||
// be made to fail or block, so cache hits and single-flight are observable.
|
||||
type countingBackend struct {
|
||||
srv *httptest.Server
|
||||
|
||||
mu sync.Mutex
|
||||
hits map[string]int
|
||||
bodies map[string]string
|
||||
fail bool
|
||||
block chan struct{}
|
||||
}
|
||||
|
||||
func newCountingBackend(t *testing.T, bodies map[string]string) *countingBackend {
|
||||
t.Helper()
|
||||
cb := &countingBackend{hits: map[string]int{}, bodies: bodies}
|
||||
cb.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
cb.mu.Lock()
|
||||
cb.hits[r.URL.Path]++
|
||||
fail, block, body := cb.fail, cb.block, cb.bodies[r.URL.Path]
|
||||
cb.mu.Unlock()
|
||||
|
||||
if block != nil {
|
||||
<-block
|
||||
}
|
||||
if fail {
|
||||
http.Error(w, "boom", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
if body == "" {
|
||||
body = "[]"
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = io.WriteString(w, body)
|
||||
}))
|
||||
t.Cleanup(cb.srv.Close)
|
||||
return cb
|
||||
}
|
||||
|
||||
func (cb *countingBackend) hitCount(path string) int {
|
||||
cb.mu.Lock()
|
||||
defer cb.mu.Unlock()
|
||||
return cb.hits[path]
|
||||
}
|
||||
|
||||
func (cb *countingBackend) setFail(v bool) {
|
||||
cb.mu.Lock()
|
||||
defer cb.mu.Unlock()
|
||||
cb.fail = v
|
||||
}
|
||||
|
||||
func (cb *countingBackend) setBody(path, body string) {
|
||||
cb.mu.Lock()
|
||||
defer cb.mu.Unlock()
|
||||
cb.bodies[path] = body
|
||||
}
|
||||
|
||||
func (cb *countingBackend) setBlock(ch chan struct{}) {
|
||||
cb.mu.Lock()
|
||||
defer cb.mu.Unlock()
|
||||
cb.block = ch
|
||||
}
|
||||
|
||||
// newCachedServer builds a server whose facts 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)
|
||||
if !ok {
|
||||
t.Fatalf("expected a memory cache, got %T", srv.factsCache)
|
||||
}
|
||||
clk := newFakeClock()
|
||||
mc.now = clk.now
|
||||
return srv, clk
|
||||
}
|
||||
|
||||
func cacheTestConfig(aURL, bURL string) Config {
|
||||
cfg := testConfig(aURL, bURL, mergeStatic)
|
||||
cfg.FactsTTL = 30 * time.Second
|
||||
cfg.CacheBytes = defaultCacheSize
|
||||
return cfg
|
||||
}
|
||||
|
||||
func TestCacheKey_CanonicalOrdering(t *testing.T) {
|
||||
a := cacheKey(factsPath, url.Values{
|
||||
"query": {`["=","name","role"]`},
|
||||
"limit": {"10"},
|
||||
"expand": {"z", "a"},
|
||||
})
|
||||
b := cacheKey(factsPath, url.Values{
|
||||
"expand": {"a", "z"},
|
||||
"limit": {"10"},
|
||||
"query": {`["=","name","role"]`},
|
||||
})
|
||||
if a != b {
|
||||
t.Errorf("param order must not change the key:\n %s\n %s", a, b)
|
||||
}
|
||||
if !strings.HasPrefix(a, factsPath+"?") {
|
||||
t.Errorf("key must start with the path, got %q", a)
|
||||
}
|
||||
if got := cacheKey(nodesPath, nil); got != nodesPath {
|
||||
t.Errorf("no params should give the bare path, got %q", got)
|
||||
}
|
||||
if cacheKey(factsPath, url.Values{"query": {"x"}}) == cacheKey(factsPath, url.Values{"query": {"y"}}) {
|
||||
t.Error("different queries must not share a key")
|
||||
}
|
||||
if cacheKey(factsPath, url.Values{"query": {"x"}}) == cacheKey(nodesPath, url.Values{"query": {"x"}}) {
|
||||
t.Error("different paths must not share a key")
|
||||
}
|
||||
}
|
||||
|
||||
func TestMemoryCache_FreshThenStale(t *testing.T) {
|
||||
c := newMemoryCache(30*time.Second, 1<<20)
|
||||
clk := newFakeClock()
|
||||
c.now = clk.now
|
||||
|
||||
if _, status, _ := c.Get(context.Background(), "k"); status != CacheMiss {
|
||||
t.Fatalf("empty cache should miss, got %s", status)
|
||||
}
|
||||
if err := c.Put(context.Background(), "k", []byte(`{"body":[]}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
ent, status, err := c.Get(context.Background(), "k")
|
||||
if err != nil || status != CacheFresh || string(ent.Body) != `{"body":[]}` {
|
||||
t.Fatalf("want a fresh hit, got %s %q (%v)", status, ent.Body, err)
|
||||
}
|
||||
|
||||
clk.advance(29 * time.Second)
|
||||
if _, status, _ := c.Get(context.Background(), "k"); status != CacheFresh {
|
||||
t.Fatalf("inside the TTL should still be fresh, got %s", status)
|
||||
}
|
||||
|
||||
clk.advance(2 * time.Second)
|
||||
ent, status, _ = c.Get(context.Background(), "k")
|
||||
if status != CacheStale {
|
||||
t.Fatalf("past the TTL should be stale, got %s", status)
|
||||
}
|
||||
if string(ent.Body) != `{"body":[]}` {
|
||||
t.Errorf("a stale entry must still carry its body, got %q", ent.Body)
|
||||
}
|
||||
if st := c.Stats(); st.Entries != 1 || st.StaleEntries != 1 {
|
||||
t.Errorf("stats should report 1 entry, 1 stale, got %+v", st)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMemoryCache_EvictsLeastRecentlyUsed(t *testing.T) {
|
||||
body := []byte("0123456789") // 10 bytes
|
||||
c := newMemoryCache(time.Minute, 25)
|
||||
ctx := context.Background()
|
||||
|
||||
for _, k := range []string{"a", "b"} {
|
||||
if err := c.Put(ctx, k, body); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
// Reading "a" makes "b" the eviction candidate.
|
||||
if _, status, _ := c.Get(ctx, "a"); status != CacheFresh {
|
||||
t.Fatalf("a should be cached, got %s", status)
|
||||
}
|
||||
if err := c.Put(ctx, "c", body); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if _, status, _ := c.Get(ctx, "b"); status != CacheMiss {
|
||||
t.Errorf("b was least recently used and should have been evicted, got %s", status)
|
||||
}
|
||||
for _, k := range []string{"a", "c"} {
|
||||
if _, status, _ := c.Get(ctx, k); status != CacheFresh {
|
||||
t.Errorf("%s should have survived eviction, got %s", k, status)
|
||||
}
|
||||
}
|
||||
if st := c.Stats(); st.Entries != 2 || st.Bytes != 20 {
|
||||
t.Errorf("stats after eviction = %+v, want 2 entries / 20 bytes", st)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMemoryCache_SkipsOversizedEntry(t *testing.T) {
|
||||
c := newMemoryCache(time.Minute, 8)
|
||||
if err := c.Put(context.Background(), "big", []byte("123456789")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, status, _ := c.Get(context.Background(), "big"); status != CacheMiss {
|
||||
t.Error("an entry larger than the whole budget must not be stored")
|
||||
}
|
||||
if st := c.Stats(); st.Bytes != 0 {
|
||||
t.Errorf("bytes = %d, want 0", st.Bytes)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNoopCache_AlwaysMisses(t *testing.T) {
|
||||
var c Cache = noopCache{}
|
||||
if err := c.Put(context.Background(), "k", []byte("x")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, status, _ := c.Get(context.Background(), "k"); status != CacheMiss {
|
||||
t.Errorf("noop cache must always miss, got %s", status)
|
||||
}
|
||||
if st := c.Stats(); st.Backend != "none" {
|
||||
t.Errorf("backend = %q, want none", st.Backend)
|
||||
}
|
||||
}
|
||||
|
||||
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: `[]`})
|
||||
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"]`)
|
||||
|
||||
if first.Code != http.StatusOK || second.Code != http.StatusOK {
|
||||
t.Fatalf("statuses %d/%d", first.Code, second.Code)
|
||||
}
|
||||
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 {
|
||||
t.Errorf("backend a saw %d requests, want 1 (second served from cache)", got)
|
||||
}
|
||||
if got := b.hitCount(factsPath); 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 {
|
||||
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: `[]`})
|
||||
srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
|
||||
|
||||
doGet(t, srv.Handler(), factsPath, "")
|
||||
clk.advance(31 * time.Second)
|
||||
doGet(t, srv.Handler(), factsPath, "")
|
||||
|
||||
if got := a.hitCount(factsPath); 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: `[]`})
|
||||
srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
|
||||
|
||||
warm := doGet(t, srv.Handler(), factsPath, "")
|
||||
if warm.Code != http.StatusOK {
|
||||
t.Fatalf("warm-up status %d", warm.Code)
|
||||
}
|
||||
if !strings.Contains(warm.Body.String(), `"old"`) {
|
||||
t.Fatalf("warm-up body = %s", warm.Body.String())
|
||||
}
|
||||
|
||||
// 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, "")
|
||||
if !strings.Contains(refetch.Body.String(), `"new"`) {
|
||||
t.Errorf("a healthy backend must not be shadowed by the stale entry: %s", refetch.Body.String())
|
||||
}
|
||||
if serving, _, _ := srv.stale.snapshot(); serving {
|
||||
t.Error("serving_stale must stay false while backends are healthy")
|
||||
}
|
||||
|
||||
// Every backend down and the entry expired: the stale copy is served.
|
||||
clk.advance(31 * time.Second)
|
||||
a.setFail(true)
|
||||
b.setFail(true)
|
||||
stale := doGet(t, srv.Handler(), factsPath, "")
|
||||
if stale.Code != http.StatusOK {
|
||||
t.Fatalf("stale fallback status %d: %s", stale.Code, stale.Body.String())
|
||||
}
|
||||
if !strings.Contains(stale.Body.String(), `"new"`) {
|
||||
t.Errorf("stale body = %s, want the last cached copy", stale.Body.String())
|
||||
}
|
||||
|
||||
serving, served, last := srv.stale.snapshot()
|
||||
if !serving || served != 1 || last.IsZero() {
|
||||
t.Errorf("stale tracker = %v/%d/%v, want serving=true served=1", serving, served, last)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandler_NoCacheEntryMeansBackendFailureIs502(t *testing.T) {
|
||||
a := newCountingBackend(t, map[string]string{factsPath: `[]`})
|
||||
b := newCountingBackend(t, map[string]string{factsPath: `[]`})
|
||||
a.setFail(true)
|
||||
b.setFail(true)
|
||||
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
|
||||
|
||||
rec := doGet(t, srv.Handler(), factsPath, "")
|
||||
if rec.Code != http.StatusBadGateway {
|
||||
t.Fatalf("status %d, want 502", rec.Code)
|
||||
}
|
||||
if !strings.Contains(rec.Body.String(), "all backends failed") {
|
||||
t.Errorf("body = %q", rec.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandler_SingleFlightCollapsesConcurrentRequests(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()
|
||||
|
||||
release := make(chan struct{})
|
||||
a.setBlock(release)
|
||||
b.setBlock(release)
|
||||
|
||||
const callers = 16
|
||||
var wg sync.WaitGroup
|
||||
codes := make([]int, callers)
|
||||
for i := range callers {
|
||||
wg.Add(1)
|
||||
go func(i int) {
|
||||
defer wg.Done()
|
||||
codes[i] = doGet(t, h, factsPath, "").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 })
|
||||
time.Sleep(250 * time.Millisecond)
|
||||
close(release)
|
||||
wg.Wait()
|
||||
|
||||
for i, code := range codes {
|
||||
if code != http.StatusOK {
|
||||
t.Fatalf("caller %d got %d", i, code)
|
||||
}
|
||||
}
|
||||
if got := a.hitCount(factsPath); got != 1 {
|
||||
t.Errorf("backend a saw %d requests, want 1 for %d concurrent callers", got, callers)
|
||||
}
|
||||
if got := b.hitCount(factsPath); got != 1 {
|
||||
t.Errorf("backend b saw %d requests, want 1 for %d concurrent callers", got, callers)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFlightGroup_LeaderRunsOnce(t *testing.T) {
|
||||
var g flightGroup
|
||||
var calls, shared atomic.Int64
|
||||
entered := make(chan struct{})
|
||||
release := make(chan struct{})
|
||||
|
||||
run := func() (cachedResponse, error) {
|
||||
if calls.Add(1) == 1 {
|
||||
close(entered)
|
||||
}
|
||||
<-release
|
||||
return cachedResponse{Body: json.RawMessage(`[]`), Records: -1}, nil
|
||||
}
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for range 16 {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
if _, _, s := g.Do("k", run); s {
|
||||
shared.Add(1)
|
||||
}
|
||||
}()
|
||||
}
|
||||
<-entered
|
||||
time.Sleep(250 * time.Millisecond)
|
||||
close(release)
|
||||
wg.Wait()
|
||||
|
||||
if calls.Load() != 1 {
|
||||
t.Errorf("fn ran %d times, want 1", calls.Load())
|
||||
}
|
||||
if shared.Load() != 15 {
|
||||
t.Errorf("%d callers shared the flight, want 15", shared.Load())
|
||||
}
|
||||
// The key is released once the flight finishes.
|
||||
if _, _, s := g.Do("k", func() (cachedResponse, error) { return cachedResponse{Records: -1}, nil }); s {
|
||||
t.Error("a later call must start its own flight")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandler_HealthzReportsCacheState(t *testing.T) {
|
||||
a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "role", "web", "") + `]`})
|
||||
b := newCountingBackend(t, map[string]string{factsPath: `[]`})
|
||||
srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
|
||||
|
||||
hr := health(t, srv)
|
||||
if hr.Cache.Backend != "memory" || hr.Cache.TTL != "30s" {
|
||||
t.Fatalf("cache health = %+v, want memory/30s", hr.Cache)
|
||||
}
|
||||
if hr.Cache.Entries != 0 || hr.Cache.ServingStale {
|
||||
t.Fatalf("a cold cache should be empty and not stale: %+v", hr.Cache)
|
||||
}
|
||||
|
||||
doGet(t, srv.Handler(), factsPath, "")
|
||||
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)
|
||||
}
|
||||
|
||||
clk.advance(31 * time.Second)
|
||||
if hr = health(t, srv); hr.Cache.StaleEntries != 1 {
|
||||
t.Fatalf("expired entry should count as stale: %+v", hr.Cache)
|
||||
}
|
||||
if hr.Cache.ServingStale {
|
||||
t.Error("an expired entry alone is not serving_stale")
|
||||
}
|
||||
|
||||
a.setFail(true)
|
||||
b.setFail(true)
|
||||
doGet(t, srv.Handler(), factsPath, "")
|
||||
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)
|
||||
}
|
||||
if hr.Status != "down" {
|
||||
t.Errorf("status = %q, want down", hr.Status)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandler_HealthzReportsDisabledCache(t *testing.T) {
|
||||
a := newCountingBackend(t, nil)
|
||||
b := newCountingBackend(t, nil)
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) // FactsTTL zero
|
||||
|
||||
hr := health(t, srv)
|
||||
if hr.Cache.Backend != "none" || hr.Cache.TTL != "0" {
|
||||
t.Errorf("disabled cache health = %+v, want none/0", hr.Cache)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandler_UncachedPathsStillRefetch(t *testing.T) {
|
||||
a := newCountingBackend(t, map[string]string{reportsPath: `[{"hash":"h"}]`})
|
||||
b := newCountingBackend(t, map[string]string{reportsPath: `[]`})
|
||||
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
|
||||
|
||||
doGet(t, srv.Handler(), reportsPath, "")
|
||||
doGet(t, srv.Handler(), reportsPath, "")
|
||||
if got := a.hitCount(reportsPath); got != 2 {
|
||||
t.Errorf("/reports is uncached: %d requests, want 2", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandler_NodesRecordSetCached(t *testing.T) {
|
||||
a := newCountingBackend(t, map[string]string{nodesPath: `[{"certname":"h1"}]`})
|
||||
b := newCountingBackend(t, map[string]string{nodesPath: `[]`})
|
||||
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
|
||||
|
||||
doGet(t, srv.Handler(), nodesPath, "")
|
||||
doGet(t, srv.Handler(), nodesPath, "")
|
||||
if got := a.hitCount(nodesPath); got != 1 {
|
||||
t.Errorf("merged /nodes is cached: %d requests, want 1", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandler_NodesAggregateNotCached(t *testing.T) {
|
||||
a := newCountingBackend(t, map[string]string{nodesPath: `[{"count":90}]`})
|
||||
b := newCountingBackend(t, map[string]string{nodesPath: `[{"count":53}]`})
|
||||
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
|
||||
|
||||
first := doGet(t, srv.Handler(), nodesPath, nodeCountQuery)
|
||||
second := doGet(t, srv.Handler(), nodesPath, nodeCountQuery)
|
||||
if first.Code != http.StatusOK || second.Code != http.StatusOK {
|
||||
t.Fatalf("statuses %d/%d", first.Code, second.Code)
|
||||
}
|
||||
if got := a.hitCount(nodesPath); got != 2 {
|
||||
t.Errorf("/nodes aggregates are uncached: %d requests, want 2", got)
|
||||
}
|
||||
if got := counts(t, second.Body.Bytes(), "count"); !slices.Equal(got, []float64{143}) {
|
||||
t.Errorf("count = %v, want [143]", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandler_ResourcesAggregateNotCached(t *testing.T) {
|
||||
a := newCountingBackend(t, map[string]string{resourcesPath: `[{"count":7}]`})
|
||||
b := newCountingBackend(t, map[string]string{resourcesPath: `[{"count":5}]`})
|
||||
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
|
||||
|
||||
doGet(t, srv.Handler(), resourcesPath, nodeCountQuery)
|
||||
doGet(t, srv.Handler(), resourcesPath, nodeCountQuery)
|
||||
if got := a.hitCount(resourcesPath); got != 2 {
|
||||
t.Errorf("/resources aggregates are uncached: %d requests, want 2", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandler_MetaAndMetricsNotCached(t *testing.T) {
|
||||
bodies := map[string]string{
|
||||
metaVersionPath: `{"version":"8.4.0"}`,
|
||||
metaServerTimePath: `{"server_time":"2026-01-01T00:00:00.000Z"}`,
|
||||
"/metrics/v2/read/x": `{"request":{},"value":{"Value":1},"status":200}`,
|
||||
}
|
||||
a := newCountingBackend(t, bodies)
|
||||
b := newCountingBackend(t, bodies)
|
||||
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
|
||||
|
||||
for path := range bodies {
|
||||
doGet(t, srv.Handler(), path, "")
|
||||
doGet(t, srv.Handler(), path, "")
|
||||
if got := a.hitCount(path); got != 2 {
|
||||
t.Errorf("%s is uncached: %d requests, want 2", path, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestConfig_FactsTTLClampedAtCap(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
t.Setenv("XDG_CONFIG_HOME", dir)
|
||||
clearEnv(t)
|
||||
|
||||
path := filepath.Join(dir, appName, configFileName)
|
||||
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(path, []byte("facts_ttl: 5m\n"), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
cfg, err := Load("")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if cfg.FactsTTL != maxFactsTTL {
|
||||
t.Errorf("file facts_ttl = %s, want it clamped to %s", cfg.FactsTTL, maxFactsTTL)
|
||||
}
|
||||
|
||||
t.Setenv(envPrefix+"FACTS_TTL", "10m")
|
||||
cfg, err = Load("")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if cfg.FactsTTL != maxFactsTTL {
|
||||
t.Errorf("env facts_ttl = %s, want it clamped to %s", cfg.FactsTTL, maxFactsTTL)
|
||||
}
|
||||
if cfg.factsTTLClamped != 10*time.Minute {
|
||||
t.Errorf("pre-clamp value = %s, want 10m0s", cfg.factsTTLClamped)
|
||||
}
|
||||
|
||||
out := captureStdout(t, func() { printConfig(cfg) })
|
||||
if !strings.Contains(out, "facts_ttl : 30s (clamped from 600s, cap 30s)") {
|
||||
t.Errorf("config show must report the clamp, got:\n%s", out)
|
||||
}
|
||||
|
||||
t.Setenv(envPrefix+"FACTS_TTL", "5s")
|
||||
cfg, err = Load("")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if cfg.FactsTTL != 5*time.Second || cfg.factsTTLClamped != 0 {
|
||||
t.Errorf("a value under the cap must pass through: %s / %s", cfg.FactsTTL, cfg.factsTTLClamped)
|
||||
}
|
||||
|
||||
t.Setenv(envPrefix+"FACTS_TTL", "0s")
|
||||
cfg, err = Load("")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if cfg.cacheEnabled() {
|
||||
t.Error("facts_ttl 0 must disable the cache")
|
||||
}
|
||||
out = captureStdout(t, func() { printConfig(cfg) })
|
||||
if !strings.Contains(out, "(cache disabled)") {
|
||||
t.Errorf("config show must say the cache is off, got:\n%s", out)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewServer_ClampsFactsTTL(t *testing.T) {
|
||||
cfg := testConfig("http://a.invalid", "http://b.invalid", mergeStatic)
|
||||
cfg.FactsTTL = time.Hour
|
||||
cfg.CacheBytes = 1 << 20
|
||||
srv := newTestServer(cfg)
|
||||
|
||||
mc, ok := srv.factsCache.(*memoryCache)
|
||||
if !ok {
|
||||
t.Fatalf("expected a memory cache, got %T", srv.factsCache)
|
||||
}
|
||||
if mc.ttl != maxFactsTTL {
|
||||
t.Errorf("cache ttl = %s, want %s", mc.ttl, maxFactsTTL)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConfig_ValidateRejectsNegativeCacheSettings(t *testing.T) {
|
||||
base := testConfig("http://a.invalid", "http://b.invalid", mergeStatic)
|
||||
for name, mutate := range map[string]func(*Config){
|
||||
"negative facts_ttl": func(c *Config) { c.FactsTTL = -time.Second },
|
||||
"negative facts_cache_bytes": func(c *Config) { c.CacheBytes = -1 },
|
||||
} {
|
||||
cfg := base
|
||||
mutate(&cfg)
|
||||
if err := cfg.Validate(); err == nil {
|
||||
t.Errorf("%s should not validate", name)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func health(t *testing.T, srv *Server) healthReport {
|
||||
t.Helper()
|
||||
rec := doGet(t, srv.Handler(), "/healthz", "")
|
||||
var hr healthReport
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &hr); err != nil {
|
||||
t.Fatalf("decode healthz: %v (%s)", err, rec.Body.String())
|
||||
}
|
||||
return hr
|
||||
}
|
||||
|
||||
func waitFor(t *testing.T, cond func() bool) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(5 * time.Second)
|
||||
for !cond() {
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatal("condition not met within 5s")
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
}
|
||||
@@ -27,6 +27,13 @@ const (
|
||||
defaultFreshnessTTL = 30 * time.Second
|
||||
|
||||
defaultSourceFact = "pdbmux_source"
|
||||
|
||||
// maxFactsTTL is a hard cap, not a default: a larger configured value is
|
||||
// clamped down to it rather than rejected, so a stray env var cannot make a
|
||||
// container crash-loop.
|
||||
maxFactsTTL = 30 * time.Second
|
||||
defaultFactsTTL = 30 * time.Second
|
||||
defaultCacheSize = int64(64 << 20)
|
||||
)
|
||||
|
||||
var exampleBackends = []Backend{
|
||||
@@ -45,11 +52,14 @@ 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
|
||||
CacheBytes int64 `yaml:"facts_cache_bytes"` // byte budget for that cache
|
||||
|
||||
SourceFact string `yaml:"source_fact"`
|
||||
SourceFactEnabled bool `yaml:"source_fact_enabled"`
|
||||
|
||||
sourcePath string // file this config was read from, empty if none was found
|
||||
sourcePath string // file this config was read from, empty if none was found
|
||||
factsTTLClamped time.Duration // pre-clamp facts_ttl, zero when nothing was clamped
|
||||
}
|
||||
|
||||
// SourcePath returns the config file Load read, or "" when none was loaded.
|
||||
@@ -66,6 +76,8 @@ func DefaultConfig() Config {
|
||||
Merge: mergeFreshness,
|
||||
Timeout: defaultTimeout,
|
||||
FreshnessTTL: defaultFreshnessTTL,
|
||||
FactsTTL: defaultFactsTTL,
|
||||
CacheBytes: defaultCacheSize,
|
||||
SourceFact: defaultSourceFact,
|
||||
SourceFactEnabled: true,
|
||||
}
|
||||
@@ -146,9 +158,19 @@ func Load(flagPath string) (Config, error) {
|
||||
}
|
||||
|
||||
applyEnv(&cfg, os.Getenv)
|
||||
cfg.clampFactsTTL()
|
||||
return cfg, nil
|
||||
}
|
||||
|
||||
// clampFactsTTL pins facts_ttl to maxFactsTTL, remembering the configured value
|
||||
// so `config show` can say the cap was applied.
|
||||
func (c *Config) clampFactsTTL() {
|
||||
if c.FactsTTL > maxFactsTTL {
|
||||
c.factsTTLClamped = c.FactsTTL
|
||||
c.FactsTTL = maxFactsTTL
|
||||
}
|
||||
}
|
||||
|
||||
func applyEnv(cfg *Config, getenv func(string) string) {
|
||||
if v := getenv(envPrefix + "LISTEN"); v != "" {
|
||||
cfg.Listen = v
|
||||
@@ -174,6 +196,16 @@ func applyEnv(cfg *Config, getenv func(string) string) {
|
||||
cfg.SourceFactEnabled = b
|
||||
}
|
||||
}
|
||||
if v := getenv(envPrefix + "FACTS_TTL"); v != "" {
|
||||
if d, err := time.ParseDuration(v); err == nil {
|
||||
cfg.FactsTTL = d
|
||||
}
|
||||
}
|
||||
if v := getenv(envPrefix + "FACTS_CACHE_BYTES"); v != "" {
|
||||
if n, err := strconv.ParseInt(v, 10, 64); err == nil {
|
||||
cfg.CacheBytes = n
|
||||
}
|
||||
}
|
||||
if v := getenv(envPrefix + "BACKENDS"); v != "" {
|
||||
if bs := parseBackends(v); len(bs) > 0 {
|
||||
cfg.Backends = bs
|
||||
@@ -234,9 +266,19 @@ func (c Config) Validate() error {
|
||||
if c.SourceFactEnabled && c.SourceFact == "" {
|
||||
return fmt.Errorf("source_fact must be non-empty, or set source_fact_enabled to false")
|
||||
}
|
||||
if c.FactsTTL < 0 {
|
||||
return fmt.Errorf("facts_ttl must not be negative (0 disables the cache)")
|
||||
}
|
||||
if c.CacheBytes < 0 {
|
||||
return fmt.Errorf("facts_cache_bytes must not be negative")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// cacheEnabled reports whether a facts/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 {
|
||||
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
||||
return fmt.Errorf("creating config dir: %w", err)
|
||||
@@ -249,8 +291,11 @@ func writeDefaultConfig(path string) error {
|
||||
"# 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_MERGE, PDBMUX_TIMEOUT,\n" +
|
||||
"# PDBMUX_FRESHNESS_TTL, PDBMUX_BACKENDS (name=url,name=url),\n" +
|
||||
"# PDBMUX_SOURCE_FACT, PDBMUX_SOURCE_FACT_ENABLED.\n\n")
|
||||
"# PDBMUX_FRESHNESS_TTL, PDBMUX_FACTS_TTL, PDBMUX_FACTS_CACHE_BYTES,\n" +
|
||||
"# PDBMUX_BACKENDS (name=url,name=url),\n" +
|
||||
"# PDBMUX_SOURCE_FACT, PDBMUX_SOURCE_FACT_ENABLED.\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\n")
|
||||
if err := os.WriteFile(path, append(header, data...), 0o644); err != nil {
|
||||
return fmt.Errorf("writing config: %w", err)
|
||||
}
|
||||
|
||||
+1
-1
@@ -456,7 +456,7 @@ func captureStdout(t *testing.T, f func()) string {
|
||||
|
||||
func clearEnv(t *testing.T) {
|
||||
t.Helper()
|
||||
for _, k := range []string{"CONFIG", "LISTEN", "MERGE", "TIMEOUT", "FRESHNESS_TTL", "BACKENDS"} {
|
||||
for _, k := range []string{"CONFIG", "LISTEN", "MERGE", "TIMEOUT", "FRESHNESS_TTL", "FACTS_TTL", "FACTS_CACHE_BYTES", "BACKENDS"} {
|
||||
t.Setenv(envPrefix+k, "")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -151,6 +151,18 @@ func runServer(cfg Config) error {
|
||||
}
|
||||
}
|
||||
|
||||
func factsTTLString(cfg Config) string {
|
||||
s := durationString(cfg.FactsTTL)
|
||||
switch {
|
||||
case cfg.factsTTLClamped > 0:
|
||||
return fmt.Sprintf("%s (clamped from %s, cap %s)",
|
||||
s, durationString(cfg.factsTTLClamped), durationString(maxFactsTTL))
|
||||
case !cfg.cacheEnabled():
|
||||
return s + " (cache disabled)"
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
func printConfig(cfg Config) {
|
||||
if p := cfg.SourcePath(); p != "" {
|
||||
fmt.Printf("config file : %s (loaded)\n", p)
|
||||
@@ -166,6 +178,8 @@ func printConfig(cfg Config) {
|
||||
} else {
|
||||
fmt.Printf("source_fact : disabled\n")
|
||||
}
|
||||
fmt.Printf("facts_ttl : %s\n", factsTTLString(cfg))
|
||||
fmt.Printf("facts_cache : %d bytes\n", cfg.CacheBytes)
|
||||
fmt.Println("backends:")
|
||||
for _, b := range cfg.Backends {
|
||||
fmt.Printf(" - %-8s %s\n", b.Name, b.URL)
|
||||
|
||||
@@ -3,6 +3,7 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
@@ -35,11 +36,18 @@ type backendResult struct {
|
||||
err error
|
||||
}
|
||||
|
||||
var errAllBackendsFailed = errors.New("all backends failed")
|
||||
|
||||
type Server struct {
|
||||
cfg Config
|
||||
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
|
||||
|
||||
// freshness cache (freshness merge only).
|
||||
mu sync.Mutex
|
||||
freshData freshness
|
||||
@@ -47,11 +55,35 @@ type Server struct {
|
||||
}
|
||||
|
||||
func NewServer(cfg Config, logger *log.Logger) *Server {
|
||||
return &Server{
|
||||
cfg.clampFactsTTL()
|
||||
s := &Server{
|
||||
cfg: cfg,
|
||||
client: &http.Client{Timeout: cfg.Timeout},
|
||||
log: logger,
|
||||
}
|
||||
if cfg.cacheEnabled() {
|
||||
s.factsCache = newMemoryCache(cfg.FactsTTL, cfg.CacheBytes)
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// cacheFor picks the cache backing a request. Merged /facts and /nodes record
|
||||
// sets 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.
|
||||
func (s *Server) cacheFor(path string, params url.Values) (Cache, bool) {
|
||||
switch path {
|
||||
case factsPath, nodesPath:
|
||||
// An aggregate row is a summed count, not the merged record set the
|
||||
// cache was built for, so it stays on the live path.
|
||||
if parseAggregate(params.Get("query")) != nil {
|
||||
return noopCache{}, false
|
||||
}
|
||||
if s.factsCache != nil {
|
||||
return s.factsCache, true
|
||||
}
|
||||
}
|
||||
return noopCache{}, false
|
||||
}
|
||||
|
||||
func (s *Server) Handler() http.Handler {
|
||||
@@ -109,11 +141,14 @@ func isReportSubResource(path string) bool {
|
||||
}
|
||||
|
||||
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 {
|
||||
return
|
||||
}
|
||||
writeJSON(w, merge(alive))
|
||||
params := queryParams(r.URL.Query().Get("query"))
|
||||
s.serveCached(w, r, path, params, func() (cachedResponse, error) {
|
||||
alive, err := s.aliveResults(r.Context(), path, params)
|
||||
if err != nil {
|
||||
return cachedResponse{}, err
|
||||
}
|
||||
return cachedResponse{Body: encodeRecords(merge(alive)), Records: -1}, nil
|
||||
})
|
||||
}
|
||||
|
||||
// Reports and events are immutable history, so both backends' records belong in the merged view.
|
||||
@@ -125,19 +160,23 @@ func (s *Server) serveUnion(w http.ResponseWriter, r *http.Request, path string,
|
||||
return
|
||||
}
|
||||
|
||||
alive, ok := s.aliveResults(w, r, path, page.upstreamParams(in))
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
merged := mergeUnion(alive, key)
|
||||
sortRecords(merged, page.order)
|
||||
if page.wantTotal {
|
||||
if total := sumTotals(alive); total >= 0 {
|
||||
w.Header().Set(recordsHeader, strconv.Itoa(total))
|
||||
// Keyed on the request's own params, not the upstream ones: upstreamParams
|
||||
// folds offset into limit, so different windows would collide.
|
||||
s.serveCached(w, r, path, in, func() (cachedResponse, error) {
|
||||
alive, err := s.aliveResults(r.Context(), path, page.upstreamParams(in))
|
||||
if err != nil {
|
||||
return cachedResponse{}, err
|
||||
}
|
||||
}
|
||||
writeJSON(w, page.apply(merged))
|
||||
merged := mergeUnion(alive, key)
|
||||
sortRecords(merged, page.order)
|
||||
resp := cachedResponse{Body: encodeRecords(page.apply(merged)), Records: -1}
|
||||
if page.wantTotal {
|
||||
if total := sumTotals(alive); total >= 0 {
|
||||
resp.Records = total
|
||||
}
|
||||
}
|
||||
return resp, nil
|
||||
})
|
||||
}
|
||||
|
||||
// A count row carries no certname, so the certname-keyed merge would collapse every backend's count into one backend's; aggregates take the summing path instead.
|
||||
@@ -176,17 +215,19 @@ func (s *Server) serveSummed(w http.ResponseWriter, r *http.Request, path string
|
||||
return
|
||||
}
|
||||
|
||||
alive, ok := s.aliveResults(w, r, path, page.upstreamParams(in))
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
merged := sumRows(alive, columns)
|
||||
sortRecords(merged, page.order)
|
||||
if page.wantTotal {
|
||||
w.Header().Set(recordsHeader, strconv.Itoa(len(merged)))
|
||||
}
|
||||
writeJSON(w, page.apply(merged))
|
||||
s.serveCached(w, r, path, in, func() (cachedResponse, error) {
|
||||
alive, err := s.aliveResults(r.Context(), path, page.upstreamParams(in))
|
||||
if err != nil {
|
||||
return cachedResponse{}, err
|
||||
}
|
||||
merged := sumRows(alive, columns)
|
||||
sortRecords(merged, page.order)
|
||||
resp := cachedResponse{Body: encodeRecords(page.apply(merged)), Records: -1}
|
||||
if page.wantTotal {
|
||||
resp.Records = len(merged)
|
||||
}
|
||||
return resp, nil
|
||||
})
|
||||
}
|
||||
|
||||
// A backend without the report answers 404, indistinguishable from a failure, so every backend is consulted before serving empty.
|
||||
@@ -214,9 +255,9 @@ func (s *Server) serveFirstHolder(w http.ResponseWriter, r *http.Request) {
|
||||
writeJSON(w, nil)
|
||||
}
|
||||
|
||||
// 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)
|
||||
// Returns errAllBackendsFailed only when every backend failed.
|
||||
func (s *Server) aliveResults(ctx context.Context, path string, params url.Values) ([]backendResult, error) {
|
||||
results := s.fanOut(ctx, path, params)
|
||||
|
||||
var alive []backendResult
|
||||
for _, res := range results {
|
||||
@@ -227,10 +268,114 @@ func (s *Server) aliveResults(w http.ResponseWriter, r *http.Request, path strin
|
||||
alive = append(alive, res)
|
||||
}
|
||||
if len(alive) == 0 {
|
||||
http.Error(w, "all backends failed", http.StatusBadGateway)
|
||||
return nil, false
|
||||
return nil, errAllBackendsFailed
|
||||
}
|
||||
return alive, true
|
||||
return alive, nil
|
||||
}
|
||||
|
||||
// cachedResponse is the stored form of a merged response: the JSON body plus the
|
||||
// X-Records value it carried, so a cache hit reproduces both.
|
||||
type cachedResponse struct {
|
||||
Body json.RawMessage `json:"body"`
|
||||
Records int `json:"records"` // -1 when the response sets no X-Records
|
||||
}
|
||||
|
||||
// serveCached answers from the cache when the entry is fresh, otherwise runs
|
||||
// build — single-flighted, so N concurrent identical requests cause one upstream
|
||||
// fan-out — and stores the result. A build failure falls back to a stale entry
|
||||
// when one exists; that is the only path on which stale data is served. Paths
|
||||
// with no cache configured run build directly, unchanged.
|
||||
func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string, params url.Values, build func() (cachedResponse, error)) {
|
||||
cache, enabled := s.cacheFor(path, params)
|
||||
if !enabled {
|
||||
resp, err := build()
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), http.StatusBadGateway)
|
||||
return
|
||||
}
|
||||
writeCached(w, resp)
|
||||
return
|
||||
}
|
||||
|
||||
key := cacheKey(path, params)
|
||||
var stale *CacheEntry
|
||||
ent, status, err := cache.Get(r.Context(), key)
|
||||
switch {
|
||||
case err != nil:
|
||||
s.log.Printf("warning: cache lookup for %s failed: %v", key, err)
|
||||
case status == CacheFresh:
|
||||
s.stale.markFresh()
|
||||
s.writeStored(w, ent.Body)
|
||||
return
|
||||
case status == CacheStale:
|
||||
stale = &ent
|
||||
}
|
||||
|
||||
resp, err, _ := s.flights.Do(key, func() (cachedResponse, error) {
|
||||
built, buildErr := build()
|
||||
if buildErr != nil {
|
||||
return cachedResponse{}, buildErr
|
||||
}
|
||||
body, marshalErr := json.Marshal(built)
|
||||
if marshalErr != nil {
|
||||
s.log.Printf("warning: encoding cache entry for %s failed: %v", key, marshalErr)
|
||||
return built, nil
|
||||
}
|
||||
// The leader's request may be cancelled while followers still wait.
|
||||
if putErr := cache.Put(context.WithoutCancel(r.Context()), key, body); putErr != nil {
|
||||
s.log.Printf("warning: cache store for %s failed: %v", key, putErr)
|
||||
}
|
||||
return built, nil
|
||||
})
|
||||
if err != nil {
|
||||
if stale != nil {
|
||||
s.stale.markStale(time.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.Body)
|
||||
return
|
||||
}
|
||||
http.Error(w, err.Error(), http.StatusBadGateway)
|
||||
return
|
||||
}
|
||||
s.stale.markFresh()
|
||||
writeCached(w, resp)
|
||||
}
|
||||
|
||||
func (s *Server) writeStored(w http.ResponseWriter, body []byte) {
|
||||
var resp cachedResponse
|
||||
if err := json.Unmarshal(body, &resp); err != nil {
|
||||
s.log.Printf("warning: unreadable cache entry: %v", err)
|
||||
http.Error(w, "unreadable cache entry", http.StatusBadGateway)
|
||||
return
|
||||
}
|
||||
writeCached(w, resp)
|
||||
}
|
||||
|
||||
func writeCached(w http.ResponseWriter, resp cachedResponse) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
if resp.Records >= 0 {
|
||||
w.Header().Set(recordsHeader, strconv.Itoa(resp.Records))
|
||||
}
|
||||
// resp.Body is shared with the cache and with every caller of a single
|
||||
// flight, so it is written, never appended to.
|
||||
body := []byte(resp.Body)
|
||||
if len(body) == 0 {
|
||||
body = []byte("[]")
|
||||
}
|
||||
_, _ = w.Write(body)
|
||||
_, _ = w.Write([]byte("\n"))
|
||||
}
|
||||
|
||||
func encodeRecords(recs []json.RawMessage) json.RawMessage {
|
||||
if recs == nil {
|
||||
recs = []json.RawMessage{}
|
||||
}
|
||||
b, err := json.Marshal(recs)
|
||||
if err != nil {
|
||||
return json.RawMessage("[]")
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
func queryParams(query string) url.Values {
|
||||
@@ -406,13 +551,48 @@ func setContentType(w http.ResponseWriter, contentType string) {
|
||||
type healthReport struct {
|
||||
Status string `json:"status"`
|
||||
Backends map[string]string `json:"backends"` // name -> "ok" | error text
|
||||
Cache cacheHealth `json:"cache"`
|
||||
}
|
||||
|
||||
type cacheHealth struct {
|
||||
Backend string `json:"backend"` // "memory" | "none"
|
||||
TTL string `json:"ttl"`
|
||||
Entries int `json:"entries"`
|
||||
StaleEntries int `json:"stale_entries"` // cached entries past their TTL
|
||||
Bytes int64 `json:"bytes"`
|
||||
ServingStale bool `json:"serving_stale"` // last cached response came from a stale entry
|
||||
StaleServed uint64 `json:"stale_served"`
|
||||
LastStale string `json:"last_stale_served,omitempty"`
|
||||
}
|
||||
|
||||
func (s *Server) cacheHealth() cacheHealth {
|
||||
stats := CacheStats{Backend: "none"}
|
||||
ttl := time.Duration(0)
|
||||
if s.factsCache != nil {
|
||||
stats = s.factsCache.Stats()
|
||||
ttl = s.cfg.FactsTTL
|
||||
}
|
||||
serving, served, last := s.stale.snapshot()
|
||||
h := cacheHealth{
|
||||
Backend: stats.Backend,
|
||||
TTL: durationString(ttl),
|
||||
Entries: stats.Entries,
|
||||
StaleEntries: stats.StaleEntries,
|
||||
Bytes: stats.Bytes,
|
||||
ServingStale: serving,
|
||||
StaleServed: served,
|
||||
}
|
||||
if !last.IsZero() {
|
||||
h.LastStale = last.UTC().Format(time.RFC3339)
|
||||
}
|
||||
return h
|
||||
}
|
||||
|
||||
func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {
|
||||
probe := `["=","certname","pdbmux-healthz-probe"]`
|
||||
results := s.fanOut(r.Context(), nodesPath, queryParams(probe))
|
||||
|
||||
report := healthReport{Backends: map[string]string{}}
|
||||
report := healthReport{Backends: map[string]string{}, Cache: s.cacheHealth()}
|
||||
healthy := 0
|
||||
for _, res := range results {
|
||||
if res.err != nil {
|
||||
|
||||
Reference in New Issue
Block a user