diff --git a/README.md b/README.md index 07fdba0..f080019 100644 --- a/README.md +++ b/README.md @@ -38,7 +38,7 @@ not PQL) is forwarded verbatim. | `GET /metrics/v2/read/` | 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[/]` | 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** — `?`, 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. diff --git a/cache.go b/cache.go new file mode 100644 index 0000000..962dcc4 --- /dev/null +++ b/cache.go @@ -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 "?" +// 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 +} diff --git a/cache_test.go b/cache_test.go new file mode 100644 index 0000000..c9231a9 --- /dev/null +++ b/cache_test.go @@ -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) + } +} diff --git a/config.go b/config.go index 6b5cddc..ae12542 100644 --- a/config.go +++ b/config.go @@ -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) } diff --git a/config_test.go b/config_test.go index bfdbf36..ef89ea4 100644 --- a/config_test.go +++ b/config_test.go @@ -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, "") } } diff --git a/main.go b/main.go index de1cd43..ebe0521 100644 --- a/main.go +++ b/main.go @@ -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) diff --git a/server.go b/server.go index 4e046e6..05f6c40 100644 --- a/server.go +++ b/server.go @@ -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 {