From c8e5f4c054d82fc5f0df2d1f5c2cce13d3e68cb8 Mon Sep 17 00:00:00 2001 From: Benjamin Knofe-Vider Date: Mon, 14 Sep 2026 13:36:57 +0200 Subject: [PATCH] feat: expose scoped Trino routing snapshots --- CLAUDE.md | 13 +- README.md | 6 + controlplane/configstore/trino_routing.go | 36 +++++ controlplane/multitenant.go | 2 +- controlplane/read_only_group.go | 7 +- controlplane/read_only_group_test.go | 10 +- controlplane/trino_inputs.go | 2 + controlplane/trino_registry.go | 5 +- controlplane/trino_registry_test.go | 12 +- controlplane/trino_routing_snapshot.go | 95 ++++++++++++ controlplane/trino_routing_snapshot_test.go | 137 ++++++++++++++++++ docs/trino-cells.md | 46 +++++- .../trino_routing_postgres_test.go | 64 ++++++++ tests/mw-dev/e2e/trino-multicell.sh | 17 +++ 14 files changed, 435 insertions(+), 17 deletions(-) create mode 100644 controlplane/configstore/trino_routing.go create mode 100644 controlplane/trino_routing_snapshot.go create mode 100644 controlplane/trino_routing_snapshot_test.go create mode 100644 tests/configstore/trino_routing_postgres_test.go diff --git a/CLAUDE.md b/CLAUDE.md index 251eafc25..ab9f2d99e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1326,13 +1326,13 @@ are load-bearing for those consumers: group in `multitenant.go` behind `admin.AnyTokenAuthMiddleware`): the read-only discovery secret (`--read-only-secret` / `DUCKGRES_READ_ONLY_SECRET`, sent in `X-Duckgres-Internal-Secret`, same - fallback-rotation semantics as the internal secret) works ONLY on these - two GETs; the admin internal secret also works here (operator/debug + + fallback-rotation semantics as the internal secret) works only on these + two GETs and `GET /api/v1/trino/routing-snapshot`; the admin internal secret also works here (operator/debug + rotation window). Never register discovery routes inside the admin `api` group and never accept `readOnlyTokens` anywhere else — external - writer pods carry this credential, and its blast radius must stay "read + writer and Gateway pods carry this credential, and its blast radius must stay "read the tenant list and its connection topology (RDS endpoints, bucket - names, k8s Secret names — never values)". Tripwires: + names, k8s Secret names — never values), plus eligible Trino principal-to-group assignments". Tripwires: `TestAnyTokenAuthMiddlewareScoping` (token matrix incl. cross-surface rejection) and `TestReadOnlyGroupTopology` (the group's exact route set, against the real `registerReadOnlyGroup` wiring). A discovery @@ -1688,6 +1688,11 @@ password/tenant/catalog changes never propagate. Admin-only initial selection runs before first enablement and refuses changes to any already owned warehouse, including a disabled one. No maintenance move, capacity model, rebalancer, drain, or Gateway routing controller is included. + The machine-authenticated routing snapshot exports eligible principal-to-group + assignments through a fresh bounded database join. It includes only ready, + enabled Trino warehouses with enabled root credentials and configured owners. + It exports no passwords or hashes. Gateway consumes this read-only snapshot; + polling, request routing, and transaction ownership live in Gateway. See [docs/trino-cells.md](docs/trino-cells.md) for configuration and recovery. - **Blue/green projections share one logical cell's namespace**, but internal communication Secrets remain distinct, chart-owned read-only references. diff --git a/README.md b/README.md index cb4936b4f..36c1c9a55 100644 --- a/README.md +++ b/README.md @@ -65,6 +65,12 @@ A PostgreSQL wire protocol compatible server backed by DuckDB. Connect with any ## Trino API identity +Gateway can obtain eligible principal-to-routing-group assignments from the +machine-only `GET /api/v1/trino/routing-snapshot`, using the scoped read-only +token rather than an admin credential. See the +[routing snapshot contract](docs/trino-cells.md#gateway-routing-snapshot) for +eligibility, limits, refresh guidance, and failure recovery. + `DUCKGRES_TRINO_HOGLAKE_URI` defaults to empty (DuckLake catalog provisioning). The frozen perf deployments set it automatically to their namespace-local Hoglake service, replacing the Trino backend in the existing cached and uncached scenarios. diff --git a/controlplane/configstore/trino_routing.go b/controlplane/configstore/trino_routing.go new file mode 100644 index 000000000..377148ddf --- /dev/null +++ b/controlplane/configstore/trino_routing.go @@ -0,0 +1,36 @@ +package configstore + +import ( + "context" + "errors" +) + +const MaxTrinoRoutingPrincipals = 100000 + +// TrinoRoutingPrincipal contains only the principal and durable cell ownership. +type TrinoRoutingPrincipal struct { + Principal string + CellID string +} + +// ListTrinoRoutingPrincipals reads eligible assignments in one database snapshot. +// Passwords are eligibility predicates only; the query never selects credentials. +func (cs *ConfigStore) ListTrinoRoutingPrincipals(ctx context.Context) ([]TrinoRoutingPrincipal, error) { + var rows []TrinoRoutingPrincipal + err := cs.db.WithContext(ctx).Table("duckgres_managed_warehouse_trino AS t"). + Select("o.database_name AS principal, t.trino_cell_id AS cell_id"). + Joins("INNER JOIN duckgres_orgs AS o ON o.name = t.org_id"). + Joins("INNER JOIN duckgres_managed_warehouses AS w ON w.org_id = t.org_id"). + Joins("INNER JOIN duckgres_org_users AS u ON u.org_id = t.org_id AND u.username = 'root'"). + Where("t.enabled = ? AND t.state = ? AND w.state = ?", true, ManagedWarehouseStateReady, ManagedWarehouseStateReady). + Where("u.disabled = ? AND u.password <> ''", false). + Where("o.database_name <> '' AND t.trino_cell_id <> ''"). + Order("o.database_name ASC").Limit(MaxTrinoRoutingPrincipals + 1).Scan(&rows).Error + if err != nil { + return nil, err + } + if len(rows) > MaxTrinoRoutingPrincipals { + return nil, errors.New("trino routing snapshot exceeds principal limit") + } + return rows, nil +} diff --git a/controlplane/multitenant.go b/controlplane/multitenant.go index 1f7f05148..893842960 100644 --- a/controlplane/multitenant.go +++ b/controlplane/multitenant.go @@ -753,7 +753,7 @@ func SetupMultiTenant( provisioning.RegisterAPIWithTrinoAdmission(api, gormStore, gormStore, cfg.DucklingBucketSuffix, liveFetcher, ingressSuffix, trinoCells.enablementCheck(store)) // Discovery endpoints live in their OWN group (see discovery_group.go // for the security rationale and the topology tripwire test). - registerReadOnlyGroup(engine, readOnlyTokens, adminTokens, provisioning.NewGormStore(store)) + registerReadOnlyGroup(engine, readOnlyTokens, adminTokens, provisioning.NewGormStore(store), newTrinoRoutingSnapshot(store, trinoCells)) // Pull-based compute-billing API (GET /billing/usage + POST /billing/ack). // The billing service authenticates with the internal secret (→ admin); // RequireAdmin keeps SSO viewers away from raw usage + the ack mutation. diff --git a/controlplane/read_only_group.go b/controlplane/read_only_group.go index 1301250e9..e6bd42550 100644 --- a/controlplane/read_only_group.go +++ b/controlplane/read_only_group.go @@ -11,18 +11,19 @@ import ( "github.com/posthog/duckgres/controlplane/provisioning" ) -// registerReadOnlyGroup mounts the read-only discovery endpoints on their +// registerReadOnlyGroup mounts discovery and routing snapshots on their // OWN gin group: token-only auth (no SSO, no roles) accepting the scoped // read-only secret OR the admin internal secret. The read-only secret // grants nothing outside this group — an external writer's pod compromise // must not escalate to the provisioning/admin surface. ALL discovery -// routes go through this function so TestReadOnlyGroupTopology can pin +// and routing routes use this function so TestReadOnlyGroupTopology can pin // the exact surface the discovery credential reaches. -func registerReadOnlyGroup(engine *gin.Engine, readOnlyTokens, adminTokens admin.TokenSet, store provisioning.Store) { +func registerReadOnlyGroup(engine *gin.Engine, readOnlyTokens, adminTokens admin.TokenSet, store provisioning.Store, routing *trinoRoutingSnapshot) { discoveryAPI := engine.Group("/api/v1", admin.AnyTokenAuthMiddleware(readOnlyTokens, adminTokens), ) provisioning.RegisterDiscoveryAPI(discoveryAPI, store) + discoveryAPI.GET("/trino/routing-snapshot", routing.handle) } // validateDistinctReadOnlySecret refuses a read-only secret (or fallback) diff --git a/controlplane/read_only_group_test.go b/controlplane/read_only_group_test.go index f21ed9769..20e306152 100644 --- a/controlplane/read_only_group_test.go +++ b/controlplane/read_only_group_test.go @@ -5,6 +5,7 @@ package controlplane import ( "net/http" "net/http/httptest" + "reflect" "sort" "testing" "time" @@ -60,7 +61,7 @@ func (stubProvisioningStore) LatestConfigChange() (time.Time, error) { return ti // reach, against the REAL wiring multitenant.go uses (registerReadOnlyGroup // is the only mount point for discovery routes). Two assertions: // -// 1. The group registers EXACTLY the two discovery GETs — a new route +// 1. The group registers exactly two discovery GETs and one routing GET. A route // added to the group shows up here and forces a deliberate decision. // 2. The auth matrix on those real routes: discovery and admin tokens // pass, junk and empty fail. (Cross-surface rejection — discovery @@ -73,7 +74,7 @@ func TestReadOnlyGroupTopology(t *testing.T) { readOnlyTokens := admin.NewTokenSet("read-only-secret", nil) engine := gin.New() - registerReadOnlyGroup(engine, readOnlyTokens, adminTokens, stubProvisioningStore{}) + registerReadOnlyGroup(engine, readOnlyTokens, adminTokens, stubProvisioningStore{}, newTrinoRoutingSnapshot(&routingSnapshotStore{}, nil)) var got []string for _, r := range engine.Routes() { @@ -81,10 +82,11 @@ func TestReadOnlyGroupTopology(t *testing.T) { } sort.Strings(got) want := []string{ + "GET /api/v1/trino/routing-snapshot", "GET /api/v1/warehouse-team-ids", "GET /api/v1/warehouses", } - if len(got) != len(want) || got[0] != want[0] || got[1] != want[1] { + if !reflect.DeepEqual(got, want) { t.Fatalf("discovery group routes = %v, want exactly %v — a new route on this group extends what the discovery credential can reach; move it or update this test deliberately", got, want) } @@ -97,7 +99,7 @@ func TestReadOnlyGroupTopology(t *testing.T) { engine.ServeHTTP(rec, req) return rec.Code } - for _, path := range []string{"/api/v1/warehouses", "/api/v1/warehouse-team-ids"} { + for _, path := range []string{"/api/v1/warehouses", "/api/v1/warehouse-team-ids", "/api/v1/trino/routing-snapshot"} { if code := serve(path, "read-only-secret"); code != http.StatusOK { t.Errorf("%s with discovery token: %d, want 200", path, code) } diff --git a/controlplane/trino_inputs.go b/controlplane/trino_inputs.go index 72940fc38..dd4db0b2c 100644 --- a/controlplane/trino_inputs.go +++ b/controlplane/trino_inputs.go @@ -129,6 +129,7 @@ func trinoProvisionerEnabled() bool { type trinoCell struct { ID string PublicID string + RoutingGroup string Namespace string Backends []trinoRegisteredBackend CoordinatorURL string @@ -165,6 +166,7 @@ func resolveTrinoCell() (trinoCell, error) { } return trinoCell{ ID: cellID, + RoutingGroup: "legacy", Namespace: strings.TrimSpace(os.Getenv(envTrinoNamespace)), CoordinatorURL: coordinatorURL, TLSServerName: strings.TrimSpace(os.Getenv(envTrinoCoordinatorServerName)), diff --git a/controlplane/trino_registry.go b/controlplane/trino_registry.go index e44ecadda..c24147aea 100644 --- a/controlplane/trino_registry.go +++ b/controlplane/trino_registry.go @@ -77,10 +77,13 @@ func resolveTrinoCells() ([]trinoCell, error) { } } for _, entry := range registered { + if !registryOnly && entry.RoutingGroup == "legacy" { + return nil, errors.New("registered cell must not share the legacy routing group") + } if !registryOnly && entry.Namespace == legacyNS { return nil, errors.New("registered cell must not share the legacy namespace") } - cell := trinoCell{ID: registeredTrinoCellPrefix + entry.ID, PublicID: entry.ID, Namespace: entry.Namespace, ClientURL: entry.ClientURL, Backends: entry.Backends} + cell := trinoCell{ID: registeredTrinoCellPrefix + entry.ID, PublicID: entry.ID, RoutingGroup: entry.RoutingGroup, Namespace: entry.Namespace, ClientURL: entry.ClientURL, Backends: entry.Backends} for _, backend := range entry.Backends { endpoint, _ := trinoEndpointKey(backend.CoordinatorURL) if !registryOnly && endpoint == legacyEndpoint { diff --git a/controlplane/trino_registry_test.go b/controlplane/trino_registry_test.go index 8b3259b12..03f365cf2 100644 --- a/controlplane/trino_registry_test.go +++ b/controlplane/trino_registry_test.go @@ -11,7 +11,8 @@ import ( func TestTrinoRegistryRuntimePreservesLegacyAndSkipsStoppedBackend(t *testing.T) { path := filepath.Join(t.TempDir(), "cells.json") - if err := os.WriteFile(path, []byte(testTrinoRegistryJSON), 0600); err != nil { + registry := strings.Replace(testTrinoRegistryJSON, `"routing_group":"cell-test"`, `"routing_group":"pool-a"`, 1) + if err := os.WriteFile(path, []byte(registry), 0600); err != nil { t.Fatal(err) } t.Setenv(envTrinoCellsFile, path) @@ -31,6 +32,15 @@ func TestTrinoRegistryRuntimePreservesLegacyAndSkipsStoppedBackend(t *testing.T) if cells[1].CoordinatorURL != "https://blue.example.test" || len(cells[1].Backends) != 2 { t.Fatal("runtime discarded blue or stopped green") } + if cells[0].RoutingGroup != "legacy" || cells[1].RoutingGroup != "pool-a" { + t.Fatalf("runtime routing group mismatch: %+v", cells) + } + if err := os.WriteFile(path, []byte(strings.Replace(registry, `"routing_group":"pool-a"`, `"routing_group":"legacy"`, 1)), 0600); err != nil { + t.Fatal(err) + } + if _, err := resolveTrinoCells(); err == nil { + t.Fatal("registered routing group must not alias the legacy group") + } t.Setenv(envTrinoCoordinatorURL, "") if _, err := resolveTrinoCells(); err == nil { t.Fatal("registry silently removed legacy") diff --git a/controlplane/trino_routing_snapshot.go b/controlplane/trino_routing_snapshot.go new file mode 100644 index 000000000..1f70c10ba --- /dev/null +++ b/controlplane/trino_routing_snapshot.go @@ -0,0 +1,95 @@ +//go:build kubernetes + +package controlplane + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "sort" + "strings" + "time" + "unicode/utf8" + + "github.com/gin-gonic/gin" + "github.com/posthog/duckgres/controlplane/configstore" +) + +const maxTrinoRoutingSnapshotBytes = 8 << 20 + +type trinoRoutingStore interface { + ListTrinoRoutingPrincipals(context.Context) ([]configstore.TrinoRoutingPrincipal, error) +} + +type trinoRoutingSnapshot struct { + store trinoRoutingStore + groups map[string]string +} + +func newTrinoRoutingSnapshot(store trinoRoutingStore, cells trinoFleet) *trinoRoutingSnapshot { + groups := make(map[string]string, len(cells)) + for _, cell := range cells { + groups[cell.Cell.ID] = cell.Cell.RoutingGroup + } + return &trinoRoutingSnapshot{store: store, groups: groups} +} + +type trinoRoutingEntry struct { + Principal string `json:"principal"` + RoutingGroup string `json:"routingGroup"` +} + +func (s *trinoRoutingSnapshot) encode(ctx context.Context) ([]byte, error) { + rows, err := s.store.ListTrinoRoutingPrincipals(ctx) + if err != nil { + return nil, err + } + if len(rows) > configstore.MaxTrinoRoutingPrincipals { + return nil, errors.New("too many routing principals") + } + routes := make([]trinoRoutingEntry, 0, len(rows)) + seen := make(map[string]bool, len(rows)) + for _, row := range rows { + if !validTrinoRoutingPrincipal(row.Principal) || seen[row.Principal] { + return nil, errors.New("invalid routing snapshot") + } + seen[row.Principal] = true + group, configured := s.groups[row.CellID] + if !configured || row.CellID == "" { + continue + } + if group == "" { + return nil, errors.New("invalid routing snapshot") + } + routes = append(routes, trinoRoutingEntry{Principal: row.Principal, RoutingGroup: group}) + } + sort.Slice(routes, func(i, j int) bool { return routes[i].Principal < routes[j].Principal }) + data, err := json.Marshal(struct { + Routes []trinoRoutingEntry `json:"routes"` + }{Routes: routes}) + if err != nil { + return nil, err + } + if len(data) > maxTrinoRoutingSnapshotBytes { + return nil, errors.New("routing snapshot exceeds response limit") + } + return data, nil +} + +func validTrinoRoutingPrincipal(principal string) bool { + return utf8.ValidString(principal) && len(principal) <= 1024 && strings.TrimSpace(principal) != "" && + !strings.Contains(principal, ":") && strings.IndexFunc(principal, func(r rune) bool { return r < 32 || r == 127 }) == -1 +} + +func (s *trinoRoutingSnapshot) handle(c *gin.Context) { + c.Header("Cache-Control", "no-store") + ctx, cancel := context.WithTimeout(c.Request.Context(), 3*time.Second) + defer cancel() + data, err := s.encode(ctx) + if err != nil { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": "routing snapshot unavailable"}) + return + } + c.Data(http.StatusOK, "application/json", data) +} diff --git a/controlplane/trino_routing_snapshot_test.go b/controlplane/trino_routing_snapshot_test.go new file mode 100644 index 000000000..98284b463 --- /dev/null +++ b/controlplane/trino_routing_snapshot_test.go @@ -0,0 +1,137 @@ +//go:build kubernetes + +package controlplane + +import ( + "context" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/gin-gonic/gin" + "github.com/posthog/duckgres/controlplane/admin" + "github.com/posthog/duckgres/controlplane/configstore" +) + +type routingSnapshotStore struct { + rows []configstore.TrinoRoutingPrincipal + err error + calls int +} + +func TestTrinoRoutingSnapshotLimits(t *testing.T) { + for _, tc := range []struct { + name string + count, size int + }{ + {"entry count", configstore.MaxTrinoRoutingPrincipals + 1, 1}, + {"response bytes", 40000, 255}, + } { + t.Run(tc.name, func(t *testing.T) { + rows := make([]configstore.TrinoRoutingPrincipal, tc.count) + for i := range rows { + rows[i] = configstore.TrinoRoutingPrincipal{Principal: fmt.Sprintf("%d%s", i, strings.Repeat("x", tc.size)), CellID: "cell-a"} + } + snapshot := newTrinoRoutingSnapshot(&routingSnapshotStore{rows: rows}, trinoFleet{&trinoWiring{Cell: trinoCell{ID: "cell-a", RoutingGroup: "pool-a"}}}) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if data, err := snapshot.encode(ctx); err == nil || data != nil { + t.Fatal("oversized snapshot did not fail atomically") + } + }) + } +} + +func (s *routingSnapshotStore) ListTrinoRoutingPrincipals(ctx context.Context) ([]configstore.TrinoRoutingPrincipal, error) { + s.calls++ + if _, ok := ctx.Deadline(); !ok { + panic("routing query needs a deadline") + } + return s.rows, s.err +} + +func TestTrinoRoutingSnapshot(t *testing.T) { + gin.SetMode(gin.TestMode) + cells := trinoFleet{ + &trinoWiring{Cell: trinoCell{ID: "cell-001", RoutingGroup: "legacy"}}, + &trinoWiring{Cell: trinoCell{ID: "registered:cell-001", PublicID: "cell-001", RoutingGroup: "pool-a"}}, + } + store := &routingSnapshotStore{rows: []configstore.TrinoRoutingPrincipal{ + {Principal: "warehouse_z", CellID: "cell-001"}, + {Principal: "warehouse_a", CellID: "registered:cell-001"}, + {Principal: "unassigned", CellID: ""}, + {Principal: "unknown", CellID: "unknown"}, + }} + r := gin.New() + registerReadOnlyGroup(r, admin.NewTokenSet("reader", []string{"reader-old"}), admin.NewTokenSet("writer", nil), stubProvisioningStore{}, newTrinoRoutingSnapshot(store, cells)) + r.POST("/api/v1/orgs/:id/trino", admin.APIAuthMiddleware(admin.NewTokenSet("writer", nil)), func(c *gin.Context) { c.Status(200) }) + request := func(token string) *httptest.ResponseRecorder { + req := httptest.NewRequest(http.MethodGet, "/api/v1/trino/routing-snapshot", nil) + req.Header.Set("X-Duckgres-Internal-Secret", token) + rec := httptest.NewRecorder() + r.ServeHTTP(rec, req) + return rec + } + for _, token := range []string{"", "bad"} { + if rec := request(token); rec.Code != 401 { + t.Fatalf("bad token status %d", rec.Code) + } + } + if store.calls != 0 { + t.Fatal("unauthorized request queried store") + } + for _, token := range []string{"reader", "reader-old", "writer"} { + rec := request(token) + if rec.Code != 200 || rec.Body.String() != `{"routes":[{"principal":"warehouse_a","routingGroup":"pool-a"},{"principal":"warehouse_z","routingGroup":"legacy"}]}` { + t.Fatalf("snapshot: %d %s", rec.Code, rec.Body.String()) + } + if rec.Header().Get("Cache-Control") != "no-store" { + t.Fatal("snapshot must not be cached by HTTP intermediaries") + } + } + for _, token := range []string{"reader", "reader-old"} { + req := httptest.NewRequest(http.MethodPost, "/api/v1/orgs/example/trino", nil) + req.Header.Set("X-Duckgres-Internal-Secret", token) + rec := httptest.NewRecorder() + r.ServeHTTP(rec, req) + if rec.Code != 401 { + t.Fatalf("routing credential reached mutation: %d", rec.Code) + } + } + for _, source := range []string{"cookie", "sso"} { + req := httptest.NewRequest(http.MethodGet, "/api/v1/trino/routing-snapshot", nil) + if source == "cookie" { + req.AddCookie(&http.Cookie{Name: "duckgres_admin_token", Value: "writer"}) + } else { + req.Header.Set("X-Amzn-Oidc-Data", "writer") + } + rec := httptest.NewRecorder() + r.ServeHTTP(rec, req) + if rec.Code != 401 { + t.Fatalf("%s was accepted on machine-only endpoint", source) + } + } + store.rows = []configstore.TrinoRoutingPrincipal{{Principal: "warehouse_z", CellID: "cell-001"}, {Principal: "warehouse_z", CellID: "registered:cell-001"}} + if rec := request("reader"); rec.Code != 503 || strings.Contains(rec.Body.String(), "warehouse_z") { + t.Fatalf("duplicate must fail atomically: %d %s", rec.Code, rec.Body.String()) + } + store.err = errors.New("database password private-secret") + if rec := request("reader"); rec.Code != 503 || strings.Contains(rec.Body.String(), "private-secret") { + t.Fatalf("query error leaked or succeeded: %d %s", rec.Code, rec.Body.String()) + } + store.err = nil + for _, principal := range []string{"", " ", "a:b", "a\nb", "a\x7fb", string([]byte{0xff}), strings.Repeat("x", 1025)} { + store.rows = []configstore.TrinoRoutingPrincipal{{Principal: principal, CellID: "cell-001"}} + if rec := request("reader"); rec.Code != 503 { + t.Fatalf("malformed principal was exported: %d", rec.Code) + } + } + store.rows = nil + if rec := request("reader"); rec.Code != 200 || rec.Body.String() != `{"routes":[]}` { + t.Fatalf("empty snapshot: %d %s", rec.Code, rec.Body.String()) + } +} diff --git a/docs/trino-cells.md b/docs/trino-cells.md index b5da376ab..a29a28dd3 100644 --- a/docs/trino-cells.md +++ b/docs/trino-cells.md @@ -99,11 +99,51 @@ The Trino pages provide an explicit cell selector; they never select an arbitrar registered cell. The legacy default remains unchanged when legacy is configured. `client_url` is the opaque client endpoint, not a cell-selection instruction to -customers. This PR does not implement authenticated Gateway assignment lookup. -Do not advertise a shared Gateway URL as usable for a new cell until server-side -routing has been separately wired and tested with an authenticated query. +customers. Gateway routing consumes the snapshot described below. Do not +advertise a shared Gateway URL until its polling integration is deployed and +tested with an authenticated query without a routing header. Coordinator/catalog readiness alone does not prove Gateway reachability. +## Gateway routing snapshot + +`GET /api/v1/trino/routing-snapshot` exports the authoritative assignment map: + +```json +{"routes":[{"principal":"warehouse_a","routingGroup":"cell-001"}]} +``` + +Send `DUCKGRES_READ_ONLY_SECRET` in `X-Duckgres-Internal-Secret`. Its rotation +fallbacks work as on the discovery endpoints. The admin internal token also +works for diagnostics, but Gateway must receive only the read-only credential. +This is a machine-only endpoint: browser cookies and SSO do not authenticate it. +The read-only credential now grants exactly the two discovery GETs and this GET; +it cannot provision, enable Trino, reset passwords, or access the admin API. + +Each request performs one fresh, context-bound database join, without a +control-plane cache. Responses carry `Cache-Control: no-store`. A route requires +an enabled, ready Trino row, a ready warehouse, a present and enabled root user +with a nonempty stored password, and a nonempty database name. The principal is +that database name, not the internal org identifier. Only configured owners are +included. Legacy's stored ownership maps to `legacy`; registered ownership maps +to the registry's explicit `routing_group`, which need not equal its cell ID. +Registered cells cannot reuse `legacy` while a legacy deployment is configured. + +The query selects no passwords or hashes. Responses contain no catalog database, +tenant metadata, or backend endpoints. Unknown/unassigned owners and disabled or +unready warehouses are absent. Database errors, duplicate principals, and size +overflow fail the whole request with HTTP 503; they never publish a partial map. +Limits are 100,000 eligible principals, 8 MiB encoded response, and a three-second +database deadline. Gateway consumers should refresh every five seconds and stop +new admissions once their last successful snapshot reaches 15 seconds of age. +They must not renew snapshot age on a failed refresh or fall back to a default +cell for an unknown principal. Existing query and transaction backend ownership +remains a Gateway responsibility, separate from this new-admission snapshot. + +For a failed refresh, check control-plane/database health and configured cell +ownership. A successful empty response is a valid empty eligible set. Do not +recover by copying assignments, supplying an admin token to Gateway, or adding a +default routing group. This endpoint does not implement warehouse migration. + ## Blue running, green stopped Both backends receive shared authentication, tenant-password, and OPA diff --git a/tests/configstore/trino_routing_postgres_test.go b/tests/configstore/trino_routing_postgres_test.go new file mode 100644 index 000000000..c884d119c --- /dev/null +++ b/tests/configstore/trino_routing_postgres_test.go @@ -0,0 +1,64 @@ +//go:build linux || darwin + +package configstore_test + +import ( + "context" + "testing" + + "github.com/posthog/duckgres/controlplane/configstore" +) + +func TestTrinoRoutingPrincipalsPostgres(t *testing.T) { + store := newIsolatedConfigStore(t) + for _, name := range []string{"ready", "disabled", "trino-pending", "warehouse-pending", "missing-root", "disabled-root", "empty-password", "unassigned"} { + seedTrinoOrg(t, store, name) + if err := store.DB().Model(&configstore.Org{}).Where("name = ?", name).Update("database_name", "principal_"+name).Error; err != nil { + t.Fatal(err) + } + if err := store.DB().Create(&configstore.ManagedWarehouse{OrgID: name, State: configstore.ManagedWarehouseStateReady}).Error; err != nil { + t.Fatal(err) + } + if err := store.EnableTrino(name, configstore.TrinoSettings{}); err != nil { + t.Fatal(err) + } + if err := store.AssignTrinoCell(name, "registered:cell-a"); err != nil { + t.Fatal(err) + } + if err := store.UpdateTrinoState(name, configstore.TrinoStateUpdate{State: configstore.ManagedWarehouseStateReady}); err != nil { + t.Fatal(err) + } + } + changes := []struct { + model any + where string + values map[string]any + }{ + {&configstore.ManagedWarehouseTrino{}, "org_id = 'disabled'", map[string]any{"enabled": false}}, + {&configstore.ManagedWarehouseTrino{}, "org_id = 'trino-pending'", map[string]any{"state": "pending"}}, + {&configstore.ManagedWarehouse{}, "org_id = 'warehouse-pending'", map[string]any{"state": "pending"}}, + {&configstore.OrgUser{}, "org_id = 'disabled-root'", map[string]any{"disabled": true}}, + {&configstore.OrgUser{}, "org_id = 'empty-password'", map[string]any{"password": ""}}, + {&configstore.ManagedWarehouseTrino{}, "org_id = 'unassigned'", map[string]any{"trino_cell_id": ""}}, + } + for _, change := range changes { + if err := store.DB().Model(change.model).Where(change.where).Updates(change.values).Error; err != nil { + t.Fatal(err) + } + } + if err := store.DB().Where("org_id = ?", "missing-root").Delete(&configstore.OrgUser{}).Error; err != nil { + t.Fatal(err) + } + rows, err := store.ListTrinoRoutingPrincipals(context.Background()) + if err != nil { + t.Fatal(err) + } + if len(rows) != 1 || rows[0].Principal != "principal_ready" || rows[0].CellID != "registered:cell-a" { + t.Fatalf("unexpected routes: %+v", rows) + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := store.ListTrinoRoutingPrincipals(ctx); err == nil { + t.Fatal("canceled database read must fail") + } +} diff --git a/tests/mw-dev/e2e/trino-multicell.sh b/tests/mw-dev/e2e/trino-multicell.sh index 105b18394..ff8f65d16 100644 --- a/tests/mw-dev/e2e/trino-multicell.sh +++ b/tests/mw-dev/e2e/trino-multicell.sh @@ -48,6 +48,9 @@ api -X PUT -H 'Content-Type: application/json' -d '{"cell":"cell-test"}' \ || fail "initial cell selection failed" api "$API/api/v1/orgs/$ORG_C/trino" | jq -e '.enabled == false and .assigned == true and .cell.id == "cell-test"' >/dev/null \ || fail "selection must not enable Trino" +api "$API/api/v1/trino/routing-snapshot" | jq -e --arg principal "$DB_C" \ + 'all(.routes[]; .principal != $principal)' >/dev/null \ + || fail "disabled warehouse appeared in routing snapshot" api -X POST -H 'Content-Type: application/json' -d '{"enabled":true,"tier":"free"}' "$API/api/v1/orgs/$ORG_C/trino" >/dev/null wait_cell_ready() { @@ -64,6 +67,17 @@ wait_cell_ready() { fail "registered cell did not reconcile tenant readiness" } wait_cell_ready +snapshot="$(api "$API/api/v1/trino/routing-snapshot")" +printf %s "$snapshot" | jq -e --arg legacy "$DB_A" --arg registered "$DB_C" \ + 'any(.routes[]; .principal == $legacy and .routingGroup == "legacy") and + any(.routes[]; .principal == $registered and .routingGroup == "cell-test") and + all(.routes[]; (keys | sort) == ["principal","routingGroup"])' >/dev/null \ + || fail "routing snapshot did not project ready principals and distinct owning groups" +for token in "" "wrong-token"; do + code="$(curl --connect-timeout 5 --max-time 30 -sS -o /dev/null -w '%{http_code}' \ + -H "X-Duckgres-Internal-Secret: $token" "$API/api/v1/trino/routing-snapshot")" + [ "$code" = 401 ] || fail "routing snapshot accepted invalid credentials" +done wait_cell_auth() { attempt=0 while [ "$attempt" -lt 36 ]; do @@ -158,6 +172,9 @@ for endpoint in "$BLUE_TRINO" "$GREEN_TRINO"; do [ "$(trino_query "$DB_C" "$pw_c" "SELECT COUNT(*), SUM(value) FROM $CAT_C.cell_test.values_test")" = '[[2,18]]' ] \ || fail "registry-only registered query failed" done +api "$API/api/v1/trino/routing-snapshot" | jq -e --arg legacy "$DB_A" --arg registered "$DB_C" \ + 'all(.routes[]; .principal != $legacy) and any(.routes[]; .principal == $registered and .routingGroup == "cell-test")' >/dev/null \ + || fail "registry-only snapshot exposed unconfigured legacy ownership" code="$(curl --connect-timeout 5 --max-time 30 -sS -o /dev/null -w '%{http_code}' -H "$H" -H 'Content-Type: application/json' \ -X POST -d '{"enabled":true,"tier":"free"}' "$API/api/v1/orgs/$ORG_A/trino")" [ "$code" = 409 ] || fail "registry-only enablement accepted legacy ownership"