From 121bfacc2fec367cd04042483aa6fa98de0f3a6f Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 13 Sep 2026 13:35:21 +1000 Subject: [PATCH 1/3] Replay every unanimous upstream status, not just 4xx openvoxdb does not reserve 5xx for its own faults: the same malformed query is a 400 on /nodes and a 500 on /facts, and /metrics answers a flat 403, so a 4xx-only replay rule made pdbmux's behaviour depend on the route. The meta and metrics handlers held their own copy of the gateway error and bypassed the replay entirely. - Replay any status from 400 up that every backend agreed on, with the backend's own body and content type. - Keep 502 for backends disagreeing on the status, or a backend that answered nothing at all. - Route /pdb/meta, /metrics and the pass-through path through the same rule as the merged query handlers. - Count a unanimous 5xx as a failed round and let it fall back to a stale cache entry; only a unanimous 4xx stays exempt from both. - Answer successful queries with openvoxdb's application/json;charset=utf-8. --- README.md | 30 +++--- cache_test.go | 16 ++- e2e_upstream_test.go | 68 ++++++++++++- meta.go | 21 +++- meta_test.go | 59 ++++++++++- metrics.go | 8 +- metrics_test.go | 56 ++++++++++- server.go | 68 ++++++------- server_test.go | 12 ++- upstream.go | 99 ++++++++++++------- upstream_test.go | 230 +++++++++++++++++++++++++++++++++++++------ 11 files changed, 541 insertions(+), 126 deletions(-) diff --git a/README.md b/README.md index 11e2907..afb5017 100644 --- a/README.md +++ b/README.md @@ -47,17 +47,24 @@ 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. + +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 +455,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 diff --git a/cache_test.go b/cache_test.go index 1f1eb25..e40672b 100644 --- a/cache_test.go +++ b/cache_test.go @@ -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, "") diff --git a/e2e_upstream_test.go b/e2e_upstream_test.go index ca6ba99..c193b40 100644 --- a/e2e_upstream_test.go +++ b/e2e_upstream_test.go @@ -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") + } +} diff --git a/meta.go b/meta.go index 427e2bd..5621d12 100644 --- a/meta.go +++ b/meta.go @@ -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, upstreamOutcome(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() diff --git a/meta_test.go b/meta_test.go index 6142592..aae28a1 100644 --- a/meta_test.go +++ b/meta_test.go @@ -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"}` diff --git a/metrics.go b/metrics.go index 321364a..5f7d3aa 100644 --- a/metrics.go +++ b/metrics.go @@ -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) } diff --git a/metrics_test.go b/metrics_test.go index 9917119..a52cb07 100644 --- a/metrics_test.go +++ b/metrics_test.go @@ -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, `[]`, `[]`) diff --git a/server.go b/server.go index 6908d6d..60008dd 100644 --- a/server.go +++ b/server.go @@ -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, upstreamOutcome(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)) } @@ -918,11 +929,12 @@ func (s *Server) proxyUnmerged(w http.ResponseWriter, r *http.Request) { // 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 + refusals := 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) + refusals = append(refusals, nil) continue } if resp.StatusCode >= 200 && resp.StatusCode < 300 { @@ -934,27 +946,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, - } - } + refusals = append(refusals, 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, upstreamOutcome(refusals)) } func (s *Server) passThrough(r *http.Request, b Backend) (*http.Response, error) { @@ -1106,7 +1100,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{} } diff --git a/server_test.go b/server_test.go index 2804ad1..c9f982f 100644 --- a/server_test.go +++ b/server_test.go @@ -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, "") diff --git a/upstream.go b/upstream.go index eb2488e..b8b36a2 100644 --- a/upstream.go +++ b/upstream.go @@ -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,8 +82,42 @@ func unanimousClientError(results []backendResult) *upstreamError { return first } +// upstreamOutcome is the error to answer a fan-out with when it produced +// nothing: the backends' own unanimous reply where there is one, and pdbmux's +// gateway error otherwise. Every handler ends here, so no route answers a +// failure differently from any other. +func upstreamOutcome(errs []*upstreamError) error { + if ue := unanimousUpstreamError(errs); ue != nil { + 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. +func clientRefusal(err error) bool { + var ue *upstreamError + return errors.As(err, &ue) && ue != nil && ue.status < 500 +} + // writeUpstreamError answers a fan-out that produced no records. A unanimous -// client-shaped rejection is replayed with the backend's own status and +// upstream error 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 @@ -98,7 +129,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, "every backend answered %d %s\n", ue.status, http.StatusText(ue.status)) return } setContentType(w, ue.contentType) diff --git a/upstream_test.go b/upstream_test.go index 58ecd74..a0a0844 100644 --- a/upstream_test.go +++ b/upstream_test.go @@ -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) + } + } +} -- 2.47.3 From 72380f27e6d91b5439a83dab3549edf14ed29fe8 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 13 Sep 2026 14:11:44 +1000 Subject: [PATCH 2/3] Serve the owner's own answer on the per-certname routes Unanimity is the right rule for a fan-out of peers, but the per-certname routes are not one: a backend that does not hold the certname answers 404 to say so, not to disagree, so requiring it to agree turned the owner's real 500 into a 502 that described neither backend. - Add askOrder, which says whether a set of backends was asked as peers or owner-first, and resolve each round's replies under its own rule. - Serve the first backend that answered on owner-routed paths, so an unreachable owner still falls back rather than collapsing to 502. - Keep unanimity for the merged, meta, metrics and pass-through routes. - Cover the owner routes: owner errors against a non-owner 404, both erroring differently, an unreachable owner, and a non-owner error behind the owner's 200. - Record what clientRefusal's 4xx exemption assumes about client certs. --- README.md | 10 +++++- certname.go | 6 +++- certname_test.go | 91 ++++++++++++++++++++++++++++++++++++++++++++++++ meta.go | 2 +- server.go | 20 ++++++----- upstream.go | 74 ++++++++++++++++++++++++++++++++++----- 6 files changed, 182 insertions(+), 21 deletions(-) diff --git a/README.md b/README.md index afb5017..daca7db 100644 --- a/README.md +++ b/README.md @@ -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//{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}/[/...]` | 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}/[/...]` | 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. | @@ -61,6 +61,14 @@ 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 diff --git a/certname.go b/certname.go index e369511..31ec2ac 100644 --- a/certname.go +++ b/certname.go @@ -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 diff --git a/certname_test.go b/certname_test.go index 576b213..a4249fc 100644 --- a/certname_test.go +++ b/certname_test.go @@ -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) + } + } + }) + } +} diff --git a/meta.go b/meta.go index 5621d12..b44d04e 100644 --- a/meta.go +++ b/meta.go @@ -86,7 +86,7 @@ func (s *Server) aliveRaw(w http.ResponseWriter, results []rawResult, path strin alive = append(alive, res) } if len(alive) == 0 { - s.writeUpstreamError(w, upstreamOutcome(rawUpstreamErrors(results))) + s.writeUpstreamError(w, peerOutcome(rawUpstreamErrors(results))) return nil, false } return alive, true diff --git a/server.go b/server.go index 60008dd..68b4e47 100644 --- a/server.go +++ b/server.go @@ -535,7 +535,7 @@ func (s *Server) serveFirstHolder(w http.ResponseWriter, r *http.Request) { if len(alive) == 0 { // 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, upstreamOutcome(backendUpstreamErrors(results))) + s.writeUpstreamError(w, peerOutcome(backendUpstreamErrors(results))) return } for _, res := range alive { @@ -922,19 +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) { - refusals := make([]*upstreamError, 0, len(backends)) +// 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) - refusals = append(refusals, nil) + replies = append(replies, nil) continue } if resp.StatusCode >= 200 && resp.StatusCode < 300 { @@ -946,9 +948,9 @@ func (s *Server) proxyOrdered(w http.ResponseWriter, r *http.Request, backends [ } body, _ := io.ReadAll(resp.Body) _ = resp.Body.Close() - refusals = append(refusals, newUpstreamError(resp.StatusCode, resp.Header.Get("Content-Type"), body)) + replies = append(replies, newUpstreamError(resp.StatusCode, resp.Header.Get("Content-Type"), body)) } - s.writeUpstreamError(w, upstreamOutcome(refusals)) + s.writeUpstreamError(w, order.outcome(replies)) } func (s *Server) passThrough(r *http.Request, b Backend) (*http.Response, error) { diff --git a/upstream.go b/upstream.go index b8b36a2..16a4afe 100644 --- a/upstream.go +++ b/upstream.go @@ -82,17 +82,66 @@ func unanimousUpstreamError(errs []*upstreamError) *upstreamError { return first } -// upstreamOutcome is the error to answer a fan-out with when it produced -// nothing: the backends' own unanimous reply where there is one, and pdbmux's -// gateway error otherwise. Every handler ends here, so no route answers a -// failure differently from any other. -func upstreamOutcome(errs []*upstreamError) error { +// 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, the backend holding it is asked + // first and the rest only as fallbacks in case it does not answer. 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 { @@ -111,14 +160,21 @@ func backendUpstreamErrors(results []backendResult) []*upstreamError { // 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 fan-out that produced no records. A unanimous -// upstream error is replayed with the backend's own status, content type and -// explanation; anything else is reported as a gateway failure. +// 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) { @@ -129,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, "every backend answered %d %s\n", ue.status, http.StatusText(ue.status)) + _, _ = fmt.Fprintf(w, "upstream answered %d %s\n", ue.status, http.StatusText(ue.status)) return } setContentType(w, ue.contentType) -- 2.47.3 From 85c9088293abd96a6246668b414d2416b260bbe1 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 13 Sep 2026 14:13:44 +1000 Subject: [PATCH 3/3] Say that owner-first covers a certname with no resolved owner --- upstream.go | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/upstream.go b/upstream.go index 16a4afe..aa6ba3f 100644 --- a/upstream.go +++ b/upstream.go @@ -94,12 +94,12 @@ const ( // 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, the backend holding it is asked - // first and the rest only as fallbacks in case it does not answer. 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: 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 ) -- 2.47.3