From eeb44057db210fe4f1c870ebfab8cc7ea2abad61 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sat, 12 Sep 2026 17:58:38 +1000 Subject: [PATCH] Resolve the per-certname routes to the node's owning backend The paths keyed on one certname took the pass-through, so a node both backends hold answered from whichever was configured first while /facts answered from whichever held its newer report. - add a route claiming /pdb/query/v4/{nodes,factsets,catalogs}/ - order the backends for it by the freshness map the /facts merge uses - try the remaining backends after the owner, replaying upstream's 404 when none holds the certname - assert ownership, the upstream 404 body, a failed backend and query forwarding against captured openvoxdb shapes --- README.md | 12 +- certname.go | 81 +++++++++++ certname_test.go | 344 +++++++++++++++++++++++++++++++++++++++++++++++ routes_test.go | 2 + server.go | 13 +- 5 files changed, 446 insertions(+), 6 deletions(-) create mode 100644 certname.go create mode 100644 certname_test.go diff --git a/README.md b/README.md index 3d11bcf..11e2907 100644 --- a/README.md +++ b/README.md @@ -34,6 +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/*` (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. | @@ -264,11 +265,12 @@ such a query correctly for every operator (`not`, `or`, subqueries) and does not pretend to for some. Read the fact from an unfiltered (or `certname`-filtered) `/facts` response, from the path route above, and filter client-side. -**Not covered:** `/factsets` and `/inventory`. Both carry facts, but `pdbmux` -does not merge either today — they take the unmerged pass-through path, where -the answer comes from whichever backend replied first rather than from a merge -winner, so there is no owner to attribute. Injecting there would state a -provenance that isn't true. +**Not covered:** `/factsets`, `/inventory` and the per-certname routes. The +first two carry facts but are not merged today — they take the unmerged +pass-through path, where the answer comes from whichever backend replied first +rather than from a merge winner, so there is no owner to attribute. The +per-certname routes do resolve to an owner, but their bodies are passed through +verbatim rather than rebuilt, so nothing is added to them either. ### Metadata and metrics diff --git a/certname.go b/certname.go new file mode 100644 index 0000000..e369511 --- /dev/null +++ b/certname.go @@ -0,0 +1,81 @@ +package main + +import ( + "context" + "net/http" + "strings" +) + +const ( + factsetsPath = "/pdb/query/v4/factsets" + catalogsPath = "/pdb/query/v4/catalogs" + + certnameRouteName = "/pdb/query/v4/{nodes,factsets,catalogs}/" +) + +// certnamePrefixes are the endpoints whose next path segment is a certname. +// openvoxdb resolves that segment to a parent row before it serves anything +// under it — status-not-found-response for the singleton +// (src/puppetlabs/puppetdb/http/handlers.clj:98-120,252-253,342-343,372 and +// http.clj:238-242) and parent-check for every child path +// (middleware.clj:381-398, mounted at handlers.clj:345-347,255-262,373-380) — +// so each of these paths describes exactly one node's data and 404s when the +// backend holds none of it. +var certnamePrefixes = []string{nodesPath + "/", factsetsPath + "/", catalogsPath + "/"} + +// certnameFor returns the certname a path is keyed on, empty when the path is +// not one of these routes. +func certnameFor(path string) string { + for _, prefix := range certnamePrefixes { + rest, ok := strings.CutPrefix(path, prefix) + if !ok { + continue + } + certname, _, _ := strings.Cut(rest, "/") + return certname + } + return "" +} + +func isCertnameRoute(path string) bool { return certnameFor(path) != "" } + +// serveByOwner answers a path keyed on one certname from the backend that owns +// that certname, so a node resolves here exactly as it resolves in the merged +// collections. Without it the answer is whichever backend is configured first, +// and a node both backends hold reports one backend's facts through /facts and +// the other's through /nodes//facts. +// +// The owner's reply is passed through whole rather than merged: openvoxdb +// answers these paths with a bare object or a parent-check 404, neither of +// 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. +func (s *Server) serveByOwner(w http.ResponseWriter, r *http.Request) { + s.proxyOrdered(w, r, s.ownerFirst(r.Context(), certnameFor(r.URL.Path))) +} + +// ownerFirst puts the backend holding a certname's newest report ahead of the +// rest, from the same freshness map the /facts merge attributes records with. +// Static merge has no per-certname owner, and neither has a certname absent +// from the map, so both keep configured order — the tie-break the merges use. +func (s *Server) ownerFirst(ctx context.Context, certname string) []Backend { + if s.cfg.Merge == mergeStatic || certname == "" { + return s.cfg.Backends + } + owner := s.freshnessMap(ctx, nil)[certname] + if owner == "" { + return s.cfg.Backends + } + ordered := make([]Backend, 0, len(s.cfg.Backends)) + for _, b := range s.cfg.Backends { + if b.Name == owner { + ordered = append(ordered, b) + } + } + for _, b := range s.cfg.Backends { + if b.Name != owner { + ordered = append(ordered, b) + } + } + return ordered +} diff --git a/certname_test.go b/certname_test.go new file mode 100644 index 0000000..576b213 --- /dev/null +++ b/certname_test.go @@ -0,0 +1,344 @@ +package main + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "sync" + "testing" +) + +// The bodies below are the shapes openvoxdb 8.9.0 actually serves, captured +// from a live backend: a bare object for /factsets/ +// (src/puppetlabs/puppetdb/http/handlers.clj:98-108,342-343), a fact array for +// /nodes//facts (handlers.clj:373-376) and a pretty-printed +// {"error": ...} body for a certname the backend has no row for +// (http.clj:238-242 for the singleton, middleware.clj:381-398 for the child +// path, which names the parent — "node" — rather than the child). + +func factsetBody(certname, timestamp, role string) string { + return fmt.Sprintf(`{ + "timestamp" : %q, + "facts" : { + "data" : [ { + "name" : "role", + "value" : %q + } ] + }, + "certname" : %q, + "hash" : "d775f92fb438b96276e721bb09524e6361156b06", + "producer_timestamp" : %q, + "producer" : "compiler.example.net", + "environment" : "develop" +}`, timestamp, role, certname, timestamp) +} + +func nodeFactsBody(certname, role string) string { + return fmt.Sprintf(`[{"certname":%q,"environment":"develop","name":"role","value":%q}]`, certname, role) +} + +func notFoundBody(kind, id string) string { + return fmt.Sprintf("{\n \"error\" : \"No information is known about %s %s\"\n}", kind, id) +} + +// voxBackend answers the per-certname routes as openvoxdb does: the certnames +// it holds get a bare object or a fact array, and every other certname gets the +// parent-check 404. Each backend labels its facts with its own name, so a +// response says which backend answered it. +type voxBackend struct { + srv *httptest.Server + name string + // nodes maps a certname it holds to that node's report_timestamp. + nodes map[string]string + fail bool + + mu sync.Mutex + gotParams map[string]url.Values +} + +func newVoxBackend(t *testing.T, name string, nodes map[string]string) *voxBackend { + t.Helper() + vb := &voxBackend{name: name, nodes: nodes, gotParams: map[string]url.Values{}} + vb.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + vb.mu.Lock() + vb.gotParams[r.URL.Path] = r.URL.Query() + vb.mu.Unlock() + if vb.fail { + http.Error(w, "boom", http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/json;charset=utf-8") + + if certname, ok := strings.CutPrefix(r.URL.Path, factsetsPath+"/"); ok { + ts, held := vb.nodes[certname] + if !held { + vb.notFound(w, "factset", certname) + return + } + _, _ = io.WriteString(w, factsetBody(certname, ts, vb.name)) + return + } + if rest, ok := strings.CutPrefix(r.URL.Path, nodesPath+"/"); ok { + certname, _, _ := strings.Cut(rest, "/") + if _, held := vb.nodes[certname]; !held { + vb.notFound(w, "node", certname) + return + } + _, _ = io.WriteString(w, nodeFactsBody(certname, vb.name)) + return + } + + switch r.URL.Path { + case nodesPath: + _, _ = io.WriteString(w, "["+strings.Join(vb.nodeRecords(), ",")+"]") + case factsPath: + _, _ = io.WriteString(w, "["+strings.Join(vb.factRecords(), ",")+"]") + default: + _, _ = io.WriteString(w, "[]") + } + })) + t.Cleanup(vb.srv.Close) + return vb +} + +func (vb *voxBackend) notFound(w http.ResponseWriter, kind, id string) { + w.WriteHeader(http.StatusNotFound) + _, _ = io.WriteString(w, notFoundBody(kind, id)) +} + +func (vb *voxBackend) nodeRecords() []string { + out := make([]string, 0, len(vb.nodes)) + for certname, ts := range vb.nodes { + out = append(out, node(certname, ts)) + } + return out +} + +func (vb *voxBackend) factRecords() []string { + out := make([]string, 0, len(vb.nodes)) + for certname := range vb.nodes { + out = append(out, fact(certname, "role", vb.name, "")) + } + return out +} + +func (vb *voxBackend) params(path string) url.Values { + vb.mu.Lock() + defer vb.mu.Unlock() + return vb.gotParams[path] +} + +// perNodePaths are the routes a certname resolves on, with the body field that +// names the backend that answered. +var perNodePaths = []struct { + name string + path func(certname string) string +}{ + {"factsets/", func(cn string) string { return factsetsPath + "/" + cn }}, + {"nodes//facts", func(cn string) string { return nodesPath + "/" + cn + "/facts" }}, +} + +func TestCertnameRoutes_ServeTheBackendHoldingTheNode(t *testing.T) { + for _, tc := range []struct { + name string + aNodes map[string]string + bNodes map[string]string + certname string + want string + }{ + { + name: "held only by a", + aNodes: map[string]string{"h1.example.net": "2026-07-01T00:00:00Z"}, + bNodes: map[string]string{}, + certname: "h1.example.net", + want: "a", + }, + { + name: "held only by b", + aNodes: map[string]string{}, + bNodes: map[string]string{"h1.example.net": "2026-07-01T00:00:00Z"}, + certname: "h1.example.net", + want: "b", + }, + { + // b reported later, so b owns the node even though a is configured + // first and would answer a plain pass-through. + name: "held by both, newer report on b", + aNodes: map[string]string{"h1.example.net": "2026-07-01T00:00:00Z"}, + bNodes: map[string]string{"h1.example.net": "2026-07-20T00:00:00Z"}, + certname: "h1.example.net", + want: "b", + }, + { + name: "held by both, newer report on a", + aNodes: map[string]string{"h1.example.net": "2026-07-20T00:00:00Z"}, + bNodes: map[string]string{"h1.example.net": "2026-07-01T00:00:00Z"}, + certname: "h1.example.net", + want: "a", + }, + } { + t.Run(tc.name, func(t *testing.T) { + a := newVoxBackend(t, "a", tc.aNodes) + b := newVoxBackend(t, "b", tc.bNodes) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness)) + + for _, rt := range perNodePaths { + rec := doGet(t, srv.Handler(), rt.path(tc.certname), "") + if rec.Code != http.StatusOK { + t.Fatalf("%s: status %d: %s", rt.name, rec.Code, rec.Body.String()) + } + if got := rec.Body.String(); !strings.Contains(got, `"`+tc.want+`"`) { + t.Errorf("%s answered by the wrong backend, want %q: %s", rt.name, tc.want, got) + } + } + + // The collection merge has to agree: one node cannot report one + // backend's facts here and another's through /facts. + rec := doGet(t, srv.Handler(), factsPath, `["=","certname","`+tc.certname+`"]`) + if got := factRoleValue(t, rec.Body.Bytes(), tc.certname); got != tc.want { + t.Errorf("/facts resolved %s to %q, want %q; the routes disagree", tc.certname, got, tc.want) + } + }) + } +} + +// factRoleValue reads the certname's "role" fact out of a merged /facts body. +func factRoleValue(t *testing.T, body []byte, certname string) string { + t.Helper() + var recs []struct { + Certname string `json:"certname"` + Name string `json:"name"` + Value string `json:"value"` + } + if err := json.Unmarshal(body, &recs); err != nil { + t.Fatalf("unmarshal %s: %v", body, err) + } + for _, rec := range recs { + if rec.Certname == certname && rec.Name == "role" { + return rec.Value + } + } + t.Fatalf("no role fact for %s in %s", certname, body) + return "" +} + +// A certname neither backend holds must still look like openvoxdb: its own 404 +// status and error body, not an empty success. +func TestCertnameRoutes_UnknownCertnameKeepsTheUpstream404(t *testing.T) { + a := newVoxBackend(t, "a", map[string]string{"h1.example.net": "2026-07-01T00:00:00Z"}) + b := newVoxBackend(t, "b", map[string]string{"h2.example.net": "2026-07-01T00:00:00Z"}) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness)) + + for _, tc := range []struct { + path string + want string + }{ + {factsetsPath + "/gone.example.net", notFoundBody("factset", "gone.example.net")}, + {nodesPath + "/gone.example.net/facts", notFoundBody("node", "gone.example.net")}, + } { + rec := doGet(t, srv.Handler(), tc.path, "") + if rec.Code != http.StatusNotFound { + t.Errorf("%s: status %d, want 404: %s", tc.path, rec.Code, rec.Body.String()) + } + if got := rec.Body.String(); got != tc.want { + t.Errorf("%s: body %q, want %q", tc.path, got, tc.want) + } + if got := rec.Header().Get("Content-Type"); got != "application/json;charset=utf-8" { + t.Errorf("%s: content type %q", tc.path, got) + } + } +} + +// A backend that is down is indistinguishable from one that has never heard of +// the node, so the surviving backend still answers. +func TestCertnameRoutes_OneBackendDown(t *testing.T) { + a := newVoxBackend(t, "a", map[string]string{"h1.example.net": "2026-07-20T00:00:00Z"}) + b := newVoxBackend(t, "b", map[string]string{"h1.example.net": "2026-07-01T00:00:00Z"}) + // a owns h1 by freshness, so the fallback is exercised only if a is skipped. + a.fail = true + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness)) + + for _, rt := range perNodePaths { + rec := doGet(t, srv.Handler(), rt.path("h1.example.net"), "") + if rec.Code != http.StatusOK { + t.Fatalf("%s: status %d, want the survivor's 200: %s", rt.name, rec.Code, rec.Body.String()) + } + if !strings.Contains(rec.Body.String(), `"b"`) { + t.Errorf("%s: want b's record, got %s", rt.name, rec.Body.String()) + } + } +} + +// Static merge attributes no certname to a backend, so these routes keep the +// configured order the /facts merge falls back to. +func TestCertnameRoutes_StaticMergeKeepsConfiguredOrder(t *testing.T) { + a := newVoxBackend(t, "a", map[string]string{"h1.example.net": "2026-07-01T00:00:00Z"}) + b := newVoxBackend(t, "b", map[string]string{"h1.example.net": "2026-07-20T00:00:00Z"}) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) + + rec := doGet(t, srv.Handler(), factsetsPath+"/h1.example.net", "") + if !strings.Contains(rec.Body.String(), `"a"`) { + t.Errorf("static merge should keep the first configured backend, got %s", rec.Body.String()) + } +} + +// openvoxdb ANDs the request's own query onto the certname restriction +// (src/puppetlabs/puppetdb/http/query.clj:136-143,158-164), so the owner has to +// receive it unchanged. +func TestCertnameRoutes_ForwardQueryToTheOwner(t *testing.T) { + a := newVoxBackend(t, "a", map[string]string{"h1.example.net": "2026-07-20T00:00:00Z"}) + b := newVoxBackend(t, "b", map[string]string{"h1.example.net": "2026-07-01T00:00:00Z"}) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness)) + + const q = `["=","name","role"]` + path := nodesPath + "/h1.example.net/facts" + if rec := doGet(t, srv.Handler(), path, q); rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + if got := a.params(path).Get("query"); got != q { + t.Errorf("owner got query %q, want %q", got, q) + } + if got := b.params(path).Get("query"); got != "" { + t.Errorf("non-owner was queried with %q; only the owner should be asked", got) + } +} + +func TestCertnameFor(t *testing.T) { + for _, tc := range []struct{ path, want string }{ + {factsetsPath + "/h1.example.net", "h1.example.net"}, + {nodesPath + "/h1.example.net", "h1.example.net"}, + {nodesPath + "/h1.example.net/facts", "h1.example.net"}, + {nodesPath + "/h1.example.net/facts/role/web", "h1.example.net"}, + {nodesPath + "/h1.example.net/resources", "h1.example.net"}, + {catalogsPath + "/h1.example.net/edges", "h1.example.net"}, + // Collections and fact-name drilldowns are merged elsewhere and must + // not be mistaken for a certname. + {nodesPath, ""}, + {nodesPath + "/", ""}, + {factsPath, ""}, + {factsPath + "/role", ""}, + {factsetsPath, ""}, + {reportsPath + "/abc123/events", ""}, + } { + if got := certnameFor(tc.path); got != tc.want { + t.Errorf("certnameFor(%q) = %q, want %q", tc.path, got, tc.want) + } + } +} + +func TestCertnameRoutes_ClaimedByTheRouteTable(t *testing.T) { + for _, path := range []string{ + factsetsPath + "/h1.example.net", + nodesPath + "/h1.example.net", + nodesPath + "/h1.example.net/facts", + catalogsPath + "/h1.example.net", + } { + if got := routeFor(path).name; got != certnameRouteName { + t.Errorf("routeFor(%q) = %q, want %q", path, got, certnameRouteName) + } + } +} diff --git a/routes_test.go b/routes_test.go index 8a3ace1..a72d3b7 100644 --- a/routes_test.go +++ b/routes_test.go @@ -21,6 +21,7 @@ var aggregateProbes = map[string]string{ eventCountsPath: eventCountsPath, aggregateEventCountsPath: aggregateEventCountsPath, reportsPath + "//": reportsPath + "/abc123/events", + certnameRouteName: nodesPath + "/h1/facts", } func TestQueryRoutes_EveryRouteIsProbed(t *testing.T) { @@ -107,6 +108,7 @@ func TestQueryRoutes_UnsummedRoutesAreTheKnownOnes(t *testing.T) { eventCountsPath, factNamesPath, reportsPath + "//", + certnameRouteName, } var got []string for _, rt := range queryRoutes { diff --git a/server.go b/server.go index da1d8ed..6908d6d 100644 --- a/server.go +++ b/server.go @@ -177,6 +177,12 @@ var queryRoutes = []route{ serve: (*Server).serveFirstHolder, unsummed: "one backend holds the report, so nothing is merged across backends", }, + { + name: certnameRouteName, + matches: isCertnameRoute, + serve: (*Server).serveByOwner, + unsummed: "one certname's own data, from the backend that owns it; nothing is merged across backends", + }, } // unmergedRoute answers every path no merged route claims. @@ -907,8 +913,13 @@ func (s *Server) queryBackend(ctx context.Context, b Backend, path string, param // 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. func (s *Server) proxyUnmerged(w http.ResponseWriter, r *http.Request) { + s.proxyOrdered(w, r, s.cfg.Backends) +} + +// 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 - for _, b := range s.cfg.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)