Replay a unanimous upstream rejection instead of a 502 #23

Merged
benvin merged 1 commits from benvin/upstream-errors into main 2026-09-07 23:07:56 +10:00
7 changed files with 676 additions and 11 deletions
+12 -1
View File
@@ -48,6 +48,16 @@ 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.
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.
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
paths add two more headers `pdbmux` sets itself, `X-Cache` and `Age` — see
@@ -435,7 +445,8 @@ backend later without further handler changes.
- **Stale on failure only** — an expired entry is kept, not dropped. When the TTL
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.
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.
- **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 -2
View File
@@ -342,6 +342,14 @@ func (b *backend) query(ctx context.Context, t fatalf, path string, params url.V
// queryStatus is query without the body, for asserting what a backend rejects.
func (b *backend) queryStatus(ctx context.Context, t fatalf, path string, params url.Values) int {
t.Helper()
status, _ := b.queryRaw(ctx, t, path, params)
return status
}
// queryRaw returns a backend's own status and body, so a test can compare what
// pdbmux served against what the backend actually said.
func (b *backend) queryRaw(ctx context.Context, t fatalf, path string, params url.Values) (int, []byte) {
t.Helper()
target := b.url + path
if len(params) > 0 {
@@ -356,8 +364,11 @@ func (b *backend) queryStatus(ctx context.Context, t fatalf, path string, params
t.Fatalf("querying %s%s: %v", b.name, path, err)
}
defer func() { _ = resp.Body.Close() }()
_, _ = io.Copy(io.Discard, resp.Body)
return resp.StatusCode
body, err := io.ReadAll(resp.Body)
if err != nil {
t.Fatalf("reading %s%s: %v", b.name, path, err)
}
return resp.StatusCode, body
}
func newNetwork(ctx context.Context, t fatalf) (string, func()) {
+58
View File
@@ -0,0 +1,58 @@
//go:build e2e
package main
import (
"context"
"strings"
"testing"
)
// Every backend gets the same query, so a query none of them can answer is the
// client's mistake rather than an outage. openvoxdb explains what is wrong with
// it; pdbmux has to hand that explanation and its status back instead of a 502
// saying the estate is down.
func TestRejectedQuerySurfacesUpstreamStatus(t *testing.T) {
ctx := context.Background()
for _, tc := range []struct{ name, path, q string }{
{"merged", nodesPath, `["=","bogus","x"]`},
{"combined", nodesPath, `["extract",[["function","count"]],["=","bogus","x"]]`},
{"union", reportsPath, `["=","bogus","x"]`},
} {
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) {
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)
}
if other, _ := h.b.queryRaw(ctx, t, tc.path, params); other != status {
t.Fatalf("backends answered %d and %d, so the rejection is not unanimous", status, other)
}
_, before := healthz(t)
resp := rawGet(t, tc.path, 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)); 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)
}
}
// A refused query is not degraded service, so it must leave the merged
// fan-out's own health counters where it found them.
_, after := healthz(t)
if after.Query.PartialRounds != before.Query.PartialRounds || after.Query.Partial {
t.Errorf("query health = %+v, want %+v unchanged by a refused query", after.Query, before.Query)
}
if after.Status != "ok" {
t.Errorf("/healthz = %q after a refused query, want ok", after.Status)
}
})
}
}
+26 -8
View File
@@ -521,6 +521,10 @@ 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)
return
}
@@ -533,17 +537,29 @@ func (s *Server) serveFirstHolder(w http.ResponseWriter, r *http.Request) {
writeJSON(w, nil)
}
// Returns errAllBackendsFailed only when every backend failed.
// Returns an *upstreamError when every backend refused the query 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)
var alive []backendResult
for _, res := range results {
if res.err == nil {
alive = append(alive, res)
}
}
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
}
}
for _, res := range results {
if res.err != nil {
s.log.Printf("warning: backend %q failed for %s: %v", res.name, path, res.err)
continue
}
alive = append(alive, res)
}
s.partial.record(len(alive), len(s.cfg.Backends), s.now())
if len(alive) == 0 {
@@ -596,7 +612,7 @@ func (s *Server) serveFiltered(w http.ResponseWriter, r *http.Request, path stri
if !enabled {
resp, err := build(r.Context())
if err != nil {
http.Error(w, err.Error(), http.StatusBadGateway)
s.writeUpstreamError(w, err)
return
}
writeCached(w, filter.apply(resp))
@@ -648,14 +664,16 @@ func (s *Server) serveFiltered(w http.ResponseWriter, r *http.Request, path stri
if errors.Is(err, errFlightAbandoned) {
return
}
if stale != nil {
// 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)) {
s.stale.markStale(s.now())
s.log.Printf("warning: serving stale %s from cache (stored %s): %v",
path, stale.StoredAt.UTC().Format(time.RFC3339), err)
s.writeStored(w, *stale, CacheStale, filter)
return
}
http.Error(w, err.Error(), http.StatusBadGateway)
s.writeUpstreamError(w, err)
return
}
s.stale.markFresh()
@@ -877,7 +895,7 @@ func (s *Server) queryBackend(ctx context.Context, b Backend, path string, param
return nil, -1, err
}
if resp.StatusCode != http.StatusOK {
return nil, -1, fmt.Errorf("HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body)))
return nil, -1, newUpstreamError(resp.StatusCode, resp.Header.Get("Content-Type"), body)
}
total := -1
if n, err := strconv.Atoi(resp.Header.Get(recordsHeader)); err == nil && n >= 0 {
@@ -919,7 +937,7 @@ func (s *Server) proxyUnmerged(w http.ResponseWriter, r *http.Request) {
}
setContentType(w, fallback.contentType)
w.WriteHeader(fallback.status)
_, _ = w.Write(fallback.body)
_, _ = w.Write(s.redactBackends(fallback.body))
}
type bufferedResponse struct {
+8
View File
@@ -31,6 +31,10 @@ type fakeBackend struct {
totals map[string]int
fail bool // return 500 for everything
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
rejectBody string
mu sync.Mutex
gotParams map[string]url.Values
@@ -60,6 +64,10 @@ func newFakeBackend(t *testing.T, nodesBody, factsBody string) *fakeBackend {
http.Error(w, "boom", http.StatusInternalServerError)
return
}
if fb.reject != 0 {
http.Error(w, fb.rejectBody, fb.reject)
return
}
if body, ok := fb.bodies[r.URL.Path]; ok {
if n, ok := fb.totals[r.URL.Path]; ok && r.URL.Query().Get("include_total") == "true" {
w.Header().Set(recordsHeader, strconv.Itoa(n))
+134
View File
@@ -0,0 +1,134 @@
package main
import (
"errors"
"fmt"
"net/http"
"net/url"
"strings"
)
// upstreamBodyLimit caps how much of a backend's error body is kept for replay.
const upstreamBodyLimit = 8 << 10
// redactedBackend stands in for anything naming a backend in a replayed body.
const redactedBackend = "<backend>"
// upstreamError is a non-2xx reply from a backend kept whole — status, content
// type and body — so a rejection the backend explained can be replayed to the
// client instead of collapsed into a gateway error that throws the explanation
// away.
type upstreamError struct {
status int
contentType string
body []byte
}
func newUpstreamError(status int, contentType string, body []byte) *upstreamError {
if len(body) > upstreamBodyLimit {
body = body[:upstreamBodyLimit]
}
return &upstreamError{status: status, contentType: contentType, body: body}
}
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.
func clientShaped(status int) bool {
switch status {
case http.StatusForbidden, http.StatusNotFound,
http.StatusRequestTimeout, http.StatusTooManyRequests:
return false
}
return status >= 400 && status < 500
}
// 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.
//
// 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 {
var first *upstreamError
for _, res := range results {
var ue *upstreamError
if !errors.As(res.err, &ue) || !clientShaped(ue.status) {
return nil
}
if first == nil {
first = ue
} else if ue.status != first.status {
return nil
}
}
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.
func (s *Server) writeUpstreamError(w http.ResponseWriter, err error) {
var ue *upstreamError
if !errors.As(err, &ue) {
http.Error(w, err.Error(), http.StatusBadGateway)
return
}
body := s.redactBackends(ue.body)
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))
return
}
setContentType(w, ue.contentType)
w.WriteHeader(ue.status)
_, _ = w.Write(body)
}
// redactBackends strips anything naming a configured backend out of text bound
// for a client. Hiding the estate behind one endpoint is the point of pdbmux, so
// a backend's own words may reach the client but its address may not.
func (s *Server) redactBackends(body []byte) []byte {
out := string(body)
for _, b := range s.cfg.Backends {
for _, term := range backendTerms(b) {
out = strings.ReplaceAll(out, term, redactedBackend)
}
}
return []byte(out)
}
// backendTerms is what identifies a backend in text: its URL, that URL's host
// and the bare hostname, longest first so replacing one cannot leave a fragment
// of another behind. The configured name is deliberately absent — no backend
// knows it, and names are short enough to match ordinary words.
func backendTerms(b Backend) []string {
terms := []string{}
if b.URL != "" {
terms = append(terms, strings.TrimRight(b.URL, "/"))
}
u, err := url.Parse(b.URL)
if err != nil || u.Host == "" {
return terms
}
terms = append(terms, u.Host)
if hn := u.Hostname(); hn != u.Host {
terms = append(terms, hn)
}
return terms
}
+425
View File
@@ -0,0 +1,425 @@
package main
import (
"errors"
"fmt"
"net/http"
"net/url"
"strings"
"testing"
"time"
)
const badOrderBy = `Unrecognized column 'bogus' specified in :order_by`
func upErr(status int, body string) error {
return newUpstreamError(status, "text/plain; charset=utf-8", []byte(body))
}
func TestUnanimousClientError_AgreementRule(t *testing.T) {
refused := errors.New("dial tcp: connection refused")
for _, tc := range []struct {
name string
errs []error
want int // 0 means "no replay"
}{
{"all agree on 400", []error{upErr(400, "bad"), upErr(400, "bad")}, 400},
{"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},
{"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},
{"no backends", nil, 0},
{"a backend succeeded", []error{upErr(400, "bad"), nil}, 0},
} {
t.Run(tc.name, func(t *testing.T) {
results := make([]backendResult, len(tc.errs))
for i, err := range tc.errs {
results[i] = backendResult{name: fmt.Sprintf("b%d", i), err: err}
}
got := unanimousClientError(results)
switch {
case tc.want == 0 && got != nil:
t.Fatalf("want no replay, got HTTP %d", got.status)
case tc.want != 0 && got == nil:
t.Fatalf("want HTTP %d replayed, got none", tc.want)
case tc.want != 0 && got.status != tc.want:
t.Fatalf("status = %d, want %d", got.status, tc.want)
}
})
}
}
// 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) {
results := []backendResult{
{name: "a", err: upErr(400, "from a")},
{name: "b", err: upErr(400, "from b")},
}
got := unanimousClientError(results)
if got == nil || !strings.Contains(string(got.body), "from a") {
t.Fatalf("body = %q, want the first backend's", got)
}
}
func TestUpstreamError_BodyIsCapped(t *testing.T) {
ue := newUpstreamError(400, "text/plain", []byte(strings.Repeat("x", upstreamBodyLimit*2)))
if len(ue.body) != upstreamBodyLimit {
t.Fatalf("kept %d bytes, want %d", len(ue.body), upstreamBodyLimit)
}
}
func TestHandler_MergedAllBackendsRejectReplaysUpstreamStatus(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
b.reject, b.rejectBody = http.StatusBadRequest, badOrderBy
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), nodesPath, url.Values{"order_by": {`[{"field":"certname"}]`}})
if rec.Code != http.StatusBadRequest {
t.Fatalf("status = %d, want the upstream 400 replayed: %s", rec.Code, rec.Body.String())
}
if !strings.Contains(rec.Body.String(), "bogus") {
t.Errorf("body = %q, want the upstream explanation", rec.Body.String())
}
}
// 400 is not the only client-shaped status openvoxdb can answer with, so the
// replay carries whatever status the backends agreed on.
func TestHandler_MergedReplaysNonBadRequestStatus(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusUnsupportedMediaType, "unsupported media type"
b.reject, b.rejectBody = http.StatusUnsupportedMediaType, "unsupported media type"
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
if rec := doGet(t, srv.Handler(), nodesPath, ""); rec.Code != http.StatusUnsupportedMediaType {
t.Fatalf("status = %d, want 415", rec.Code)
}
}
// 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))
if rec := doGet(t, srv.Handler(), nodesPath, ""); rec.Code != http.StatusBadGateway {
t.Fatalf("status = %d, want 502", rec.Code)
}
}
// 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 := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusNotFound, "Not Found"
b.reject, b.rejectBody = http.StatusNotFound, "Not Found"
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)
}
}
func TestHandler_MergedMixedRejectionAndServerErrorStays502(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
b.fail = true
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 when only one backend blamed the query", rec.Code)
}
}
func TestHandler_MergedMixedRejectionAndUnreachableStays502(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
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_MergedDisagreeingRejectionsStay502(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
b.reject, b.rejectBody = http.StatusNotFound, "no such entity"
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 when backends disagree on the rejection", rec.Code)
}
}
// A backend that cannot answer a query must not fail one another backend can.
func TestHandler_MergedOneRejectionOneAnswerServesTheAnswer(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
b := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`)
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGet(t, srv.Handler(), nodesPath, "")
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want the surviving backend's 200: %s", rec.Code, rec.Body.String())
}
if !strings.Contains(rec.Body.String(), "h1") {
t.Errorf("body = %s, want the survivor's record", rec.Body.String())
}
}
func TestHandler_CombinedAllBackendsRejectReplaysUpstreamStatus(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
b.reject, b.rejectBody = http.StatusBadRequest, badOrderBy
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.StatusBadRequest {
t.Fatalf("status = %d, want the aggregate path to replay 400: %s", rec.Code, rec.Body.String())
}
if !strings.Contains(rec.Body.String(), "bogus") {
t.Errorf("body = %q, want the upstream explanation", rec.Body.String())
}
}
func TestHandler_CombinedMixedRejectionAndServerErrorStays502(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
b.fail = true
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.StatusBadGateway {
t.Fatalf("status = %d, want 502", rec.Code)
}
}
func TestHandler_UnionAllBackendsRejectReplaysUpstreamStatus(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
b.reject, b.rejectBody = http.StatusBadRequest, badOrderBy
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{"order_by": {receiveDesc}})
if rec.Code != http.StatusBadRequest {
t.Fatalf("status = %d, want the union path to replay 400: %s", rec.Code, rec.Body.String())
}
}
// /fact-names orders locally, so its fan-out is the third shape aliveResults serves.
func TestHandler_FactNamesAllBackendsRejectReplaysUpstreamStatus(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
b.reject, b.rejectBody = http.StatusBadRequest, badOrderBy
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
if rec := doGet(t, srv.Handler(), factNamesPath, ""); rec.Code != http.StatusBadRequest {
t.Fatalf("status = %d, want 400", rec.Code)
}
}
// A backend answering 404 means it does not hold the report, which is the whole
// premise of this route, so it must keep producing pdbmux's own 404.
func TestHandler_ReportSubResourceRejectionReplayedButNotItsAbsence(t *testing.T) {
a := newFakeBackend(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)
}
a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
b.reject, b.rejectBody = http.StatusBadRequest, badOrderBy
rec := doGet(t, newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)).Handler(),
reportsPath+"/nope/events", "")
if rec.Code != http.StatusBadRequest {
t.Fatalf("status = %d, want the upstream 400 replayed: %s", rec.Code, rec.Body.String())
}
}
func TestHandler_RejectedQueryIsNotADegradedRound(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
b.reject, b.rejectBody = http.StatusBadRequest, badOrderBy
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
if rec := doGet(t, srv.Handler(), nodesPath, ""); rec.Code != http.StatusBadRequest {
t.Fatalf("status = %d, want 400", rec.Code)
}
hr := health(t, srv)
if hr.Query.Partial || hr.Query.PartialRounds != 0 || hr.Query.LastPartial != "" {
t.Fatalf("a refused query was counted as degraded service: %+v", hr.Query)
}
// The report's own reachability probe is a separate query the backends also
// refuse, so they read as unreachable while the estate itself is fine.
if hr.Status != "down" {
t.Errorf("status = %q, want the probe's own verdict", hr.Status)
}
}
func TestHandler_RejectedQueryIsNotCached(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
b.reject, b.rejectBody = http.StatusBadRequest, badOrderBy
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
rec := doGet(t, srv.Handler(), factsPath, "")
if rec.Code != http.StatusBadRequest {
t.Fatalf("status = %d, want the upstream 400 replayed: %s", rec.Code, rec.Body.String())
}
if n := srv.factsCache.Stats().Entries; n != 0 {
t.Fatalf("cache holds %d entries after a rejected query", n)
}
// The rejection left nothing behind, so the next good query is served fresh.
a.reject, b.reject = 0, 0
a.factsBody = `[` + fact("h1", "role", "web", "") + `]`
ok := doGet(t, srv.Handler(), factsPath, "")
if ok.Code != http.StatusOK || !strings.Contains(ok.Body.String(), "web") {
t.Fatalf("follow-up = %d %s", ok.Code, ok.Body.String())
}
}
func TestHandler_AllBackendsFailedIsNotCached(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.fail, b.fail = true, true
srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
if rec := doGet(t, srv.Handler(), factsPath, ""); rec.Code != http.StatusBadGateway {
t.Fatalf("status = %d, want 502", rec.Code)
}
if n := srv.factsCache.Stats().Entries; n != 0 {
t.Fatalf("cache holds %d entries after a failed fan-out", n)
}
}
// Stale records answer an outage. They do not answer a query the estate refused:
// the client has to be told why, not handed data for a question it did not ask.
func TestHandler_RejectedQueryIsNotAnsweredFromStale(t *testing.T) {
a := newFakeBackend(t, `[]`, `[`+fact("h1", "role", "old", "")+`]`)
b := newFakeBackend(t, `[]`, `[]`)
srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
if warm := doGet(t, srv.Handler(), factsPath, ""); warm.Code != http.StatusOK {
t.Fatalf("warm-up status %d", warm.Code)
}
clk.advance(31 * time.Second)
a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
b.reject, b.rejectBody = http.StatusBadRequest, badOrderBy
rec := doGet(t, srv.Handler(), factsPath, "")
if rec.Code != http.StatusBadRequest {
t.Fatalf("status = %d, want the rejection rather than the stale entry: %s", rec.Code, rec.Body.String())
}
if strings.Contains(rec.Body.String(), "old") {
t.Errorf("body = %s, want the rejection, not cached records", rec.Body.String())
}
if serving, _, _ := srv.stale.snapshot(); serving {
t.Error("a refused query must not mark the cache as serving stale")
}
}
// An outage still falls back to the stale copy, unchanged by the replay path.
func TestHandler_OutageStillFallsBackToStale(t *testing.T) {
a := newFakeBackend(t, `[]`, `[`+fact("h1", "role", "old", "")+`]`)
b := newFakeBackend(t, `[]`, `[]`)
srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
if warm := doGet(t, srv.Handler(), factsPath, ""); warm.Code != http.StatusOK {
t.Fatalf("warm-up status %d", warm.Code)
}
clk.advance(31 * time.Second)
a.fail, b.fail = true, true
rec := doGet(t, srv.Handler(), factsPath, "")
if rec.Code != http.StatusOK || !strings.Contains(rec.Body.String(), "old") {
t.Fatalf("stale fallback = %d %s", rec.Code, rec.Body.String())
}
}
func TestWriteUpstreamError_RedactsBackendAddresses(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
leak := "connection to " + a.srv.URL + "/pdb/query/v4/nodes refused"
a.reject, a.rejectBody = http.StatusBadRequest, leak
b.reject, b.rejectBody = http.StatusBadRequest, leak
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGet(t, srv.Handler(), nodesPath, "")
body := rec.Body.String()
if strings.Contains(body, a.srv.URL) || strings.Contains(body, hostOf(t, a.srv.URL)) {
t.Fatalf("replayed body names a backend: %q", body)
}
if !strings.Contains(body, redactedBackend) {
t.Errorf("body = %q, want the address redacted", body)
}
}
// The pass-through path replays upstream errors too, so it redacts the same way.
func TestProxyUnmerged_RedactsBackendAddresses(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusBadRequest, "upstream "+a.srv.URL+" said no"
b.reject, b.rejectBody = http.StatusBadRequest, "upstream "+b.srv.URL+" said no"
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGet(t, srv.Handler(), resourcesPath, `["=","type","File"]`)
if rec.Code != http.StatusBadRequest {
t.Fatalf("status = %d, want the upstream 400", 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 TestWriteUpstreamError_EmptyBodyGetsAMessage(t *testing.T) {
a := newFakeBackend(t, `[]`, `[]`)
b := newFakeBackend(t, `[]`, `[]`)
a.reject, a.rejectBody = http.StatusBadRequest, ""
b.reject, b.rejectBody = http.StatusBadRequest, ""
srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
rec := doGet(t, srv.Handler(), nodesPath, "")
if rec.Code != http.StatusBadRequest {
t.Fatalf("status = %d, want 400", rec.Code)
}
if strings.TrimSpace(rec.Body.String()) == "" {
t.Error("a replayed rejection must carry some explanation")
}
}
func hostOf(t *testing.T, raw string) string {
t.Helper()
u, err := url.Parse(raw)
if err != nil {
t.Fatalf("parsing %q: %v", raw, err)
}
return u.Host
}