Serve PuppetDB meta and metrics endpoints, sum node and resource counts
ci/woodpecker/pr/build Pipeline was successful
ci/woodpecker/pr/test Pipeline was successful
ci/woodpecker/pr/pre-commit Pipeline was successful

## Why
Puppetboard 7.0.1 cannot run against pdbmux: it exits at import when
/pdb/meta/v1/version 404s, and its landing page, metrics and radiator views
404 on the Jolokia surface.

## How
- Serve /pdb/meta/v1/version, reporting the lowest version any backend runs,
  and /pdb/meta/v1/server-time from the first reachable backend.
- Merge the Jolokia surface (/metrics/v2/read, /metrics/v2/list,
  /metrics/v1/mbeans): objects union, numeric attributes sum by default, and
  Min/Max/Uptime/StartTime plus the distribution stats take a bound or a mean.
- Route /nodes extract-count queries to the summing path ahead of the certname
  merge, and give /resources aggregates the same path.
- Document the endpoints and merge semantics in the README.
- Cover version disagreement, metric rules, escaped MBean names, count summing
  and the non-aggregate /nodes merge with httptest backends.
This commit is contained in:
2026-09-05 20:41:35 +10:00
parent 1ee7a2c07c
commit b6d59af7ef
7 changed files with 1130 additions and 8 deletions
+54 -7
View File
@@ -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`. |
| `GET /pdb/query/v4/nodes` | Fan out to all backends, dedupe by `certname`, keep the record with the newer `report_timestamp`. 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). |
| `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/<hash>/{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/<mbean>` | 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[/<mbean>]` | 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
@@ -77,6 +91,39 @@ unknown fields survive untouched.
- `include_total=true` on a summed endpoint reports the **merged** row count,
not the sum of the backends' `X-Records`, since shared keys collapse.
### 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
+195
View File
@@ -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)
}
+170
View File
@@ -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)
}
}
+272
View File
@@ -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/<mbean>,
// lists via /metrics/v2/list, and falls back to /metrics/v1/mbeans/<mbean>.
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/<mbean>/<attribute>), 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/<mbean>[/<attribute>]
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
}
+285
View File
@@ -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/<mbean>/<attribute> 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)
}
}
}
+25 -1
View File
@@ -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)
s.serveNodes(w, r)
case resourcesPath:
s.serveResources(w, r)
case factsPath:
s.serveMerged(w, r, factsPath, s.mergeFactsResponse)
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)
}
// 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 {
+129
View File
@@ -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)
@@ -858,3 +870,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")
}
}