Merge pull request 'Merge the /facts/<name> and /fact-names routes' (#18) from benvin/merge-fact-routes into main
Reviewed-on: #18
This commit was merged in pull request #18.
This commit is contained in:
@@ -26,6 +26,8 @@ not PQL) is forwarded verbatim.
|
||||
|---|---|
|
||||
| `GET /pdb/query/v4/nodes` | Fan out to all backends, dedupe by `certname`, keep the record with the newer `report_timestamp`, stamped with the winning backend's name (see provenance). An `extract`/`count` query is **summed** instead. |
|
||||
| `GET /pdb/query/v4/facts` | Fan out to all, and per `certname` keep **all** facts from the backend that owns that node (see merge semantics), plus a synthetic `pdbmux_source` fact naming it. An `extract`/`count` query is **summed** instead. |
|
||||
| `GET /pdb/query/v4/facts/<name>[/<value>]` | Same fan-out and merge as `/facts`, and an `extract`/`count` query is **summed** the same way. The path segment is a `name` constraint, so no synthetic `pdbmux_source` record is added — except on the fact's own path, which is **synthesised** from the `/facts` merge (see provenance). |
|
||||
| `GET /pdb/query/v4/fact-names` | Fan out to all and serve the **union** of the flat name arrays, deduped and re-sorted, re-paged across backends, plus the `pdbmux_source` name while injection is on. `order_by` is only valid on `name`. |
|
||||
| `GET /pdb/query/v4/resources` | An `extract`/`count` query is fanned out and **summed**; any other query is an unmerged pass-through. |
|
||||
| `GET /pdb/query/v4/reports` | Fan out to all and serve the **union**, deduped by report `hash`, re-ordered and re-paged across backends. |
|
||||
| `GET /pdb/query/v4/events` | Fan out to all and serve the **union**, deduped by record identity, re-ordered and re-paged. |
|
||||
@@ -67,6 +69,18 @@ paths add two more headers `pdbmux` sets itself, `X-Cache` and `Age` — see
|
||||
facts from the first backend in configured order that holds it.
|
||||
- A node present in only one backend always appears (falls back to whichever
|
||||
backend actually returned facts for it).
|
||||
- `/facts/<name>` and `/facts/<name>/<value>` are the same records with one
|
||||
more constraint applied upstream, so they take the same rule — and the same
|
||||
aggregate branch, since a count row has no `certname` there either. The
|
||||
`pdbmux_source` path is the exception: no backend holds that name, so it is
|
||||
synthesised from the `/facts` merge (see provenance).
|
||||
- **`/fact-names`** — a flat array of strings, not records: **union**, deduped by
|
||||
the name and re-sorted, ascending unless `order_by` says otherwise. `name` is
|
||||
the only column the entity projects, so an `order_by` on any other field is
|
||||
rejected with `400`, as the backends reject it. While injection is on the
|
||||
`pdbmux_source` name is listed too, exactly once (see provenance).
|
||||
`include_total=true` reports the deduped union's size, so the `limit` is applied
|
||||
to the merged list rather than pushed upstream.
|
||||
- **`/reports`, `/events`** — **union**, not a per-node winner. Reports are
|
||||
immutable history, so a node's reports can legitimately exist in more than one
|
||||
backend and all of them belong in the merged view. Reports dedupe on `hash`;
|
||||
@@ -75,17 +89,18 @@ paths add two more headers `pdbmux` sets itself, `X-Cache` and `Age` — see
|
||||
- **Aggregates** — `extract`/`group_by` rows are counts, not records, so each
|
||||
backend returns a partial answer that has to be **added**, not deduped. This
|
||||
covers `/event-counts`, `/aggregate-event-counts`, and any `/reports`,
|
||||
`/nodes`, `/facts` or `/resources` query whose `extract` carries a
|
||||
`["function", ...]` column.
|
||||
`/nodes`, `/resources`, `/facts` or `/facts/<name>[/<value>]` query whose
|
||||
`extract` carries a `["function", ...]` column.
|
||||
- The grouping key is the row's non-aggregate fields: for `/reports`,
|
||||
`/nodes`, `/facts` and `/resources` they come from the query — the plain
|
||||
`extract` fields plus any `group_by` clause — and for the event-count
|
||||
endpoints from the row itself (`subject_type`/`subject`, or
|
||||
`/nodes`, `/resources`, `/facts` and `/facts/<name>` they come from the
|
||||
query — the plain `extract` fields plus any `group_by` clause — and for the
|
||||
event-count endpoints from the row itself (`subject_type`/`subject`, or
|
||||
`summarize_by`), whose remaining fields are all counts.
|
||||
- On `/nodes` and `/facts` this takes precedence over the `certname` merge: an
|
||||
aggregate row has no `certname`, so deduping would collapse every backend's
|
||||
rows into one backend's numbers. A query with no `function` column —
|
||||
including a plain `extract` projection — still merges by `certname`.
|
||||
- On `/nodes`, `/facts` and `/facts/<name>[/<value>]` this takes precedence
|
||||
over the `certname` merge: an aggregate row has no `certname`, so deduping
|
||||
would collapse every backend's rows into one backend's numbers. A query with
|
||||
no `function` column — including a plain `extract` projection — still merges
|
||||
by `certname`.
|
||||
- PuppetDB accepts `count`, `sum`, `avg`, `min`, `max`, `to_string` and
|
||||
`jsonb_typeof` as `extract` functions. Only `count` and `sum` are additive,
|
||||
so only those two merge correctly. `avg`, `min` and `max` are folded like any
|
||||
@@ -155,16 +170,38 @@ matters more.
|
||||
Only the outer query is inspected: a `name` filter inside an `in`/`select_facts`
|
||||
subquery narrows which *nodes* match, not which facts come back, so injection
|
||||
still happens;
|
||||
- the path is `/facts/<name>` for any other fact. The path segment is the same
|
||||
outer `name` constraint, so only the fact's own path carries the record;
|
||||
- injection is turned off (see `source_fact_enabled`).
|
||||
|
||||
**The fact's own path is synthesised.** `/pdb/query/v4/fact-names` lists the name
|
||||
while injection is on, so a client that discovers names there can click through
|
||||
to it, and `/pdb/query/v4/facts/pdbmux_source` has to answer. No backend holds a
|
||||
record of that name, so the route does not serve its own fan-out: it takes the
|
||||
records from the `/facts` merge that produces them, which makes the `certname`
|
||||
set, the owner and the `environment` identical to the ones an unfiltered `/facts`
|
||||
response reports, and lets the request's own `query` narrow the result upstream.
|
||||
That costs one `/facts` fan-out per cache miss — the widest fan-out `pdbmux`
|
||||
makes — on a rare, user-initiated path.
|
||||
|
||||
`/facts/pdbmux_source/<value>` pins the backend name, so it answers with the
|
||||
nodes that backend owns. The `<value>` 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
|
||||
the backends like any other, and they return nothing, because the fact does not
|
||||
exist upstream. `pdbmux` does not evaluate the AST itself, so it cannot answer
|
||||
such a query correctly for every operator (`not`, `or`, subqueries) and does not
|
||||
pretend to for some. Read the fact from an unfiltered (or `certname`-filtered)
|
||||
`/facts` response and filter client-side. The same applies to the
|
||||
`/pdb/query/v4/facts/<name>` route, which is served unmerged pass-through.
|
||||
`/facts` response, from the path route above, and filter client-side.
|
||||
|
||||
**Not covered:** `/factsets` and `/inventory`. Both carry facts, but `pdbmux`
|
||||
does not merge either today — they take the unmerged pass-through path, where
|
||||
@@ -323,9 +360,11 @@ not answering.
|
||||
|
||||
## 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
|
||||
`pdbmux` caches merged `/nodes`, `/facts`, `/facts/<name>[/<value>]` and
|
||||
`/fact-names` record sets **in memory** so a busy Puppetboard does not re-fan-out
|
||||
the same query every few seconds — its facts overview and fact drilldown are two
|
||||
of the pages that hit hardest. Everything else runs uncached — including
|
||||
`extract`/`count` aggregates on those paths, and
|
||||
the `/pdb/meta/v1/*` and `/metrics/*` endpoints, which are served live on every
|
||||
request. The cache is an interface, and `/reports` gets its own (S3-backed)
|
||||
backend later without further handler changes.
|
||||
@@ -374,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 `<value>` is dropped from
|
||||
the key and applied to the shared entry instead. `source_fact` and
|
||||
`source_fact_enabled` are read once at startup, and the cache lives for the
|
||||
same process, so changing either cannot leave differently-shaped entries
|
||||
behind.
|
||||
@@ -481,11 +522,7 @@ have drained.
|
||||
| `PDBMUX_E2E_TUNNEL_IMAGE` | `docker.io/library/alpine:3` |
|
||||
| `PDBMUX_E2E_NODE_LOOKUP` | path to a `node-lookup` binary (else `PATH`, else skipped) |
|
||||
|
||||
Two tests skip rather than assert, each naming a known gap and failing if that
|
||||
gap closes: `/pdb/query/v4/facts/<name>` is served unmerged (so Puppetboard's
|
||||
single-fact drilldown silently loses the other backend's nodes), and
|
||||
`/fact-names` is likewise unmerged (so the facts overview lists only the first
|
||||
backend's fact names).
|
||||
Every test asserts; the suite records no known gaps.
|
||||
|
||||
## Deployment
|
||||
|
||||
|
||||
@@ -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()
|
||||
@@ -340,6 +352,43 @@ func TestHandler_ServesStaleOnlyWhenBackendsFail(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// The stored entry is the whole estate's record set and the pinned value narrows
|
||||
// it per request, so the fallback has to keep narrowing: a client asking for one
|
||||
// backend's records must not be handed every backend's because the entry expired.
|
||||
func TestHandler_StaleSourceFactDrilldownStaysFilteredByOwner(t *testing.T) {
|
||||
a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "osfamily", "RedHat", "") + `]`})
|
||||
b := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h2", "osfamily", "Debian", "") + `]`})
|
||||
srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
|
||||
|
||||
drilldown := sourceFactURL + "/a"
|
||||
want := map[string]string{"h1": "a"}
|
||||
|
||||
warm := doGet(t, srv.Handler(), drilldown, "")
|
||||
if warm.Code != http.StatusOK {
|
||||
t.Fatalf("warm-up status %d: %s", warm.Code, warm.Body.String())
|
||||
}
|
||||
if got, n := sourceValues(t, warm.Body.Bytes(), defaultSourceFact); n != len(want) || !reflect.DeepEqual(got, want) {
|
||||
t.Fatalf("warm-up %s = %v (%d records), want %v", drilldown, got, n, want)
|
||||
}
|
||||
|
||||
// Every backend down and the entry expired: the stale copy is served.
|
||||
clk.advance(31 * time.Second)
|
||||
a.setFail(true)
|
||||
b.setFail(true)
|
||||
|
||||
rec := doGet(t, srv.Handler(), drilldown, "")
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("stale fallback status %d: %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
if got := rec.Header().Get(cacheStatusHeader); got != "stale" {
|
||||
t.Fatalf("%s = %q, want stale: the request did not take the fallback path", cacheStatusHeader, got)
|
||||
}
|
||||
got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact)
|
||||
if n != len(want) || !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("stale %s = %v (%d records), want only backend a's %v", drilldown, got, n, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandler_NoCacheEntryMeansBackendFailureIs502(t *testing.T) {
|
||||
a := newCountingBackend(t, map[string]string{factsPath: `[]`})
|
||||
b := newCountingBackend(t, map[string]string{factsPath: `[]`})
|
||||
|
||||
@@ -340,6 +340,26 @@ func (b *backend) query(ctx context.Context, t fatalf, path string, params url.V
|
||||
return rows
|
||||
}
|
||||
|
||||
// queryStatus is query without the body, for asserting what a backend rejects.
|
||||
func (b *backend) queryStatus(ctx context.Context, t fatalf, path string, params url.Values) int {
|
||||
t.Helper()
|
||||
target := b.url + path
|
||||
if len(params) > 0 {
|
||||
target += "?" + params.Encode()
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("building query for %s: %v", b.name, err)
|
||||
}
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
t.Fatalf("querying %s%s: %v", b.name, path, err)
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
_, _ = io.Copy(io.Discard, resp.Body)
|
||||
return resp.StatusCode
|
||||
}
|
||||
|
||||
func newNetwork(ctx context.Context, t fatalf) (string, func()) {
|
||||
t.Helper()
|
||||
nw, err := tcnet.New(ctx)
|
||||
|
||||
+35
-32
@@ -143,43 +143,43 @@ func TestPuppetboardNodeDetail(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// The facts overview lists fact names, which come from the unmerged /fact-names
|
||||
// route, so only the first backend's names reach the page today.
|
||||
// The facts overview lists fact names, which come from the merged /fact-names
|
||||
// route, so both backends' names have to reach the page.
|
||||
func TestPuppetboardFactsOverview(t *testing.T) {
|
||||
body := pbPage(t, pbAllEnvs+"/facts")
|
||||
for _, name := range []string{"osfamily", "kernel", "only_a"} {
|
||||
// only_a lives in backend A alone; only_b and extra_b in backend B alone.
|
||||
for _, name := range []string{"osfamily", "kernel", "only_a", "only_b", "extra_b"} {
|
||||
if !strings.Contains(body, name) {
|
||||
t.Errorf("Puppetboard facts overview does not list %s", name)
|
||||
}
|
||||
}
|
||||
|
||||
t.Run("names from the second backend are missing", func(t *testing.T) {
|
||||
// Known gap: /fact-names has no merge rule, so handleQuery falls through to
|
||||
// proxyUnmerged and the first backend to answer supplies the whole list.
|
||||
// Recorded rather than asserted; fails once the gap closes.
|
||||
if strings.Contains(body, "only_b") {
|
||||
t.Fatal("the facts overview now lists a fact name only the second backend holds: /fact-names is merged, so assert this properly and drop the skip")
|
||||
// A name the overview lists is a link a user can click, so the one pdbmux
|
||||
// owns has to lead to a page with every node on it rather than an empty one.
|
||||
t.Run("the owned fact is listed and its link resolves", func(t *testing.T) {
|
||||
if !strings.Contains(body, defaultSourceFact) {
|
||||
t.Fatalf("Puppetboard facts overview does not list %s", defaultSourceFact)
|
||||
}
|
||||
listed := pbFactRows(t, defaultSourceFact)
|
||||
for _, cn := range allNodes {
|
||||
if !listed[cn] {
|
||||
t.Errorf("the %s drilldown omits %s, so the overview links to a dead page", defaultSourceFact, cn)
|
||||
}
|
||||
}
|
||||
t.Skip("known gap: /fact-names is an unmerged pass-through, so only_b and extra_b never reach the facts overview")
|
||||
})
|
||||
}
|
||||
|
||||
// The single-fact drilldown is the page that exercises the /facts/<name> path
|
||||
// route.
|
||||
func TestPuppetboardFactDrilldown(t *testing.T) {
|
||||
if body := pbPage(t, pbAllEnvs+"/fact/osfamily"); !strings.Contains(body, "osfamily") {
|
||||
t.Errorf("Puppetboard fact page for osfamily does not name it")
|
||||
}
|
||||
|
||||
// The page's table is filled from this endpoint, so it is what a user sees.
|
||||
// pbFactRows reads the JSON table a Puppetboard fact page renders from, and
|
||||
// returns the certnames it lists.
|
||||
func pbFactRows(t *testing.T, name string) map[string]bool {
|
||||
t.Helper()
|
||||
var payload struct {
|
||||
Data [][]string `json:"data"`
|
||||
}
|
||||
body := pbPage(t, pbAllEnvs+"/fact/osfamily/json")
|
||||
body := pbPage(t, pbAllEnvs+"/fact/"+name+"/json")
|
||||
if err := json.Unmarshal([]byte(body), &payload); err != nil {
|
||||
t.Fatalf("decoding the fact drilldown table: %v: %s", err, body)
|
||||
t.Fatalf("decoding the %s drilldown table: %v: %s", name, err, body)
|
||||
}
|
||||
|
||||
listed := map[string]bool{}
|
||||
for _, row := range payload.Data {
|
||||
for _, cn := range allNodes {
|
||||
@@ -188,25 +188,28 @@ func TestPuppetboardFactDrilldown(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
return listed
|
||||
}
|
||||
|
||||
// Known gap: Puppetboard fetches a single fact through
|
||||
// GET /pdb/query/v4/facts/<name>, a path route with no merge rule, so
|
||||
// proxyUnmerged streams back the first backend's answer alone. The other
|
||||
// backend's nodes go missing with no error. Tracked separately; this test
|
||||
// records the gap and fails once it closes.
|
||||
// The single-fact drilldown is the page that exercises the merged /facts/<name>
|
||||
// path route, so every backend's nodes have to appear on it.
|
||||
func TestPuppetboardFactDrilldown(t *testing.T) {
|
||||
if body := pbPage(t, pbAllEnvs+"/fact/osfamily"); !strings.Contains(body, "osfamily") {
|
||||
t.Errorf("Puppetboard fact page for osfamily does not name it")
|
||||
}
|
||||
|
||||
// Puppetboard fetches a single fact through GET /pdb/query/v4/facts/<name>,
|
||||
// so the page is only whole if that path route merges every backend.
|
||||
listed := pbFactRows(t, "osfamily")
|
||||
missing := []string{}
|
||||
for _, cn := range allNodes {
|
||||
if !listed[cn] {
|
||||
missing = append(missing, cn)
|
||||
}
|
||||
}
|
||||
if len(missing) == 0 {
|
||||
t.Fatal("the fact drilldown now lists every node: the /facts/<name> path route is merged, so assert this properly and drop the skip")
|
||||
if len(missing) > 0 {
|
||||
t.Fatalf("the fact drilldown omits %v; /pdb/query/v4/facts/<name> must serve every backend's nodes", missing)
|
||||
}
|
||||
if !listed[nodeAlpha] {
|
||||
t.Fatalf("the fact drilldown lists neither backend's nodes (%v missing), which is not the pass-through behaviour under test", missing)
|
||||
}
|
||||
t.Skipf("known gap: /pdb/query/v4/facts/<name> is served unmerged, so the drilldown silently omits %v", missing)
|
||||
}
|
||||
|
||||
// Reports are the least exercised merged path, and Puppetboard reads them
|
||||
|
||||
@@ -5,6 +5,8 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"reflect"
|
||||
"sort"
|
||||
@@ -59,6 +61,208 @@ func TestFactsUnionAcrossBackends(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// The /facts/<name> path route carries the same records /facts does, so it takes
|
||||
// the same merge: every node in the estate, each resolved to one backend.
|
||||
func TestFactsByNamePathIsMerged(t *testing.T) {
|
||||
resp := get(t, factsPath+"/osfamily", nil)
|
||||
rows := resp.rows(t)
|
||||
|
||||
if got := e2eCertnames(rows); !equalStrings(got, allNodes) {
|
||||
t.Fatalf("merged /facts/osfamily certnames = %v, want %v", got, allNodes)
|
||||
}
|
||||
if len(rows) != len(allNodes) {
|
||||
t.Fatalf("/facts/osfamily returned %d records for %d nodes, so a shared node was not deduped", len(rows), len(allNodes))
|
||||
}
|
||||
if got, _ := factValue(rows, nodeShared, "osfamily"); got != "Debian" {
|
||||
t.Errorf("osfamily of %s = %v, want Debian from the fresher backend", nodeShared, got)
|
||||
}
|
||||
if got := resp.header.Get(backendsHeader); got != "2/2" {
|
||||
t.Errorf("%s = %q, want %q", backendsHeader, got, "2/2")
|
||||
}
|
||||
|
||||
t.Run("the name constraint gates injection", func(t *testing.T) {
|
||||
for _, row := range rows {
|
||||
if row["name"] == defaultSourceFact {
|
||||
t.Fatalf("%s was injected into a /facts/<name> response: %v", defaultSourceFact, row)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("the value sub-route merges too", func(t *testing.T) {
|
||||
rows := get(t, factsPath+"/osfamily/Debian", nil).rows(t)
|
||||
want := []string{nodeBeta, nodeGamma, nodeShared}
|
||||
if got := e2eCertnames(rows); !equalStrings(got, want) {
|
||||
t.Errorf("/facts/osfamily/Debian certnames = %v, want %v", got, want)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("a fact only one backend holds survives", func(t *testing.T) {
|
||||
rows := get(t, factsPath+"/only_b", nil).rows(t)
|
||||
if got, ok := factValue(rows, nodeBeta, "only_b"); !ok || got != "yes" {
|
||||
t.Errorf("/facts/only_b for %s = %v (present %v), want yes", nodeBeta, got, ok)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// /fact-names is a flat array of strings, so it needs its own union: the merged
|
||||
// list is every backend's names, deduped and sorted.
|
||||
func TestFactNamesAreMerged(t *testing.T) {
|
||||
resp := get(t, factNamesPath, nil)
|
||||
var got []string
|
||||
if err := json.Unmarshal(resp.body, &got); err != nil {
|
||||
t.Fatalf("/fact-names is not a flat string array: %v: %s", err, resp.body)
|
||||
}
|
||||
|
||||
for _, name := range []string{"osfamily", "kernel", "role", "owner", "only_a", "only_b", "extra_b"} {
|
||||
if !containsString(got, name) {
|
||||
t.Errorf("merged /fact-names is missing %s: %v", name, got)
|
||||
}
|
||||
}
|
||||
if !sort.StringsAreSorted(got) {
|
||||
t.Errorf("merged /fact-names is not sorted ascending: %v", got)
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
for _, name := range got {
|
||||
if seen[name] {
|
||||
t.Errorf("merged /fact-names lists %s twice: %v", name, got)
|
||||
}
|
||||
seen[name] = true
|
||||
}
|
||||
// The merged /facts response carries a record of the name pdbmux owns, so a
|
||||
// client discovering names here has to see it.
|
||||
if !seen[defaultSourceFact] {
|
||||
t.Errorf("merged /fact-names omits %s, which every merged /facts response carries: %v", defaultSourceFact, got)
|
||||
}
|
||||
if got := resp.header.Get(backendsHeader); got != "2/2" {
|
||||
t.Errorf("%s = %q, want %q", backendsHeader, got, "2/2")
|
||||
}
|
||||
|
||||
t.Run("paging and ordering are redone across the union", func(t *testing.T) {
|
||||
params := url.Values{
|
||||
"order_by": {`[{"field":"name","order":"desc"}]`},
|
||||
"limit": {"3"},
|
||||
"include_total": {"true"},
|
||||
}
|
||||
resp := get(t, factNamesPath, params)
|
||||
var page []string
|
||||
if err := json.Unmarshal(resp.body, &page); err != nil {
|
||||
t.Fatalf("decoding the paged /fact-names response: %v: %s", err, resp.body)
|
||||
}
|
||||
want := append([]string{}, got...)
|
||||
sort.Sort(sort.Reverse(sort.StringSlice(want)))
|
||||
if len(want) > 3 {
|
||||
want = want[:3]
|
||||
}
|
||||
if !equalStrings(page, want) {
|
||||
t.Errorf("descending /fact-names page = %v, want %v", page, want)
|
||||
}
|
||||
if n := resp.header.Get(recordsHeader); n != fmt.Sprint(len(got)) {
|
||||
t.Errorf("%s = %q, want the merged count %d", recordsHeader, n, len(got))
|
||||
}
|
||||
})
|
||||
|
||||
// name is the only column the entity projects, so pdbmux rejects any other
|
||||
// order_by field the way the backends do.
|
||||
t.Run("an order_by on another field is rejected", func(t *testing.T) {
|
||||
params := url.Values{"order_by": {`[{"field":"bogus","order":"desc"}]`}}
|
||||
if r := rawGet(t, factNamesPath, params); r.status != http.StatusBadRequest {
|
||||
t.Errorf("merged /fact-names order_by bogus = HTTP %d, want 400: %s", r.status, r.body)
|
||||
}
|
||||
if r := h.a.queryStatus(context.Background(), t, factNamesPath, params); r != http.StatusBadRequest {
|
||||
t.Errorf("backend %s answered HTTP %d for the same order_by, so 400 is not what it does", h.a.name, r)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// A fact aggregate row carries no certname, so the per-certname merge would keep
|
||||
// one backend's row and drop the other's; only adding the numbers is right.
|
||||
func TestFactsByNamePathAggregatesAreSummed(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
const q = `["extract",[["function","count"]]]`
|
||||
|
||||
for _, path := range []string{factsPath + "/kernel", factsPath + "/kernel/Linux"} {
|
||||
t.Run(path, func(t *testing.T) {
|
||||
wantA := backendCount(ctx, t, h.a, path, q)
|
||||
wantB := backendCount(ctx, t, h.b, path, q)
|
||||
if wantA == wantB {
|
||||
t.Fatalf("the fixture gives both backends %d records on %s, so a sum is indistinguishable from one backend's number", wantA, path)
|
||||
}
|
||||
|
||||
resp := get(t, path, query(q))
|
||||
if got := countOf(t, resp.rows(t)); got != wantA+wantB {
|
||||
t.Fatalf("%s count = %d, want %d (%s=%d + %s=%d)", path, got, wantA+wantB, h.a.name, wantA, h.b.name, wantB)
|
||||
}
|
||||
if got := resp.header.Get(backendsHeader); got != "2/2" {
|
||||
t.Errorf("%s = %q, want %q", backendsHeader, got, "2/2")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
t.Run("grouped counts are summed per key", func(t *testing.T) {
|
||||
const grouped = `["extract",[["function","count"],"value"],["group_by","value"]]`
|
||||
path := factsPath + "/osfamily"
|
||||
want := map[string]int{}
|
||||
for _, b := range []*backend{h.a, h.b} {
|
||||
for value, n := range countsByValue(t, b.query(ctx, t, path, query(grouped))) {
|
||||
want[value] += n
|
||||
}
|
||||
}
|
||||
got := countsByValue(t, get(t, path, query(grouped)).rows(t))
|
||||
if !equalCounts(got, want) {
|
||||
t.Errorf("grouped %s counts = %v, want %v", path, got, want)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("a query with no function column still merges by certname", func(t *testing.T) {
|
||||
rows := get(t, factsPath+"/kernel", nil).rows(t)
|
||||
if got := e2eCertnames(rows); !equalStrings(got, allNodes) {
|
||||
t.Errorf("/facts/kernel certnames = %v, want %v", got, allNodes)
|
||||
}
|
||||
if len(rows) != len(allNodes) {
|
||||
t.Errorf("/facts/kernel returned %d records for %d nodes, so a shared node was not deduped", len(rows), len(allNodes))
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// countsByValue reads a ["function","count"] + group_by "value" result set.
|
||||
func countsByValue(t *testing.T, rows []map[string]any) map[string]int {
|
||||
t.Helper()
|
||||
out := map[string]int{}
|
||||
for _, row := range rows {
|
||||
value, ok := row["value"].(string)
|
||||
if !ok {
|
||||
t.Fatalf("grouped aggregate row has no value: %v", row)
|
||||
}
|
||||
n, ok := row["count"].(float64)
|
||||
if !ok {
|
||||
t.Fatalf("grouped aggregate row has no numeric count: %v", row)
|
||||
}
|
||||
out[value] = int(n)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func equalCounts(a, b map[string]int) bool {
|
||||
if len(a) != len(b) {
|
||||
return false
|
||||
}
|
||||
for k, v := range a {
|
||||
if b[k] != v {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func containsString(s []string, v string) bool {
|
||||
for _, x := range s {
|
||||
if x == v {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// The shared node's two copies disagree on every fact value; the backend holding
|
||||
// its newer report_timestamp has to win both endpoints.
|
||||
func TestSharedNodeResolvesToTheFresherBackend(t *testing.T) {
|
||||
@@ -284,6 +488,83 @@ func TestSourceFactInjectionAndGating(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
// /fact-names advertises the fact, so its drilldown has to answer with the same
|
||||
// records /facts carries rather than the empty set the backends hold.
|
||||
func TestSourceFactDrilldown(t *testing.T) {
|
||||
want := map[string]string{
|
||||
nodeAlpha: backendAName,
|
||||
nodeBeta: backendBName,
|
||||
nodeGamma: backendBName,
|
||||
nodeShared: backendBName, // won on freshness, not on configured order
|
||||
}
|
||||
path := factsPath + "/" + defaultSourceFact
|
||||
|
||||
t.Run("one record per node, attributed as /facts attributes it", func(t *testing.T) {
|
||||
resp := get(t, path, nil)
|
||||
rows := resp.rows(t)
|
||||
if got := e2eCertnames(rows); !equalStrings(got, allNodes) {
|
||||
t.Fatalf("%s certnames = %v, want %v", path, got, allNodes)
|
||||
}
|
||||
if len(rows) != len(allNodes) {
|
||||
t.Fatalf("%s returned %d records for %d nodes", path, len(rows), len(allNodes))
|
||||
}
|
||||
unfiltered := get(t, factsPath, nil).rows(t)
|
||||
for _, row := range rows {
|
||||
cn, _ := row["certname"].(string)
|
||||
if row["name"] != defaultSourceFact {
|
||||
t.Errorf("%s returned a record named %v", path, row["name"])
|
||||
}
|
||||
if row["value"] != want[cn] {
|
||||
t.Errorf("%s for %s = %v, want %q", path, cn, row["value"], want[cn])
|
||||
}
|
||||
// The drilldown must not disagree with the response it stands for.
|
||||
if got, ok := factValue(unfiltered, cn, defaultSourceFact); !ok || got != row["value"] {
|
||||
t.Errorf("%s for %s = %v, want the %s value %v", path, cn, row["value"], factsPath, got)
|
||||
}
|
||||
if _, ok := row["environment"]; !ok {
|
||||
t.Errorf("%s record for %s has no environment key", path, cn)
|
||||
}
|
||||
}
|
||||
if got := resp.header.Get(backendsHeader); got != "2/2" {
|
||||
t.Errorf("%s = %q, want %q", backendsHeader, got, "2/2")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("the value sub-route filters by owning backend", func(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
value string
|
||||
want []string
|
||||
}{
|
||||
{backendAName, []string{nodeAlpha}},
|
||||
{backendBName, []string{nodeBeta, nodeGamma, nodeShared}},
|
||||
{"nosuchbackend", nil},
|
||||
} {
|
||||
rows := get(t, path+"/"+tc.value, nil).rows(t)
|
||||
if got := e2eCertnames(rows); !equalStrings(got, tc.want) {
|
||||
t.Errorf("%s/%s certnames = %v, want %v", path, tc.value, got, tc.want)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("a certname query narrows the drilldown", func(t *testing.T) {
|
||||
rows := get(t, path, query(`["=","certname","`+nodeAlpha+`"]`)).rows(t)
|
||||
if got := e2eCertnames(rows); !equalStrings(got, []string{nodeAlpha}) {
|
||||
t.Errorf("certname-filtered %s = %v, want only %s", path, got, nodeAlpha)
|
||||
}
|
||||
})
|
||||
|
||||
// An aggregate carries no certname to attribute, so it stays summed.
|
||||
t.Run("an aggregate is still summed", func(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
const q = `["extract",[["function","count"]]]`
|
||||
wantA := backendCount(ctx, t, h.a, path, q)
|
||||
wantB := backendCount(ctx, t, h.b, path, q)
|
||||
if got := countOf(t, get(t, path, query(q)).rows(t)); got != wantA+wantB {
|
||||
t.Errorf("%s count = %d, want %d (%s=%d + %s=%d)", path, got, wantA+wantB, h.a.name, wantA, h.b.name, wantB)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// X-Backends has to report what the response was actually built from, not what
|
||||
// is configured.
|
||||
func TestBackendsHeaderReportsContributors(t *testing.T) {
|
||||
|
||||
@@ -0,0 +1,699 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"reflect"
|
||||
"slices"
|
||||
"testing"
|
||||
)
|
||||
|
||||
const (
|
||||
roleFactPath = factsPath + "/role"
|
||||
osFactPath = factsPath + "/osfamily"
|
||||
sourceFactURL = factsPath + "/" + defaultSourceFact
|
||||
)
|
||||
|
||||
func names(t *testing.T, body []byte) []string {
|
||||
t.Helper()
|
||||
var out []string
|
||||
if err := json.Unmarshal(body, &out); err != nil {
|
||||
t.Fatalf("unmarshal %s: %v", body, err)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func factNamesBody(names ...string) string {
|
||||
b, err := json.Marshal(names)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return string(b)
|
||||
}
|
||||
|
||||
func TestFactsSubPath(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
path string
|
||||
wantName string
|
||||
wantValue string
|
||||
wantValued bool
|
||||
}{
|
||||
{factsPath, "", "", false},
|
||||
{factsPath + "/", "", "", false},
|
||||
{roleFactPath, "role", "", false},
|
||||
{roleFactPath + "/web", "role", "web", true},
|
||||
{roleFactPath + "/", "", "", false},
|
||||
{roleFactPath + "/web/extra", "", "", false},
|
||||
{nodesPath + "/h1/facts/role", "", "", false},
|
||||
{factNamesPath, "", "", false},
|
||||
} {
|
||||
name, value, valued := factsSubPath(tc.path)
|
||||
if name != tc.wantName || value != tc.wantValue || valued != tc.wantValued {
|
||||
t.Errorf("factsSubPath(%q) = (%q, %q, %v), want (%q, %q, %v)",
|
||||
tc.path, name, value, valued, tc.wantName, tc.wantValue, tc.wantValued)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The path route carries the same records /facts does, so it merges the same
|
||||
// way: every backend's nodes survive and a shared node resolves to one owner.
|
||||
func TestHandler_FactsByNameMerged(t *testing.T) {
|
||||
a := newFakeBackend(t, `[]`, `[]`)
|
||||
a.bodies[roleFactPath] = `[` + fact("h1", "role", "web-a", "") + `,` + fact("h2", "role", "db-a", "") + `]`
|
||||
b := newFakeBackend(t, `[]`, `[]`)
|
||||
b.bodies[roleFactPath] = `[` + fact("h1", "role", "web-b", "") + `,` + fact("h3", "role", "db-b", "") + `]`
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
|
||||
|
||||
rec := doGet(t, srv.Handler(), roleFactPath, "")
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
got := map[string]string{}
|
||||
var rows []struct {
|
||||
Certname string `json:"certname"`
|
||||
Value string `json:"value"`
|
||||
}
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &rows); err != nil {
|
||||
t.Fatalf("unmarshal %s: %v", rec.Body.String(), err)
|
||||
}
|
||||
for _, row := range rows {
|
||||
got[row.Certname] = row.Value
|
||||
}
|
||||
want := map[string]string{"h1": "web-a", "h2": "db-a", "h3": "db-b"}
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("merged %s = %v, want %v", roleFactPath, got, want)
|
||||
}
|
||||
if h := rec.Header().Get(backendsHeader); h != "2/2" {
|
||||
t.Errorf("%s = %q, want 2/2", backendsHeader, h)
|
||||
}
|
||||
}
|
||||
|
||||
// The /<name>/<value> sub-route is the same entity with one more constraint, so
|
||||
// it takes the same merge rather than falling through to the pass-through path.
|
||||
func TestHandler_FactsByNameAndValueMerged(t *testing.T) {
|
||||
path := roleFactPath + "/web"
|
||||
a := newFakeBackend(t, `[]`, `[]`)
|
||||
a.bodies[path] = `[` + fact("h1", "role", "web", "") + `]`
|
||||
b := newFakeBackend(t, `[]`, `[]`)
|
||||
b.bodies[path] = `[` + fact("h3", "role", "web", "") + `]`
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
|
||||
|
||||
rec := doGet(t, srv.Handler(), path, "")
|
||||
var rows []recordMeta
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &rows); err != nil {
|
||||
t.Fatalf("unmarshal %s: %v", rec.Body.String(), err)
|
||||
}
|
||||
if len(rows) != 2 {
|
||||
t.Fatalf("merged %s returned %d records, want both backends': %s", path, len(rows), rec.Body.String())
|
||||
}
|
||||
if h := rec.Header().Get(backendsHeader); h != "2/2" {
|
||||
t.Errorf("%s = %q, want 2/2", backendsHeader, h)
|
||||
}
|
||||
}
|
||||
|
||||
// An aggregate row carries no certname, so the per-certname merge would keep one
|
||||
// backend's row and drop the other's without any error or partial-backend signal.
|
||||
func TestHandler_FactsByNameAggregateSummed(t *testing.T) {
|
||||
for _, path := range []string{roleFactPath, roleFactPath + "/web"} {
|
||||
t.Run(path, func(t *testing.T) {
|
||||
a := newFakeBackend(t, `[]`, `[]`)
|
||||
a.bodies[path] = `[{"count":7}]`
|
||||
b := newFakeBackend(t, `[]`, `[]`)
|
||||
b.bodies[path] = `[{"count":3}]`
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
|
||||
|
||||
rec := doGet(t, srv.Handler(), path, `["extract",[["function","count"]]]`)
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
if got := counts(t, rec.Body.Bytes(), "count"); !slices.Equal(got, []float64{10}) {
|
||||
t.Errorf("count = %v, want [10]", got)
|
||||
}
|
||||
if h := rec.Header().Get(backendsHeader); h != "2/2" {
|
||||
t.Errorf("%s = %q, want 2/2", backendsHeader, h)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandler_FactsByNameAggregateGroupedSummed(t *testing.T) {
|
||||
a := newFakeBackend(t, `[]`, `[]`)
|
||||
a.bodies[roleFactPath] = `[{"count":2,"value":"web"},{"count":1,"value":"db"}]`
|
||||
b := newFakeBackend(t, `[]`, `[]`)
|
||||
b.bodies[roleFactPath] = `[{"count":5,"value":"web"}]`
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
|
||||
|
||||
rec := doGet(t, srv.Handler(), roleFactPath,
|
||||
`["extract",[["function","count"],"value"],["group_by","value"]]`)
|
||||
var rows []map[string]any
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &rows); err != nil {
|
||||
t.Fatalf("unmarshal %s: %v", rec.Body.String(), err)
|
||||
}
|
||||
got := map[string]float64{}
|
||||
for _, row := range rows {
|
||||
v, _ := row["value"].(string)
|
||||
n, _ := row["count"].(float64)
|
||||
got[v] = n
|
||||
}
|
||||
want := map[string]float64{"web": 7, "db": 1}
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("grouped counts = %v, want %v", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// The summing branch must not divert a plain query, including an extract
|
||||
// projection that carries no function column.
|
||||
func TestHandler_FactsByNameNonAggregateStillMergedByCertname(t *testing.T) {
|
||||
for _, q := range []string{"", `["extract",["certname","value"],["~","certname",".*"]]`} {
|
||||
a := newFakeBackend(t, `[]`, `[]`)
|
||||
a.bodies[roleFactPath] = `[` + fact("h1", "role", "web-a", "") + `,` + fact("h2", "role", "db-a", "") + `]`
|
||||
b := newFakeBackend(t, `[]`, `[]`)
|
||||
b.bodies[roleFactPath] = `[` + fact("h1", "role", "web-b", "") + `,` + fact("h3", "role", "db-b", "") + `]`
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
|
||||
|
||||
rec := doGet(t, srv.Handler(), roleFactPath, q)
|
||||
var rows []recordMeta
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &rows); err != nil {
|
||||
t.Fatalf("query %q: unmarshal %s: %v", q, rec.Body.String(), err)
|
||||
}
|
||||
got := map[string]bool{}
|
||||
for _, row := range rows {
|
||||
got[row.Certname] = true
|
||||
}
|
||||
if len(rows) != 3 || !got["h1"] || !got["h2"] || !got["h3"] {
|
||||
t.Errorf("query %q: merged %s = %s, want one record per certname", q, roleFactPath, rec.Body.String())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A summed row is not the merged record set the cache stores, so it stays live.
|
||||
func TestHandler_FactsByNameAggregateNotCached(t *testing.T) {
|
||||
for _, path := range []string{roleFactPath, roleFactPath + "/web", sourceFactURL} {
|
||||
t.Run(path, func(t *testing.T) {
|
||||
a := newCountingBackend(t, map[string]string{path: `[{"count":7}]`})
|
||||
b := newCountingBackend(t, map[string]string{path: `[{"count":3}]`})
|
||||
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
|
||||
|
||||
const q = `["extract",[["function","count"]]]`
|
||||
doGet(t, srv.Handler(), path, q)
|
||||
second := doGet(t, srv.Handler(), path, q)
|
||||
if got := a.hitCount(path); got != 2 {
|
||||
t.Errorf("%s aggregates are uncached: %d requests, want 2", path, got)
|
||||
}
|
||||
if got := counts(t, second.Body.Bytes(), "count"); !slices.Equal(got, []float64{10}) {
|
||||
t.Errorf("count = %v, want [10]", got)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandler_FactsByNameForwardsPathAndQuery(t *testing.T) {
|
||||
a := newFakeBackend(t, `[]`, `[]`)
|
||||
a.bodies[roleFactPath] = `[]`
|
||||
b := newFakeBackend(t, `[]`, `[]`)
|
||||
b.bodies[roleFactPath] = `[]`
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
|
||||
|
||||
q := `["=","certname","h1"]`
|
||||
doGet(t, srv.Handler(), roleFactPath, q)
|
||||
for _, fb := range []*fakeBackend{a, b} {
|
||||
if got := fb.gotQuery(roleFactPath); got != q {
|
||||
t.Errorf("backend got query %q on %s, want %q", got, roleFactPath, q)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The path is a name constraint, the shape the injection gate already excludes.
|
||||
func TestHandler_FactsByNameNotInjected(t *testing.T) {
|
||||
a := newFakeBackend(t, `[]`, `[]`)
|
||||
a.bodies[osFactPath] = `[` + fact("h1", "osfamily", "RedHat", "") + `]`
|
||||
b := newFakeBackend(t, `[]`, `[]`)
|
||||
b.bodies[osFactPath] = `[` + fact("h2", "osfamily", "Debian", "") + `]`
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
|
||||
|
||||
rec := doGet(t, srv.Handler(), osFactPath, "")
|
||||
if _, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact); n != 0 {
|
||||
t.Errorf("%s was injected into %s: %s", defaultSourceFact, osFactPath, rec.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
// sourceFactBackends is a pair of fake backends whose /facts bodies give h1 to a
|
||||
// and h2/h3 to b, the shape every drilldown assertion below rests on.
|
||||
func sourceFactBackends(t *testing.T) (*fakeBackend, *fakeBackend) {
|
||||
t.Helper()
|
||||
a := newFakeBackend(t,
|
||||
`[`+node("h1", "2026-01-01T00:00:00Z")+`,`+node("h3", "2026-01-01T00:00:00Z")+`]`,
|
||||
`[`+factEnv("h1", "osfamily", "RedHat", "prod")+`,`+factEnv("h3", "osfamily", "RedHat", "prod")+`]`)
|
||||
b := newFakeBackend(t,
|
||||
`[`+node("h2", "2026-01-01T00:00:00Z")+`,`+node("h3", "2026-06-01T00:00:00Z")+`]`,
|
||||
`[`+factEnv("h2", "osfamily", "Debian", "dev")+`,`+factEnv("h3", "osfamily", "Debian", "dev")+`]`)
|
||||
return a, b
|
||||
}
|
||||
|
||||
// The drilldown has to answer with the records /facts carries, which no backend
|
||||
// holds: the certname set and the per-node owner come from the same merge.
|
||||
func TestHandler_FactsBySourceNamePathSynthesisesOneRecordPerNode(t *testing.T) {
|
||||
for _, merge := range []string{mergeStatic, mergeFreshness} {
|
||||
t.Run(merge, func(t *testing.T) {
|
||||
a, b := sourceFactBackends(t)
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, merge))
|
||||
|
||||
// h3 lives in both: static keeps the first configured backend, freshness
|
||||
// the one holding its newer report.
|
||||
wantH3 := "a"
|
||||
if merge == mergeFreshness {
|
||||
wantH3 = "b"
|
||||
}
|
||||
want := map[string]string{"h1": "a", "h2": "b", "h3": wantH3}
|
||||
|
||||
rec := doGet(t, srv.Handler(), sourceFactURL, "")
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact)
|
||||
if n != len(want) || !reflect.DeepEqual(got, want) {
|
||||
t.Fatalf("%s = %v (%d records), want %v", sourceFactURL, got, n, want)
|
||||
}
|
||||
|
||||
// The drilldown must agree with the records the unfiltered route reports.
|
||||
unfiltered, _ := sourceValues(t, doGet(t, srv.Handler(), factsPath, "").Body.Bytes(), defaultSourceFact)
|
||||
if !reflect.DeepEqual(got, unfiltered) {
|
||||
t.Errorf("%s = %v, want the same attribution %s reports: %v", sourceFactURL, got, factsPath, unfiltered)
|
||||
}
|
||||
if h := rec.Header().Get(backendsHeader); h != "2/2" {
|
||||
t.Errorf("%s = %q, want 2/2", backendsHeader, h)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// A fact record's four keys are always present, and environment is the owning
|
||||
// backend's, since that is the record the merge attributed the node to.
|
||||
func TestHandler_FactsBySourceNamePathCarriesTheOwnersEnvironment(t *testing.T) {
|
||||
a, b := sourceFactBackends(t)
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness))
|
||||
|
||||
rec := doGet(t, srv.Handler(), sourceFactURL, "")
|
||||
var rows []map[string]any
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &rows); err != nil {
|
||||
t.Fatalf("unmarshal %s: %v", rec.Body.String(), err)
|
||||
}
|
||||
want := map[string]string{"h1": "prod", "h2": "dev", "h3": "dev"}
|
||||
for _, row := range rows {
|
||||
for _, key := range []string{"certname", "name", "value", "environment"} {
|
||||
if _, ok := row[key]; !ok {
|
||||
t.Fatalf("synthetic record %v is missing %s", row, key)
|
||||
}
|
||||
}
|
||||
cn, _ := row["certname"].(string)
|
||||
if got := row["environment"]; got != want[cn] {
|
||||
t.Errorf("environment of %s = %v, want the owner's %q", cn, got, want[cn])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The pinned value names a backend, so the sub-route is the drilldown filtered
|
||||
// by owner; a value naming no backend matches nothing.
|
||||
func TestHandler_FactsBySourceNameAndValueFiltersByOwner(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
value string
|
||||
want map[string]string
|
||||
}{
|
||||
{"a", map[string]string{"h1": "a"}},
|
||||
{"b", map[string]string{"h2": "b", "h3": "b"}},
|
||||
{"nosuchbackend", map[string]string{}},
|
||||
} {
|
||||
t.Run(tc.value, func(t *testing.T) {
|
||||
a, b := sourceFactBackends(t)
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness))
|
||||
|
||||
rec := doGet(t, srv.Handler(), sourceFactURL+"/"+tc.value, "")
|
||||
got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact)
|
||||
if n != len(tc.want) || !reflect.DeepEqual(got, tc.want) {
|
||||
t.Errorf("%s/%s = %v (%d records), want %v", sourceFactURL, tc.value, got, n, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// 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)
|
||||
}
|
||||
}
|
||||
// No fan-out happened, so none was heard from partially either.
|
||||
if got := health(t, srv).Query.PartialRounds; got != 0 {
|
||||
t.Errorf("partial_rounds = %d, 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) {
|
||||
a, b := sourceFactBackends(t)
|
||||
a.bodies[factNamesPath] = factNamesBody("osfamily")
|
||||
b.bodies[factNamesPath] = `[]`
|
||||
cfg := testConfig(a.srv.URL, b.srv.URL, mergeFreshness)
|
||||
cfg.SourceFactEnabled = false
|
||||
srv := newTestServer(cfg)
|
||||
|
||||
// No backend holds the fact, so the route falls through to the plain merge.
|
||||
for _, fb := range []*fakeBackend{a, b} {
|
||||
fb.bodies[sourceFactURL] = `[]`
|
||||
fb.bodies[sourceFactURL+"/a"] = `[]`
|
||||
}
|
||||
for _, path := range []string{sourceFactURL, sourceFactURL + "/a"} {
|
||||
if got := doGet(t, srv.Handler(), path, "").Body.String(); got != "[]\n" {
|
||||
t.Errorf("disabled %s = %q, want %q", path, got, "[]\n")
|
||||
}
|
||||
}
|
||||
if got := names(t, doGet(t, srv.Handler(), factNamesPath, "").Body.Bytes()); !reflect.DeepEqual(got, []string{"osfamily"}) {
|
||||
t.Errorf("disabled %s = %v, want no %s entry", factNamesPath, got, defaultSourceFact)
|
||||
}
|
||||
}
|
||||
|
||||
// The query narrows the certnames upstream, so the drilldown only covers what it
|
||||
// selects.
|
||||
func TestHandler_FactsBySourceNamePathRespectsQuery(t *testing.T) {
|
||||
a, b := sourceFactBackends(t)
|
||||
a.bodies[factsPath] = `[` + factEnv("h1", "osfamily", "RedHat", "prod") + `]`
|
||||
b.bodies[factsPath] = `[]`
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness))
|
||||
|
||||
const q = `["=","certname","h1"]`
|
||||
rec := doGet(t, srv.Handler(), sourceFactURL, q)
|
||||
got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact)
|
||||
if n != 1 || got["h1"] != "a" {
|
||||
t.Fatalf("%s?query=%s = %v (%d records), want only h1: %s", sourceFactURL, q, got, n, rec.Body.String())
|
||||
}
|
||||
// The query has to reach the backends for them to narrow anything.
|
||||
for _, fb := range []*fakeBackend{a, b} {
|
||||
if seen := fb.gotQuery(factsPath); seen != q {
|
||||
t.Errorf("backend got query %q on %s, want %q", seen, factsPath, q)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// pdbmux owns the name, so an upstream record of it is dropped rather than
|
||||
// listed beside the synthetic one.
|
||||
func TestHandler_FactsBySourceNamePathSuppressesUpstream(t *testing.T) {
|
||||
a, b := sourceFactBackends(t)
|
||||
a.bodies[factsPath] = `[` + factEnv("h1", "osfamily", "RedHat", "prod") + `,` +
|
||||
fact("h1", defaultSourceFact, "stale", "") + `]`
|
||||
b.bodies[factsPath] = `[]`
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness))
|
||||
|
||||
rec := doGet(t, srv.Handler(), sourceFactURL, "")
|
||||
got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact)
|
||||
if n != 1 || got["h1"] != "a" {
|
||||
t.Fatalf("%s = %v (%d records), want exactly the synthetic one: %s", sourceFactURL, got, n, rec.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
// A name-constrained query on the source fact's own path is still gated by the
|
||||
// query, not just the path.
|
||||
func TestHandler_FactsBySourceNamePathRespectsQueryGate(t *testing.T) {
|
||||
a, b := sourceFactBackends(t)
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
|
||||
|
||||
rec := doGet(t, srv.Handler(), sourceFactURL, `["extract",["certname","value"]]`)
|
||||
if _, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact); n != 0 {
|
||||
t.Errorf("%s survived an extract projection: %s", defaultSourceFact, rec.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
// An aggregate carries no certname to attribute, so the route stays on the
|
||||
// summing path rather than synthesising records.
|
||||
func TestHandler_FactsBySourceNameAggregateStillSummed(t *testing.T) {
|
||||
for _, path := range []string{sourceFactURL, sourceFactURL + "/a"} {
|
||||
t.Run(path, func(t *testing.T) {
|
||||
a, b := sourceFactBackends(t)
|
||||
a.bodies[path] = `[{"count":7}]`
|
||||
b.bodies[path] = `[{"count":3}]`
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness))
|
||||
|
||||
rec := doGet(t, srv.Handler(), path, `["extract",[["function","count"]]]`)
|
||||
if got := counts(t, rec.Body.Bytes(), "count"); !slices.Equal(got, []float64{10}) {
|
||||
t.Errorf("count = %v, want [10]", got)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandler_FactNamesUnionDedupedAndSorted(t *testing.T) {
|
||||
a := newFakeBackend(t, `[]`, `[]`)
|
||||
a.bodies[factNamesPath] = factNamesBody("kernel", "only_a", "osfamily")
|
||||
b := newFakeBackend(t, `[]`, `[]`)
|
||||
b.bodies[factNamesPath] = factNamesBody("extra_b", "kernel", "only_b", "osfamily")
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
|
||||
|
||||
rec := doGet(t, srv.Handler(), factNamesPath, "")
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
|
||||
}
|
||||
want := []string{"extra_b", "kernel", "only_a", "only_b", "osfamily", defaultSourceFact}
|
||||
if got := names(t, rec.Body.Bytes()); !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("merged %s = %v, want %v", factNamesPath, got, want)
|
||||
}
|
||||
if h := rec.Header().Get(backendsHeader); h != "2/2" {
|
||||
t.Errorf("%s = %q, want 2/2", backendsHeader, h)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandler_FactNamesOrderByDescending(t *testing.T) {
|
||||
a := newFakeBackend(t, `[]`, `[]`)
|
||||
a.bodies[factNamesPath] = factNamesBody("kernel", "osfamily")
|
||||
b := newFakeBackend(t, `[]`, `[]`)
|
||||
b.bodies[factNamesPath] = factNamesBody("zone")
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
|
||||
|
||||
rec := doGetParams(t, srv.Handler(), factNamesPath, url.Values{
|
||||
"order_by": {`[{"field":"name","order":"desc"}]`},
|
||||
})
|
||||
want := []string{"zone", defaultSourceFact, "osfamily", "kernel"}
|
||||
if got := names(t, rec.Body.Bytes()); !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("descending %s = %v, want %v", factNamesPath, got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// Paging is applied to the merged list: each backend can only page its own.
|
||||
func TestHandler_FactNamesPagedAcrossBackends(t *testing.T) {
|
||||
a := newFakeBackend(t, `[]`, `[]`)
|
||||
a.bodies[factNamesPath] = factNamesBody("a1", "a2", "a3")
|
||||
b := newFakeBackend(t, `[]`, `[]`)
|
||||
b.bodies[factNamesPath] = factNamesBody("b1", "b2")
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
|
||||
|
||||
rec := doGetParams(t, srv.Handler(), factNamesPath, url.Values{
|
||||
"limit": {"2"},
|
||||
"offset": {"1"},
|
||||
})
|
||||
want := []string{"a2", "a3"}
|
||||
if got := names(t, rec.Body.Bytes()); !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("paged %s = %v, want %v", factNamesPath, got, want)
|
||||
}
|
||||
// Backends are asked for limit+offset with no offset, so the union has enough
|
||||
// records to page locally.
|
||||
if got, _ := a.params(factNamesPath); got.Get("limit") != "3" || got.Get("offset") != "" {
|
||||
t.Errorf("backend a got limit=%q offset=%q, want 3 and none", got.Get("limit"), got.Get("offset"))
|
||||
}
|
||||
}
|
||||
|
||||
// A merged total counts the union, so the limit cannot be pushed upstream.
|
||||
func TestHandler_FactNamesTotalCountsTheWholeUnion(t *testing.T) {
|
||||
a := newFakeBackend(t, `[]`, `[]`)
|
||||
a.bodies[factNamesPath] = factNamesBody("a1", "a2", "a3")
|
||||
b := newFakeBackend(t, `[]`, `[]`)
|
||||
b.bodies[factNamesPath] = factNamesBody("a3", "b1")
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
|
||||
|
||||
rec := doGetParams(t, srv.Handler(), factNamesPath, url.Values{
|
||||
"limit": {"2"},
|
||||
"include_total": {"true"},
|
||||
})
|
||||
if got := names(t, rec.Body.Bytes()); !reflect.DeepEqual(got, []string{"a1", "a2"}) {
|
||||
t.Errorf("paged %s = %v, want [a1 a2]", factNamesPath, got)
|
||||
}
|
||||
if got := rec.Header().Get(recordsHeader); got != "5" {
|
||||
t.Errorf("%s = %q, want the deduped union size 5", recordsHeader, got)
|
||||
}
|
||||
if got, _ := a.params(factNamesPath); got.Get("limit") != "" {
|
||||
t.Errorf("backend a got limit=%q, want none: a total cannot be counted from a window", got.Get("limit"))
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandler_FactNamesRejectsBadPaging(t *testing.T) {
|
||||
a := newFakeBackend(t, `[]`, `[]`)
|
||||
b := newFakeBackend(t, `[]`, `[]`)
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
|
||||
|
||||
for _, params := range []url.Values{
|
||||
{"limit": {"-1"}},
|
||||
// name is the only column the entity projects, so anything else is a 400
|
||||
// upstream and must be one here.
|
||||
{"order_by": {`[{"field":"bogus","order":"desc"}]`}},
|
||||
} {
|
||||
rec := doGetParams(t, srv.Handler(), factNamesPath, params)
|
||||
if rec.Code != http.StatusBadRequest {
|
||||
t.Errorf("%v: status %d, want 400: %s", params, rec.Code, rec.Body.String())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The merged /facts response carries records of the name pdbmux owns, so the
|
||||
// overview has to list it — and must not once nothing produces it.
|
||||
func TestHandler_FactNamesListsTheOwnedName(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
enabled bool
|
||||
upstream string
|
||||
want []string
|
||||
}{
|
||||
{"enabled", true, factNamesBody("osfamily"), []string{"osfamily", defaultSourceFact}},
|
||||
{"disabled", false, factNamesBody("osfamily"), []string{"osfamily"}},
|
||||
{"listed once when a backend reports it too", true, factNamesBody(defaultSourceFact, "osfamily"), []string{"osfamily", defaultSourceFact}},
|
||||
{"kept when a backend reports it and the feature is off", false, factNamesBody(defaultSourceFact, "osfamily"), []string{"osfamily", defaultSourceFact}},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
a := newFakeBackend(t, `[]`, `[]`)
|
||||
a.bodies[factNamesPath] = tc.upstream
|
||||
b := newFakeBackend(t, `[]`, `[]`)
|
||||
b.bodies[factNamesPath] = `[]`
|
||||
cfg := testConfig(a.srv.URL, b.srv.URL, mergeStatic)
|
||||
cfg.SourceFactEnabled = tc.enabled
|
||||
srv := newTestServer(cfg)
|
||||
|
||||
rec := doGet(t, srv.Handler(), factNamesPath, "")
|
||||
if got := names(t, rec.Body.Bytes()); !reflect.DeepEqual(got, tc.want) {
|
||||
t.Errorf("%s = %v, want %v", factNamesPath, got, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// The owned name is part of the union, so it is counted and ordered with it.
|
||||
func TestHandler_FactNamesOwnedNameIsCountedAndOrdered(t *testing.T) {
|
||||
a := newFakeBackend(t, `[]`, `[]`)
|
||||
a.bodies[factNamesPath] = factNamesBody("osfamily", "zone")
|
||||
b := newFakeBackend(t, `[]`, `[]`)
|
||||
b.bodies[factNamesPath] = `[]`
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
|
||||
|
||||
rec := doGetParams(t, srv.Handler(), factNamesPath, url.Values{
|
||||
"order_by": {`[{"field":"name","order":"desc"}]`},
|
||||
"include_total": {"true"},
|
||||
})
|
||||
want := []string{"zone", defaultSourceFact, "osfamily"}
|
||||
if got := names(t, rec.Body.Bytes()); !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("descending %s = %v, want %v", factNamesPath, got, want)
|
||||
}
|
||||
if got := rec.Header().Get(recordsHeader); got != "3" {
|
||||
t.Errorf("%s = %q, want 3", recordsHeader, got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandler_FactNamesServesEmptyArray(t *testing.T) {
|
||||
a := newFakeBackend(t, `[]`, `[]`)
|
||||
a.bodies[factNamesPath] = `[]`
|
||||
b := newFakeBackend(t, `[]`, `[]`)
|
||||
b.bodies[factNamesPath] = `[]`
|
||||
cfg := testConfig(a.srv.URL, b.srv.URL, mergeStatic)
|
||||
cfg.SourceFactEnabled = false
|
||||
srv := newTestServer(cfg)
|
||||
|
||||
rec := doGet(t, srv.Handler(), factNamesPath, "")
|
||||
if got := rec.Body.String(); got != "[]\n" {
|
||||
t.Errorf("empty %s body = %q, want %q", factNamesPath, got, "[]\n")
|
||||
}
|
||||
}
|
||||
|
||||
// A backend that is down must not black-hole the other's names.
|
||||
func TestHandler_FactNamesSurvivesOneBackend(t *testing.T) {
|
||||
a := newFakeBackend(t, `[]`, `[]`)
|
||||
a.fail = true
|
||||
b := newFakeBackend(t, `[]`, `[]`)
|
||||
b.bodies[factNamesPath] = factNamesBody("kernel")
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
|
||||
|
||||
rec := doGet(t, srv.Handler(), factNamesPath, "")
|
||||
if got := names(t, rec.Body.Bytes()); !reflect.DeepEqual(got, []string{"kernel", defaultSourceFact}) {
|
||||
t.Errorf("%s = %v, want the surviving backend's names", factNamesPath, got)
|
||||
}
|
||||
if h := rec.Header().Get(backendsHeader); h != "1/2" {
|
||||
t.Errorf("%s = %q, want 1/2", backendsHeader, h)
|
||||
}
|
||||
}
|
||||
|
||||
// Both routes are merged record sets, so they use the same cache /facts does.
|
||||
func TestHandler_FactRoutesAreCached(t *testing.T) {
|
||||
for _, path := range []string{roleFactPath, roleFactPath + "/web", sourceFactURL, factNamesPath} {
|
||||
t.Run(path, func(t *testing.T) {
|
||||
a := newFakeBackend(t, `[]`, `[]`)
|
||||
a.bodies[path] = `[]`
|
||||
b := newFakeBackend(t, `[]`, `[]`)
|
||||
b.bodies[path] = `[]`
|
||||
srv := newTestServer(cacheTestConfig(a.srv.URL, b.srv.URL))
|
||||
|
||||
if got := doGet(t, srv.Handler(), path, "").Header().Get(cacheStatusHeader); got != "miss" {
|
||||
t.Errorf("first %s = %q, want miss", cacheStatusHeader, got)
|
||||
}
|
||||
if got := doGet(t, srv.Handler(), path, "").Header().Get(cacheStatusHeader); got != "hit" {
|
||||
t.Errorf("second %s = %q, want hit", cacheStatusHeader, got)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -2,6 +2,7 @@ package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"sort"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -170,6 +171,86 @@ func mergeFacts(results []backendResult, owner func(certname string) string, inj
|
||||
return out
|
||||
}
|
||||
|
||||
// sourceFactRecords keeps the merged /facts records naming the fact pdbmux
|
||||
// owns, and, when the path pins a value, only those naming that backend. The
|
||||
// merge drops every upstream record of that name, so what survives is exactly
|
||||
// the synthetic set — one record per certname the merge attributed.
|
||||
func sourceFactRecords(merged []json.RawMessage, name, value string, valued bool) []json.RawMessage {
|
||||
out := []json.RawMessage{}
|
||||
for _, raw := range merged {
|
||||
var m struct {
|
||||
Name string `json:"name"`
|
||||
Value string `json:"value"`
|
||||
}
|
||||
if json.Unmarshal(raw, &m) != nil || m.Name != name {
|
||||
continue
|
||||
}
|
||||
if valued && m.Value != value {
|
||||
continue
|
||||
}
|
||||
out = append(out, raw)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// mergeFactNames unions the backends' /fact-names arrays, dedupes by the name
|
||||
// itself and re-sorts, since each backend only ordered its own slice. owned is
|
||||
// the fact name pdbmux injects, or "" while injection is off: it is listed
|
||||
// whether or not a backend reports it, because the merged /facts response
|
||||
// carries records of that name.
|
||||
func mergeFactNames(results []backendResult, owned string, desc bool) []json.RawMessage {
|
||||
seen := map[string]bool{}
|
||||
raws := []json.RawMessage{}
|
||||
values := []any{}
|
||||
add := func(raw json.RawMessage) {
|
||||
key := factNameKey(raw)
|
||||
if seen[key] {
|
||||
return
|
||||
}
|
||||
seen[key] = true
|
||||
var v any
|
||||
_ = json.Unmarshal(raw, &v) // an undecodable element sorts as null
|
||||
raws = append(raws, raw)
|
||||
values = append(values, v)
|
||||
}
|
||||
for _, res := range results {
|
||||
for _, rec := range res.records {
|
||||
add(rec.Raw)
|
||||
}
|
||||
}
|
||||
if owned != "" {
|
||||
if raw, err := json.Marshal(owned); err == nil {
|
||||
add(raw)
|
||||
}
|
||||
}
|
||||
idx := make([]int, len(raws))
|
||||
for i := range idx {
|
||||
idx[i] = i
|
||||
}
|
||||
sort.SliceStable(idx, func(a, b int) bool {
|
||||
c := compareValues(values[idx[a]], values[idx[b]])
|
||||
if desc {
|
||||
return c > 0
|
||||
}
|
||||
return c < 0
|
||||
})
|
||||
out := make([]json.RawMessage, 0, len(raws))
|
||||
for _, i := range idx {
|
||||
out = append(out, raws[i])
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Two encoders can spell one name differently — Go escapes <, > and & — so a
|
||||
// name is deduped on its decoded value, not on its bytes.
|
||||
func factNameKey(raw json.RawMessage) string {
|
||||
var name string
|
||||
if json.Unmarshal(raw, &name) == nil {
|
||||
return "s\x00" + name
|
||||
}
|
||||
return "r\x00" + string(raw)
|
||||
}
|
||||
|
||||
func environmentOf(recs []record) string {
|
||||
for _, rec := range recs {
|
||||
if rec.Environment != "" {
|
||||
|
||||
@@ -17,6 +17,7 @@ import (
|
||||
|
||||
const (
|
||||
factsPath = "/pdb/query/v4/facts"
|
||||
factNamesPath = "/pdb/query/v4/fact-names"
|
||||
nodesPath = "/pdb/query/v4/nodes"
|
||||
resourcesPath = "/pdb/query/v4/resources"
|
||||
reportsPath = "/pdb/query/v4/reports"
|
||||
@@ -25,6 +26,9 @@ const (
|
||||
aggregateEventCountsPath = "/pdb/query/v4/aggregate-event-counts"
|
||||
queryV4 = "/pdb/query/v4/"
|
||||
|
||||
// The only column /fact-names projects, so the only one it can be ordered on.
|
||||
factNamesColumn = "name"
|
||||
|
||||
// PuppetDB only sends this when the request carries include_total=true.
|
||||
recordsHeader = "X-Records"
|
||||
|
||||
@@ -91,13 +95,13 @@ func (s *Server) StartProbes(ctx context.Context) { s.health.Start(ctx) }
|
||||
// StopProbes stops the probing goroutines and waits for them to exit.
|
||||
func (s *Server) StopProbes() { s.health.Stop() }
|
||||
|
||||
// cacheFor picks the cache backing a request. 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.
|
||||
// cacheFor picks the cache backing a request. The merged /nodes and fact record
|
||||
// sets — /facts, /facts/<name>[/<value>] and /fact-names — share the in-memory
|
||||
// cache; every other path is uncached until the reports cache lands, and a new
|
||||
// backend is a case here rather than a change to any handler.
|
||||
func (s *Server) cacheFor(path string, params url.Values) (Cache, bool) {
|
||||
switch path {
|
||||
case factsPath, nodesPath:
|
||||
switch {
|
||||
case path == factsPath, path == nodesPath, path == factNamesPath, isFactsSubPath(path):
|
||||
// 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 {
|
||||
@@ -132,6 +136,8 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) {
|
||||
s.serveResources(w, r)
|
||||
case factsPath:
|
||||
s.serveFacts(w, r)
|
||||
case factNamesPath:
|
||||
s.serveFactNames(w, r)
|
||||
case reportsPath:
|
||||
s.serveReports(w, r)
|
||||
case eventsPath:
|
||||
@@ -139,6 +145,10 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) {
|
||||
case eventCountsPath, aggregateEventCountsPath:
|
||||
s.serveSummed(w, r, r.URL.Path, inferredColumns)
|
||||
default:
|
||||
if name, value, valued := factsSubPath(r.URL.Path); name != "" {
|
||||
s.serveFactsByName(w, r, name, value, valued)
|
||||
return
|
||||
}
|
||||
if isReportSubResource(r.URL.Path) {
|
||||
s.serveFirstHolder(w, r)
|
||||
return
|
||||
@@ -147,6 +157,88 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
}
|
||||
|
||||
// factsSubPath matches /pdb/query/v4/facts/<name> and /pdb/query/v4/facts/<name>/<value>,
|
||||
// returning the fact name the path constrains, the value it pins and whether it
|
||||
// pins one at all. openvoxdb serves both from the facts entity, ANDing
|
||||
// ["=","name",<name>] (and ["=","value",<value>]) onto the request's own query,
|
||||
// so the record shape is exactly /facts' —
|
||||
// src/puppetlabs/puppetdb/http/handlers.clj:283-302 and
|
||||
// src/puppetlabs/puppetdb/http/query.clj:136-143,193-209.
|
||||
func factsSubPath(path string) (name, value string, valued bool) {
|
||||
rest, ok := strings.CutPrefix(path, factsPath+"/")
|
||||
if !ok {
|
||||
return "", "", false
|
||||
}
|
||||
name, value, hasValue := strings.Cut(rest, "/")
|
||||
if name == "" {
|
||||
return "", "", false
|
||||
}
|
||||
if hasValue && (value == "" || strings.Contains(value, "/")) {
|
||||
return "", "", false
|
||||
}
|
||||
if !hasValue {
|
||||
value = ""
|
||||
}
|
||||
return name, value, hasValue
|
||||
}
|
||||
|
||||
func isFactsSubPath(path string) bool {
|
||||
name, _, _ := factsSubPath(path)
|
||||
return name != ""
|
||||
}
|
||||
|
||||
// /fact-names answers with a flat array of fact-name strings rather than
|
||||
// certname-keyed records, so it gets its own union: dedupe by name and re-sort,
|
||||
// because each backend only ordered its own slice. openvoxdb projects a single
|
||||
// DISTINCT `name` column and defaults to name-ascending
|
||||
// (src/puppetlabs/puppetdb/query_eng/engine.clj:488-498,
|
||||
// src/puppetlabs/puppetdb/http/handlers.clj:349-362), so an order_by on any
|
||||
// other field is rejected as openvoxdb rejects it, and the direction is all this
|
||||
// route reads.
|
||||
func (s *Server) serveFactNames(w http.ResponseWriter, r *http.Request) {
|
||||
in := r.URL.Query()
|
||||
page, err := parsePaging(in)
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
for _, f := range page.order {
|
||||
if f.Field != factNamesColumn {
|
||||
http.Error(w, fmt.Sprintf("order_by field must be %q, got %q", factNamesColumn, f.Field), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
}
|
||||
desc := len(page.order) > 0 && page.order[0].Desc
|
||||
|
||||
// The merged /facts response carries the injected fact, so the list of names
|
||||
// has to carry its name; nothing produces it while injection is off.
|
||||
owned := ""
|
||||
if s.cfg.SourceFactEnabled {
|
||||
owned = s.cfg.SourceFact
|
||||
}
|
||||
|
||||
upstream := page.upstreamParams(in)
|
||||
if page.wantTotal {
|
||||
// A merged total counts the whole union, so the window cannot be pushed
|
||||
// upstream; the full name list is small enough to fetch.
|
||||
upstream.Del("limit")
|
||||
}
|
||||
|
||||
s.serveCached(w, r, factNamesPath, in, func(ctx context.Context) (cachedResponse, error) {
|
||||
alive, err := s.aliveResults(ctx, factNamesPath, upstream)
|
||||
if err != nil {
|
||||
return cachedResponse{}, err
|
||||
}
|
||||
merged := mergeFactNames(alive, owned, desc)
|
||||
resp := cachedResponse{Body: encodeRecords(page.apply(merged)), Records: -1}
|
||||
s.countBackends(&resp, alive)
|
||||
if page.wantTotal {
|
||||
resp.Records = len(merged)
|
||||
}
|
||||
return resp, nil
|
||||
})
|
||||
}
|
||||
|
||||
// Matches /pdb/query/v4/reports/<hash>/{events,logs,metrics}, whose data lives in exactly one backend.
|
||||
func isReportSubResource(path string) bool {
|
||||
rest, ok := strings.CutPrefix(path, reportsPath+"/")
|
||||
@@ -221,6 +313,77 @@ func (s *Server) serveFacts(w http.ResponseWriter, r *http.Request) {
|
||||
s.serveMerged(w, r, factsPath, s.mergeFactsResponse(r))
|
||||
}
|
||||
|
||||
// Both path forms are the facts entity with a name (and value) constraint ANDed
|
||||
// on, so an aggregate over them carries no certname for the per-certname merge
|
||||
// to key on, and is summed instead — the same split /nodes makes. A plain query
|
||||
// on the source fact's own path is the one name no backend can answer for, so it
|
||||
// is synthesised instead of fanned out as-is.
|
||||
func (s *Server) serveFactsByName(w http.ResponseWriter, r *http.Request, name, value string, valued bool) {
|
||||
if spec := parseAggregate(r.URL.Query().Get("query")); spec != nil {
|
||||
s.serveSummed(w, r, r.URL.Path, spec.columns)
|
||||
return
|
||||
}
|
||||
if inject := s.newSourceInjector(r.URL.Query().Get("query"), true); inject.claims(name) && inject.injects() {
|
||||
s.serveSourceFact(w, r, inject, value, valued)
|
||||
return
|
||||
}
|
||||
s.serveMerged(w, r, r.URL.Path, s.mergeFactsByNameResponse(r))
|
||||
}
|
||||
|
||||
// serveSourceFact answers the drilldown on pdbmux's own fact. No backend holds a
|
||||
// record of that name, so the route's own fan-out would return nothing; the
|
||||
// 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 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)
|
||||
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, "", 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 {
|
||||
@@ -327,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())
|
||||
@@ -340,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
|
||||
}
|
||||
|
||||
@@ -352,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
|
||||
@@ -393,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)
|
||||
@@ -401,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
|
||||
@@ -413,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)
|
||||
@@ -421,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
|
||||
@@ -498,7 +685,20 @@ func (s *Server) mergeNodesResponse(r *http.Request) func([]backendResult) []jso
|
||||
}
|
||||
|
||||
func (s *Server) mergeFactsResponse(r *http.Request) func([]backendResult) []json.RawMessage {
|
||||
return s.mergeFactsWith(s.newSourceInjector(r.URL.Query().Get("query"), true))
|
||||
}
|
||||
|
||||
// The path segment of /facts/<name> is the same outer `name` constraint the
|
||||
// query gate already rules injection out on, so nothing is synthesised here:
|
||||
// the source fact's own path is diverted to serveSourceFact before this.
|
||||
// Suppression of an upstream record of the owned name stays on.
|
||||
func (s *Server) mergeFactsByNameResponse(r *http.Request) func([]backendResult) []json.RawMessage {
|
||||
inject := s.newSourceInjector(r.URL.Query().Get("query"), true)
|
||||
inject.disableInject()
|
||||
return s.mergeFactsWith(inject)
|
||||
}
|
||||
|
||||
func (s *Server) mergeFactsWith(inject *sourceInjector) func([]backendResult) []json.RawMessage {
|
||||
return func(results []backendResult) []json.RawMessage {
|
||||
var merged []json.RawMessage
|
||||
if s.cfg.Merge == mergeStatic {
|
||||
|
||||
@@ -39,6 +39,14 @@ func (si *sourceInjector) injects() bool {
|
||||
return si != nil && si.inject
|
||||
}
|
||||
|
||||
// disableInject rules the synthetic record out for a request shape the query
|
||||
// gate cannot see, leaving suppression on.
|
||||
func (si *sourceInjector) disableInject() {
|
||||
if si != nil {
|
||||
si.inject = false
|
||||
}
|
||||
}
|
||||
|
||||
// logSuppressed reports, once per request, that upstream records were dropped.
|
||||
func (si *sourceInjector) logSuppressed(l *log.Logger) {
|
||||
if si == nil || si.suppressed == 0 || l == nil {
|
||||
|
||||
Reference in New Issue
Block a user