PostgreSQL v14 compatibility campaign

adminapi/adminapi.go

@@ -0,0 +1,147 @@
+// Package adminapi is the admin listener PostgREST runs next to the API when
+// admin-server-port is set: GET /live and /ready for orchestrator probes,
+// GET /schema_cache for the loaded cache, and GET /metrics in Prometheus text
+// format. The endpoints, paths, and status codes mirror PostgREST v14's
+// admin server (PostgREST.Admin): /live is 200 while the API listener accepts
+// connections and 500 otherwise; /ready adds the backend and schema cache
+// health and degrades to 503; any other path is 404 with an empty body.
+//
+// Spec 20 sketches a POST /schema_cache for an on-demand reload; upstream has
+// no such endpoint (reload is SIGUSR1 or NOTIFY), so it is not served here.
+// If the reload entry point lands it belongs next to the signal handler, not
+// in this package's GET-only surface.
+package adminapi
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+)
+
+// Server serves the admin endpoints. The health checks are injected by the
+// command, which knows the API listener address and owns the backend; the
+// zero value of any field degrades gracefully (a nil check reports healthy,
+// a nil SchemaCache serves an empty body, matching upstream's "no cache yet").
+type Server struct {
+	// Live reports whether the API listener accepts connections. PostgREST
+	// implements this as a TCP dial of its own socket; the command wires the
+	// same here.
+	Live func(ctx context.Context) error
+
+	// Ready reports whether the backend connection and the schema cache are
+	// usable. It is consulted in addition to Live.
+	Ready func(ctx context.Context) error
+
+	// SchemaCache returns the loaded schema cache rendered as JSON.
+	SchemaCache func() ([]byte, error)
+
+	// Metrics holds the counters rendered at /metrics.
+	Metrics *Metrics
+}
+
+func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	switch strings.TrimSuffix(r.URL.Path, "/") {
+	case "/live":
+		w.WriteHeader(s.liveStatus(r.Context()))
+	case "/ready":
+		w.WriteHeader(s.readyStatus(r.Context()))
+	case "/schema_cache":
+		body, err := s.schemaCacheJSON()
+		if err != nil {
+			w.WriteHeader(http.StatusInternalServerError)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json; charset=utf-8")
+		w.WriteHeader(http.StatusOK)
+		w.Write(body)
+	case "/metrics":
+		w.Header().Set("Content-Type", "text/plain; charset=utf-8")
+		w.WriteHeader(http.StatusOK)
+		if s.Metrics != nil {
+			w.Write([]byte(s.Metrics.Text()))
+		}
+	default:
+		w.WriteHeader(http.StatusNotFound)
+	}
+}
+
+func (s *Server) liveStatus(ctx context.Context) int {
+	if s.Live != nil && s.Live(ctx) != nil {
+		return http.StatusInternalServerError
+	}
+	return http.StatusOK
+}
+
+func (s *Server) readyStatus(ctx context.Context) int {
+	if status := s.liveStatus(ctx); status != http.StatusOK {
+		return status
+	}
+	if s.Ready != nil && s.Ready(ctx) != nil {
+		return http.StatusServiceUnavailable
+	}
+	return http.StatusOK
+}
+
+func (s *Server) schemaCacheJSON() ([]byte, error) {
+	if s.SchemaCache == nil {
+		return nil, nil
+	}
+	return s.SchemaCache()
+}
+
+// Metrics is a small Prometheus-text registry covering what dbrest measures
+// today: schema cache loads and the configured pool ceiling. The names follow
+// PostgREST's metric names where the concept matches.
+type Metrics struct {
+	mu              sync.Mutex
+	loads           map[string]int64 // by status label: SUCCESS / FAIL
+	lastLoadSeconds float64
+	poolMax         int
+}
+
+// NewMetrics builds a registry; poolMax is the db-pool setting.
+func NewMetrics(poolMax int) *Metrics {
+	return &Metrics{loads: map[string]int64{}, poolMax: poolMax}
+}
+
+// ObserveSchemaCacheLoad records one schema cache load attempt.
+func (m *Metrics) ObserveSchemaCacheLoad(d time.Duration, err error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	status := "SUCCESS"
+	if err != nil {
+		status = "FAIL"
+	}
+	m.loads[status]++
+	if err == nil {
+		m.lastLoadSeconds = d.Seconds()
+	}
+}
+
+// Text renders the registry in the Prometheus text exposition format.
+func (m *Metrics) Text() string {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	var b strings.Builder
+	b.WriteString("# HELP pgrst_schema_cache_query_time_seconds The query time in seconds of the last schema cache load\n")
+	b.WriteString("# TYPE pgrst_schema_cache_query_time_seconds gauge\n")
+	fmt.Fprintf(&b, "pgrst_schema_cache_query_time_seconds %g\n", m.lastLoadSeconds)
+	b.WriteString("# HELP pgrst_schema_cache_loads_total The total number of schema cache loads\n")
+	b.WriteString("# TYPE pgrst_schema_cache_loads_total counter\n")
+	statuses := make([]string, 0, len(m.loads))
+	for status := range m.loads {
+		statuses = append(statuses, status)
+	}
+	sort.Strings(statuses)
+	for _, status := range statuses {
+		fmt.Fprintf(&b, "pgrst_schema_cache_loads_total{status=%q} %d\n", status, m.loads[status])
+	}
+	b.WriteString("# HELP pgrst_db_pool_max Max pool connections\n")
+	b.WriteString("# TYPE pgrst_db_pool_max gauge\n")
+	fmt.Fprintf(&b, "pgrst_db_pool_max %d\n", m.poolMax)
+	return b.String()
+}

