Sum aggregates on the /facts/<name> routes
An aggregate row carries no certname, so the per-certname merge collapsed every backend's row into one bucket and served a single backend's count as the estate's — no error, no warning, X-Backends still 2/2. - Route an aggregate query on /facts/<name>[/<value>] to the summing path. - List the injected fact's name in /fact-names instead of hiding it. - Reject an order_by on any field but name, as the backends do.
This commit is contained in:
@@ -26,8 +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. |
|
||||
| `GET /pdb/query/v4/facts/<name>[/<value>]` | Same fan-out and merge as `/facts`. The path segment is a `name` constraint, so no synthetic `pdbmux_source` record is added unless the path names it. |
|
||||
| `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. |
|
||||
| `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 unless the path names it. |
|
||||
| `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. |
|
||||
@@ -70,9 +70,13 @@ paths add two more headers `pdbmux` sets itself, `X-Cache` and `Age` — see
|
||||
- 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.
|
||||
more constraint applied upstream, so they take the same rule — and the same
|
||||
aggregate branch, since a count row has no `certname` there either.
|
||||
- **`/fact-names`** — a flat array of strings, not records: **union**, deduped by
|
||||
the name and re-sorted, ascending unless `order_by` says otherwise.
|
||||
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
|
||||
@@ -83,17 +87,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` or `/resources` query whose `extract` carries a `["function", ...]`
|
||||
column.
|
||||
`/nodes`, `/resources` or `/facts/<name>[/<value>]` query whose `extract`
|
||||
carries a `["function", ...]` column.
|
||||
- The grouping key is the row's non-aggregate fields: for `/reports`,
|
||||
`/nodes` 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 `summarize_by`), whose
|
||||
remaining fields are all counts.
|
||||
- On `/nodes` this takes precedence over the `certname` merge: a count row has
|
||||
no `certname`, so deduping would collapse every backend's count into one
|
||||
backend's number. A `/nodes` query with no `function` column — including a
|
||||
plain `extract` projection — still merges by `certname`.
|
||||
`/nodes`, `/resources` 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/<name>[/<value>]` this takes precedence over the
|
||||
`certname` merge: a count row has no `certname`, so deduping would collapse
|
||||
every backend's count into one backend's number. A query with no `function`
|
||||
column — including a plain `extract` projection — still merges by
|
||||
`certname`.
|
||||
- `/resources` has no cross-backend record identity to dedupe on, so only its
|
||||
aggregate queries merge; everything else stays an unmerged pass-through.
|
||||
- Rows sharing a key collapse into one with their numeric columns summed. A key
|
||||
@@ -170,9 +175,11 @@ 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. `/pdb/query/v4/facts/pdbmux_source` is
|
||||
the same story: the route is merged and injection is allowed there, but the
|
||||
backends hold no record to attach it to. `/pdb/query/v4/fact-names` therefore
|
||||
leaves the name out of its list — advertising a name no `/facts` response serves
|
||||
would offer a drilldown with nothing behind it.
|
||||
backends hold no record to attach it to, so it answers empty.
|
||||
`/pdb/query/v4/fact-names` still lists the name while injection is on, because
|
||||
every merged `/facts` response does carry records of it — leaving it out hides a
|
||||
fact every node has from any client that discovers names there. Turning injection
|
||||
off removes it, since nothing then produces it.
|
||||
|
||||
**Not covered:** `/factsets` and `/inventory`. Both carry facts, but `pdbmux`
|
||||
does not merge either today — they take the unmerged pass-through path, where
|
||||
|
||||
@@ -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)
|
||||
|
||||
+97
-4
@@ -6,6 +6,7 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"sort"
|
||||
"testing"
|
||||
@@ -126,10 +127,10 @@ func TestFactNamesAreMerged(t *testing.T) {
|
||||
}
|
||||
seen[name] = true
|
||||
}
|
||||
// No merged /facts response can serve the name pdbmux owns, so the list must
|
||||
// not advertise it.
|
||||
if seen[defaultSourceFact] {
|
||||
t.Errorf("merged /fact-names advertises %s, which no /facts response carries upstream", defaultSourceFact)
|
||||
// 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")
|
||||
@@ -158,6 +159,98 @@ func TestFactNamesAreMerged(t *testing.T) {
|
||||
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 {
|
||||
|
||||
+148
-19
@@ -5,6 +5,7 @@ import (
|
||||
"net/http"
|
||||
"net/url"
|
||||
"reflect"
|
||||
"slices"
|
||||
"testing"
|
||||
)
|
||||
|
||||
@@ -109,6 +110,102 @@ func TestHandler_FactsByNameAndValueMerged(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// 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"} {
|
||||
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] = `[]`
|
||||
@@ -200,7 +297,7 @@ func TestHandler_FactNamesUnionDedupedAndSorted(t *testing.T) {
|
||||
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"}
|
||||
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)
|
||||
}
|
||||
@@ -219,7 +316,7 @@ func TestHandler_FactNamesOrderByDescending(t *testing.T) {
|
||||
rec := doGetParams(t, srv.Handler(), factNamesPath, url.Values{
|
||||
"order_by": {`[{"field":"name","order":"desc"}]`},
|
||||
})
|
||||
want := []string{"zone", "osfamily", "kernel"}
|
||||
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)
|
||||
}
|
||||
@@ -263,8 +360,8 @@ func TestHandler_FactNamesTotalCountsTheWholeUnion(t *testing.T) {
|
||||
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 != "4" {
|
||||
t.Errorf("%s = %q, want the deduped union size 4", recordsHeader, 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"))
|
||||
@@ -276,27 +373,36 @@ func TestHandler_FactNamesRejectsBadPaging(t *testing.T) {
|
||||
b := newFakeBackend(t, `[]`, `[]`)
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
|
||||
|
||||
rec := doGetParams(t, srv.Handler(), factNamesPath, url.Values{"limit": {"-1"}})
|
||||
if rec.Code != http.StatusBadRequest {
|
||||
t.Errorf("status %d, want 400: %s", rec.Code, rec.Body.String())
|
||||
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())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// No merged /facts response can serve the name pdbmux owns, so the overview must
|
||||
// not advertise it — and must keep it once the feature is off.
|
||||
func TestHandler_FactNamesDropsTheOwnedName(t *testing.T) {
|
||||
body := factNamesBody(defaultSourceFact, "osfamily")
|
||||
// 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
|
||||
want []string
|
||||
name string
|
||||
enabled bool
|
||||
upstream string
|
||||
want []string
|
||||
}{
|
||||
{"enabled", true, []string{"osfamily"}},
|
||||
{"disabled", false, []string{"osfamily", defaultSourceFact}},
|
||||
{"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] = body
|
||||
a.bodies[factNamesPath] = tc.upstream
|
||||
b := newFakeBackend(t, `[]`, `[]`)
|
||||
b.bodies[factNamesPath] = `[]`
|
||||
cfg := testConfig(a.srv.URL, b.srv.URL, mergeStatic)
|
||||
@@ -311,12 +417,35 @@ func TestHandler_FactNamesDropsTheOwnedName(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// 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] = `[]`
|
||||
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
|
||||
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" {
|
||||
@@ -333,7 +462,7 @@ func TestHandler_FactNamesSurvivesOneBackend(t *testing.T) {
|
||||
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"}) {
|
||||
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" {
|
||||
|
||||
@@ -172,29 +172,33 @@ func mergeFacts(results []backendResult, owner func(certname string) string, inj
|
||||
}
|
||||
|
||||
// mergeFactNames unions the backends' /fact-names arrays, dedupes by the name
|
||||
// itself and re-sorts, since each backend only ordered its own slice. The name
|
||||
// pdbmux owns is dropped for the same reason its fact records are: while the
|
||||
// feature is on, no merged /facts response can serve it.
|
||||
func mergeFactNames(results []backendResult, suppress *sourceInjector, desc bool) []json.RawMessage {
|
||||
// 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 {
|
||||
var name string
|
||||
if json.Unmarshal(rec.Raw, &name) == nil && suppress.claims(name) {
|
||||
suppress.suppressed++
|
||||
continue
|
||||
}
|
||||
key := string(rec.Raw)
|
||||
if seen[key] {
|
||||
continue
|
||||
}
|
||||
seen[key] = true
|
||||
var v any
|
||||
_ = json.Unmarshal(rec.Raw, &v) // an undecodable element sorts as null
|
||||
raws = append(raws, rec.Raw)
|
||||
values = append(values, v)
|
||||
add(rec.Raw)
|
||||
}
|
||||
}
|
||||
if owned != "" {
|
||||
if raw, err := json.Marshal(owned); err == nil {
|
||||
add(raw)
|
||||
}
|
||||
}
|
||||
idx := make([]int, len(raws))
|
||||
@@ -215,6 +219,16 @@ func mergeFactNames(results []backendResult, suppress *sourceInjector, desc bool
|
||||
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 != "" {
|
||||
|
||||
@@ -26,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"
|
||||
|
||||
@@ -143,7 +146,7 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) {
|
||||
s.serveSummed(w, r, r.URL.Path, inferredColumns)
|
||||
default:
|
||||
if name, valued := factsSubPath(r.URL.Path); name != "" {
|
||||
s.serveMerged(w, r, r.URL.Path, s.mergeFactsByNameResponse(r, name, valued))
|
||||
s.serveFactsByName(w, r, name, valued)
|
||||
return
|
||||
}
|
||||
if isReportSubResource(r.URL.Path) {
|
||||
@@ -185,8 +188,9 @@ func isFactsSubPath(path string) bool {
|
||||
// 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 can only be
|
||||
// on that column and its direction is all this route reads.
|
||||
// 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)
|
||||
@@ -194,12 +198,20 @@ func (s *Server) serveFactNames(w http.ResponseWriter, r *http.Request) {
|
||||
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
|
||||
|
||||
// Built only for the name pdbmux owns: a string list has no query shape for
|
||||
// the injection gate to read, and nothing is ever injected here.
|
||||
suppress := s.newSourceInjector("", true)
|
||||
suppress.disableInject()
|
||||
// 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 {
|
||||
@@ -213,8 +225,7 @@ func (s *Server) serveFactNames(w http.ResponseWriter, r *http.Request) {
|
||||
if err != nil {
|
||||
return cachedResponse{}, err
|
||||
}
|
||||
merged := mergeFactNames(alive, suppress, desc)
|
||||
suppress.logSuppressed(s.log)
|
||||
merged := mergeFactNames(alive, owned, desc)
|
||||
resp := cachedResponse{Body: encodeRecords(page.apply(merged)), Records: -1}
|
||||
s.countBackends(&resp, alive)
|
||||
if page.wantTotal {
|
||||
@@ -289,6 +300,17 @@ func (s *Server) serveUnion(w http.ResponseWriter, r *http.Request, path string,
|
||||
})
|
||||
}
|
||||
|
||||
// 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.
|
||||
func (s *Server) serveFactsByName(w http.ResponseWriter, r *http.Request, name string, valued bool) {
|
||||
if spec := parseAggregate(r.URL.Query().Get("query")); spec != nil {
|
||||
s.serveSummed(w, r, r.URL.Path, spec.columns)
|
||||
return
|
||||
}
|
||||
s.serveMerged(w, r, r.URL.Path, s.mergeFactsByNameResponse(r, name, valued))
|
||||
}
|
||||
|
||||
// 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 {
|
||||
|
||||
Reference in New Issue
Block a user