feat: merge /reports and /events across both PuppetDBs

Reports are immutable history, so a node that migrated has reports in the old PuppetDB and the new one; serve the union rather than picking a single owning backend as /facts does. Re-apply order_by/limit/offset over the merged set and sum X-Records, since each backend only orders and pages its own slice.
2026-09-05 11:22:36 +10:00
parent e2e9004784
commit ed2e5b73d6
8 changed files with 1025 additions and 57 deletions
@@ -7,7 +7,10 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"net/url"
+	"slices"
+	"strconv"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 )
@@ -15,26 +18,55 @@ import (
 // fakeBackend is an httptest PuppetDB that returns canned bodies per path and
 // records the query params it received.
 type fakeBackend struct {
-	srv        *httptest.Server
-	nodesBody  string
-	factsBody  string
-	fail       bool          // return 500 for everything
-	delay      time.Duration // artificial latency
-	gotQueries map[string]string
+	srv       *httptest.Server
+	nodesBody string
+	factsBody string
+	// bodies holds extra canned responses keyed by path (reports, events, a
+	// report's sub-resources). A path under /reports/ that is absent from bodies
+	// answers 404, like a PuppetDB that does not hold that report.
+	bodies map[string]string
+	// 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
+
+	mu        sync.Mutex
+	gotParams map[string]url.Values
 }

 func newFakeBackend(t *testing.T, nodesBody, factsBody string) *fakeBackend {
 	t.Helper()
-	fb := &fakeBackend{nodesBody: nodesBody, factsBody: factsBody, gotQueries: map[string]string{}}
+	fb := &fakeBackend{
+		nodesBody: nodesBody,
+		factsBody: factsBody,
+		bodies:    map[string]string{},
+		totals:    map[string]int{},
+		gotParams: map[string]url.Values{},
+	}
 	fb.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		if fb.delay > 0 {
 			time.Sleep(fb.delay)
 		}
-		fb.gotQueries[r.URL.Path] = r.URL.Query().Get("query")
+		fb.mu.Lock()
+		fb.gotParams[r.URL.Path] = r.URL.Query()
+		fb.mu.Unlock()
 		if fb.fail {
 			http.Error(w, "boom", http.StatusInternalServerError)
 			return
 		}
+		if body, ok := fb.bodies[r.URL.Path]; ok {
+			if n, ok := fb.totals[r.URL.Path]; ok && r.URL.Query().Get("include_total") == "true" {
+				w.Header().Set(recordsHeader, strconv.Itoa(n))
+			}
+			w.Header().Set("Content-Type", "application/json")
+			_, _ = io.WriteString(w, truncate(t, body, r.URL.Query().Get("limit")))
+			return
+		}
+		if strings.HasPrefix(r.URL.Path, reportsPath+"/") {
+			http.Error(w, "no report with that hash", http.StatusNotFound)
+			return
+		}
 		w.Header().Set("Content-Type", "application/json")
 		switch r.URL.Path {
 		case nodesPath:
@@ -49,6 +81,43 @@ func newFakeBackend(t *testing.T, nodesBody, factsBody string) *fakeBackend {
 	return fb
 }

+// params returns the query params the backend saw for a path, and whether it was
+// asked for that path at all.
+func (fb *fakeBackend) params(path string) (url.Values, bool) {
+	fb.mu.Lock()
+	defer fb.mu.Unlock()
+	v, ok := fb.gotParams[path]
+	return v, ok
+}
+
+// gotQuery returns the PuppetDB query param the backend saw for a path.
+func (fb *fakeBackend) gotQuery(path string) string {
+	v, _ := fb.params(path)
+	return v.Get("query")
+}
+
+// truncate applies an upstream limit param to a canned JSON array body, the way
+// a real PuppetDB would, so paging tests exercise the proxy's re-paging.
+func truncate(t *testing.T, body, limit string) string {
+	t.Helper()
+	n, err := strconv.Atoi(limit)
+	if err != nil {
+		return body
+	}
+	var raws []json.RawMessage
+	if err := json.Unmarshal([]byte(body), &raws); err != nil {
+		return body
+	}
+	if n < len(raws) {
+		raws = raws[:n]
+	}
+	out, err := json.Marshal(raws)
+	if err != nil {
+		t.Fatalf("re-marshal truncated body: %v", err)
+	}
+	return string(out)
+}
+
 func testConfig(oldURL, newURL, merge string) Config {
 	return Config{
 		Listen:       ":0",
@@ -109,11 +178,11 @@ func TestHandler_QueryPassthrough(t *testing.T) {

 	q := `["=","certname","abc.example.net"]`
 	doGet(t, srv.Handler(), factsPath, q)
-	if old.gotQueries[factsPath] != q {
-		t.Errorf("old backend got query %q, want %q", old.gotQueries[factsPath], q)
+	if old.gotQuery(factsPath) != q {
+		t.Errorf("old backend got query %q, want %q", old.gotQuery(factsPath), q)
 	}
-	if nw.gotQueries[factsPath] != q {
-		t.Errorf("new backend got query %q, want %q", nw.gotQueries[factsPath], q)
+	if nw.gotQuery(factsPath) != q {
+		t.Errorf("new backend got query %q, want %q", nw.gotQuery(factsPath), q)
 	}
 }

@@ -190,23 +259,24 @@ func TestHandler_BothBackendsDown(t *testing.T) {
 }

 func TestHandler_PassThroughToPrimary(t *testing.T) {
-	// A non-merged v4 path (e.g. /reports) goes only to the primary (new).
+	// A non-merged v4 path (e.g. /resources) goes only to the primary (new).
 	old := newFakeBackend(t, `[]`, `[]`)
 	nw := newFakeBackend(t, `[]`, `[]`)
 	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))

-	rec := doGet(t, srv.Handler(), "/pdb/query/v4/reports", `["=","certname","h1"]`)
+	const path = "/pdb/query/v4/resources"
+	rec := doGet(t, srv.Handler(), path, `["=","certname","h1"]`)
 	if rec.Code != http.StatusOK {
 		t.Fatalf("status %d", rec.Code)
 	}
-	if !strings.Contains(rec.Body.String(), "/pdb/query/v4/reports") {
+	if !strings.Contains(rec.Body.String(), path) {
 		t.Errorf("expected pass-through body, got %s", rec.Body.String())
 	}
 	// Only primary (new) should have been queried.
-	if _, hit := old.gotQueries["/pdb/query/v4/reports"]; hit {
+	if _, hit := old.params(path); hit {
 		t.Errorf("non-primary backend should not be queried for pass-through")
 	}
-	if _, hit := nw.gotQueries["/pdb/query/v4/reports"]; !hit {
+	if _, hit := nw.params(path); !hit {
 		t.Errorf("primary backend should be queried for pass-through")
 	}
 }
@@ -284,3 +354,241 @@ func TestFreshnessCache_Reused(t *testing.T) {
 		}
 	}
 }
+
+// doGetParams issues a GET with an arbitrary param set, for the paging/ordering
+// params the reports endpoints accept.
+func doGetParams(t *testing.T, h http.Handler, path string, params url.Values) *httptest.ResponseRecorder {
+	t.Helper()
+	target := path
+	if len(params) > 0 {
+		target += "?" + params.Encode()
+	}
+	req := httptest.NewRequest(http.MethodGet, target, nil)
+	rec := httptest.NewRecorder()
+	h.ServeHTTP(rec, req)
+	return rec
+}
+
+// hashes extracts report hashes from a merged response body, in order.
+func hashes(t *testing.T, body []byte) []string {
+	t.Helper()
+	var raws []json.RawMessage
+	if err := json.Unmarshal(body, &raws); err != nil {
+		t.Fatalf("unmarshal %s: %v", body, err)
+	}
+	return hashesOf(t, raws)
+}
+
+const receiveDesc = `[{"field":"receive_time","order":"desc"}]`
+
+func TestHandler_ReportsUnioned(t *testing.T) {
+	// h1 migrated: its pre-migration reports are in old, later ones in new.
+	// Both must show up, unlike /facts where one backend wins the node.
+	old := newFakeBackend(t, `[]`, `[]`)
+	old.bodies[reportsPath] = `[` + report("h1", "r2", "2026-07-10T00:00:00Z") + `,` +
+		report("h1", "r1", "2026-07-01T00:00:00Z") + `]`
+	nw := newFakeBackend(t, `[]`, `[]`)
+	nw.bodies[reportsPath] = `[` + report("h1", "r4", "2026-07-30T00:00:00Z") + `,` +
+		report("h1", "r3", "2026-07-20T00:00:00Z") + `]`
+	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
+
+	rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{
+		"query":    {`["=","certname","h1"]`},
+		"order_by": {receiveDesc},
+	})
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
+	}
+	got := hashes(t, rec.Body.Bytes())
+	want := []string{"r4", "r3", "r2", "r1"}
+	if !slices.Equal(got, want) {
+		t.Errorf("merged reports = %v, want %v (union re-sorted by receive_time desc)", got, want)
+	}
+}
+
+func TestHandler_ReportsDedupedByHash(t *testing.T) {
+	// A node reporting to both PuppetDBs mid-migration stores the same report
+	// hash in each; the merged view must show it once.
+	dup := report("h1", "r1", "2026-07-01T00:00:00Z")
+	old := newFakeBackend(t, `[]`, `[]`)
+	old.bodies[reportsPath] = `[` + dup + `]`
+	nw := newFakeBackend(t, `[]`, `[]`)
+	nw.bodies[reportsPath] = `[` + dup + `]`
+	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
+
+	rec := doGetParams(t, srv.Handler(), reportsPath, nil)
+	if got := hashes(t, rec.Body.Bytes()); !slices.Equal(got, []string{"r1"}) {
+		t.Errorf("merged reports = %v, want one r1", got)
+	}
+}
+
+func TestHandler_ReportsPagedAcrossBackends(t *testing.T) {
+	old := newFakeBackend(t, `[]`, `[]`)
+	old.bodies[reportsPath] = `[` + report("h1", "r5", "2026-07-05T00:00:00Z") + `,` +
+		report("h1", "r3", "2026-07-03T00:00:00Z") + `,` +
+		report("h1", "r1", "2026-07-01T00:00:00Z") + `]`
+	nw := newFakeBackend(t, `[]`, `[]`)
+	nw.bodies[reportsPath] = `[` + report("h1", "r6", "2026-07-06T00:00:00Z") + `,` +
+		report("h1", "r4", "2026-07-04T00:00:00Z") + `,` +
+		report("h1", "r2", "2026-07-02T00:00:00Z") + `]`
+	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
+
+	rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{
+		"order_by": {receiveDesc},
+		"limit":    {"2"},
+		"offset":   {"2"},
+	})
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
+	}
+	// Globally-ordered page 2 of the union, not each backend's own page 2.
+	if got := hashes(t, rec.Body.Bytes()); !slices.Equal(got, []string{"r4", "r3"}) {
+		t.Errorf("page = %v, want [r4 r3]", got)
+	}
+	// Each backend must be asked for the first offset+limit records so the
+	// merged window is fully covered.
+	for name, fb := range map[string]*fakeBackend{"old": old, "new": nw} {
+		p, ok := fb.params(reportsPath)
+		if !ok {
+			t.Fatalf("%s backend was not queried", name)
+		}
+		if p.Get("limit") != "4" {
+			t.Errorf("%s backend got limit=%q, want 4 (offset+limit)", name, p.Get("limit"))
+		}
+		if p.Has("offset") {
+			t.Errorf("%s backend got offset=%q, want it applied locally instead", name, p.Get("offset"))
+		}
+	}
+}
+
+func TestHandler_ReportsIncludeTotalSummed(t *testing.T) {
+	old := newFakeBackend(t, `[]`, `[]`)
+	old.bodies[reportsPath] = `[` + report("h1", "r1", "2026-07-01T00:00:00Z") + `]`
+	old.totals[reportsPath] = 40
+	nw := newFakeBackend(t, `[]`, `[]`)
+	nw.bodies[reportsPath] = `[` + report("h1", "r2", "2026-07-02T00:00:00Z") + `]`
+	nw.totals[reportsPath] = 60
+	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
+
+	rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{
+		"include_total": {"true"},
+		"limit":         {"1"},
+	})
+	if got := rec.Header().Get(recordsHeader); got != "100" {
+		t.Errorf("%s = %q, want 100 (sum of both backends)", recordsHeader, got)
+	}
+}
+
+func TestHandler_ReportsNoTotalWhenNotRequested(t *testing.T) {
+	old := newFakeBackend(t, `[]`, `[]`)
+	old.bodies[reportsPath] = `[]`
+	old.totals[reportsPath] = 40
+	nw := newFakeBackend(t, `[]`, `[]`)
+	nw.bodies[reportsPath] = `[]`
+	nw.totals[reportsPath] = 60
+	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
+
+	rec := doGetParams(t, srv.Handler(), reportsPath, nil)
+	if got := rec.Header().Get(recordsHeader); got != "" {
+		t.Errorf("%s = %q, want it unset without include_total", recordsHeader, got)
+	}
+}
+
+func TestHandler_ReportsBadPagingParam(t *testing.T) {
+	old := newFakeBackend(t, `[]`, `[]`)
+	nw := newFakeBackend(t, `[]`, `[]`)
+	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
+
+	for _, params := range []url.Values{
+		{"limit": {"lots"}},
+		{"offset": {"-1"}},
+		{"order_by": {"receive_time"}},
+	} {
+		rec := doGetParams(t, srv.Handler(), reportsPath, params)
+		if rec.Code != http.StatusBadRequest {
+			t.Errorf("%v: expected 400, got %d", params, rec.Code)
+		}
+	}
+}
+
+func TestHandler_EventsUnioned(t *testing.T) {
+	// Puppetboard fetches a report's events as /events?query=["=","report",hash],
+	// and the report may live in either backend.
+	old := newFakeBackend(t, `[]`, `[]`)
+	old.bodies[eventsPath] = `[` + event("h1", "r1", "Package[nginx]") + `]`
+	nw := newFakeBackend(t, `[]`, `[]`)
+	nw.bodies[eventsPath] = `[` + event("h1", "r2", "Service[nginx]") + `]`
+	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), eventsPath, `["=","certname","h1"]`)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
+	}
+	body := rec.Body.String()
+	if !strings.Contains(body, "Package[nginx]") || !strings.Contains(body, "Service[nginx]") {
+		t.Errorf("expected both backends' events: %s", body)
+	}
+}
+
+func TestHandler_EventsDedupedByIdentity(t *testing.T) {
+	dup := event("h1", "r1", "Package[nginx]")
+	old := newFakeBackend(t, `[]`, `[]`)
+	old.bodies[eventsPath] = `[` + dup + `]`
+	nw := newFakeBackend(t, `[]`, `[]`)
+	nw.bodies[eventsPath] = `[` + dup + `]`
+	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), eventsPath, "")
+	var got []json.RawMessage
+	if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil {
+		t.Fatal(err)
+	}
+	if len(got) != 1 {
+		t.Errorf("expected the duplicate event once, got %d: %s", len(got), rec.Body.String())
+	}
+}
+
+func TestHandler_ReportSubResourceFromHoldingBackend(t *testing.T) {
+	// Only old holds report r1, so its logs must come from old even though new
+	// is the primary — a pass-through would have 404'd.
+	const path = reportsPath + "/r1/logs"
+	old := newFakeBackend(t, `[]`, `[]`)
+	old.bodies[path] = `[{"level":"notice","message":"from-old"}]`
+	nw := newFakeBackend(t, `[]`, `[]`)
+	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), path, "")
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
+	}
+	if !strings.Contains(rec.Body.String(), "from-old") {
+		t.Errorf("expected old's logs, got %s", rec.Body.String())
+	}
+}
+
+func TestHandler_ReportSubResourceMissingEverywhere(t *testing.T) {
+	old := newFakeBackend(t, `[]`, `[]`)
+	nw := newFakeBackend(t, `[]`, `[]`)
+	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), reportsPath+"/nope/events", "")
+	if rec.Code != http.StatusNotFound {
+		t.Fatalf("expected 404 when no backend holds the report, got %d", rec.Code)
+	}
+}
+
+func TestHandler_ReportsOneBackendDown(t *testing.T) {
+	old := newFakeBackend(t, `[]`, `[]`)
+	old.fail = true
+	nw := newFakeBackend(t, `[]`, `[]`)
+	nw.bodies[reportsPath] = `[` + report("h1", "r1", "2026-07-01T00:00:00Z") + `]`
+	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
+
+	rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{"order_by": {receiveDesc}})
+	if rec.Code != http.StatusOK {
+		t.Fatalf("expected 200 serving the survivor, got %d", rec.Code)
+	}
+	if got := hashes(t, rec.Body.Bytes()); !slices.Equal(got, []string{"r1"}) {
+		t.Errorf("merged reports = %v, want [r1]", got)
+	}
+}