Replay every unanimous upstream status, not just 4xx #27

Merged
benvin merged 3 commits from benvin/replay-upstream-errors into main 2026-09-13 14:31:56 +10:00
13 changed files with 710 additions and 134 deletions
+28 -12
View File
@@ -34,7 +34,7 @@ not PQL) is forwarded verbatim.
| `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/{nodes,factsets,catalogs}/<certname>[/...]` | Ask the backend that owns that `certname` — the same owner the `/facts` merge attributes records to — and serve its reply verbatim. The remaining backends are tried after it, so a node only one backend holds is still served, and openvoxdb's own `404` body is replayed when none holds it. |
| `GET /pdb/query/v4/{nodes,factsets,catalogs}/<certname>[/...]` | Ask the backend that owns that `certname` — the same owner the `/facts` merge attributes records to — and serve its reply verbatim, success or error. The remaining backends are tried after it, so a node only one backend holds is still served, and openvoxdb's own `404` body is replayed when none holds it. |
| `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. |
@@ -47,17 +47,32 @@ Fan-out is concurrent, and goes only to the backends the health prober currently
believes are up — see [Backend health](#backend-health). If one backend errors or
times out, `pdbmux` serves the surviving backends' results and logs a warning; a
merged endpoint only returns `502` when **every** backend fails. Response records
are passed through as raw JSON so unknown fields survive untouched.
are passed through as raw JSON so unknown fields survive untouched, under
openvoxdb's own `application/json;charset=utf-8`.
Every backend is asked the same question, so a query all of them *refuse* with
the same client-shaped status — a `400` naming an unknown field, say — is the
query's fault rather than an outage: that status and openvoxdb's own explanation
are replayed to the client instead of a `502`, with any backend address stripped
out of the body first. Backends disagreeing on the status, a `403` (`pdbmux`'s
own credentials, not the client's), a `404` (which records a backend holds is
exactly what backends disagree about), `408`, `429` and every `5xx` still return
`502`. A refused query is not counted as a partial round on `/healthz`, and
nothing about it is cached.
Every backend is asked the same question, so a status **every** backend answered
with is the estate's own answer, not an outage: that status, openvoxdb's own
explanation and its content type are replayed to the client instead of a `502`,
with any backend address stripped out of the body first. This holds for every
status from `400` up — openvoxdb answers `["=","name"]` with `400` on `/nodes`
but `500` on `/facts`, and `/metrics/v2` with a flat `403`, so a rule drawn at
`500` would replay one and swallow the other. `502` is kept for what it actually
describes: backends **disagreeing** on the status, or a backend that answered
nothing at all. The rule is the same on `/pdb/query`, `/pdb/meta` and `/metrics`,
so no route answers a failure differently from any other.
Unanimity is asked for only where the backends are peers answering the same
question. The per-certname routes are not: one backend **owns** the node and the
others are fallbacks, so the owner's reply is the answer whatever they said — a
backend that does not hold the certname answers `404` to say the node is not its
own, not to disagree about the request. When the owner answers nothing at all the
first fallback that did answer stands in, so an unreachable owner does not take
the node with it, and `502` is left for a request no backend answered.
A unanimous `4xx` blames the request, so it is not counted as a partial round on
`/healthz` and nothing about it is cached. A unanimous `5xx` is the backends
reporting their own fault, so it is replayed just as faithfully but still counts
as a failed round and still falls back to a stale cache entry where there is one.
Responses carry PuppetDB's `X-Records` when the query asked for a total, and on
the merged paths `X-Backends` (see [Backend health](#backend-health)). Cached
@@ -448,7 +463,8 @@ backend later without further handler changes.
has passed `pdbmux` always re-queries the backends; the expired copy is served
**only** if every backend fails, which turns a `502` into slightly-old data. A
healthy backend is never shadowed by a stale entry, and a query every backend
refuses is answered with the refusal rather than the stale copy.
*refuses* with a `4xx` is answered with the refusal rather than the stale copy
a unanimous `5xx` is an outage like any other and still takes the stale copy.
- **Bounded** — `facts_cache_bytes` (default 64 MiB) is a byte budget, evicted
least-recently-used; reads count as use, so a stale entry that is still being
asked for survives. A single response larger than the whole budget is not
+13 -3
View File
@@ -61,6 +61,7 @@ type countingBackend struct {
hits map[string]int
bodies map[string]string
fail bool
dead bool
block chan struct{}
}
@@ -70,12 +71,15 @@ func newCountingBackend(t *testing.T, bodies map[string]string) *countingBackend
cb.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
cb.mu.Lock()
cb.hits[r.URL.Path]++
fail, block, body := cb.fail, cb.block, cb.bodies[r.URL.Path]
fail, dead, block, body := cb.fail, cb.dead, cb.block, cb.bodies[r.URL.Path]
cb.mu.Unlock()
if block != nil {
<-block
}
if dead {
panic(http.ErrAbortHandler)
}
if fail {
http.Error(w, "boom", http.StatusInternalServerError)
return
@@ -108,6 +112,12 @@ func (cb *countingBackend) totalHits() int {
return n
}
func (cb *countingBackend) setDead(v bool) {
cb.mu.Lock()
defer cb.mu.Unlock()
cb.dead = v
}
func (cb *countingBackend) setFail(v bool) {
cb.mu.Lock()
defer cb.mu.Unlock()
@@ -392,8 +402,8 @@ func TestHandler_StaleSourceFactDrilldownStaysFilteredByOwner(t *testing.T) {
func TestHandler_NoCacheEntryMeansBackendFailureIs502(t *testing.T) {
a := newCountingBackend(t, map[string]string{factsPath: `[]`})
b := newCountingBackend(t, map[string]string{factsPath: `[]`})
a.setFail(true)
b.setFail(true)
a.setDead(true)
b.setDead(true)
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
rec := doGet(t, srv.Handler(), factsPath, "")
+5 -1
View File
@@ -50,8 +50,12 @@ func isCertnameRoute(path string) bool { return certnameFor(path) != "" }
// which is the record array the merges are built on. The remaining backends are
// tried after it, so a node only one backend holds is still served and a
// certname no backend holds still answers with openvoxdb's own 404 body.
//
// askOwnerFirst is what makes the owner's reply the answer even when it is an
// error. A backend that does not hold the certname 404s to say so, so it has no
// opinion to weigh against the owner's.
func (s *Server) serveByOwner(w http.ResponseWriter, r *http.Request) {
s.proxyOrdered(w, r, s.ownerFirst(r.Context(), certnameFor(r.URL.Path)))
s.proxyOrdered(w, r, s.ownerFirst(r.Context(), certnameFor(r.URL.Path)), askOwnerFirst)
}
// ownerFirst puts the backend holding a certname's newest report ahead of the
+91
View File
@@ -55,6 +55,13 @@ type voxBackend struct {
// nodes maps a certname it holds to that node's report_timestamp.
nodes map[string]string
fail bool
// perNodeStatus and perNodeBody answer the per-certname routes with a fixed
// reply, leaving /nodes alone so the freshness map still resolves an owner.
perNodeStatus int
perNodeBody string
// perNodeDead aborts the per-certname routes without a status, standing in
// for a backend that is unreachable only for that request.
perNodeDead bool
mu sync.Mutex
gotParams map[string]url.Values
@@ -73,6 +80,17 @@ func newVoxBackend(t *testing.T, name string, nodes map[string]string) *voxBacke
}
w.Header().Set("Content-Type", "application/json;charset=utf-8")
if isCertnameRoute(r.URL.Path) {
if vb.perNodeDead {
panic(http.ErrAbortHandler)
}
if vb.perNodeStatus != 0 {
w.WriteHeader(vb.perNodeStatus)
_, _ = io.WriteString(w, vb.perNodeBody)
return
}
}
if certname, ok := strings.CutPrefix(r.URL.Path, factsetsPath+"/"); ok {
ts, held := vb.nodes[certname]
if !held {
@@ -342,3 +360,76 @@ func TestCertnameRoutes_ClaimedByTheRouteTable(t *testing.T) {
}
}
}
// The owner holds the node, so its answer is the answer: a non-owner's 404 only
// says the node is not its own, and demanding it agree turns the owner's real
// error into a 502 that describes neither backend.
func TestCertnameRoutes_OwnerAnswerWins(t *testing.T) {
const certname = "h1.example.net"
for _, tc := range []struct {
name string
// setUp configures the owner (a, which reported later) and the backend
// that does not hold the node (b).
setUp func(a, b *voxBackend)
wantCode int
wantBody string
}{
{
name: "owner errors while the non-owner 404s",
setUp: func(a, b *voxBackend) { a.perNodeStatus, a.perNodeBody = http.StatusInternalServerError, "boom" },
wantCode: http.StatusInternalServerError,
wantBody: "boom",
},
{
name: "owner and non-owner error differently",
setUp: func(a, b *voxBackend) {
a.perNodeStatus, a.perNodeBody = http.StatusInternalServerError, "boom"
b.perNodeStatus, b.perNodeBody = http.StatusServiceUnavailable, "busy"
},
wantCode: http.StatusInternalServerError,
wantBody: "boom",
},
{
// The fallback chain is there so an unreachable owner does not take the
// node with it; a backend that did answer explains more than a 502.
name: "owner unreachable, fallback answers",
setUp: func(a, b *voxBackend) {
a.perNodeDead = true
b.perNodeStatus, b.perNodeBody = http.StatusServiceUnavailable, "busy"
},
wantCode: http.StatusServiceUnavailable,
wantBody: "busy",
},
{
name: "owner succeeds while the non-owner errors",
setUp: func(a, b *voxBackend) {
b.perNodeStatus, b.perNodeBody = http.StatusInternalServerError, "boom"
},
wantCode: http.StatusOK,
wantBody: `"a"`,
},
{
// Nothing answered at all, so there is no upstream reply to replay.
name: "every backend unreachable",
setUp: func(a, b *voxBackend) { a.perNodeDead, b.perNodeDead = true, true },
wantCode: http.StatusBadGateway,
},
} {
t.Run(tc.name, func(t *testing.T) {
a := newVoxBackend(t, "a", map[string]string{certname: "2026-07-20T00:00:00Z"})
b := newVoxBackend(t, "b", map[string]string{})
tc.setUp(a, b)
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness))
for _, rt := range perNodePaths {
rec := doGet(t, srv.Handler(), rt.path(certname), "")
if rec.Code != tc.wantCode {
t.Fatalf("%s: status = %d, want %d: %s", rt.name, rec.Code, tc.wantCode, rec.Body.String())
}
if tc.wantBody != "" && !strings.Contains(rec.Body.String(), tc.wantBody) {
t.Errorf("%s: body = %q, want it to contain %q", rt.name, rec.Body.String(), tc.wantBody)
}
}
})
}
}
+67 -1
View File
@@ -4,6 +4,7 @@ package main
import (
"context"
"net/url"
"strings"
"testing"
)
@@ -22,7 +23,7 @@ func TestRejectedQuerySurfacesUpstreamStatus(t *testing.T) {
t.Run(tc.name, func(t *testing.T) {
params := query(tc.q)
status, upstream := h.a.queryRaw(ctx, t, tc.path, params)
if !clientShaped(status) {
if !replayableStatus(status) || status >= 500 {
t.Fatalf("backend %s answered HTTP %d for %s, which is not the client error this test needs: %s",
h.a.name, status, tc.q, upstream)
}
@@ -56,3 +57,68 @@ func TestRejectedQuerySurfacesUpstreamStatus(t *testing.T) {
})
}
}
// openvoxdb answers the same malformed clause with 400 on /nodes and 500 on
// /facts — engine.clj's rewrite-fact-query runs an unguarded nth only for the
// facts entity. Both are the backends' real answer, so both have to reach the
// client rather than one of them becoming a 502.
func TestUnanimousServerErrorSurfacesUpstreamStatus(t *testing.T) {
ctx := context.Background()
params := query(`["=","name"]`)
status, upstream := h.a.queryRaw(ctx, t, factsPath, params)
if status < 500 {
t.Skipf("backend %s answered HTTP %d for the arity bug this test needs", h.a.name, status)
}
if other, _ := h.b.queryRaw(ctx, t, factsPath, params); other != status {
t.Fatalf("backends answered %d and %d, so the failure is not unanimous", status, other)
}
resp := rawGet(t, factsPath, params)
if resp.status != status {
t.Fatalf("pdbmux answered HTTP %d, want the upstream %d: %s", resp.status, status, resp.body)
}
if got, want := strings.TrimSpace(string(resp.body)), strings.TrimSpace(string(upstream)); want != "" && got != want {
t.Errorf("pdbmux body = %q, want openvoxdb's own explanation %q", got, want)
}
for _, b := range []*backend{h.a, h.b} {
if strings.Contains(string(resp.body), b.url) {
t.Errorf("replayed body names backend %s: %q", b.name, resp.body)
}
}
}
// /metrics and /pdb/meta had their own copy of the gateway error, so a status
// both backends agreed on never reached the client on those routes.
func TestUnanimousMetricsStatusIsReplayed(t *testing.T) {
ctx := context.Background()
for _, path := range []string{"/metrics/v1/mbeans", "/metrics/v2/list"} {
t.Run(path, func(t *testing.T) {
status, _ := h.a.queryRaw(ctx, t, path, nil)
if status < 400 {
t.Skipf("%s answers HTTP %d on this estate, so there is nothing to replay", path, status)
}
if other, _ := h.b.queryRaw(ctx, t, path, nil); other != status {
t.Fatalf("backends answered %d and %d, so the status is not unanimous", status, other)
}
if resp := rawGet(t, path, nil); resp.status != status {
t.Fatalf("pdbmux answered HTTP %d, want the upstream %d: %s", resp.status, status, resp.body)
}
})
}
}
// A client cannot tell pdbmux from a PuppetDB by the content type either.
func TestSuccessContentTypeMatchesUpstream(t *testing.T) {
ctx := context.Background()
params := url.Values{"limit": {"1"}}
_, _ = h.a.queryRaw(ctx, t, nodesPath, params)
resp := rawGet(t, nodesPath, params)
if resp.status != 200 {
t.Fatalf("status %d: %s", resp.status, resp.body)
}
if got := resp.header.Get("Content-Type"); got != "application/json;charset=utf-8" {
t.Errorf("Content-Type = %q, want openvoxdb's own %q", got, "application/json;charset=utf-8")
}
}
+18 -3
View File
@@ -72,8 +72,10 @@ func (s *Server) rawBackend(ctx context.Context, b Backend, path, rawQuery strin
}
}
// aliveRaw drops backends that errored or answered non-2xx, writing a 502 and
// returning ok=false only when none is left.
// aliveRaw drops backends that errored or answered non-2xx, returning ok=false
// when none is left. The reply it writes then is the same one the merged query
// handlers write, so /pdb/meta and /metrics answer a failure exactly as
// /pdb/query does: a unanimous upstream error replayed, anything else a 502.
func (s *Server) aliveRaw(w http.ResponseWriter, results []rawResult, path string) ([]rawResult, bool) {
var alive []rawResult
for _, res := range results {
@@ -84,12 +86,25 @@ func (s *Server) aliveRaw(w http.ResponseWriter, results []rawResult, path strin
alive = append(alive, res)
}
if len(alive) == 0 {
http.Error(w, "all backends failed", http.StatusBadGateway)
s.writeUpstreamError(w, peerOutcome(rawUpstreamErrors(results)))
return nil, false
}
return alive, true
}
// rawUpstreamErrors reduces a raw fan-out to one entry per backend, nil where
// the backend answered 2xx or never answered at all.
func rawUpstreamErrors(results []rawResult) []*upstreamError {
errs := make([]*upstreamError, len(results))
for i, res := range results {
if res.err != nil || res.ok() {
continue
}
errs[i] = newUpstreamError(res.status, res.contentType, res.body)
}
return errs
}
func (r rawResult) reason() string {
if r.err != nil {
return r.err.Error()
+57 -2
View File
@@ -4,6 +4,7 @@ import (
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
)
@@ -85,9 +86,9 @@ func TestMetaVersion_OneBackendDown(t *testing.T) {
func TestMetaVersion_AllBackendsDown(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
a.fail = true
a.dead = true
b := newFakeBackend(t, `[]`, `[]`)
b.fail = true
b.dead = true
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
if rec := metaGet(t, srv.Handler(), metaVersionPath); rec.Code != http.StatusBadGateway {
@@ -95,6 +96,60 @@ func TestMetaVersion_AllBackendsDown(t *testing.T) {
}
}
// /pdb/meta had its own copy of the gateway error, so a status both backends
// agreed on never reached the client. It answers failures the way the query
// routes do now.
func TestMeta_ReplaysUnanimousUpstreamStatus(t *testing.T) {
for _, path := range []string{metaVersionPath, metaServerTimePath} {
for _, status := range []int{http.StatusNotFound, http.StatusForbidden, http.StatusInternalServerError} {
t.Run(path+"/"+http.StatusText(status), func(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = status, "upstream said no"
b.reject, b.rejectBody = status, "upstream said no"
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := metaGet(t, srv.Handler(), path)
if rec.Code != status {
t.Fatalf("status %d, want the upstream %d: %s", rec.Code, status, rec.Body.String())
}
if !strings.Contains(rec.Body.String(), "upstream said no") {
t.Errorf("body = %q, want the upstream explanation", rec.Body.String())
}
})
}
}
}
func TestMeta_DisagreeingStatusesStay502(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusNotFound, "gone"
b.reject, b.rejectBody = http.StatusInternalServerError, "boom"
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 when backends disagree", rec.Code)
}
}
// A replayed body must not name a backend, on /pdb/meta as anywhere else.
func TestMeta_ReplayedBodyIsRedacted(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusInternalServerError, "upstream "+a.srv.URL+" blew up"
b.reject, b.rejectBody = http.StatusInternalServerError, "upstream "+b.srv.URL+" blew up"
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := metaGet(t, srv.Handler(), metaVersionPath)
if rec.Code != http.StatusInternalServerError {
t.Fatalf("status %d, want the upstream 500", rec.Code)
}
if strings.Contains(rec.Body.String(), hostOf(t, a.srv.URL)) {
t.Errorf("replayed body names a backend: %q", rec.Body.String())
}
}
func TestMetaServerTime_FirstReachableBackend(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
a.bodies[metaServerTimePath] = `{"server_time":"2026-08-29T01:00:00.000Z"}`
+7 -1
View File
@@ -65,7 +65,13 @@ func (s *Server) handleMetrics(w http.ResponseWriter, r *http.Request) {
writeRaw(w, alive[0])
return
}
w.Header().Set("Content-Type", "application/json")
// The merged body is the backends' own payload reshaped, so it keeps their
// content type rather than announcing a different one.
contentType := alive[0].contentType
if contentType == "" {
contentType = jsonContentType
}
w.Header().Set("Content-Type", contentType)
_, _ = w.Write(body)
}
+54 -2
View File
@@ -6,6 +6,7 @@ import (
"net/http/httptest"
"reflect"
"strconv"
"strings"
"testing"
)
@@ -229,9 +230,9 @@ func TestMetrics_OneBackendDown(t *testing.T) {
func TestMetrics_AllBackendsDown(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
a.fail = true
a.dead = true
b := newFakeBackend(t, `[]`, `[]`)
b.fail = true
b.dead = true
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
if rec := metaGet(t, srv.Handler(), numNodesEscaped); rec.Code != http.StatusBadGateway {
@@ -239,6 +240,57 @@ func TestMetrics_AllBackendsDown(t *testing.T) {
}
}
// /metrics/v1/mbeans is a unanimous 404 on this estate and /metrics/v2 a
// unanimous 403; both used to come back as 502.
func TestMetrics_ReplaysUnanimousUpstreamStatus(t *testing.T) {
for _, status := range []int{http.StatusNotFound, http.StatusForbidden, http.StatusInternalServerError} {
t.Run(http.StatusText(status), func(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = status, "Forbidden request: /metrics/v2/list (method :get)."
b.reject, b.rejectBody = status, "Forbidden request: /metrics/v2/list (method :get)."
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := metaGet(t, srv.Handler(), numNodesEscaped)
if rec.Code != status {
t.Fatalf("status %d, want the upstream %d: %s", rec.Code, status, rec.Body.String())
}
if !strings.Contains(rec.Body.String(), "Forbidden request") {
t.Errorf("body = %q, want the upstream explanation", rec.Body.String())
}
})
}
}
func TestMetrics_DisagreeingStatusesStay502(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusNotFound, "gone"
b.reject, b.rejectBody = http.StatusForbidden, "denied"
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 when backends disagree", rec.Code)
}
}
// One backend refusing is not the estate's answer, so the other still serves.
func TestMetrics_OneRefusalStillServesTheOther(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusForbidden, "denied"
b.bodies[numNodesPath] = `{"request":{},"value":{"Value":7},"status":200}`
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, want the surviving backend's 200: %s", rec.Code, rec.Body.String())
}
if got := metricValue(t, rec.Body.Bytes())["Value"]; got != float64(7) {
t.Errorf("Value = %v, want 7", got)
}
}
func TestMetrics_RejectsNonGET(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
+37 -41
View File
@@ -50,6 +50,12 @@ type backendResult struct {
var errAllBackendsFailed = errors.New("all backends failed")
// jsonContentType is the spelling openvoxdb puts on the wire for a query
// response, charset and all. Its own reader treats the space after the
// semicolon as insignificant (http.clj:55-66, simple-utf8-ctype?), but a client
// comparing the raw header must see no difference between pdbmux and a PuppetDB.
const jsonContentType = "application/json;charset=utf-8"
type Server struct {
cfg Config
client *http.Client
@@ -527,11 +533,9 @@ func (s *Server) serveFirstHolder(w http.ResponseWriter, r *http.Request) {
alive = append(alive, res)
}
if len(alive) == 0 {
if ue := unanimousClientError(results); ue != nil {
s.writeUpstreamError(w, ue)
return
}
http.Error(w, "no backend holds this report", http.StatusNotFound)
// Unanimity is the whole answer here too: every backend saying 404 means
// nobody holds the report, while one silent backend leaves that unknown.
s.writeUpstreamError(w, peerOutcome(backendUpstreamErrors(results)))
return
}
for _, res := range alive {
@@ -543,8 +547,8 @@ func (s *Server) serveFirstHolder(w http.ResponseWriter, r *http.Request) {
writeJSON(w, nil)
}
// Returns an *upstreamError when every backend refused the query the same way,
// and errAllBackendsFailed when every backend failed for any other reason.
// Returns an *upstreamError when every backend answered the same way, and
// errAllBackendsFailed when every backend failed for any other reason.
func (s *Server) aliveResults(ctx context.Context, path string, params url.Values) ([]backendResult, error) {
results := s.fanOut(ctx, path, params)
@@ -554,13 +558,17 @@ func (s *Server) aliveResults(ctx context.Context, path string, params url.Value
alive = append(alive, res)
}
}
var unanimous *upstreamError
if len(alive) == 0 {
// A query every backend refuses identically is the client's mistake, not an
// outage, so it is neither logged as one nor counted as degraded service.
if ue := unanimousClientError(results); ue != nil {
s.log.Printf("info: every backend refused %s: %v", path, ue)
return nil, ue
}
unanimous = unanimousUpstreamError(backendUpstreamErrors(results))
}
// A query every backend refuses identically is the client's mistake, not an
// outage, so it is neither logged as one nor counted as degraded service. A
// unanimous 5xx is replayed just as faithfully, but it is the backends
// reporting their own fault, so it still counts against service health.
if clientRefusal(unanimous) {
s.log.Printf("info: every backend refused %s: %v", path, unanimous)
return nil, unanimous
}
for _, res := range results {
if res.err != nil {
@@ -569,6 +577,9 @@ func (s *Server) aliveResults(ctx context.Context, path string, params url.Value
}
s.partial.record(len(alive), len(s.cfg.Backends), s.now())
if len(alive) == 0 {
if unanimous != nil {
return nil, unanimous
}
return nil, errAllBackendsFailed
}
return alive, nil
@@ -672,7 +683,7 @@ func (s *Server) serveFiltered(w http.ResponseWriter, r *http.Request, path stri
}
// A refused query is answered, not degraded, so stale records are no reply
// to it: the client has to see why the query was rejected.
if stale != nil && !errors.As(err, new(*upstreamError)) {
if stale != nil && !clientRefusal(err) {
s.stale.markStale(s.now())
s.log.Printf("warning: serving stale %s from cache (stored %s): %v",
path, stale.StoredAt.UTC().Format(time.RFC3339), err)
@@ -730,7 +741,7 @@ func (s *Server) setCacheHeaders(w http.ResponseWriter, status CacheStatus, stor
}
func writeCached(w http.ResponseWriter, resp cachedResponse) {
w.Header().Set("Content-Type", "application/json")
w.Header().Set("Content-Type", jsonContentType)
if resp.Records >= 0 {
w.Header().Set(recordsHeader, strconv.Itoa(resp.Records))
}
@@ -911,18 +922,21 @@ func (s *Server) queryBackend(ctx context.Context, b Backend, path string, param
return recs, total, err
}
// The record shape is unknown, so a union would be guesswork: the first 2xx wins and the first error response is replayed when none succeeds.
// The record shape is unknown, so a union would be guesswork: the first 2xx wins, and every backend was asked the same question, so only a reply they all gave is replayed.
func (s *Server) proxyUnmerged(w http.ResponseWriter, r *http.Request) {
s.proxyOrdered(w, r, s.cfg.Backends)
s.proxyOrdered(w, r, s.cfg.Backends, askPeers)
}
// proxyOrdered asks backends in the given order, which is what decides the answer when more than one of them holds the path.
func (s *Server) proxyOrdered(w http.ResponseWriter, r *http.Request, backends []Backend) {
var fallback *bufferedResponse
// proxyOrdered asks backends in the given order, which is what decides the
// answer when more than one of them holds the path. order says why they are in
// that order, which is what decides whose reply is served when none answers 2xx.
func (s *Server) proxyOrdered(w http.ResponseWriter, r *http.Request, backends []Backend, order askOrder) {
replies := make([]*upstreamError, 0, len(backends))
for _, b := range backends {
resp, err := s.passThrough(r, b)
if err != nil {
s.log.Printf("warning: backend %q pass-through failed for %s: %v", b.Name, r.URL.Path, err)
replies = append(replies, nil)
continue
}
if resp.StatusCode >= 200 && resp.StatusCode < 300 {
@@ -934,27 +948,9 @@ func (s *Server) proxyOrdered(w http.ResponseWriter, r *http.Request, backends [
}
body, _ := io.ReadAll(resp.Body)
_ = resp.Body.Close()
if fallback == nil {
fallback = &bufferedResponse{
status: resp.StatusCode,
contentType: resp.Header.Get("Content-Type"),
body: body,
}
}
replies = append(replies, newUpstreamError(resp.StatusCode, resp.Header.Get("Content-Type"), body))
}
if fallback == nil {
http.Error(w, "all backends failed", http.StatusBadGateway)
return
}
setContentType(w, fallback.contentType)
w.WriteHeader(fallback.status)
_, _ = w.Write(s.redactBackends(fallback.body))
}
type bufferedResponse struct {
status int
contentType string
body []byte
s.writeUpstreamError(w, order.outcome(replies))
}
func (s *Server) passThrough(r *http.Request, b Backend) (*http.Response, error) {
@@ -1106,7 +1102,7 @@ func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {
}
func writeJSON(w http.ResponseWriter, recs []json.RawMessage) {
w.Header().Set("Content-Type", "application/json")
w.Header().Set("Content-Type", jsonContentType)
if recs == nil {
recs = []json.RawMessage{}
}
+9 -3
View File
@@ -29,8 +29,11 @@ type fakeBackend struct {
// totals is the X-Records count advertised per path when the request asks
// for include_total.
totals map[string]int
fail bool // return 500 for everything
delay time.Duration // artificial latency
fail bool // answer 500 for everything, as a backend admitting a fault
// dead aborts the connection instead of answering, standing in for a backend
// that is unreachable rather than one that replies badly.
dead bool
delay time.Duration // artificial latency
// reject answers every request with this status and rejectBody, standing in
// for a PuppetDB refusing a query it cannot answer.
reject int
@@ -60,6 +63,9 @@ func newFakeBackend(t *testing.T, nodesBody, factsBody string) *fakeBackend {
fb.gotParams[r.URL.Path] = r.URL.Query()
fb.rawPaths = append(fb.rawPaths, r.URL.EscapedPath())
fb.mu.Unlock()
if fb.dead {
panic(http.ErrAbortHandler)
}
if fb.fail {
http.Error(w, "boom", http.StatusInternalServerError)
return
@@ -285,7 +291,7 @@ func TestHandler_OneBackendDown(t *testing.T) {
func TestHandler_BothBackendsDown(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.fail, b.fail = true, true
a.dead, b.dead = true, true
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGet(t, srv.Handler(), nodesPath, "")
+123 -36
View File
@@ -35,45 +35,42 @@ func (e *upstreamError) Error() string {
return fmt.Sprintf("HTTP %d: %s", e.status, strings.TrimSpace(string(e.body)))
}
// clientShaped reports whether a status blames the request rather than the
// backend or pdbmux itself. openvoxdb answers every bad query with 400
// src/puppetlabs/puppetdb/middleware.clj:98-116 and http.clj:115-129 — and its
// other 4xx say something a client of pdbmux neither caused nor can fix: 403 is
// pdbmux's own certificate being refused (middleware.clj:44-58); 404 is either
// an absent object or an unknown path (http.clj:238-242,
// middleware.clj:381-398), and which objects a backend holds is the one thing
// backends are meant to disagree about; 408 and 429 report a backend's timing
// and capacity. A backend that times out mid-query answers 200 with a truncated
// body rather than any 4xx (query_eng.clj:463-483), so no timeout reaches here.
// replayableStatus reports whether a status is an error a backend explained and
// so can stand as the answer when every backend gave it. Everything from 400 up
// qualifies, because openvoxdb does not reserve 5xx for its own faults: a
// malformed query reaches the client as a 400 only when something on the way
// down happens to throw a class the query engine catches (query_eng.clj:571-583
// catches IllegalArgumentException and the :puppetlabs.puppetdb.query/invalid
// ExceptionInfo), and as a 500 otherwise. ["=","name"] is a 400 on /nodes and a
// 500 with an empty body on /facts, because only the facts entity runs
// rewrite-fact-query, whose name-constraint does an unguarded (nth clause 2)
// (engine.clj:2988-3003); a bare string or number for query is a 500 reading
// "Output of convert-query-params does not match schema" from the schema check
// in front of the parser. Drawing the line at 500 would replay one of those and
// swallow the other, which is the route-dependence clients notice.
//
// Every 4xx left over blames the request and so is safe to replay, though none
// is reachable while the fan-out forwards no client header: 406 needs an Accept
// the query app refuses (http/server.clj:72; must-accept-type in
// http.clj:136-145 is unwired), and 415 a Content-Encoding on a POST to
// /commands (middleware.clj:165-178), which pdbmux never proxies.
func clientShaped(status int) bool {
switch status {
case http.StatusForbidden, http.StatusNotFound,
http.StatusRequestTimeout, http.StatusTooManyRequests:
return false
}
return status >= 400 && status < 500
// Below 400 nothing is replayed. A 3xx carries its meaning in Location, which
// the fan-out does not keep, and a non-200 2xx is not an error at all.
func replayableStatus(status int) bool {
return status >= 400
}
// unanimousClientError returns the rejection to replay when every backend in a
// fan-out refused the same query with the same client-shaped status, and nil
// otherwise. Every backend is asked the same question, so unanimity is what
// distinguishes a bad query from a sick estate: a transport failure, a 5xx, or
// two backends disagreeing on the status all leave at least one backend whose
// answer is evidence about the backend rather than about the request.
// unanimousUpstreamError returns the reply to replay when every backend answered
// the same request the same way, and nil otherwise. Every backend is asked the
// same question, so unanimity is what separates the estate's real answer from a
// sick estate: a transport failure, or two backends disagreeing on the status,
// leaves at least one backend whose answer is evidence about the backend rather
// than about the request, and that is what 502 describes.
//
// A nil entry stands for a backend that did not answer with an HTTP status at
// all, which defeats unanimity.
//
// The reply returned is the first in configured order, so a client retrying a
// rejected query is told the same thing every time.
func unanimousClientError(results []backendResult) *upstreamError {
func unanimousUpstreamError(errs []*upstreamError) *upstreamError {
var first *upstreamError
for _, res := range results {
var ue *upstreamError
if !errors.As(res.err, &ue) || !clientShaped(ue.status) {
for _, ue := range errs {
if ue == nil || !replayableStatus(ue.status) {
return nil
}
if first == nil {
@@ -85,9 +82,99 @@ func unanimousClientError(results []backendResult) *upstreamError {
return first
}
// writeUpstreamError answers a fan-out that produced no records. A unanimous
// client-shaped rejection is replayed with the backend's own status and
// explanation; anything else is reported as a gateway failure.
// askOrder says why a set of backends was asked, which is what decides whose
// reply becomes the client's answer when none of them answered 2xx. The two
// cases are not settings on one rule, they are different questions, and reading
// a reply under the wrong one is how a real answer turns into a 502.
type askOrder int
const (
// askPeers: every backend was asked the same question and any of them could
// have answered it, so each reply is an opinion about that question. Only a
// reply they all gave is the estate's answer; anything else leaves a backend
// whose reply is evidence about the backend, which is what 502 reports.
askPeers askOrder = iota
// askOwnerFirst: the path names one entity, so the backend holding it is
// asked first — or, where no owner is known, they are asked in configured
// order — and the rest only as fallbacks. A backend that does not hold the
// entity replies 404 to say exactly that, which is not a dissent from the
// holder's reply, so requiring the two to agree asks a question nobody was
// posed. The first backend that answered is the one that knows.
askOwnerFirst
)
// outcome turns a round's replies — one entry per backend in the order they were
// asked, nil where a backend produced no HTTP status at all — into the error to
// answer with.
func (o askOrder) outcome(replies []*upstreamError) error {
if o == askOwnerFirst {
return ownerOutcome(replies)
}
return peerOutcome(replies)
}
// peerOutcome is the error to answer a fan-out of peers with when it produced
// nothing: their unanimous reply where there is one, and pdbmux's gateway error
// otherwise.
func peerOutcome(errs []*upstreamError) error {
if ue := unanimousUpstreamError(errs); ue != nil {
return ue
}
return errAllBackendsFailed
}
// ownerOutcome is the error to answer an owner-routed path with: the reply of
// the first backend that answered. The owner is asked first, so that is the
// owner's own reply — it holds the entity, so its 500 is the truth about this
// request whatever a backend that does not hold the entity said.
//
// An owner that produced no status at all leaves a nil entry and the next
// backend's reply stands instead. The fallback chain exists so a node whose
// owner is unreachable is still served, and a backend that did answer explains
// more than a 502 that describes neither. Only a round where nothing answered is
// pdbmux's own gateway error.
func ownerOutcome(replies []*upstreamError) error {
for _, ue := range replies {
if ue != nil && replayableStatus(ue.status) {
return ue
}
}
return errAllBackendsFailed
}
// backendUpstreamErrors reduces a merged fan-out's results to one entry per
// backend, nil where the backend answered or failed without a status.
func backendUpstreamErrors(results []backendResult) []*upstreamError {
errs := make([]*upstreamError, len(results))
for i, res := range results {
var ue *upstreamError
if errors.As(res.err, &ue) {
errs[i] = ue
}
}
return errs
}
// clientRefusal reports whether err is a unanimous upstream rejection blaming
// the request. Only those mean the estate is healthy and the query was wrong,
// which is why they are neither counted as degraded service nor answered from
// the cache; a unanimous 5xx is replayed just the same but is the backends
// reporting their own fault.
//
// Every 4xx openvoxdb answers a fan-out with is about the request, including the
// 403 on /metrics/v2/list, which is that backend's own policy rather than an
// authentication failure. That holds only while pdbmux presents no client
// certificate: authenticate to backends and a rejected certificate becomes a
// 403 on every route at once, which this would read as a healthy estate refusing
// a bad query and hide a total outage behind a replayed 403.
func clientRefusal(err error) bool {
var ue *upstreamError
return errors.As(err, &ue) && ue != nil && ue.status < 500
}
// writeUpstreamError answers a round that produced no records. An upstream
// error the askOrder resolved to is replayed with the backend's own status,
// content type and explanation; anything else is reported as a gateway failure.
func (s *Server) writeUpstreamError(w http.ResponseWriter, err error) {
var ue *upstreamError
if !errors.As(err, &ue) {
@@ -98,7 +185,7 @@ func (s *Server) writeUpstreamError(w http.ResponseWriter, err error) {
if len(strings.TrimSpace(string(body))) == 0 {
setContentType(w, "text/plain; charset=utf-8")
w.WriteHeader(ue.status)
_, _ = fmt.Fprintf(w, "upstream rejected the query: %s\n", http.StatusText(ue.status))
_, _ = fmt.Fprintf(w, "upstream answered %d %s\n", ue.status, http.StatusText(ue.status))
return
}
setContentType(w, ue.contentType)
+201 -29
View File
@@ -14,11 +14,16 @@ import (
const badOrderBy = `Unrecognized column 'bogus' specified in :order_by`
// upstreamJSONContentType is spelled out rather than taken from jsonContentType,
// so the assertion is against what openvoxdb answers and not against whatever
// pdbmux happens to be configured with.
const upstreamJSONContentType = "application/json;charset=utf-8"
func upErr(status int, body string) error {
return newUpstreamError(status, "text/plain; charset=utf-8", []byte(body))
}
func TestUnanimousClientError_AgreementRule(t *testing.T) {
func TestUnanimousUpstreamError_AgreementRule(t *testing.T) {
refused := errors.New("dial tcp: connection refused")
for _, tc := range []struct {
name string
@@ -29,18 +34,25 @@ func TestUnanimousClientError_AgreementRule(t *testing.T) {
{"all agree on 415", []error{upErr(415, "bad media"), upErr(415, "bad media")}, 415},
{"differing bodies still agree", []error{upErr(400, "one"), upErr(400, "two")}, 400},
{"disagreeing 4xx", []error{upErr(400, "bad"), upErr(404, "gone")}, 0},
// Both client-shaped, so only the same-status rule refuses these.
{"agree on shape, disagree on code", []error{upErr(400, "x"), upErr(415, "y")}, 0},
{"a majority agrees", []error{upErr(400, "x"), upErr(400, "y"), upErr(422, "z")}, 0},
{"the first differs", []error{upErr(422, "z"), upErr(400, "x"), upErr(400, "y")}, 0},
{"4xx with a 5xx", []error{upErr(400, "bad"), upErr(500, "boom")}, 0},
{"4xx with a transport failure", []error{upErr(400, "bad"), refused}, 0},
{"all 403", []error{upErr(403, "denied"), upErr(403, "denied")}, 0},
{"all 404", []error{upErr(404, "gone"), upErr(404, "gone")}, 0},
{"all 429", []error{upErr(429, "slow down"), upErr(429, "slow down")}, 0},
{"all 408", []error{upErr(408, "too slow"), upErr(408, "too slow")}, 0},
{"all 500", []error{upErr(500, "boom"), upErr(500, "boom")}, 0},
{"all 503", []error{upErr(503, "unavailable"), upErr(503, "unavailable")}, 0},
{"5xx with a transport failure", []error{upErr(500, "boom"), refused}, 0},
// Statuses the old 4xx-only rule swallowed. Every one of them is what
// both backends actually said, so every one of them is the answer.
{"all 403", []error{upErr(403, "denied"), upErr(403, "denied")}, 403},
{"all 404", []error{upErr(404, "gone"), upErr(404, "gone")}, 404},
{"all 429", []error{upErr(429, "slow down"), upErr(429, "slow down")}, 429},
{"all 408", []error{upErr(408, "too slow"), upErr(408, "too slow")}, 408},
{"all 500", []error{upErr(500, "boom"), upErr(500, "boom")}, 500},
{"all 503", []error{upErr(503, "unavailable"), upErr(503, "unavailable")}, 503},
{"disagreeing 5xx", []error{upErr(500, "boom"), upErr(503, "later")}, 0},
// Nothing below 400 is an error a backend explained, and a 3xx's meaning
// lives in a Location header the fan-out never kept.
{"all 302", []error{upErr(302, ""), upErr(302, "")}, 0},
{"all 204", []error{upErr(204, ""), upErr(204, "")}, 0},
{"no backends", nil, 0},
{"a backend succeeded", []error{upErr(400, "bad"), nil}, 0},
} {
@@ -49,7 +61,7 @@ func TestUnanimousClientError_AgreementRule(t *testing.T) {
for i, err := range tc.errs {
results[i] = backendResult{name: fmt.Sprintf("b%d", i), err: err}
}
got := unanimousClientError(results)
got := unanimousUpstreamError(backendUpstreamErrors(results))
switch {
case tc.want == 0 && got != nil:
t.Fatalf("want no replay, got HTTP %d", got.status)
@@ -64,17 +76,33 @@ func TestUnanimousClientError_AgreementRule(t *testing.T) {
// Two backends can explain the same rejection differently; the reply is the
// first in configured order so a retried query is answered the same way twice.
func TestUnanimousClientError_PicksFirstInConfiguredOrder(t *testing.T) {
func TestUnanimousUpstreamError_PicksFirstInConfiguredOrder(t *testing.T) {
results := []backendResult{
{name: "a", err: upErr(400, "from a")},
{name: "b", err: upErr(400, "from b")},
}
got := unanimousClientError(results)
got := unanimousUpstreamError(backendUpstreamErrors(results))
if got == nil || !strings.Contains(string(got.body), "from a") {
t.Fatalf("body = %q, want the first backend's", got)
}
}
// Only a 4xx says the estate is well and the query was wrong. A 5xx is replayed
// too, but it must keep counting as a backend fault.
func TestClientRefusal_OnlyFourXX(t *testing.T) {
for status, want := range map[int]bool{400: true, 404: true, 429: true, 499: true, 500: false, 503: false} {
if got := clientRefusal(upErr(status, "x")); got != want {
t.Errorf("clientRefusal(%d) = %v, want %v", status, got, want)
}
}
if clientRefusal(errAllBackendsFailed) {
t.Error("a transport failure is not a client refusal")
}
if clientRefusal(unanimousUpstreamError(nil)) {
t.Error("an absent unanimous error is not a client refusal")
}
}
func TestUpstreamError_BodyIsCapped(t *testing.T) {
ue := newUpstreamError(400, "text/plain", []byte(strings.Repeat("x", upstreamBodyLimit*2)))
if len(ue.body) != upstreamBodyLimit {
@@ -112,31 +140,78 @@ func TestHandler_MergedReplaysNonBadRequestStatus(t *testing.T) {
}
}
// A unanimous 403 is pdbmux's own credentials being refused, not the client's
// query, so it must not be handed back as the client's fault.
func TestHandler_MergedForbiddenStays502(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusForbidden, "certificate not allowed"
b.reject, b.rejectBody = http.StatusForbidden, "certificate not allowed"
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
// A status every backend agreed on is the estate's own answer whatever it is,
// so the statuses the 4xx-only rule used to swallow now reach the client.
func TestHandler_MergedReplaysEveryUnanimousStatus(t *testing.T) {
for _, tc := range []struct {
status int
body string
}{
{http.StatusForbidden, "certificate not allowed"},
{http.StatusNotFound, "Not Found"},
{http.StatusTooManyRequests, "slow down"},
{http.StatusInternalServerError, "Output of convert-query-params does not match schema"},
{http.StatusServiceUnavailable, "starting up"},
} {
t.Run(http.StatusText(tc.status), func(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = tc.status, tc.body
b.reject, b.rejectBody = tc.status, tc.body
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
if rec := doGet(t, srv.Handler(), nodesPath, ""); rec.Code != http.StatusBadGateway {
t.Fatalf("status = %d, want 502", rec.Code)
rec := doGet(t, srv.Handler(), nodesPath, "")
if rec.Code != tc.status {
t.Fatalf("status = %d, want the upstream %d: %s", rec.Code, tc.status, rec.Body.String())
}
if !strings.Contains(rec.Body.String(), tc.body) {
t.Errorf("body = %q, want the upstream explanation", rec.Body.String())
}
})
}
}
// Every backend answering 404 is ambiguous between a bad path and a record
// nobody holds, so it stays a gateway error on a merged route.
func TestHandler_MergedNotFoundStays502(t *testing.T) {
// A backend that answered nothing at all leaves no unanimity to replay, whatever
// the others said.
func TestHandler_MergedServerErrorWithUnreachableStays502(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusInternalServerError, "boom"
dead := newFakeBackend(t, `[]`, `[]`)
deadURL := dead.srv.URL
dead.srv.Close()
srv := newTestServer(testConfig(a.srv.URL, deadURL, mergeStatic))
if rec := doGet(t, srv.Handler(), nodesPath, ""); rec.Code != http.StatusBadGateway {
t.Fatalf("status = %d, want 502 when a backend never answered", rec.Code)
}
}
func TestHandler_MergedDisagreeingServerErrorsStay502(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusNotFound, "Not Found"
b.reject, b.rejectBody = http.StatusNotFound, "Not Found"
a.reject, a.rejectBody = http.StatusInternalServerError, "boom"
b.reject, b.rejectBody = http.StatusServiceUnavailable, "later"
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
if rec := doGet(t, srv.Handler(), nodesPath, ""); rec.Code != http.StatusBadGateway {
t.Fatalf("status = %d, want 502", rec.Code)
t.Fatalf("status = %d, want 502 when backends disagree", rec.Code)
}
}
// A unanimous 5xx is replayed, but it is the backends admitting a fault, so it
// still has to read as degraded service.
func TestHandler_UnanimousServerErrorIsStillADegradedRound(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusInternalServerError, "boom"
b.reject, b.rejectBody = http.StatusInternalServerError, "boom"
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
if rec := doGet(t, srv.Handler(), nodesPath, ""); rec.Code != http.StatusInternalServerError {
t.Fatalf("status = %d, want the upstream 500", rec.Code)
}
if hr := health(t, srv); hr.Query.PartialRounds == 0 {
t.Errorf("query health = %+v, want the 5xx counted as a failed round", hr.Query)
}
}
@@ -255,7 +330,7 @@ func TestHandler_ReportSubResourceRejectionReplayedButNotItsAbsence(t *testing.T
b := newFakeBackend(t, `[]`, `[]`)
if rec := doGet(t, newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)).Handler(),
reportsPath+"/nope/events", ""); rec.Code != http.StatusNotFound {
t.Fatalf("status = %d, want pdbmux's own 404 when nobody holds the report", rec.Code)
t.Fatalf("status = %d, want 404 when nobody holds the report", rec.Code)
}
a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
@@ -316,7 +391,7 @@ func TestHandler_RejectedQueryIsNotCached(t *testing.T) {
func TestHandler_AllBackendsFailedIsNotCached(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.fail, b.fail = true, true
a.dead, b.dead = true, true
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
if rec := doGet(t, srv.Handler(), factsPath, ""); rec.Code != http.StatusBadGateway {
@@ -493,3 +568,100 @@ func hostOf(t *testing.T, raw string) string {
}
return u.Host
}
// Every backend answering 500 is the estate's answer on the first-holder route
// as much as on a merged one.
func TestHandler_ReportSubResourceReplaysUnanimousServerError(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusInternalServerError, "boom"
b.reject, b.rejectBody = http.StatusInternalServerError, "boom"
rec := doGet(t, newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)).Handler(),
reportsPath+"/nope/events", "")
if rec.Code != http.StatusInternalServerError {
t.Fatalf("status = %d, want the upstream 500: %s", rec.Code, rec.Body.String())
}
}
// One backend saying 404 and one saying nothing leaves it unknown whether the
// report exists, which is the gateway error's job to say.
func TestHandler_ReportSubResourceUnreachableBackendIs502(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
b.dead = true
rec := doGet(t, newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)).Handler(),
reportsPath+"/nope/events", "")
if rec.Code != http.StatusBadGateway {
t.Fatalf("status = %d, want 502 when a backend never answered", rec.Code)
}
}
// The pass-through route used to replay whichever backend answered first, even
// when the others disagreed or never answered. It follows the same rule now.
func TestProxyUnmerged_ReplayRule(t *testing.T) {
for _, tc := range []struct {
name string
aStatus int
bStatus int
bDead bool
want int
}{
{"unanimous 500 is replayed", 500, 500, false, 500},
{"unanimous 404 is replayed", 404, 404, false, 404},
{"unanimous 403 is replayed", 403, 403, false, 403},
{"disagreement is a gateway error", 400, 404, false, http.StatusBadGateway},
{"a silent backend is a gateway error", 404, 0, true, http.StatusBadGateway},
} {
t.Run(tc.name, func(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = tc.aStatus, "upstream said no"
b.reject, b.rejectBody = tc.bStatus, "upstream said no"
b.dead = tc.bDead
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGet(t, srv.Handler(), resourcesPath, `["=","type","File"]`)
if rec.Code != tc.want {
t.Fatalf("status = %d, want %d: %s", rec.Code, tc.want, rec.Body.String())
}
})
}
}
// pdbmux has to be indistinguishable from a PuppetDB, and openvoxdb answers a
// query with a charset on the content type.
func TestHandler_SuccessContentTypeMatchesUpstream(t *testing.T) {
a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
h := srv.Handler()
for _, path := range []string{nodesPath, factsPath, factNamesPath, reportsPath} {
t.Run(path, func(t *testing.T) {
rec := doGet(t, h, path, "")
if rec.Code != http.StatusOK {
t.Fatalf("status = %d: %s", rec.Code, rec.Body.String())
}
if got := rec.Header().Get("Content-Type"); got != upstreamJSONContentType {
t.Errorf("Content-Type = %q, want %q", got, upstreamJSONContentType)
}
})
}
}
func TestHandler_CachedSuccessContentTypeMatchesUpstream(t *testing.T) {
a := newFakeBackend(t, `[]`, `[`+fact("h1", "role", "web", "")+`]`)
b := newFakeBackend(t, `[]`, `[]`)
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
h := srv.Handler()
// The miss builds the entry and the hit replays it; both are the client's view.
for _, label := range []string{"miss", "hit"} {
rec := doGet(t, h, factsPath, "")
if got := rec.Header().Get("Content-Type"); got != upstreamJSONContentType {
t.Errorf("%s Content-Type = %q, want %q", label, got, upstreamJSONContentType)
}
}
}