Merge pull request 'Resolve the per-certname routes to the node's owning backend' (#25) from benvin/merge-per-node-fact-routes into main

Reviewed-on: #25

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/<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/*` (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
 

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}/<certname>"
+)
+
+// 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/<certname>/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
+}

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/<certname>
+// (src/puppetlabs/puppetdb/http/handlers.clj:98-108,342-343), a fact array for
+// /nodes/<certname>/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/<certname>", func(cn string) string { return factsetsPath + "/" + cn }},
+	{"nodes/<certname>/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)
+		}
+	}
+}

routes_test.go

@@ -21,6 +21,7 @@ var aggregateProbes = map[string]string{
 	eventCountsPath:               eventCountsPath,
 	aggregateEventCountsPath:      aggregateEventCountsPath,
 	reportsPath + "/<hash>/<sub>": 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 + "/<hash>/<sub>",
+		certnameRouteName,
 	}
 	var got []string
 	for _, rt := range queryRoutes {

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)