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