From bd085842b74ed6e7e7c87825eaaba61fe759990f Mon Sep 17 00:00:00 2001 From: James Greenhill Date: Wed, 23 Sep 2026 20:33:55 +0000 Subject: [PATCH] fix(trino): observe shared pools through their live instances A shared-pool cell has no fixed coordinator, but its console observer and usage collector were built from the cell's CoordinatorURL, which is empty. Every pooled org read as unavailable with no advertised connection, and Trino usage on the pool was never collected (#1216). The pool observer lists the pool's instances from the config store on each call and asks every instance that can hold queries on its own Service, declaring the Gateway's forwarded HTTPS hop and mapping the forwarded nextUri back onto the plain Service. Queries and nodes are unioned across members; one unreachable member does not hide the others. Closes #1216. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01Nic9bR3LPTRLk56ntJ4M5Z --- CLAUDE.md | 6 +- controlplane/admin/trino_client.go | 49 +++- .../admin/trino_pool_member_client_test.go | 94 +++++++ controlplane/trino_inputs.go | 15 ++ .../trino_pool_hoglake_wiring_test.go | 2 +- controlplane/trino_pool_observer.go | 237 ++++++++++++++++++ controlplane/trino_pool_observer_test.go | 235 +++++++++++++++++ controlplane/trino_registry.go | 3 + docs/trino-cells.md | 5 + tests/mw-dev/e2e/harness.sh | 21 +- 10 files changed, 663 insertions(+), 4 deletions(-) create mode 100644 controlplane/admin/trino_pool_member_client_test.go create mode 100644 controlplane/trino_pool_observer.go create mode 100644 controlplane/trino_pool_observer_test.go diff --git a/CLAUDE.md b/CLAUDE.md index f7e0f29ba..6ccc940c9 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -994,7 +994,11 @@ impersonation, audit log; sliceable by org + user). Design + decisions: passwords remain write-only provision/reset results. Every read is cached + timeout-bounded and degrades to `available:false` plus a reason rather than erroring the page: the console must render - during exactly the incident it exists for. Unset + during exactly the incident it exists for. A shared-pool cell has no fixed + coordinator, so its observer (`trino_pool_observer.go`) lists the pool's + live instances on every call and fans out to each instance's Service over + forwarded HTTPS; usage metering polls the same observer. Never build a + pool observer from `CoordinatorURL` - it is empty (#1216). Unset legacy URL and registry configuration leaves the routes unregistered. - Touching any of the above → update `controlplane/admin/*_test.go` (esp `authz_test.go`, `kill_switch_test.go`, `operators_api_test.go`, diff --git a/controlplane/admin/trino_client.go b/controlplane/admin/trino_client.go index 1e7fcd1b3..845736442 100644 --- a/controlplane/admin/trino_client.go +++ b/controlplane/admin/trino_client.go @@ -251,6 +251,11 @@ type trinoCoordinatorHTTPClient struct { baseURL string hc *http.Client creds TrinoCredentialSource + // forwardedHTTPS declares the TLS hop the Gateway terminated. A pool + // member serves plain HTTP on its in-cluster Service and, like every + // other client of that Service, must say so rather than have the + // coordinator relax password authentication for cleartext. + forwardedHTTPS bool } // newTrinoCoordinatorClient builds a client with a fixed password. Used by @@ -287,6 +292,19 @@ func NewTrinoCoordinatorClient(baseURL, tlsServerName string, creds TrinoCredent } } +// NewTrinoPoolMemberClient builds an observer client for one shared-pool +// instance, addressed on its in-cluster Service over plain HTTP. TLS for a +// pool terminates at the Gateway, so the request declares that forwarded +// hop the same way the pool operator's own probes do. +func NewTrinoPoolMemberClient(baseURL string, creds TrinoCredentialSource) TrinoCoordinatorClient { + return &trinoCoordinatorHTTPClient{ + baseURL: strings.TrimSuffix(baseURL, "/"), + hc: &http.Client{Timeout: trinoCoordinatorTimeout}, + creds: creds, + forwardedHTTPS: true, + } +} + // do issues one authenticated request and returns the body. Non-2xx is an // error carrying the status, except two cases that are not cell failures: // @@ -316,6 +334,10 @@ func (c *trinoCoordinatorHTTPClient) doURL(ctx context.Context, method, rawURL, req.Header.Set("X-Trino-Source", TrinoAdminSource) req.Header.Set("Accept", "application/json") req.SetBasicAuth(user, password) + if c.forwardedHTTPS { + req.Header.Set("X-Forwarded-Proto", "https") + req.Header.Set("X-Forwarded-Port", "443") + } resp, err := c.hc.Do(req) if err != nil { @@ -645,13 +667,38 @@ func (c *trinoCoordinatorHTTPClient) runStatement(ctx context.Context, sql strin if r.NextURI == "" { return all, nil } - if body, err = c.doURL(ctx, http.MethodGet, r.NextURI, "/v1/statement", nil); err != nil { + next, err := c.continuationURL(r.NextURI) + if err != nil { + return nil, err + } + if body, err = c.doURL(ctx, http.MethodGet, next, "/v1/statement", nil); err != nil { return nil, err } } return nil, fmt.Errorf("statement drain exceeded %d hops without completing", maxDrainHops) } +// continuationURL maps a nextUri back onto the transport this client dials. +// +// A pool member honors the forwarded headers, so its nextUri names the HTTPS +// origin the Gateway would serve (https://:443) rather than the +// plain-HTTP Service the request actually reached. Only a statement path on +// the same host is rewritten; anything else is refused rather than followed. +func (c *trinoCoordinatorHTTPClient) continuationURL(nextURI string) (string, error) { + if !c.forwardedHTTPS { + return nextURI, nil + } + base, baseErr := url.Parse(c.baseURL) + next, err := url.Parse(nextURI) + if baseErr != nil || err != nil || next.User != nil || + !strings.EqualFold(next.Hostname(), base.Hostname()) || + !strings.HasPrefix(next.Path, "/v1/statement/") { + return "", errors.New("trino: statement continuation points outside the pool member") + } + next.Scheme, next.Host = base.Scheme, base.Host + return next.String(), nil +} + // announcedNodes reads the ANNOUNCE inventory: the set of node URIs that // have announced themselves to this coordinator. // diff --git a/controlplane/admin/trino_pool_member_client_test.go b/controlplane/admin/trino_pool_member_client_test.go new file mode 100644 index 000000000..d2f797ecb --- /dev/null +++ b/controlplane/admin/trino_pool_member_client_test.go @@ -0,0 +1,94 @@ +//go:build kubernetes + +package admin + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +// A pool member serves plain HTTP behind the Gateway's TLS. Every request, +// including a statement continuation, must declare that forwarded hop, and +// the HTTPS nextUri the coordinator hands back must be followed on the plain +// Service rather than dialled as https://:443. +func TestTrinoPoolMemberClientDeclaresForwardedHTTPSAndRewritesContinuations(t *testing.T) { + var sawPaths []string + var srv *httptest.Server + srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + sawPaths = append(sawPaths, r.Method+" "+r.URL.Path) + if r.Header.Get("X-Forwarded-Proto") != "https" || r.Header.Get("X-Forwarded-Port") != "443" { + t.Errorf("%s %s: missing forwarded HTTPS headers", r.Method, r.URL.Path) + http.Error(w, "insecure", http.StatusForbidden) + return + } + host := strings.Split(strings.TrimPrefix(srv.URL, "http://"), ":")[0] + w.Header().Set("Content-Type", "application/json") + switch { + case r.Method == http.MethodPost && r.URL.Path == "/v1/statement": + _ = json.NewEncoder(w).Encode(map[string]any{ + // What a coordinator with process-forwarded=true returns. + "nextUri": "https://" + host + ":443/v1/statement/executing/q1/abc/1", + }) + case r.Method == http.MethodGet && r.URL.Path == "/v1/statement/executing/q1/abc/1": + _ = json.NewEncoder(w).Encode(map[string]any{"data": [][]any{{"node-1", "http://10.0.0.1:8080", "471", true, "active"}}}) + default: + http.NotFound(w, r) + } + })) + defer srv.Close() + + client := NewTrinoPoolMemberClient(srv.URL, func() (string, string) { return trinoObserverUser, "pw" }).(*trinoCoordinatorHTTPClient) + rows, err := client.runStatement(context.Background(), "SELECT 1") + if err != nil { + t.Fatalf("runStatement: %v", err) + } + if len(rows) != 1 { + t.Fatalf("rows = %v, want the continuation page's one row", rows) + } + want := []string{"POST /v1/statement", "GET /v1/statement/executing/q1/abc/1"} + if strings.Join(sawPaths, ",") != strings.Join(want, ",") { + t.Fatalf("requests = %v, want %v", sawPaths, want) + } +} + +// A continuation that leaves the member (another host, or a non-statement +// path) is refused rather than followed with the observer's credentials. +func TestTrinoPoolMemberClientRefusesForeignContinuation(t *testing.T) { + client := NewTrinoPoolMemberClient("http://cell-1.trino-cells.svc.cluster.local:8080", func() (string, string) { return "u", "p" }).(*trinoCoordinatorHTTPClient) + for _, next := range []string{ + "https://elsewhere.example:443/v1/statement/executing/q/1", + "https://cell-1.trino-cells.svc.cluster.local:443/v1/query/q", + "https://user@cell-1.trino-cells.svc.cluster.local:443/v1/statement/executing/q/1", + } { + if _, err := client.continuationURL(next); err == nil { + t.Errorf("continuationURL(%q) accepted a continuation outside the member", next) + } + } + got, err := client.continuationURL("https://cell-1.trino-cells.svc.cluster.local:443/v1/statement/executing/q/1?x=1") + if err != nil || got != "http://cell-1.trino-cells.svc.cluster.local:8080/v1/statement/executing/q/1?x=1" { + t.Fatalf("continuationURL = %q, %v", got, err) + } +} + +// The fixed-coordinator client is unchanged: no forwarded headers, and the +// nextUri is followed verbatim. +func TestTrinoCoordinatorClientSendsNoForwardedHeaders(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("X-Forwarded-Proto") != "" { + t.Errorf("fixed coordinator client sent X-Forwarded-Proto") + } + _, _ = w.Write([]byte(`{"nodeVersion":{"version":"471"},"environment":"e","coordinator":true}`)) + })) + defer srv.Close() + client := newTrinoCoordinatorClient(srv.URL, "pw", "") + if _, err := client.ServerInfo(context.Background()); err != nil { + t.Fatalf("ServerInfo: %v", err) + } + if next, _ := client.continuationURL("https://x:443/v1/statement/1"); next != "https://x:443/v1/statement/1" { + t.Fatalf("fixed client rewrote a continuation: %q", next) + } +} diff --git a/controlplane/trino_inputs.go b/controlplane/trino_inputs.go index 01703d270..0b666a60d 100644 --- a/controlplane/trino_inputs.go +++ b/controlplane/trino_inputs.go @@ -142,6 +142,10 @@ type trinoCell struct { CoordinatorURL string TLSServerName string ClientURL string + // PoolCoordinatorPort is a shared-pool cell's per-instance coordinator + // Service port. A pool has no CoordinatorURL; its observer reaches each + // instance on this port instead. + PoolCoordinatorPort int32 } // consoleCell preserves legacy ownership and exposes each logical identity. @@ -386,6 +390,17 @@ func buildTrinoCellWiring(store trinoWiringStore, kc kubernetes.Interface, duckl // with a token a real client could match by accident. bundleHandler := opa.NewHandler(bundleStore, opa.BearerTokenAuth(bundleToken)) observer := admin.NewTrinoCoordinatorClient(cell.CoordinatorURL, cell.TLSServerName, trinoProv.ObserverCredential) + if cell.Mode == trinoPoolModeShared { + // A pool has no fixed coordinator, so the client above would dial an + // empty URL: the console would report every pooled org unavailable + // and usage metering would skip the pool (#1216). Observe the pool's + // current instances instead. + lister, ok := store.(trinoPoolInstanceLister) + if !ok { + return nil, fmt.Errorf("Trino cell %s is a shared pool but its store cannot list pool instances", cell.PublicID) + } + observer = newTrinoPoolObserver(cell.ID, cell.Namespace, cell.PoolCoordinatorPort, lister, trinoProv.ObserverCredential) + } observers := []admin.TrinoCoordinatorClient{observer} for _, backend := range cell.Backends { if backend.Running && !backend.RoutingActive { diff --git a/controlplane/trino_pool_hoglake_wiring_test.go b/controlplane/trino_pool_hoglake_wiring_test.go index 1c9d01cc0..9c354e3b2 100644 --- a/controlplane/trino_pool_hoglake_wiring_test.go +++ b/controlplane/trino_pool_hoglake_wiring_test.go @@ -24,7 +24,7 @@ func TestPooledCellCarriesTheManagedHoglakeInputs(t *testing.T) { t.Setenv(envTrinoHoglakeDataPath, "s3://example-bucket/trino/") t.Setenv(envTrinoHoglakeNamespace, "") - store := &fleetBootstrapStore{initialized: map[string]bool{}} + store := &poolObserverWiringStore{fleetBootstrapStore: &fleetBootstrapStore{initialized: map[string]bool{}}} kc := kubefake.NewClientset() ducklings := func(context.Context, string) (*provisioner.DucklingStatus, error) { return nil, nil } storage := func(context.Context, string) (*provisioner.DucklingStatus, error) { return nil, nil } diff --git a/controlplane/trino_pool_observer.go b/controlplane/trino_pool_observer.go new file mode 100644 index 000000000..1bc94cf58 --- /dev/null +++ b/controlplane/trino_pool_observer.go @@ -0,0 +1,237 @@ +//go:build kubernetes + +package controlplane + +import ( + "context" + "errors" + "fmt" + "sort" + "sync" + + "github.com/posthog/duckgres/controlplane/admin" + "github.com/posthog/duckgres/controlplane/configstore" + "github.com/posthog/duckgres/controlplane/trinopool" +) + +// errTrinoPoolNoMembers means the pool has no instance with a running +// coordinator to observe. The console reports it as unavailable with this +// reason instead of a transport error against an empty URL. +var errTrinoPoolNoMembers = errors.New("trino pool has no running member to observe") + +// trinoPoolInstanceLister is the one config-store read the observer needs. +type trinoPoolInstanceLister interface { + ListTrinoPoolInstances(ctx context.Context, poolID string) ([]configstore.TrinoPoolInstance, error) +} + +// trinoPoolObservedPhases are the phases whose coordinator is up and can hold +// queries. DRAINING and SEALED stay observed so a query that started before a +// drain is still seen to finish (usage metering), and SUSPECT stays observed +// because a failing probe is not proof that the process is gone. Nothing +// before ADMITTED has served a tenant. +var trinoPoolObservedPhases = map[string]bool{ + string(trinopool.PhaseAdmitted): true, + string(trinopool.PhaseServing): true, + string(trinopool.PhaseDraining): true, + string(trinopool.PhaseSealed): true, + string(trinopool.PhaseSuspect): true, +} + +// trinoPoolObserver is the console and usage observer for a shared pool. +// +// A pool has no fixed coordinator: its instances are replaced over time. So +// instead of one client built at startup from a URL the pool does not have, +// the observer reads the pool's current instances from the config store on +// every call and asks each instance's own Service. Any control-plane replica +// can do this; it does not need the pool operator's lease. +type trinoPoolObserver struct { + poolID string + namespace string + port int32 + instances trinoPoolInstanceLister + creds admin.TrinoCredentialSource + newClient func(baseURL string, creds admin.TrinoCredentialSource) admin.TrinoCoordinatorClient + + mu sync.Mutex + clients map[string]admin.TrinoCoordinatorClient +} + +func newTrinoPoolObserver(poolID, namespace string, port int32, instances trinoPoolInstanceLister, creds admin.TrinoCredentialSource) *trinoPoolObserver { + return &trinoPoolObserver{ + poolID: poolID, + namespace: namespace, + port: port, + instances: instances, + creds: creds, + newClient: admin.NewTrinoPoolMemberClient, + clients: make(map[string]admin.TrinoCoordinatorClient), + } +} + +type trinoPoolObservedMember struct { + instanceID string + serving bool + client admin.TrinoCoordinatorClient +} + +// members returns the observable instances, serving ones first, in a stable +// order. Clients are cached per instance and dropped once it leaves the set. +func (o *trinoPoolObserver) members(ctx context.Context) ([]trinoPoolObservedMember, error) { + instances, err := o.instances.ListTrinoPoolInstances(ctx, o.poolID) + if err != nil { + return nil, fmt.Errorf("list trino pool instances: %w", err) + } + o.mu.Lock() + defer o.mu.Unlock() + live := make(map[string]bool) + var members []trinoPoolObservedMember + for _, instance := range instances { + if !trinoPoolObservedPhases[instance.Phase] { + continue + } + live[instance.InstanceID] = true + client, ok := o.clients[instance.InstanceID] + if !ok { + // Same address the pool operator probes: the instance's own Service. + endpoint := fmt.Sprintf("http://%s.%s.svc.cluster.local:%d", instance.InstanceID, o.namespace, o.port) + client = o.newClient(endpoint, o.creds) + o.clients[instance.InstanceID] = client + } + members = append(members, trinoPoolObservedMember{ + instanceID: instance.InstanceID, + serving: instance.Phase == string(trinopool.PhaseServing), + client: client, + }) + } + for id := range o.clients { + if !live[id] { + delete(o.clients, id) + } + } + sort.SliceStable(members, func(i, j int) bool { + if members[i].serving != members[j].serving { + return members[i].serving + } + return members[i].instanceID < members[j].instanceID + }) + return members, nil +} + +// Queries is the union across members. One unreachable member does not hide +// the others' queries; only when every member fails is the call an error. +func (o *trinoPoolObserver) Queries(ctx context.Context) ([]admin.TrinoQuery, error) { + members, err := o.members(ctx) + if err != nil { + return nil, err + } + if len(members) == 0 { + return nil, nil + } + var all []admin.TrinoQuery + var firstErr error + succeeded := 0 + for _, member := range members { + queries, err := member.client.Queries(ctx) + if err != nil { + if firstErr == nil { + firstErr = fmt.Errorf("pool member %s: %w", member.instanceID, err) + } + continue + } + succeeded++ + all = append(all, queries...) + } + if succeeded == 0 { + return nil, firstErr + } + return all, nil +} + +// Query asks each member until one holds the query. A member that does not +// know it answers not-found, which is the answer only if nobody else has it. +func (o *trinoPoolObserver) Query(ctx context.Context, queryID string) (*admin.TrinoQuery, error) { + members, err := o.members(ctx) + if err != nil { + return nil, err + } + var lastErr error = errTrinoPoolNoMembers + for _, member := range members { + query, err := member.client.Query(ctx, queryID) + if err == nil { + return query, nil + } + lastErr = err + } + return nil, lastErr +} + +// KillQuery goes to the member that holds the query. Trino query ids are +// generated per coordinator, so at most one member can kill a given id. +func (o *trinoPoolObserver) KillQuery(ctx context.Context, queryID, message string) error { + members, err := o.members(ctx) + if err != nil { + return err + } + var lastErr error = errTrinoPoolNoMembers + for _, member := range members { + if _, err := member.client.Query(ctx, queryID); err != nil { + lastErr = err + continue + } + return member.client.KillQuery(ctx, queryID, message) + } + return lastErr +} + +// Nodes is the union of every member's inventory. The source is reported +// only when all members answered from the same one; a mixed inventory keeps +// the first member's source, which is the least-detailed claim the console +// can make about the rows it received. +func (o *trinoPoolObserver) Nodes(ctx context.Context) (admin.TrinoNodeInventory, error) { + members, err := o.members(ctx) + if err != nil { + return admin.TrinoNodeInventory{}, err + } + if len(members) == 0 { + return admin.TrinoNodeInventory{}, errTrinoPoolNoMembers + } + var inventory admin.TrinoNodeInventory + var firstErr error + succeeded := 0 + for _, member := range members { + nodes, err := member.client.Nodes(ctx) + if err != nil { + if firstErr == nil { + firstErr = fmt.Errorf("pool member %s: %w", member.instanceID, err) + } + continue + } + if succeeded == 0 { + inventory.Source = nodes.Source + } + succeeded++ + inventory.Nodes = append(inventory.Nodes, nodes.Nodes...) + } + if succeeded == 0 { + return admin.TrinoNodeInventory{}, firstErr + } + return inventory, nil +} + +// ServerInfo answers from the first member that responds, serving members +// first. All members run the pool's one release, so any answer describes it. +func (o *trinoPoolObserver) ServerInfo(ctx context.Context) (*admin.TrinoServerInfo, error) { + members, err := o.members(ctx) + if err != nil { + return nil, err + } + var lastErr error = errTrinoPoolNoMembers + for _, member := range members { + info, err := member.client.ServerInfo(ctx) + if err == nil { + return info, nil + } + lastErr = err + } + return nil, lastErr +} diff --git a/controlplane/trino_pool_observer_test.go b/controlplane/trino_pool_observer_test.go new file mode 100644 index 000000000..3b23073e9 --- /dev/null +++ b/controlplane/trino_pool_observer_test.go @@ -0,0 +1,235 @@ +//go:build kubernetes + +package controlplane + +import ( + "context" + "errors" + "sort" + "testing" + + "github.com/posthog/duckgres/controlplane/admin" + "github.com/posthog/duckgres/controlplane/configstore" + "github.com/posthog/duckgres/controlplane/provisioner" + "github.com/posthog/duckgres/controlplane/trinopool" + kubefake "k8s.io/client-go/kubernetes/fake" +) + +type fakePoolInstanceLister struct { + instances []configstore.TrinoPoolInstance + err error + poolIDs []string +} + +func (f *fakePoolInstanceLister) ListTrinoPoolInstances(_ context.Context, poolID string) ([]configstore.TrinoPoolInstance, error) { + f.poolIDs = append(f.poolIDs, poolID) + return f.instances, f.err +} + +type fakePoolMember struct { + queries []admin.TrinoQuery + err error + nodes admin.TrinoNodeInventory + killed []string +} + +func (m *fakePoolMember) Queries(context.Context) ([]admin.TrinoQuery, error) { + return m.queries, m.err +} +func (m *fakePoolMember) Query(_ context.Context, id string) (*admin.TrinoQuery, error) { + if m.err != nil { + return nil, m.err + } + for _, q := range m.queries { + if q.QueryID == id { + q := q + return &q, nil + } + } + return nil, errors.New("not found") +} +func (m *fakePoolMember) KillQuery(_ context.Context, id, _ string) error { + m.killed = append(m.killed, id) + return nil +} +func (m *fakePoolMember) Nodes(context.Context) (admin.TrinoNodeInventory, error) { + return m.nodes, m.err +} +func (m *fakePoolMember) ServerInfo(context.Context) (*admin.TrinoServerInfo, error) { + if m.err != nil { + return nil, m.err + } + return &admin.TrinoServerInfo{Version: "471", Coordinator: true}, nil +} + +func poolObserverFixture(instances []configstore.TrinoPoolInstance, members map[string]*fakePoolMember) (*trinoPoolObserver, *fakePoolInstanceLister, map[string]string) { + lister := &fakePoolInstanceLister{instances: instances} + endpoints := map[string]string{} + o := newTrinoPoolObserver("registered:cell-001", "trino-cells", 8080, lister, func() (string, string) { return "u", "p" }) + o.newClient = func(baseURL string, _ admin.TrinoCredentialSource) admin.TrinoCoordinatorClient { + for id, member := range members { + if baseURL == "http://"+id+".trino-cells.svc.cluster.local:8080" { + endpoints[id] = baseURL + return member + } + } + return &fakePoolMember{err: errors.New("unexpected endpoint " + baseURL)} + } + return o, lister, endpoints +} + +func instance(id string, phase trinopool.Phase) configstore.TrinoPoolInstance { + return configstore.TrinoPoolInstance{InstanceID: id, Phase: string(phase)} +} + +// The pool observer queries every live instance's own Service and unions the +// result, so a pooled org's queries are visible to the console and metered. +func TestTrinoPoolObserverUnionsLiveMembers(t *testing.T) { + members := map[string]*fakePoolMember{ + "cell-a": {queries: []admin.TrinoQuery{{QueryID: "qa"}}}, + "cell-b": {queries: []admin.TrinoQuery{{QueryID: "qb"}}}, + "cell-c": {queries: []admin.TrinoQuery{{QueryID: "qc"}}}, + } + o, lister, endpoints := poolObserverFixture([]configstore.TrinoPoolInstance{ + instance("cell-a", trinopool.PhaseServing), + instance("cell-b", trinopool.PhaseDraining), + instance("cell-c", trinopool.PhaseCreating), // no coordinator yet: not observed + instance("cell-d", trinopool.PhaseRetired), // gone: not observed + }, members) + + queries, err := o.Queries(context.Background()) + if err != nil { + t.Fatalf("Queries: %v", err) + } + var ids []string + for _, q := range queries { + ids = append(ids, q.QueryID) + } + sort.Strings(ids) + if len(ids) != 2 || ids[0] != "qa" || ids[1] != "qb" { + t.Fatalf("queries = %v, want qa and qb (serving + draining only)", ids) + } + if len(endpoints) != 2 { + t.Fatalf("dialled %v, want only the two observed instances", endpoints) + } + if lister.poolIDs[0] != "registered:cell-001" { + t.Fatalf("listed pool %q, want the stored pool id", lister.poolIDs[0]) + } +} + +// One unreachable member must not hide the rest; only a pool where every +// member fails is an error. +func TestTrinoPoolObserverToleratesPartialFailure(t *testing.T) { + members := map[string]*fakePoolMember{ + "cell-a": {err: errors.New("connection refused")}, + "cell-b": {queries: []admin.TrinoQuery{{QueryID: "qb"}}}, + } + o, _, _ := poolObserverFixture([]configstore.TrinoPoolInstance{ + instance("cell-a", trinopool.PhaseServing), + instance("cell-b", trinopool.PhaseServing), + }, members) + queries, err := o.Queries(context.Background()) + if err != nil || len(queries) != 1 || queries[0].QueryID != "qb" { + t.Fatalf("Queries = %v, %v; want qb despite cell-a failing", queries, err) + } + + members["cell-b"].err = errors.New("connection refused") + if _, err := o.Queries(context.Background()); err == nil { + t.Fatal("Queries succeeded with every member failing") + } +} + +// With no running member the console gets a named reason, not a transport +// error against an empty URL (#1216). +func TestTrinoPoolObserverWithoutMembers(t *testing.T) { + o, _, _ := poolObserverFixture(nil, nil) + if queries, err := o.Queries(context.Background()); err != nil || len(queries) != 0 { + t.Fatalf("Queries = %v, %v; want empty and no error", queries, err) + } + if _, err := o.ServerInfo(context.Background()); !errors.Is(err, errTrinoPoolNoMembers) { + t.Fatalf("ServerInfo err = %v, want errTrinoPoolNoMembers", err) + } + if _, err := o.Nodes(context.Background()); !errors.Is(err, errTrinoPoolNoMembers) { + t.Fatalf("Nodes err = %v, want errTrinoPoolNoMembers", err) + } +} + +// A kill goes only to the member that holds the query. +func TestTrinoPoolObserverKillsOnOwningMember(t *testing.T) { + members := map[string]*fakePoolMember{ + "cell-a": {queries: []admin.TrinoQuery{{QueryID: "qa"}}}, + "cell-b": {queries: []admin.TrinoQuery{{QueryID: "qb"}}}, + } + o, _, _ := poolObserverFixture([]configstore.TrinoPoolInstance{ + instance("cell-a", trinopool.PhaseServing), + instance("cell-b", trinopool.PhaseServing), + }, members) + if err := o.KillQuery(context.Background(), "qb", "operator kill"); err != nil { + t.Fatalf("KillQuery: %v", err) + } + if len(members["cell-a"].killed) != 0 || len(members["cell-b"].killed) != 1 { + t.Fatalf("kills a=%v b=%v, want only cell-b", members["cell-a"].killed, members["cell-b"].killed) + } + if err := o.KillQuery(context.Background(), "missing", "x"); err == nil { + t.Fatal("KillQuery of an unknown query succeeded") + } +} + +// A retired instance's client is dropped, so the cache cannot grow without +// bound as the pool replaces members. +func TestTrinoPoolObserverDropsRetiredClients(t *testing.T) { + members := map[string]*fakePoolMember{"cell-a": {}, "cell-b": {}} + o, lister, _ := poolObserverFixture([]configstore.TrinoPoolInstance{ + instance("cell-a", trinopool.PhaseServing), + instance("cell-b", trinopool.PhaseServing), + }, members) + if _, err := o.Queries(context.Background()); err != nil { + t.Fatal(err) + } + lister.instances = []configstore.TrinoPoolInstance{instance("cell-b", trinopool.PhaseServing), instance("cell-a", trinopool.PhaseRetired)} + if _, err := o.Queries(context.Background()); err != nil { + t.Fatal(err) + } + if _, ok := o.clients["cell-a"]; ok || len(o.clients) != 1 { + t.Fatalf("clients = %v, want only cell-b", o.clients) + } +} + +type poolObserverWiringStore struct { + *fleetBootstrapStore + fakePoolInstanceLister +} + +// The wired observer for a pooled cell is the pool observer, for the console +// and for usage metering alike; a fixed cell keeps its coordinator client. +func TestPooledCellWiresThePoolObserver(t *testing.T) { + t.Setenv(envTrinoFilesystemCacheEnabled, "false") + store := &poolObserverWiringStore{fleetBootstrapStore: &fleetBootstrapStore{initialized: map[string]bool{}}} + kc := kubefake.NewClientset() + ducklings := func(context.Context, string) (*provisioner.DucklingStatus, error) { return nil, nil } + + pooled := trinoCell{ID: registeredTrinoCellPrefix + "cell-001", PublicID: "cell-001", Namespace: "trino-cells", Mode: trinoPoolModeShared, PoolCoordinatorPort: 8080} + wire, err := buildTrinoCellWiring(store, kc, ducklings, pooled) + if err != nil { + t.Fatalf("wire pooled cell: %v", err) + } + observer, ok := wire.Console.Observer.(*trinoPoolObserver) + if !ok { + t.Fatalf("console observer is %T, want *trinoPoolObserver", wire.Console.Observer) + } + if observer.poolID != pooled.ID || observer.namespace != "trino-cells" || observer.port != 8080 { + t.Fatalf("observer = %s/%s:%d, want the pool's stored id, namespace and port", observer.poolID, observer.namespace, observer.port) + } + if len(wire.Observers) != 1 || wire.Observers[0] != wire.Console.Observer { + t.Fatalf("usage observers = %v, want exactly the pool observer", wire.Observers) + } + + fixed := trinoCell{ID: "cell-legacy", Namespace: "legacy", CoordinatorURL: "https://legacy.example.test"} + wire, err = buildTrinoCellWiring(store, kc, ducklings, fixed) + if err != nil { + t.Fatalf("wire fixed cell: %v", err) + } + if _, ok := wire.Console.Observer.(*trinoPoolObserver); ok { + t.Fatal("a fixed cell was given the pool observer") + } +} diff --git a/controlplane/trino_registry.go b/controlplane/trino_registry.go index 8647e8946..efc9b17e8 100644 --- a/controlplane/trino_registry.go +++ b/controlplane/trino_registry.go @@ -84,6 +84,9 @@ func resolveTrinoCells() ([]trinoCell, string, error) { return nil, "", errors.New("registered cell must not share the legacy namespace") } cell := trinoCell{Mode: strings.TrimSpace(entry.Mode), ID: registeredTrinoCellPrefix + entry.ID, PublicID: entry.ID, RoutingGroup: entry.RoutingGroup, Namespace: entry.Namespace, ClientURL: entry.ClientURL, Backends: entry.Backends, CatalogManagement: entry.CatalogManagement} + if entry.Pool != nil { + cell.PoolCoordinatorPort = entry.Pool.CoordinatorServicePort + } for _, backend := range entry.Backends { endpoint, _ := trinoEndpointKey(backend.CoordinatorURL) if !registryOnly && endpoint == legacyEndpoint { diff --git a/docs/trino-cells.md b/docs/trino-cells.md index bba8e8ab2..85772ac2f 100644 --- a/docs/trino-cells.md +++ b/docs/trino-cells.md @@ -72,6 +72,11 @@ before deploying the readiness-aware control plane. New-cell OPA sidecars poll `/bundles/trino/` with that namespace's bundle token. Legacy keeps `/bundles/trino`. Tokens cannot read another cell's bundle. The observer credential remains separate from the catalog administrator. +A shared-pool cell has no fixed coordinator, so its observer reads the pool's +current instances from the config store on each call and asks every instance +that can hold queries (ADMITTED, SERVING, DRAINING, SEALED or SUSPECT) on its +own Service, declaring the Gateway's forwarded HTTPS hop. The console and usage +metering see the union; one unreachable member does not hide the others. In registry-only mode, `/bundles/trino` is absent and returns HTTP 404. Missing bundle URLs never fall back to the admin UI page. diff --git a/tests/mw-dev/e2e/harness.sh b/tests/mw-dev/e2e/harness.sh index 33a4cc886..646306310 100755 --- a/tests/mw-dev/e2e/harness.sh +++ b/tests/mw-dev/e2e/harness.sh @@ -1031,13 +1031,15 @@ pool_object_count() { # kind selector # does not provision by default: a registry entry with mode "shared-pool", a # blueprint artifact carrying a real image digest, a Gateway speaking the pooled # protocol, and the feature flags on. When those exist this is the acceptance -# check, in three +# check, in four # separately reported stages, because each is evidence for less than the next: # # [structure] ready, non-terminating coordinator pods, each with its own # Service and its own workers. A ready pod routes nothing. # [admission] the warehouse reports ready, which with the Gateway gate on # means its publication committed on every serving member. +# [observer] the console observes the pool through its live instances and +# advertises the org's connection (#1216). # [query] a statement run with an EXISTING login of that org returns its # result. Needs caller-supplied credentials; skipped, and said to # be skipped, when they are not given. No canary warehouse and no @@ -1139,6 +1141,23 @@ trino_shared_pool_active() { [ "$state" = "ready" ] || fail "shared pool: $org is '$state', want ready once its publication commits" log "shared pool OK [admission]: $org is admitted (publication committed on every serving member)" + # The console must be able to SEE the pool (#1216). A pool has no fixed + # coordinator, so the observer reads the pool's live instances; before that + # it dialled an empty URL, reported every pooled org unavailable and never + # advertised its connection, and usage metering skipped the pool entirely. + # Retried because the observer's first poll can race a member replacement. + a=0 detail="" + while [ "$a" -lt 12 ]; do + detail="$(curl -fsS -H "$H" "$API/api/v1/orgs/$org/trino")" || detail="" + printf %s "$detail" | jq -e '.available == true and ((.status.connection.host // "") != "")' >/dev/null && break + sleep 5; a=$((a + 1)) + done + printf %s "$detail" | jq -e '.available == true' >/dev/null \ + || fail "shared pool: console cannot observe the pool for $org: $(printf %s "$detail" | head -c 400)" + printf %s "$detail" | jq -e '(.status.connection.host // "") != ""' >/dev/null \ + || fail "shared pool: $org is ready and observed but advertises no connection: $(printf %s "$detail" | head -c 400)" + log "shared pool OK [observer]: console observes the pool and advertises $(printf %s "$detail" | jq -r '.status.connection.host') for $org" + # And the end the user actually experiences: a statement, run with the # caller's OWN credentials, returning a result. Structure and admission are # both upstream of this and neither implies it - a member can be admitted and