fix(trino): observe shared pools through their live instances

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`,

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://<service host>: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.
 //

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://<host>: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)
+	}
+}

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 {

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 }