From bfe28b488de6b215b83a8821fce426590829bc27 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Mon, 7 Sep 2026 22:39:58 +1000 Subject: [PATCH] Replay a unanimous upstream rejection instead of a 502 Every backend gets the same query, so one they all refuse is the client's mistake; flattening it into "all backends failed" threw openvoxdb's own explanation away and logged a typo as an outage. - Carry status, content type and body on a typed upstreamError - Replay the status and explanation when every backend refuses alike - Redact backend addresses from replayed bodies - Keep a refused query out of the partial counters and the cache --- README.md | 13 +- e2e_backend_test.go | 15 +- e2e_upstream_test.go | 58 ++++++ server.go | 34 +++- server_test.go | 8 + upstream.go | 134 ++++++++++++++ upstream_test.go | 425 +++++++++++++++++++++++++++++++++++++++++++ 7 files changed, 676 insertions(+), 11 deletions(-) create mode 100644 e2e_upstream_test.go create mode 100644 upstream.go create mode 100644 upstream_test.go diff --git a/README.md b/README.md index 46005e5..3d11bcf 100644 --- a/README.md +++ b/README.md @@ -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 diff --git a/e2e_backend_test.go b/e2e_backend_test.go index cfa2523..72eefb3 100644 --- a/e2e_backend_test.go +++ b/e2e_backend_test.go @@ -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()) { diff --git a/e2e_upstream_test.go b/e2e_upstream_test.go new file mode 100644 index 0000000..ca6ba99 --- /dev/null +++ b/e2e_upstream_test.go @@ -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) + } + }) + } +} diff --git a/server.go b/server.go index 9ad914c..da1d8ed 100644 --- a/server.go +++ b/server.go @@ -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 { diff --git a/server_test.go b/server_test.go index 7f39cde..2804ad1 100644 --- a/server_test.go +++ b/server_test.go @@ -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)) diff --git a/upstream.go b/upstream.go new file mode 100644 index 0000000..8324976 --- /dev/null +++ b/upstream.go @@ -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 = "" + +// 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 +} diff --git a/upstream_test.go b/upstream_test.go new file mode 100644 index 0000000..8e8f97f --- /dev/null +++ b/upstream_test.go @@ -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 +}