From abf565b0f68bab13647f4d42e8999a4ed696da51 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 6 Sep 2026 16:51:00 +1000 Subject: [PATCH] Stop the source-fact drilldown fanning out per pinned value The path segment is client-supplied and reached a full unfiltered /facts fan-out, keyed per value, so every distinct value was a fresh whole-estate query and a fresh cache entry. - validate against the configured backend names, answering [] with no fan-out when it names none - key the drilldown's fetch on the fact name alone and apply to the shared record set, so all values share one entry and one fan-out - report every configured backend on the no-fan-out empty response, which is complete rather than partial --- README.md | 22 +++++++++----- cache_test.go | 12 ++++++++ factroutes_test.go | 60 +++++++++++++++++++++++++++++++++++++ server.go | 75 ++++++++++++++++++++++++++++++++++++++++------ 4 files changed, 153 insertions(+), 16 deletions(-) diff --git a/README.md b/README.md index a5597aa..a45d752 100644 --- a/README.md +++ b/README.md @@ -182,12 +182,18 @@ records from the `/facts` merge that produces them, which makes the `certname` set, the owner and the `environment` identical to the ones an unfiltered `/facts` response reports, and lets the request's own `query` narrow the result upstream. That costs one `/facts` fan-out per cache miss — the widest fan-out `pdbmux` -makes — on a rare, user-initiated path. `/facts/pdbmux_source/` pins the -backend name, so it answers with the nodes that backend owns, and with `[]` for a -value naming no backend. An `extract`/`count` query still takes the summing -branch, and the gated query shapes above still answer `[]`, as does every form -while injection is off — with the name kept out of `/fact-names`, since nothing -then produces it. +makes — on a rare, user-initiated path. + +`/facts/pdbmux_source/` pins the backend name, so it answers with the +nodes that backend owns. The `` segment never reaches that fan-out: the +synthetic record's value is always a backend name, so a value naming none is +answered `[]` from the configured names alone, with no fan-out at all, and a +value naming one filters a record set fetched under a key the value is not part +of. The record set is a property of the estate rather than of the filter, so +every value of it — and the unfiltered path — share one entry and one fetch. +An `extract`/`count` query still takes the summing branch, and the gated query +shapes above still answer `[]`, as does every form while injection is off — with +the name kept out of `/fact-names`, since nothing then produces it. **Not supported in v1: server-side filtering on the fact.** A query that selects it — `["=","name","pdbmux_source"]`, or an `extract` naming it — is forwarded to @@ -407,7 +413,9 @@ backend later without further handler changes. requests can only share an entry when they share a key, and the key is path plus query, which is exactly what decides whether injection applies; a name-filtered `/facts` query and a plain one therefore cache separately and - neither is ever served the other's shape. `source_fact` and + neither is ever served the other's shape. The one path that keys on less than + it is asked is the `pdbmux_source` drilldown, whose `` is dropped from + the key and applied to the shared entry instead. `source_fact` and `source_fact_enabled` are read once at startup, and the cache lives for the same process, so changing either cannot leave differently-shaped entries behind. diff --git a/cache_test.go b/cache_test.go index 9f29901..33c8618 100644 --- a/cache_test.go +++ b/cache_test.go @@ -96,6 +96,18 @@ func (cb *countingBackend) hitCount(path string) int { return cb.hits[path] } +// totalHits counts requests across every path, so a test can assert a route +// reached no backend at all. +func (cb *countingBackend) totalHits() int { + cb.mu.Lock() + defer cb.mu.Unlock() + n := 0 + for _, c := range cb.hits { + n += c + } + return n +} + func (cb *countingBackend) setFail(v bool) { cb.mu.Lock() defer cb.mu.Unlock() diff --git a/factroutes_test.go b/factroutes_test.go index 722dd21..9f76369 100644 --- a/factroutes_test.go +++ b/factroutes_test.go @@ -337,6 +337,66 @@ func TestHandler_FactsBySourceNameAndValueFiltersByOwner(t *testing.T) { } } +// The pinned value is client-supplied and the drilldown's fetch is the widest +// query pdbmux makes, so a value naming no backend is answered from the +// configured names alone: no fan-out, and so no per-value cache key either. +func TestHandler_FactsBySourceNameUnknownValueSkipsFanOut(t *testing.T) { + facts := map[string]string{factsPath: `[` + fact("h1", "osfamily", "RedHat", "") + `]`} + a := newCountingBackend(t, facts) + b := newCountingBackend(t, facts) + srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + + for _, value := range []string{"nosuchbackend", "nonce-1", "nonce-2"} { + rec := doGet(t, srv.Handler(), sourceFactURL+"/"+value, "") + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + if got := rec.Body.String(); got != "[]\n" { + t.Errorf("%s/%s = %q, want %q", sourceFactURL, value, got, "[]\n") + } + // The answer is complete, not built from a subset of backends. + if h := rec.Header().Get(backendsHeader); h != "2/2" { + t.Errorf("%s/%s %s = %q, want 2/2", sourceFactURL, value, backendsHeader, h) + } + } + for _, cb := range []*countingBackend{a, b} { + if got := cb.totalHits(); got != 0 { + t.Errorf("unknown value reached a backend %d times, want 0", got) + } + } +} + +// The record set is the estate's, not the pinned value's, so every value shares +// one entry: two valid values must not cost two whole-estate fan-outs. +func TestHandler_FactsBySourceNameValuesShareOneFetch(t *testing.T) { + a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "osfamily", "RedHat", "") + `]`}) + b := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h2", "osfamily", "Debian", "") + `]`}) + srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + + for _, tc := range []struct { + path string + want map[string]string + }{ + {sourceFactURL + "/a", map[string]string{"h1": "a"}}, + {sourceFactURL + "/b", map[string]string{"h2": "b"}}, + {sourceFactURL, map[string]string{"h1": "a", "h2": "b"}}, + } { + rec := doGet(t, srv.Handler(), tc.path, "") + got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact) + if n != len(tc.want) || !reflect.DeepEqual(got, tc.want) { + t.Fatalf("%s = %v (%d records), want %v", tc.path, got, n, tc.want) + } + } + for _, cb := range []*countingBackend{a, b} { + if got := cb.hitCount(factsPath); got != 1 { + t.Errorf("%s fetched %d times, want 1 shared fetch", factsPath, got) + } + if got := cb.hitCount(sourceFactURL); got != 0 { + t.Errorf("%s was fanned out %d times, want 0", sourceFactURL, got) + } + } +} + // Nothing produces the fact while injection is off, so the drilldown is empty // and the name is not advertised for a client to click through to. func TestHandler_FactsBySourceNamePathEmptyWhenDisabled(t *testing.T) { diff --git a/server.go b/server.go index 19914dc..0a4fa5d 100644 --- a/server.go +++ b/server.go @@ -335,22 +335,55 @@ func (s *Server) serveFactsByName(w http.ResponseWriter, r *http.Request, name, // records are taken from the /facts merge that produces them instead, which // makes the certname set, the owner and the environment identical to the ones an // unfiltered /facts response reports, and lets the request's query narrow the -// result upstream. That costs one extra fan-out on a rare, user-initiated path. +// result upstream. +// +// That merge is the widest query pdbmux makes and the pinned value is +// client-supplied, so the value never reaches it: a value naming no configured +// backend is answered empty without any fan-out, and a value naming one filters +// a record set fetched under a value-independent key. Otherwise each distinct +// value would be a fresh cache key, a fresh flight and a fresh whole-estate +// fan-out. func (s *Server) serveSourceFact(w http.ResponseWriter, r *http.Request, inject *sourceInjector, value string, valued bool) { + // The synthetic record's value is always a backend name, so any other value + // matches zero records. The response is complete rather than degraded, so it + // reports every configured backend. + if valued && !s.hasBackend(value) { + n := len(s.cfg.Backends) + writeCached(w, cachedResponse{Records: -1, Backends: n, Configured: n}) + return + } + params := queryParams(r.URL.Query().Get("query")) merge := s.mergeFactsWith(inject) - s.serveCached(w, r, r.URL.Path, params, func(ctx context.Context) (cachedResponse, error) { + var filter recordFilter + if valued { + filter = func(recs []json.RawMessage) []json.RawMessage { + return sourceFactRecords(recs, inject.name, value, true) + } + } + // The stored set is the whole owned-fact record set, so every value of it + // keys, and waits on, the same fetch. + s.serveFiltered(w, r, factsPath+"/"+inject.name, params, filter, func(ctx context.Context) (cachedResponse, error) { alive, err := s.aliveResults(ctx, factsPath, params) if err != nil { return cachedResponse{}, err } - recs := sourceFactRecords(merge(alive), inject.name, value, valued) + recs := sourceFactRecords(merge(alive), inject.name, "", false) resp := cachedResponse{Body: encodeRecords(recs), Records: -1} s.countBackends(&resp, alive) return resp, nil }) } +func (s *Server) hasBackend(name string) bool { + for _, b := range s.cfg.Backends { + if b.Name == name { + return true + } + } + return false +} + // 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. func (s *Server) serveNodes(w http.ResponseWriter, r *http.Request) { if spec := parseAggregate(r.URL.Query().Get("query")); spec != nil { @@ -457,12 +490,36 @@ type cachedResponse struct { Configured int `json:"configured"` // backends configured at build time } +// recordFilter narrows a response's records after it has been built or read back +// from the cache, so requests differing only in the filter share one stored entry +// and one fan-out. It leaves Records alone, so it only suits responses that set +// no X-Records. +type recordFilter func([]json.RawMessage) []json.RawMessage + +func (f recordFilter) apply(resp cachedResponse) cachedResponse { + if f == nil { + return resp + } + var recs []json.RawMessage + if json.Unmarshal(resp.Body, &recs) != nil { + return resp + } + resp.Body = encodeRecords(f(recs)) + return resp +} + // 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(context.Context) (cachedResponse, error)) { + s.serveFiltered(w, r, path, params, nil, build) +} + +// serveFiltered is serveCached with a per-request narrowing applied to whatever +// the shared entry holds. +func (s *Server) serveFiltered(w http.ResponseWriter, r *http.Request, path string, params url.Values, filter recordFilter, build func(context.Context) (cachedResponse, error)) { cache, enabled := s.cacheFor(path, params) if !enabled { resp, err := build(r.Context()) @@ -470,7 +527,7 @@ func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string http.Error(w, err.Error(), http.StatusBadGateway) return } - writeCached(w, resp) + writeCached(w, filter.apply(resp)) return } @@ -482,7 +539,7 @@ func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string s.log.Printf("warning: cache lookup for %s failed: %v", key, err) case status == CacheFresh: s.stale.markFresh() - s.writeStored(w, ent, CacheFresh) + s.writeStored(w, ent, CacheFresh, filter) return case status == CacheStale: stale = &ent @@ -523,7 +580,7 @@ func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string s.stale.markStale(s.now()) s.log.Printf("warning: serving stale %s from cache (stored %s): %v", path, stale.StoredAt.UTC().Format(time.RFC3339), err) - s.writeStored(w, *stale, CacheStale) + s.writeStored(w, *stale, CacheStale, filter) return } http.Error(w, err.Error(), http.StatusBadGateway) @@ -531,7 +588,7 @@ func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string } s.stale.markFresh() s.setCacheHeaders(w, CacheMiss, time.Time{}) - writeCached(w, resp) + writeCached(w, filter.apply(resp)) } // http.Client reads a zero Timeout as "no deadline", but it would expire a @@ -543,7 +600,7 @@ func (s *Server) flightTimeout() time.Duration { return defaultTimeout } -func (s *Server) writeStored(w http.ResponseWriter, ent CacheEntry, status CacheStatus) { +func (s *Server) writeStored(w http.ResponseWriter, ent CacheEntry, status CacheStatus, filter recordFilter) { var resp cachedResponse if err := json.Unmarshal(ent.Body, &resp); err != nil { s.log.Printf("warning: unreadable cache entry: %v", err) @@ -551,7 +608,7 @@ func (s *Server) writeStored(w http.ResponseWriter, ent CacheEntry, status Cache return } s.setCacheHeaders(w, status, ent.StoredAt) - writeCached(w, resp) + writeCached(w, filter.apply(resp)) } // setCacheHeaders labels a response from a cache-backed path: X-Cache is