diff --git a/README.md b/README.md index d9f05f5..95e8712 100644 --- a/README.md +++ b/README.md @@ -24,14 +24,20 @@ not PQL) is forwarded verbatim. | Path | Behaviour | |---|---| -| `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). | +| `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/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. | | `GET /pdb/query/v4/event-counts` | Fan out to all and **sum** each subject's counts into one row per subject. | | `GET /pdb/query/v4/aggregate-event-counts` | Fan out to all and **sum** the summary object's counts. | | `GET /pdb/query/v4/reports//{events,logs,metrics}` | Ask every backend; serve the answer from whichever backend actually holds that report. `404` when none does. | | `GET /pdb/query/v4/*` (any other) | No merge rule, so backends are tried in configured order and the first success is streamed back verbatim; if all reject it, the first upstream error response is replayed. | +| `GET /pdb/meta/v1/version` | Fan out to all and report the **lowest** version any backend runs. | +| `GET /pdb/meta/v1/server-time` | Fan out to all and serve the first reachable backend's clock. | +| `GET /metrics/v2/read/` | Fan out to all and merge the Jolokia response; numeric attributes are **summed** by default (see merge semantics). | +| `GET /metrics/v2/list` | Fan out to all and serve the **union** of the backends' MBean trees. | +| `GET /metrics/v1/mbeans[/]` | Same merge, applied to the legacy envelope-less body. | | `GET /healthz` | Per-backend reachability. `200 {"status":"ok"}` if all reachable, `200 degraded` if some fail, `503 down` if all fail. | Fan-out is concurrent. If one backend errors or times out, `pdbmux` serves the @@ -62,12 +68,20 @@ unknown fields survive untouched. reporting to more than one backend stores identical records in each). - **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 a `/reports` query whose - `extract` carries a `["function", ...]` column. - - The grouping key is the row's non-aggregate fields: for `/reports` 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. + covers `/event-counts`, `/aggregate-event-counts`, and any `/reports`, + `/nodes` or `/resources` 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`. + - `/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 only one backend reported is passed through byte-for-byte. An aggregate column that is absent or non-numeric in a row is skipped, never zeroed, so the @@ -133,6 +147,39 @@ the answer comes from whichever backend replied first rather than from a merge winner, so there is no owner to attribute. Injecting there would state a provenance that isn't true. +### Metadata and metrics + +- **`/pdb/meta/v1/version`** — when the backends agree, that version is served. + When they differ, `pdbmux` reports the **lowest**: a client reads this as the + feature level it may rely on, and the estate can only be relied on for what its + oldest PuppetDB implements. Versions compare segment by segment, numerically + where both segments are numbers (`7.9.0` < `7.12.0`), lexically otherwise. + A backend whose body is unparseable is skipped rather than treated as lowest. +- **`/pdb/meta/v1/server-time`** — the clock of whichever PuppetDB answered is + not estate state and has no meaningful merge, so the first **reachable** + backend in configured order supplies it, the same tie-break used elsewhere. +- **`/metrics/...`** — the Jolokia envelope's `value` is merged and the rest of + the envelope comes from the first backend (with the newest `timestamp`). + Values merge recursively: + - Objects merge over the **union** of their keys, so an MBean attribute only + one backend exposes still survives. + - Numbers combine by the attribute's own name. The default is a **sum** — + almost everything here is a population count (`num-nodes`, `num-resources`, + queue depth, command totals) whose estate-wide value is the total, and rates + are additive throughput. The exceptions describe a distribution or a bound, + where adding two servers' numbers yields a figure that was never true of + either: `Min` takes the minimum; `Max`, `Uptime` and `StartTime` take the + maximum; `Mean`, `Median`, `StdDev` and `*Percentile` take the unweighted + arithmetic mean (`pdbmux` has no per-backend sample counts to weight by). + Matching is case-insensitive. + - Strings, booleans, arrays, nulls and mixed kinds keep the first backend's + value — there is no sound way to add them. + - Jolokia signals a bad MBean as a non-2xx `status` **inside** an HTTP 200. + Such a backend is skipped; if every backend does so, the first one's error + envelope is replayed verbatim so the client sees the real reason. + - MBean names arrive percent-encoded over Jolokia's own `!`-escaping; the raw + path is forwarded so neither layer is lost. + ### Paging and ordering on the merged endpoints Each backend applies `order_by`/`limit`/`offset` to its own slice only, so diff --git a/meta.go b/meta.go new file mode 100644 index 0000000..427e2bd --- /dev/null +++ b/meta.go @@ -0,0 +1,195 @@ +package main + +import ( + "context" + "encoding/json" + "io" + "net/http" + "strconv" + "strings" + "sync" +) + +const ( + metaVersionPath = "/pdb/meta/v1/version" + metaServerTimePath = "/pdb/meta/v1/server-time" +) + +// rawResult is one backend's verbatim response, for endpoints whose payload is +// not a PuppetDB record array. +type rawResult struct { + name string + status int + contentType string + body []byte + err error +} + +// ok reports whether the backend answered 2xx. +func (r rawResult) ok() bool { + return r.err == nil && r.status >= 200 && r.status < 300 +} + +// fanOutRaw asks every backend for path concurrently and returns one result per +// backend, in configured order, without interpreting the bodies. +func (s *Server) fanOutRaw(ctx context.Context, path, rawQuery string) []rawResult { + results := make([]rawResult, len(s.cfg.Backends)) + var wg sync.WaitGroup + for i, b := range s.cfg.Backends { + wg.Add(1) + go func(i int, b Backend) { + defer wg.Done() + results[i] = s.rawBackend(ctx, b, path, rawQuery) + }(i, b) + } + wg.Wait() + return results +} + +func (s *Server) rawBackend(ctx context.Context, b Backend, path, rawQuery string) rawResult { + target := strings.TrimRight(b.URL, "/") + path + if rawQuery != "" { + target += "?" + rawQuery + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil) + if err != nil { + return rawResult{name: b.Name, err: err} + } + resp, err := s.client.Do(req) + if err != nil { + return rawResult{name: b.Name, err: err} + } + defer func() { _ = resp.Body.Close() }() + body, err := io.ReadAll(resp.Body) + if err != nil { + return rawResult{name: b.Name, err: err} + } + return rawResult{ + name: b.Name, + status: resp.StatusCode, + contentType: resp.Header.Get("Content-Type"), + body: body, + } +} + +// aliveRaw drops backends that errored or answered non-2xx, writing a 502 and +// returning ok=false only when none is left. +func (s *Server) aliveRaw(w http.ResponseWriter, results []rawResult, path string) ([]rawResult, bool) { + var alive []rawResult + for _, res := range results { + if !res.ok() { + s.log.Printf("warning: backend %q failed for %s: %s", res.name, path, res.reason()) + continue + } + alive = append(alive, res) + } + if len(alive) == 0 { + http.Error(w, "all backends failed", http.StatusBadGateway) + return nil, false + } + return alive, true +} + +func (r rawResult) reason() string { + if r.err != nil { + return r.err.Error() + } + return "HTTP " + strconv.Itoa(r.status) + ": " + strings.TrimSpace(string(r.body)) +} + +// handleMetaVersion serves /pdb/meta/v1/version. Clients (pypuppetdb, and so +// Puppetboard's startup check) treat the answer as the feature level they may +// rely on, so the merged answer is the *lowest* version any backend reports: +// the estate can only be counted on for what its oldest member implements. +func (s *Server) handleMetaVersion(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "only GET is supported", http.StatusMethodNotAllowed) + return + } + alive, ok := s.aliveRaw(w, s.fanOutRaw(r.Context(), metaVersionPath, r.URL.RawQuery), metaVersionPath) + if !ok { + return + } + + lowest := alive[0] + lowestVer, hasVer := metaField(lowest.body, "version") + for _, res := range alive[1:] { + v, ok := metaField(res.body, "version") + if !ok { + continue + } + if !hasVer || compareVersions(v, lowestVer) < 0 { + lowest, lowestVer, hasVer = res, v, true + } + } + writeRaw(w, lowest) +} + +// handleMetaServerTime serves /pdb/meta/v1/server-time. The clock of whichever +// PuppetDB answered is not estate state and does not merge, so the first +// reachable backend in configured order supplies it — the same tie-break rule +// used everywhere else. +func (s *Server) handleMetaServerTime(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "only GET is supported", http.StatusMethodNotAllowed) + return + } + alive, ok := s.aliveRaw(w, s.fanOutRaw(r.Context(), metaServerTimePath, r.URL.RawQuery), metaServerTimePath) + if !ok { + return + } + writeRaw(w, alive[0]) +} + +// metaField pulls a string field out of a `{"version": "..."}`-shaped body. +func metaField(body []byte, field string) (string, bool) { + var obj map[string]json.RawMessage + if json.Unmarshal(body, &obj) != nil { + return "", false + } + var s string + if json.Unmarshal(obj[field], &s) != nil || s == "" { + return "", false + } + return s, true +} + +// compareVersions orders dotted version strings segment by segment, comparing +// numerically where both segments are numbers and lexically otherwise, so +// "7.12.1" sorts below "8.4.0" and below "7.12.2". A prefix is lower than a +// longer string sharing it ("7.12" < "7.12.1"), and a pre-release suffix is +// compared as text within its segment ("8.0.0" < "8.0.0-SNAPSHOT"). +func compareVersions(a, b string) int { + as, bs := strings.Split(a, "."), strings.Split(b, ".") + for i := 0; i < len(as) && i < len(bs); i++ { + an, aok := strconv.Atoi(as[i]) + bn, bok := strconv.Atoi(bs[i]) + if aok == nil && bok == nil { + if an != bn { + return sign(an - bn) + } + continue + } + if c := strings.Compare(as[i], bs[i]); c != 0 { + return c + } + } + return sign(len(as) - len(bs)) +} + +func sign(n int) int { + switch { + case n < 0: + return -1 + case n > 0: + return 1 + default: + return 0 + } +} + +func writeRaw(w http.ResponseWriter, res rawResult) { + setContentType(w, res.contentType) + w.WriteHeader(res.status) + _, _ = w.Write(res.body) +} diff --git a/meta_test.go b/meta_test.go new file mode 100644 index 0000000..6142592 --- /dev/null +++ b/meta_test.go @@ -0,0 +1,170 @@ +package main + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +func metaGet(t *testing.T, h http.Handler, path string) *httptest.ResponseRecorder { + t.Helper() + rec := httptest.NewRecorder() + h.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, path, nil)) + return rec +} + +func metaString(t *testing.T, body []byte, field string) string { + t.Helper() + var obj map[string]string + if err := json.Unmarshal(body, &obj); err != nil { + t.Fatalf("unmarshal %s: %v", body, err) + } + return obj[field] +} + +func TestMetaVersion_BackendsAgree(t *testing.T) { + // Puppetboard's check_db_version() calls this at import and exits 2 on any + // non-200, so a 404 here is the difference between a running dashboard and + // CrashLoopBackOff. + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[metaVersionPath] = `{"version":"7.12.1"}` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[metaVersionPath] = `{"version":"7.12.1"}` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := metaGet(t, srv.Handler(), metaVersionPath) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + if got := metaString(t, rec.Body.Bytes(), "version"); got != "7.12.1" { + t.Errorf("version = %q, want 7.12.1", got) + } +} + +func TestMetaVersion_DisagreementReportsLowest(t *testing.T) { + // The estate can only be relied on for what its oldest PuppetDB implements. + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[metaVersionPath] = `{"version":"8.4.0"}` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[metaVersionPath] = `{"version":"7.12.1"}` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + if got := metaString(t, metaGet(t, srv.Handler(), metaVersionPath).Body.Bytes(), "version"); got != "7.12.1" { + t.Errorf("version = %q, want the lower 7.12.1", got) + } +} + +func TestMetaVersion_LowestIsIndependentOfBackendOrder(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[metaVersionPath] = `{"version":"7.12.1"}` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[metaVersionPath] = `{"version":"8.4.0"}` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + if got := metaString(t, metaGet(t, srv.Handler(), metaVersionPath).Body.Bytes(), "version"); got != "7.12.1" { + t.Errorf("version = %q, want the lower 7.12.1", got) + } +} + +func TestMetaVersion_OneBackendDown(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.fail = true + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[metaVersionPath] = `{"version":"8.4.0"}` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := metaGet(t, srv.Handler(), metaVersionPath) + if rec.Code != http.StatusOK { + t.Fatalf("expected 200 serving the survivor, got %d", rec.Code) + } + if got := metaString(t, rec.Body.Bytes(), "version"); got != "8.4.0" { + t.Errorf("version = %q, want 8.4.0", got) + } +} + +func TestMetaVersion_AllBackendsDown(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.fail = true + b := newFakeBackend(t, `[]`, `[]`) + b.fail = true + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + if rec := metaGet(t, srv.Handler(), metaVersionPath); rec.Code != http.StatusBadGateway { + t.Errorf("status %d, want 502", rec.Code) + } +} + +func TestMetaServerTime_FirstReachableBackend(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[metaServerTimePath] = `{"server_time":"2026-08-29T01:00:00.000Z"}` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[metaServerTimePath] = `{"server_time":"2026-08-29T02:00:00.000Z"}` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + got := metaString(t, metaGet(t, srv.Handler(), metaServerTimePath).Body.Bytes(), "server_time") + if got != "2026-08-29T01:00:00.000Z" { + t.Errorf("server_time = %q, want the first backend's", got) + } +} + +func TestMetaServerTime_SkipsDeadBackend(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.fail = true + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[metaServerTimePath] = `{"server_time":"2026-08-29T02:00:00.000Z"}` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := metaGet(t, srv.Handler(), metaServerTimePath) + if rec.Code != http.StatusOK { + t.Fatalf("expected 200 serving the survivor, got %d", rec.Code) + } + if got := metaString(t, rec.Body.Bytes(), "server_time"); got != "2026-08-29T02:00:00.000Z" { + t.Errorf("server_time = %q, want the survivor's", got) + } +} + +func TestMetaVersion_RejectsNonGET(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + b := newFakeBackend(t, `[]`, `[]`) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := httptest.NewRecorder() + srv.Handler().ServeHTTP(rec, httptest.NewRequest(http.MethodPost, metaVersionPath, nil)) + if rec.Code != http.StatusMethodNotAllowed { + t.Errorf("status %d, want 405", rec.Code) + } +} + +func TestCompareVersions(t *testing.T) { + cases := []struct { + a, b string + want int + }{ + {"7.12.1", "7.12.1", 0}, + {"7.12.1", "8.4.0", -1}, + {"8.4.0", "7.12.1", 1}, + {"7.9.0", "7.12.0", -1}, // numeric, not lexical: 9 < 12 + {"7.12", "7.12.1", -1}, + {"8.0.0", "8.0.0-SNAPSHOT", -1}, + {"8.0.0-SNAPSHOT", "8.0.0", 1}, + } + for _, c := range cases { + if got := compareVersions(c.a, c.b); got != c.want { + t.Errorf("compareVersions(%q, %q) = %d, want %d", c.a, c.b, got, c.want) + } + } +} + +func TestMetaField_MalformedBodyIgnored(t *testing.T) { + // A backend serving junk must not become the "lowest" version. + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[metaVersionPath] = `not json` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[metaVersionPath] = `{"version":"8.4.0"}` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + if got := metaString(t, metaGet(t, srv.Handler(), metaVersionPath).Body.Bytes(), "version"); got != "8.4.0" { + t.Errorf("version = %q, want 8.4.0 from the only parseable backend", got) + } +} diff --git a/metrics.go b/metrics.go new file mode 100644 index 0000000..321364a --- /dev/null +++ b/metrics.go @@ -0,0 +1,272 @@ +package main + +import ( + "bytes" + "encoding/json" + "net/http" + "strconv" + "strings" +) + +// metricsPrefix covers PuppetDB's Jolokia surface, which sits at the server root +// rather than under /pdb: pypuppetdb's metric() reads /metrics/v2/read/, +// lists via /metrics/v2/list, and falls back to /metrics/v1/mbeans/. +const metricsPrefix = "/metrics/" + +type mergeRule int + +const ( + ruleSum mergeRule = iota + ruleMin + ruleMax + ruleMean +) + +// mergeRuleFor picks how one numeric MBean attribute combines across backends. +// The default is a sum: the metrics Puppetboard renders are population counts +// (num-nodes, num-resources, queue depth, command totals) whose estate-wide +// value is the total. The exceptions are attributes describing a distribution or +// a bound rather than a quantity, where adding two servers' numbers yields a +// figure that was never true of either. +func mergeRuleFor(attr string) mergeRule { + a := strings.ToLower(attr) + switch a { + case "min": + return ruleMin + case "max", "uptime", "starttime": + return ruleMax + case "mean", "median", "stddev": + return ruleMean + } + if strings.HasSuffix(a, "percentile") { + return ruleMean + } + return ruleSum +} + +func (s *Server) handleMetrics(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "only GET is supported", http.StatusMethodNotAllowed) + return + } + // MBean names carry Jolokia's !-escapes under percent-encoding; the raw path + // is forwarded so neither layer is lost. + path := r.URL.EscapedPath() + alive, ok := s.aliveRaw(w, s.fanOutRaw(r.Context(), path, r.URL.RawQuery), path) + if !ok { + return + } + + body, ok := mergeMetrics(alive, metricAttribute(r.URL.Path)) + if !ok { + // Every backend answered 2xx but none carried a mergeable payload — a + // Jolokia error envelope, or a body that is not a JSON object. Replaying + // the first keeps the upstream error text the client expects. + writeRaw(w, alive[0]) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write(body) +} + +// metricAttribute names the single attribute a read asked for, when the URL +// carries one (/metrics/v2/read//), so a scalar response body +// still gets the right numeric rule. Empty means the response is an object whose +// own keys name its attributes. +func metricAttribute(path string) string { + rest, ok := strings.CutPrefix(path, metricsPrefix) + if !ok { + return "" + } + parts := strings.Split(rest, "/") // v2/read/[/] + if len(parts) < 4 { + return "" + } + return parts[len(parts)-1] +} + +// mergeMetrics folds the backends' Jolokia responses into one. A response is +// either a Jolokia envelope ({"request":…,"value":…,"status":200}), where only +// "value" merges and the rest comes from the first backend, or a bare attribute +// object (metrics/v1), which merges whole. ok=false means nothing was mergeable. +func mergeMetrics(alive []rawResult, attr string) ([]byte, bool) { + var objs []map[string]json.RawMessage + for _, res := range alive { + var obj map[string]json.RawMessage + if decodeJSON(res.body, &obj) != nil || obj == nil { + continue + } + // Jolokia reports a per-request failure inside an HTTP 200. + if n, ok := numberOf(obj["status"]); ok && (n < 200 || n >= 300) { + continue + } + objs = append(objs, obj) + } + if len(objs) == 0 { + return nil, false + } + + _, hasValue := objs[0]["value"] + _, hasStatus := objs[0]["status"] + if !hasValue || !hasStatus { + vals := make([]any, 0, len(objs)) + for _, obj := range objs { + vals = append(vals, decodedObject(obj)) + } + return marshal(mergeMetricValue(vals, attr)) + } + + vals := make([]any, 0, len(objs)) + for _, obj := range objs { + var v any + if decodeJSON(obj["value"], &v) == nil { + vals = append(vals, v) + } + } + if len(vals) == 0 { + return nil, false + } + merged, ok := marshal(mergeMetricValue(vals, attr)) + if !ok { + return nil, false + } + + out := make(map[string]json.RawMessage, len(objs[0])) + for k, v := range objs[0] { + out[k] = v + } + out["value"] = merged + // The envelope timestamp says when the answer was produced; the newest one + // describes the merged answer. + if ts, ok := maxField(objs, "timestamp"); ok { + out["timestamp"] = ts + } + return marshal(out) +} + +// maxField returns the largest numeric value of a field across the responses. +func maxField(objs []map[string]json.RawMessage, field string) (json.RawMessage, bool) { + var best json.RawMessage + var bestN float64 + for _, obj := range objs { + n, ok := numberOf(obj[field]) + if !ok { + continue + } + if best == nil || n > bestN { + best, bestN = obj[field], n + } + } + return best, best != nil +} + +// mergeMetricValue folds one attribute's value from every backend into one. +// Objects merge key by key over the union of keys, so a backend missing an +// attribute still contributes the rest. Numbers combine by the attribute's rule. +// Anything else — strings, booleans, arrays, nulls, or a mix of kinds — keeps +// the first backend's value, there being no sound way to add them. +func mergeMetricValue(vals []any, attr string) any { + if len(vals) == 0 { + return nil + } + if len(vals) == 1 { + return vals[0] + } + + objs := make([]map[string]any, 0, len(vals)) + for _, v := range vals { + if m, ok := v.(map[string]any); ok { + objs = append(objs, m) + } + } + if len(objs) == len(vals) { + out := map[string]any{} + for _, m := range objs { + for k := range m { + if _, done := out[k]; done { + continue + } + sub := make([]any, 0, len(objs)) + for _, o := range objs { + if v, ok := o[k]; ok { + sub = append(sub, v) + } + } + out[k] = mergeMetricValue(sub, k) + } + } + return out + } + + nums := make([]float64, 0, len(vals)) + for _, v := range vals { + n, ok := v.(json.Number) + if !ok { + break + } + f, err := n.Float64() + if err != nil { + break + } + nums = append(nums, f) + } + if len(nums) != len(vals) { + return vals[0] + } + return combineNumbers(nums, attr) +} + +func combineNumbers(nums []float64, attr string) json.RawMessage { + acc := nums[0] + switch mergeRuleFor(attr) { + case ruleMin: + for _, n := range nums[1:] { + if n < acc { + acc = n + } + } + case ruleMax: + for _, n := range nums[1:] { + if n > acc { + acc = n + } + } + case ruleMean: + for _, n := range nums[1:] { + acc += n + } + acc /= float64(len(nums)) + default: + for _, n := range nums[1:] { + acc += n + } + } + return json.RawMessage(strconv.FormatFloat(acc, 'f', -1, 64)) +} + +// decodedObject re-reads an object's fields as generic values so the whole thing +// can go through mergeMetricValue. +func decodedObject(obj map[string]json.RawMessage) any { + out := make(map[string]any, len(obj)) + for k, raw := range obj { + var v any + if decodeJSON(raw, &v) == nil { + out[k] = v + } + } + return out +} + +// decodeJSON keeps integers exact by decoding numbers as json.Number. +func decodeJSON(data []byte, v any) error { + dec := json.NewDecoder(bytes.NewReader(data)) + dec.UseNumber() + return dec.Decode(v) +} + +// marshal reports ok=false rather than an error: an unmarshalable merge result +// has only one recovery, replaying a backend's body verbatim. +func marshal(v any) ([]byte, bool) { + b, err := json.Marshal(v) + return b, err == nil +} diff --git a/metrics_test.go b/metrics_test.go new file mode 100644 index 0000000..9917119 --- /dev/null +++ b/metrics_test.go @@ -0,0 +1,285 @@ +package main + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "reflect" + "strconv" + "testing" +) + +const ( + numNodesMBean = "puppetlabs.puppetdb.population:name=num-nodes" + numNodesPath = metricsPrefix + "v2/read/" + numNodesMBean + // What pypuppetdb actually sends: quote() percent-encodes ':' and '='. + numNodesEscaped = metricsPrefix + "v2/read/puppetlabs.puppetdb.population%3Aname%3Dnum-nodes" +) + +// jolokiaRead wraps an MBean value in the envelope PuppetDB's Jolokia returns. +func jolokiaRead(mbean, value string, timestamp int) string { + return `{"request":{"mbean":"` + mbean + `","type":"read"},` + + `"value":` + value + `,"timestamp":` + strconv.Itoa(timestamp) + `,"status":200}` +} + +func metricValue(t *testing.T, body []byte) map[string]any { + t.Helper() + var env map[string]json.RawMessage + if err := json.Unmarshal(body, &env); err != nil { + t.Fatalf("unmarshal envelope %s: %v", body, err) + } + var val map[string]any + if err := json.Unmarshal(env["value"], &val); err != nil { + t.Fatalf("unmarshal value %s: %v", env["value"], err) + } + return val +} + +func TestMetrics_ReadSumsPopulationCounts(t *testing.T) { + // Puppetboard's landing page and radiator read num-nodes when + // DEFAULT_ENVIRONMENT is '*'; each backend only knows its own nodes. + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[numNodesPath] = jolokiaRead(numNodesMBean, `{"Value":90}`, 1000) + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[numNodesPath] = jolokiaRead(numNodesMBean, `{"Value":53}`, 2000) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := metaGet(t, srv.Handler(), numNodesEscaped) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + if got := metricValue(t, rec.Body.Bytes())["Value"]; got != float64(143) { + t.Errorf("Value = %v, want 143", got) + } +} + +func TestMetrics_EscapedMBeanNameSurvives(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[numNodesPath] = jolokiaRead(numNodesMBean, `{"Value":1}`, 1) + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[numNodesPath] = jolokiaRead(numNodesMBean, `{"Value":1}`, 1) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + metaGet(t, srv.Handler(), numNodesEscaped) + if !a.sawRawPath(numNodesEscaped) { + t.Errorf("backend saw %v, want the percent-encoded path %q", a.rawPaths, numNodesEscaped) + } +} + +func TestMetrics_EnvelopeKeepsNewestTimestamp(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[numNodesPath] = jolokiaRead(numNodesMBean, `{"Value":1}`, 1000) + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[numNodesPath] = jolokiaRead(numNodesMBean, `{"Value":1}`, 2000) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + var env map[string]any + if err := json.Unmarshal(metaGet(t, srv.Handler(), numNodesEscaped).Body.Bytes(), &env); err != nil { + t.Fatal(err) + } + if env["timestamp"] != float64(2000) { + t.Errorf("timestamp = %v, want 2000", env["timestamp"]) + } + if env["status"] != float64(200) { + t.Errorf("status = %v, want 200", env["status"]) + } +} + +func TestMetrics_PerAttributeRules(t *testing.T) { + const mbean = "puppetlabs.puppetdb.mq:name=global.processing-time" + path := metricsPrefix + "v2/read/" + mbean + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[path] = jolokiaRead(mbean, + `{"Count":10,"Min":2,"Max":9,"Mean":4,"StdDev":1,"50thPercentile":3,"MeanRate":1.5}`, 1) + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[path] = jolokiaRead(mbean, + `{"Count":6,"Min":1,"Max":20,"Mean":6,"StdDev":3,"50thPercentile":5,"MeanRate":0.5}`, 1) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + got := metricValue(t, metaGet(t, srv.Handler(), path).Body.Bytes()) + want := map[string]any{ + "Count": float64(16), // counts add + "Min": float64(1), // a bound stays a bound + "Max": float64(20), + "Mean": float64(5), // distribution stats average + "StdDev": float64(2), + "50thPercentile": float64(4), + "MeanRate": float64(2), // throughput adds + } + if !reflect.DeepEqual(got, want) { + t.Errorf("merged value = %v, want %v", got, want) + } +} + +func TestMetrics_ScalarReadUsesURLAttributeName(t *testing.T) { + // /metrics/v2/read// answers with a bare number, so the + // rule has to come from the URL rather than an object key. + const mbean = "puppetlabs.puppetdb.population:name=num-resources" + sumPath := metricsPrefix + "v2/read/" + mbean + "/Value" + maxPath := metricsPrefix + "v2/read/" + mbean + "/Max" + + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[sumPath] = jolokiaRead(mbean, `1000`, 1) + a.bodies[maxPath] = jolokiaRead(mbean, `1000`, 1) + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[sumPath] = jolokiaRead(mbean, `234`, 1) + b.bodies[maxPath] = jolokiaRead(mbean, `234`, 1) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + for _, c := range []struct { + path string + want float64 + }{{sumPath, 1234}, {maxPath, 1000}} { + var env map[string]any + if err := json.Unmarshal(metaGet(t, srv.Handler(), c.path).Body.Bytes(), &env); err != nil { + t.Fatal(err) + } + if env["value"] != c.want { + t.Errorf("%s value = %v, want %v", c.path, env["value"], c.want) + } + } +} + +func TestMetrics_ListUnionsDomains(t *testing.T) { + // Puppetboard's /metrics page calls metric() with no name, which is a + // Jolokia list; a backend-local MBean must not vanish from the browse tree. + const path = metricsPrefix + "v2/list" + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[path] = `{"value":{"java.lang":{"type=Memory":{"attr":{"HeapMemoryUsage":{"rw":false}}}}},"status":200,"timestamp":1}` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[path] = `{"value":{"puppetlabs.puppetdb.population":{"name=num-nodes":{"attr":{"Value":{"rw":false}}}}},"status":200,"timestamp":1}` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + got := metricValue(t, metaGet(t, srv.Handler(), path).Body.Bytes()) + if _, ok := got["java.lang"]; !ok { + t.Errorf("java.lang missing from merged list: %v", got) + } + if _, ok := got["puppetlabs.puppetdb.population"]; !ok { + t.Errorf("puppetlabs.puppetdb.population missing from merged list: %v", got) + } +} + +func TestMetrics_V1BareObjectMerged(t *testing.T) { + // metrics/v1/mbeans has no Jolokia envelope; the whole body is the value. + const path = metricsPrefix + "v1/mbeans/" + numNodesMBean + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[path] = `{"Value":90}` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[path] = `{"Value":53}` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + var got map[string]any + if err := json.Unmarshal(metaGet(t, srv.Handler(), path).Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + if got["Value"] != float64(143) { + t.Errorf("Value = %v, want 143", got["Value"]) + } +} + +func TestMetrics_MissingMBeanKeepsUpstreamError(t *testing.T) { + // Jolokia reports a bad MBean as a 200 with an error envelope, which + // pypuppetdb turns into DoesNotComputeError; the client must still see it. + const path = metricsPrefix + "v2/read/nope:name=nothing" + errEnv := `{"request":{"mbean":"nope:name=nothing"},"error_type":"javax.management.InstanceNotFoundException","error":"nope:name=nothing is not registered","status":404}` + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[path] = errEnv + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[path] = errEnv + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := metaGet(t, srv.Handler(), path) + var env map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &env); err != nil { + t.Fatal(err) + } + if env["error"] == nil { + t.Errorf("expected the upstream Jolokia error to be replayed, got %s", rec.Body.String()) + } +} + +func TestMetrics_ErroringBackendIgnoredWhenAnotherAnswers(t *testing.T) { + const path = metricsPrefix + "v2/read/" + numNodesMBean + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[path] = `{"request":{},"error":"boom","status":500}` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[path] = jolokiaRead(numNodesMBean, `{"Value":53}`, 1) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + if got := metricValue(t, metaGet(t, srv.Handler(), path).Body.Bytes())["Value"]; got != float64(53) { + t.Errorf("Value = %v, want 53 from the backend that answered", got) + } +} + +func TestMetrics_OneBackendDown(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.fail = true + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[numNodesPath] = jolokiaRead(numNodesMBean, `{"Value":53}`, 1) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := metaGet(t, srv.Handler(), numNodesEscaped) + if rec.Code != http.StatusOK { + t.Fatalf("expected 200 serving the survivor, got %d", rec.Code) + } + if got := metricValue(t, rec.Body.Bytes())["Value"]; got != float64(53) { + t.Errorf("Value = %v, want 53", got) + } +} + +func TestMetrics_AllBackendsDown(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.fail = true + b := newFakeBackend(t, `[]`, `[]`) + b.fail = true + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + if rec := metaGet(t, srv.Handler(), numNodesEscaped); rec.Code != http.StatusBadGateway { + t.Errorf("status %d, want 502", rec.Code) + } +} + +func TestMetrics_RejectsNonGET(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + b := newFakeBackend(t, `[]`, `[]`) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := httptest.NewRecorder() + srv.Handler().ServeHTTP(rec, httptest.NewRequest(http.MethodPost, numNodesEscaped, nil)) + if rec.Code != http.StatusMethodNotAllowed { + t.Errorf("status %d, want 405", rec.Code) + } +} + +func TestMergeMetricValue_NonNumericKeepsFirst(t *testing.T) { + a := map[string]any{"Name": "pdb-a", "Enabled": true} + b := map[string]any{"Name": "pdb-b", "Enabled": false} + got, ok := mergeMetricValue([]any{a, b}, "").(map[string]any) + if !ok { + t.Fatalf("expected an object, got %T", got) + } + if got["Name"] != "pdb-a" || got["Enabled"] != true { + t.Errorf("merged = %v, want the first backend's strings and booleans", got) + } +} + +func TestMergeRuleFor(t *testing.T) { + cases := map[string]mergeRule{ + "Count": ruleSum, + "Value": ruleSum, + "MeanRate": ruleSum, + "queue-depth": ruleSum, + "min": ruleMin, + "Max": ruleMax, + "Uptime": ruleMax, + "StartTime": ruleMax, + "Mean": ruleMean, + "StdDev": ruleMean, + "99thPercentile": ruleMean, + } + for attr, want := range cases { + if got := mergeRuleFor(attr); got != want { + t.Errorf("mergeRuleFor(%q) = %v, want %v", attr, got, want) + } + } +} diff --git a/server.go b/server.go index e3ab0b2..8231b14 100644 --- a/server.go +++ b/server.go @@ -17,6 +17,7 @@ import ( const ( factsPath = "/pdb/query/v4/facts" nodesPath = "/pdb/query/v4/nodes" + resourcesPath = "/pdb/query/v4/resources" reportsPath = "/pdb/query/v4/reports" eventsPath = "/pdb/query/v4/events" eventCountsPath = "/pdb/query/v4/event-counts" @@ -57,6 +58,9 @@ func (s *Server) Handler() http.Handler { mux := http.NewServeMux() mux.HandleFunc("/healthz", s.handleHealth) mux.HandleFunc("/pdb/query/v4/", s.handleQuery) + mux.HandleFunc(metaVersionPath, s.handleMetaVersion) + mux.HandleFunc(metaServerTimePath, s.handleMetaServerTime) + mux.HandleFunc(metricsPrefix, s.handleMetrics) return mux } @@ -67,7 +71,9 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) { } switch r.URL.Path { case nodesPath: - s.serveMerged(w, r, nodesPath, s.mergeNodesResponse(r)) + s.serveNodes(w, r) + case resourcesPath: + s.serveResources(w, r) case factsPath: s.serveMerged(w, r, factsPath, s.mergeFactsResponse(r)) case reportsPath: @@ -134,6 +140,24 @@ func (s *Server) serveUnion(w http.ResponseWriter, r *http.Request, path string, writeJSON(w, page.apply(merged)) } +// 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 { + s.serveSummed(w, r, nodesPath, spec.columns) + return + } + s.serveMerged(w, r, nodesPath, s.mergeNodesResponse(r)) +} + +// Only aggregates merge: a resource record has no cross-backend identity to dedupe on, so a plain query stays on the pass-through path. +func (s *Server) serveResources(w http.ResponseWriter, r *http.Request) { + if spec := parseAggregate(r.URL.Query().Get("query")); spec != nil { + s.serveSummed(w, r, resourcesPath, spec.columns) + return + } + s.proxyUnmerged(w, r) +} + // An `extract` query with a `function` column returns synthetic aggregate rows that carry no identity, so they are summed rather than unioned. func (s *Server) serveReports(w http.ResponseWriter, r *http.Request) { if spec := parseAggregate(r.URL.Query().Get("query")); spec != nil { diff --git a/server_test.go b/server_test.go index 253ef0b..a426837 100644 --- a/server_test.go +++ b/server_test.go @@ -34,6 +34,9 @@ type fakeBackend struct { mu sync.Mutex gotParams map[string]url.Values + // rawPaths records the still-escaped request paths, so tests can assert an + // MBean name's percent-encoding survived the proxy. + rawPaths []string } func newFakeBackend(t *testing.T, nodesBody, factsBody string) *fakeBackend { @@ -51,6 +54,7 @@ func newFakeBackend(t *testing.T, nodesBody, factsBody string) *fakeBackend { } fb.mu.Lock() fb.gotParams[r.URL.Path] = r.URL.Query() + fb.rawPaths = append(fb.rawPaths, r.URL.EscapedPath()) fb.mu.Unlock() if fb.fail { http.Error(w, "boom", http.StatusInternalServerError) @@ -91,6 +95,14 @@ func (fb *fakeBackend) params(path string) (url.Values, bool) { return v, ok } +// sawRawPath reports whether the backend was asked for a path with exactly that +// escaping. +func (fb *fakeBackend) sawRawPath(p string) bool { + fb.mu.Lock() + defer fb.mu.Unlock() + return slices.Contains(fb.rawPaths, p) +} + // gotQuery returns the PuppetDB query param the backend saw for a path. func (fb *fakeBackend) gotQuery(path string) string { v, _ := fb.params(path) @@ -860,3 +872,120 @@ func TestHandler_EventCountsBadPagingParam(t *testing.T) { t.Errorf("expected 400 for a malformed limit, got %d", rec.Code) } } + +// What Puppetboard's landing page sends when DEFAULT_ENVIRONMENT names a real +// environment: an extract/count with no group_by, so every backend returns one +// anonymous row. +const nodeCountQuery = `["extract",[["function","count"]],["and",["=","catalog_environment","production"]]]` + +func TestHandler_NodesAggregateSummed(t *testing.T) { + // A count row has no certname, so the certname-keyed merge would have + // collapsed both backends' counts into one backend's number. + a := newFakeBackend(t, `[{"count":90}]`, `[]`) + b := newFakeBackend(t, `[{"count":53}]`, `[]`) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), nodesPath, nodeCountQuery) + 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{143}) { + t.Errorf("count = %v, want [143]", got) + } +} + +func TestHandler_NodesAggregateGroupedSummed(t *testing.T) { + a := newFakeBackend(t, `[{"count":4,"catalog_environment":"production"},{"count":1,"catalog_environment":"dev"}]`, `[]`) + b := newFakeBackend(t, `[{"count":3,"catalog_environment":"production"}]`, `[]`) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), nodesPath, + `["extract",[["function","count"],"catalog_environment"],["~","certname",".*"],["group_by","catalog_environment"]]`) + var got []map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + byEnv := map[string]float64{} + for _, row := range got { + e, _ := row["catalog_environment"].(string) + n, _ := row["count"].(float64) + byEnv[e] = n + } + want := map[string]float64{"production": 7, "dev": 1} + if !reflect.DeepEqual(byEnv, want) { + t.Errorf("counts = %v, want %v", byEnv, want) + } +} + +func TestHandler_NodesNonAggregateStillMergedByCertname(t *testing.T) { + // Regression: routing aggregates to the summing path must not divert plain + // queries, including an extract projection that carries no function column. + a := newFakeBackend(t, + `[`+node("h1", "2026-07-01T00:00:00Z")+`,`+node("h2", "2026-07-10T00:00:00Z")+`]`, `[]`) + b := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + for _, q := range []string{ + `["=","certname","h1"]`, + `["extract",["certname","report_timestamp"],["~","certname",".*"]]`, + } { + rec := doGet(t, srv.Handler(), nodesPath, q) + var got []recordMeta + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatalf("query %s: %v", q, err) + } + if len(got) != 2 { + t.Fatalf("query %s: expected 2 deduped nodes, got %d: %s", q, len(got), rec.Body.String()) + } + for _, m := range got { + if m.Certname == "h1" && m.ReportTimestamp != "2026-07-20T00:00:00Z" { + t.Errorf("query %s: h1 should be the newer record, got %s", q, m.ReportTimestamp) + } + } + } +} + +func TestHandler_ResourcesAggregateSummed(t *testing.T) { + // /resources is otherwise an unmerged pass-through, so before this the + // landing page's resource total was whichever backend answered first. + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[resourcesPath] = `[{"count":1000}]` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[resourcesPath] = `[{"count":234}]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), resourcesPath, + `["extract",[["function","count"]],["=","environment","production"]]`) + 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{1234}) { + t.Errorf("count = %v, want [1234]", got) + } +} + +func TestHandler_ResourcesNonAggregateStillPassesThrough(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[resourcesPath] = `[{"certname":"h1","type":"File"}]` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[resourcesPath] = `[{"certname":"h2","type":"File"}]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), resourcesPath, `["=","type","File"]`) + if got := rec.Body.String(); !strings.Contains(got, `"h1"`) || strings.Contains(got, `"h2"`) { + t.Errorf("body = %s, want the first backend's response verbatim", got) + } +} + +func TestHandler_ResourcesAggregateAsksEveryBackend(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + a.bodies[resourcesPath] = `[{"count":1}]` + b := newFakeBackend(t, `[]`, `[]`) + b.bodies[resourcesPath] = `[{"count":1}]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + doGet(t, srv.Handler(), resourcesPath, `["extract",[["function","count"]],["=","environment","production"]]`) + if _, ok := b.params(resourcesPath); !ok { + t.Error("second backend was never asked for the resource count") + } +} diff --git a/source_test.go b/source_test.go index 6eb7e7b..89aca22 100644 --- a/source_test.go +++ b/source_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "net/http" "slices" + "strings" "testing" ) @@ -246,6 +247,38 @@ func TestHandler_SourceNotInjectedOnAggregate(t *testing.T) { } } +// /nodes aggregates are summed rather than merged, so nothing may stamp them. +func TestHandler_NodesAggregateNotStamped(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + b := newFakeBackend(t, `[]`, `[]`) + a.bodies[nodesPath] = `[{"count":3}]` + b.bodies[nodesPath] = `[{"count":2}]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), nodesPath, `["extract",[["function","count"]]]`) + if strings.Contains(rec.Body.String(), defaultSourceFact) { + t.Errorf("aggregate rows were stamped: %s", rec.Body.String()) + } + if got := rec.Body.String(); !strings.Contains(got, `"count":5`) { + t.Errorf("count = %s, want the summed 5", got) + } +} + +// A plain extract projects columns and skips the aggregate path, so the stamp +// must not add a key the client did not ask for. +func TestHandler_NodesProjectionNotStamped(t *testing.T) { + a := newFakeBackend(t, `[]`, `[]`) + b := newFakeBackend(t, `[]`, `[]`) + a.bodies[nodesPath] = `[{"certname":"h1"}]` + b.bodies[nodesPath] = `[]` + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), nodesPath, `["extract",["certname"]]`) + if strings.Contains(rec.Body.String(), defaultSourceFact) { + t.Errorf("projection gained a stamp: %s", rec.Body.String()) + } +} + // A query naming a specific fact asked for that fact only. func TestHandler_SourceNotInjectedWhenNameFiltered(t *testing.T) { a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`,