adminapi/adminapi_test.go

@@ -0,0 +1,132 @@
+package adminapi
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+)
+
+func get(t *testing.T, s *Server, path string) *http.Response {
+	t.Helper()
+	rec := httptest.NewRecorder()
+	s.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, path, nil))
+	return rec.Result()
+}
+
+// TestLive covers both sides of the liveness probe: 200 while the API socket
+// answers, 500 once it does not, matching the PostgREST admin server.
+func TestLive(t *testing.T) {
+	up := &Server{Live: func(context.Context) error { return nil }}
+	if resp := get(t, up, "/live"); resp.StatusCode != http.StatusOK {
+		t.Errorf("live up: status = %d, want 200", resp.StatusCode)
+	}
+	down := &Server{Live: func(context.Context) error { return errors.New("refused") }}
+	if resp := get(t, down, "/live"); resp.StatusCode != http.StatusInternalServerError {
+		t.Errorf("live down: status = %d, want 500", resp.StatusCode)
+	}
+}
+
+// TestReady covers the three readiness answers: 500 when the API is not
+// reachable, 503 when it is up but the backend is not usable, 200 otherwise.
+func TestReady(t *testing.T) {
+	ok := func(context.Context) error { return nil }
+	bad := func(context.Context) error { return errors.New("down") }
+
+	cases := []struct {
+		name string
+		srv  *Server
+		want int
+	}{
+		{"loaded", &Server{Live: ok, Ready: ok}, http.StatusOK},
+		{"backend pending", &Server{Live: ok, Ready: bad}, http.StatusServiceUnavailable},
+		{"api unreachable", &Server{Live: bad, Ready: ok}, http.StatusInternalServerError},
+	}
+	for _, tc := range cases {
+		if resp := get(t, tc.srv, "/ready"); resp.StatusCode != tc.want {
+			t.Errorf("%s: status = %d, want %d", tc.name, resp.StatusCode, tc.want)
+		}
+	}
+}
+
+// TestSchemaCache checks the dump is served as JSON, and that a failing dump
+// degrades to 500 rather than half a body.
+func TestSchemaCache(t *testing.T) {
+	srv := &Server{SchemaCache: func() ([]byte, error) {
+		return json.Marshal(map[string]any{"relations": []string{"films"}})
+	}}
+	resp := get(t, srv, "/schema_cache")
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("status = %d, want 200", resp.StatusCode)
+	}
+	if ct := resp.Header.Get("Content-Type"); !strings.HasPrefix(ct, "application/json") {
+		t.Errorf("Content-Type = %q, want application/json", ct)
+	}
+	var body map[string]any
+	if err := json.NewDecoder(resp.Body).Decode(&body); err != nil {
+		t.Fatalf("decode: %v", err)
+	}
+
+	broken := &Server{SchemaCache: func() ([]byte, error) { return nil, errors.New("nope") }}
+	if resp := get(t, broken, "/schema_cache"); resp.StatusCode != http.StatusInternalServerError {
+		t.Errorf("broken dump: status = %d, want 500", resp.StatusCode)
+	}
+}
+
+// TestMetrics checks the Prometheus text rendering: content type, the load
+// counters by status, the last query time, and the pool gauge.
+func TestMetrics(t *testing.T) {
+	m := NewMetrics(10)
+	m.ObserveSchemaCacheLoad(250*time.Millisecond, nil)
+	m.ObserveSchemaCacheLoad(0, errors.New("introspect failed"))
+	srv := &Server{Metrics: m}
+
+	resp := get(t, srv, "/metrics")
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("status = %d, want 200", resp.StatusCode)
+	}
+	if ct := resp.Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/plain") {
+		t.Errorf("Content-Type = %q, want text/plain", ct)
+	}
+	raw, err := io.ReadAll(resp.Body)
+	if err != nil {
+		t.Fatalf("read body: %v", err)
+	}
+	body := string(raw)
+	for _, want := range []string{
+		`pgrst_schema_cache_loads_total{status="SUCCESS"} 1`,
+		`pgrst_schema_cache_loads_total{status="FAIL"} 1`,
+		"pgrst_schema_cache_query_time_seconds 0.25",
+		"pgrst_db_pool_max 10",
+	} {
+		if !strings.Contains(body, want) {
+			t.Errorf("metrics body missing %q\n%s", want, body)
+		}
+	}
+}
+
+// TestUnknownPathIs404 checks the fall-through, including the root.
+func TestUnknownPathIs404(t *testing.T) {
+	srv := &Server{}
+	for _, path := range []string{"/", "/config", "/live/extra"} {
+		if resp := get(t, srv, path); resp.StatusCode != http.StatusNotFound {
+			t.Errorf("%s: status = %d, want 404", path, resp.StatusCode)
+		}
+	}
+}
+
+// TestNilChecksDegradeGracefully checks the zero-value server: health reports
+// up (nothing to check), the cache dump is empty, metrics body is empty.
+func TestNilChecksDegradeGracefully(t *testing.T) {
+	srv := &Server{}
+	for _, path := range []string{"/live", "/ready", "/schema_cache", "/metrics"} {
+		if resp := get(t, srv, path); resp.StatusCode != http.StatusOK {
+			t.Errorf("%s: status = %d, want 200", path, resp.StatusCode)
+		}
+	}
+}