From 49ad7bb504031e65497f5506282c356bad50b7e0 Mon Sep 17 00:00:00 2001
From: eric <eric@posthog.com>
Date: Thu, 1 Oct 2026 14:22:21 -0400
Subject: [PATCH 1/5] feat: mask literal values in Trino query text for tenant
 views

---
 controlplane/admin/trino_sql_mask.go      | 153 ++++++++++++++++++++++
 controlplane/admin/trino_sql_mask_test.go |  51 ++++++++
 2 files changed, 204 insertions(+)
 create mode 100644 controlplane/admin/trino_sql_mask.go
 create mode 100644 controlplane/admin/trino_sql_mask_test.go

diff --git a/controlplane/admin/trino_sql_mask.go b/controlplane/admin/trino_sql_mask.go
new file mode 100644
index 00000000..14d55086
--- /dev/null
+++ b/controlplane/admin/trino_sql_mask.go
@@ -0,0 +1,153 @@
+//go:build kubernetes
+
+package admin
+
+import (
+	"strings"
+	"unicode/utf8"
+)
+
+const (
+	trinoMaskedQueryMaxRunes    = 2000
+	trinoMaskedQueryUnavailable = "(query text unavailable)"
+)
+
+// maskTrinoSQLLiterals returns query with every literal value replaced by `?`
+// and every comment removed, so the shape of a statement can be shown to the
+// tenant's own users without the values they filtered on.
+//
+// It is a lexer, not a parser: it only has to find where literals, comments
+// and quoted identifiers start and end. Input it cannot lex (an unterminated
+// string, comment or quoted identifier) returns a fixed placeholder, because
+// a wrong guess about where a literal ends would print part of it.
+func maskTrinoSQLLiterals(query string) string {
+	var out strings.Builder
+	out.Grow(len(query))
+	lastWasSpace := true // drops leading whitespace
+	emit := func(text string) {
+		out.WriteString(text)
+		lastWasSpace = false
+	}
+	space := func() {
+		if !lastWasSpace {
+			out.WriteByte(' ')
+			lastWasSpace = true
+		}
+	}
+
+	n := len(query)
+	for i := 0; i < n; {
+		c := query[i]
+		switch {
+		case c == ' ' || c == '\t' || c == '\n' || c == '\r' || c == '\f':
+			space()
+			i++
+		case c == '-' && i+1 < n && query[i+1] == '-':
+			for i < n && query[i] != '\n' {
+				i++
+			}
+			space()
+		case c == '/' && i+1 < n && query[i+1] == '*':
+			end := strings.Index(query[i+2:], "*/")
+			if end < 0 {
+				return trinoMaskedQueryUnavailable
+			}
+			i += end + 4
+			space()
+		case c == '\'':
+			next, ok := skipSQLQuoted(query, i, '\'')
+			if !ok {
+				return trinoMaskedQueryUnavailable
+			}
+			emit("?")
+			i = next
+		case c == '"' || c == '`':
+			next, ok := skipSQLQuoted(query, i, c)
+			if !ok {
+				return trinoMaskedQueryUnavailable
+			}
+			emit(query[i:next])
+			i = next
+		case isSQLDigit(c) || (c == '.' && i+1 < n && isSQLDigit(query[i+1])):
+			j := i + 1
+			for j < n && isSQLNumberByte(query, j) {
+				j++
+			}
+			emit("?")
+			i = j
+		case isSQLIdentStart(c):
+			j := i + 1
+			for j < n && isSQLIdentByte(query[j]) {
+				j++
+			}
+			word := query[i:j]
+			// X'..' and U&'..' are literals that carry a prefix.
+			literalStart := -1
+			switch {
+			case (word == "X" || word == "x") && j < n && query[j] == '\'':
+				literalStart = j
+			case (word == "U" || word == "u") && j+1 < n && query[j] == '&' && query[j+1] == '\'':
+				literalStart = j + 1
+			}
+			if literalStart < 0 {
+				emit(word)
+				i = j
+				continue
+			}
+			next, ok := skipSQLQuoted(query, literalStart, '\'')
+			if !ok {
+				return trinoMaskedQueryUnavailable
+			}
+			emit("?")
+			i = next
+		default:
+			emit(query[i : i+1])
+			i++
+		}
+	}
+
+	masked := strings.TrimRight(out.String(), " ")
+	if utf8.RuneCountInString(masked) > trinoMaskedQueryMaxRunes {
+		masked = string([]rune(masked)[:trinoMaskedQueryMaxRunes]) + "…"
+	}
+	return masked
+}
+
+// skipSQLQuoted returns the index after the closing quote of the token that
+// opens at start. A doubled quote is an escaped quote, not the end.
+func skipSQLQuoted(s string, start int, quote byte) (int, bool) {
+	for i := start + 1; i < len(s); i++ {
+		if s[i] != quote {
+			continue
+		}
+		if i+1 < len(s) && s[i+1] == quote {
+			i++
+			continue
+		}
+		return i + 1, true
+	}
+	return 0, false
+}
+
+func isSQLDigit(c byte) bool { return c >= '0' && c <= '9' }
+
+// Bytes at or above 0x80 belong to a multi-byte UTF-8 character. Treating
+// them as identifier bytes keeps the character whole.
+func isSQLIdentStart(c byte) bool {
+	return c == '_' || c == '$' || c >= 0x80 || (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z')
+}
+
+func isSQLIdentByte(c byte) bool {
+	return isSQLIdentStart(c) || isSQLDigit(c) || c == '@'
+}
+
+// isSQLNumberByte reports whether s[i] continues a numeric literal: digits,
+// a decimal point, hex and exponent letters, digit separators, and a sign
+// directly after an exponent marker.
+func isSQLNumberByte(s string, i int) bool {
+	c := s[i]
+	if isSQLDigit(c) || c == '.' || c == '_' || (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') {
+		return true
+	}
+	return (c == '+' || c == '-') && (s[i-1] == 'e' || s[i-1] == 'E')
+}
diff --git a/controlplane/admin/trino_sql_mask_test.go b/controlplane/admin/trino_sql_mask_test.go
new file mode 100644
index 00000000..7d4e2554
--- /dev/null
+++ b/controlplane/admin/trino_sql_mask_test.go
@@ -0,0 +1,51 @@
+//go:build kubernetes
+
+package admin
+
+import (
+	"strings"
+	"testing"
+	"unicode/utf8"
+)
+
+func TestMaskTrinoSQLLiterals(t *testing.T) {
+	cases := []struct {
+		name  string
+		query string
+		want  string
+	}{
+		{"string and number", `SELECT * FROM t WHERE email = 'a@example.com' AND n > 42`, `SELECT * FROM t WHERE email = ? AND n > ?`},
+		{"escaped quote stays inside the literal", `SELECT 'it''s', "col""x" FROM t`, `SELECT ?, "col""x" FROM t`},
+		{"number forms and identifiers with digits", `SELECT 1.5e-3, .5, 0x1F, 1_000, col1, t2.c3 FROM t2 LIMIT 10`, `SELECT ?, ?, ?, ?, col1, t2.c3 FROM t2 LIMIT ?`},
+		{"subtraction is not an exponent sign", `SELECT a-1, 2-b FROM t`, `SELECT a-?, ?-b FROM t`},
+		{"typed and prefixed literals", `SELECT DATE '2026-01-01', X'CAFE', U&'\0041', INTERVAL '3' DAY FROM t`, `SELECT DATE ?, ?, ?, INTERVAL ? DAY FROM t`},
+		{"comments are dropped", "SELECT 1 -- secret 'x'\nFROM t /* hidden 42 */ WHERE a = 2", `SELECT ? FROM t WHERE a = ?`},
+		{"whitespace collapses", "  SELECT\n\t a\n FROM   t  ", `SELECT a FROM t`},
+		{"quoted identifiers are kept", "SELECT \"my col\", `other` FROM \"my schema\".t", "SELECT \"my col\", `other` FROM \"my schema\".t"},
+		{"non-ASCII identifier stays whole", `SELECT "prénom", größe FROM t WHERE x = 'é'`, `SELECT "prénom", größe FROM t WHERE x = ?`},
+		{"secret redaction placeholder passes through", `CREATE SECRET s (…redacted)`, `CREATE SECRET s (…redacted)`},
+		{"empty", ``, ``},
+		{"unterminated string", `SELECT 'abc`, trinoMaskedQueryUnavailable},
+		{"unterminated block comment", `SELECT 1 /* never closed`, trinoMaskedQueryUnavailable},
+		{"unterminated quoted identifier", `SELECT "abc FROM t`, trinoMaskedQueryUnavailable},
+		{"unterminated prefixed literal", `SELECT X'CAFE`, trinoMaskedQueryUnavailable},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := maskTrinoSQLLiterals(tc.query); got != tc.want {
+				t.Errorf("maskTrinoSQLLiterals(%q)\n got: %q\nwant: %q", tc.query, got, tc.want)
+			}
+		})
+	}
+}
+
+func TestMaskTrinoSQLLiteralsTruncatesLongStatements(t *testing.T) {
+	got := maskTrinoSQLLiterals("SELECT " + strings.Repeat("column_name, ", 400) + "x FROM t")
+
+	if count := utf8.RuneCountInString(got); count != trinoMaskedQueryMaxRunes+1 {
+		t.Fatalf("rune count = %d, want %d", count, trinoMaskedQueryMaxRunes+1)
+	}
+	if !strings.HasSuffix(got, "…") {
+		t.Fatalf("truncated statement must end with an ellipsis: %q", got[len(got)-10:])
+	}
+}

From 7cce9a1802d938bf398490b7190e7c8fd0c660e9 Mon Sep 17 00:00:00 2001
From: eric <eric@posthog.com>
Date: Thu, 1 Oct 2026 14:27:40 -0400
Subject: [PATCH 2/5] feat: add tenant-safe Trino monitoring snapshot route

---
 controlplane/admin/README.md                  |  20 ++
 controlplane/admin/extras.go                  |   1 +
 controlplane/admin/monitoring_trino.go        | 295 ++++++++++++++++++
 controlplane/admin/monitoring_trino_test.go   | 263 ++++++++++++++++
 controlplane/provisioner/trino_provisioner.go |   8 +
 .../provisioner/trino_tier_limits_test.go     |  24 ++
 6 files changed, 611 insertions(+)
 create mode 100644 controlplane/admin/monitoring_trino.go
 create mode 100644 controlplane/admin/monitoring_trino_test.go
 create mode 100644 controlplane/provisioner/trino_tier_limits_test.go

diff --git a/controlplane/admin/README.md b/controlplane/admin/README.md
index 03ef8fbb..edb6cc1d 100644
--- a/controlplane/admin/README.md
+++ b/controlplane/admin/README.md
@@ -186,6 +186,26 @@ Prometheus response and retains only the `status`, `reason`, or `source` labels
 the corresponding chart. Unknown metrics and windows return 400; unknown orgs
 return 404 with `code: "managed_warehouse_not_found"` before Prometheus is called.
 
+#### Trino monitoring (`monitoring_trino.go`)
+
+The PostHog backend reads `GET /api/v1/orgs/:id/monitoring/trino/snapshot` for
+organizations whose customers query through Trino. Auth and scoping follow the
+rules above: internal secret only, and the org is fixed by the path.
+
+The snapshot reports the org's Trino lifecycle state (`not_enabled`, `pending`,
+`provisioning`, `ready`, `failed`), its tier's concurrency and queue limits,
+in-flight totals, and at most 200 in-flight queries, longest-running first.
+`available` is false when the coordinator or the org index cannot be read; the
+totals are then zero and must not be shown as an idle warehouse. `in_flight`
+counts every non-terminal query. `running` counts every in-flight query past
+the queue, which is what the concurrency limit counts, so
+`in_flight = running + queued`. `blocked` is the subset of `running` whose
+drivers are all blocked. Query text has literals replaced with `?` and comments
+removed (`trino_sql_mask.go`), and becomes a fixed placeholder when it cannot
+be lexed. The snapshot intentionally omits cell ids, Trino principals,
+resource-group and tier names, status messages, driver counts, error codes, and
+connection details.
+
 ### Trino cell views (`trino.go` + `trino_client.go`)
 
 The **Trino cell → Instance recovery** panel lets admins select a shared-pool
diff --git a/controlplane/admin/extras.go b/controlplane/admin/extras.go
index 4e9527ff..bdf4e911 100644
--- a/controlplane/admin/extras.go
+++ b/controlplane/admin/extras.go
@@ -38,6 +38,7 @@ func RegisterExtras(r *gin.RouterGroup, x Extras) {
 	r.GET("/me", meHandler)
 	registerLiveAPI(r, x.Live, x.Fetcher, x.Users)
 	registerMonitoringAPI(r, x.Monitoring, x.Live, x.Fetcher, x.Metrics, x.MonitoringWorkerDefaults)
+	registerTrinoMonitoringAPI(r, x.Monitoring, x.Trino, x.Metrics)
 	if x.ClusterClient != nil {
 		registerClusterAPI(r, x.ClusterClient)
 	}
diff --git a/controlplane/admin/monitoring_trino.go b/controlplane/admin/monitoring_trino.go
new file mode 100644
index 00000000..1822caeb
--- /dev/null
+++ b/controlplane/admin/monitoring_trino.go
@@ -0,0 +1,295 @@
+//go:build kubernetes
+
+package admin
+
+import (
+	"net/http"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"github.com/posthog/duckgres/controlplane/configstore"
+	"github.com/posthog/duckgres/controlplane/provisioner"
+)
+
+const (
+	trinoMonitoringSchemaVersion   = 1
+	trinoMonitoringQueryLimit      = 200
+	trinoMonitoringStateNotEnabled = "not_enabled"
+)
+
+// In-flight states reported to tenants and used as the `state` label of the
+// per-org in-flight gauge. They partition every query that has not finished.
+const (
+	TrinoInFlightQueued  = "queued"
+	TrinoInFlightRunning = "running"
+	TrinoInFlightBlocked = "blocked"
+	TrinoInFlightOther   = "other"
+)
+
+// TrinoInFlightStates lists every value TrinoInFlightState returns.
+var TrinoInFlightStates = []string{TrinoInFlightQueued, TrinoInFlightRunning, TrinoInFlightBlocked, TrinoInFlightOther}
+
+// TrinoInFlightState classifies a query that has not finished. ok is false
+// for a finished or failed query.
+//
+// "other" is a query past the queue that is not executing yet or any more
+// (planning, starting, finishing). It still holds a concurrency slot.
+func TrinoInFlightState(q TrinoQuery) (state string, ok bool) {
+	switch {
+	case !isActiveTrinoState(q.State):
+		return "", false
+	case q.State == trinoStateQueued:
+		return TrinoInFlightQueued, true
+	case q.State == trinoStateRunning && q.FullyBlocked:
+		return TrinoInFlightBlocked, true
+	case q.State == trinoStateRunning:
+		return TrinoInFlightRunning, true
+	default:
+		return TrinoInFlightOther, true
+	}
+}
+
+type trinoMonitoringStore interface {
+	Snapshot() *configstore.Snapshot
+}
+
+type trinoMonitoringState struct {
+	State    string     `json:"state"`
+	ReadyAt  *time.Time `json:"ready_at"`
+	FailedAt *time.Time `json:"failed_at"`
+}
+
+type trinoMonitoringLimits struct {
+	MaxRunningQueries int `json:"max_running_queries"`
+	MaxQueuedQueries  int `json:"max_queued_queries"`
+}
+
+type trinoMonitoringTotals struct {
+	InFlight int `json:"in_flight"`
+	// Running is every in-flight query past the queue, which is what the
+	// tier's concurrency limit counts. InFlight = Running + Queued.
+	Running int `json:"running"`
+	Queued  int `json:"queued"`
+	// Blocked is the subset of Running whose drivers are all blocked.
+	Blocked            int   `json:"blocked"`
+	LongestRunningMS   int64 `json:"longest_running_ms"`
+	PhysicalInputBytes int64 `json:"physical_input_bytes"`
+}
+
+type trinoMonitoringQuery struct {
+	QueryID            string     `json:"query_id"`
+	State              string     `json:"state"`
+	User               string     `json:"user"`
+	Source             string     `json:"source"`
+	Query              string     `json:"query"`
+	CreatedAt          *time.Time `json:"created_at"`
+	ElapsedMS          int64      `json:"elapsed_ms"`
+	QueuedMS           int64      `json:"queued_ms"`
+	CPUMS              int64      `json:"cpu_ms"`
+	PhysicalInputBytes int64      `json:"physical_input_bytes"`
+	PeakMemoryBytes    int64      `json:"peak_memory_bytes"`
+	ProcessedInputRows int64      `json:"processed_input_rows"`
+	ProgressPercentage *float64   `json:"progress_percentage"`
+	Blocked            bool       `json:"blocked"`
+}
+
+type trinoMonitoringSnapshotResponse struct {
+	SchemaVersion int                  `json:"schema_version"`
+	OrgID         string               `json:"org_id"`
+	AsOf          time.Time            `json:"as_of"`
+	Trino         trinoMonitoringState `json:"trino"`
+	// Available is false when the coordinator could not be read. Totals are
+	// then zero, and a consumer must not present them as an idle warehouse.
+	Available        bool                   `json:"available"`
+	Limits           trinoMonitoringLimits  `json:"limits"`
+	Totals           trinoMonitoringTotals  `json:"totals"`
+	Queries          []trinoMonitoringQuery `json:"queries"`
+	QueriesTruncated bool                   `json:"queries_truncated"`
+}
+
+type trinoMonitoringHandler struct {
+	store   trinoMonitoringStore
+	trino   *TrinoAPI
+	metrics *MetricsProxy
+}
+
+// registerTrinoMonitoringAPI mounts the tenant-safe Trino monitoring contract
+// that the PostHog backend reads. A nil trino means the deployment has no
+// Trino cell; the snapshot then reports every org as not enabled.
+func registerTrinoMonitoringAPI(r *gin.RouterGroup, store trinoMonitoringStore, trino *TrinoAPI, metrics *MetricsProxy) {
+	h := &trinoMonitoringHandler{store: store, trino: trino, metrics: metrics}
+	group := r.Group("/orgs/:id/monitoring/trino", requireInternalSecret())
+	group.GET("/snapshot", h.snapshot)
+}
+
+// orgCell returns the cell API that owns orgID's Trino assignment, with the
+// org's Trino row. The API is nil when the org is not enabled or no
+// configured cell owns it.
+func (a *TrinoAPI) orgCell(orgID string) (*TrinoAPI, *configstore.ManagedWarehouseTrino, error) {
+	row, err := a.orgs.GetManagedWarehouseTrino(orgID)
+	if err != nil {
+		return nil, nil, err
+	}
+	if row == nil || !row.Enabled {
+		return nil, row, nil
+	}
+	if a.fleet == nil {
+		return a, row, nil
+	}
+	for _, candidate := range a.fleet {
+		if row.TrinoCellID != "" && candidate.cell.storedID() == row.TrinoCellID {
+			return candidate, row, nil
+		}
+	}
+	return nil, row, nil
+}
+
+func (h *trinoMonitoringHandler) warehouseExists(c *gin.Context) bool {
+	if h.store == nil {
+		c.JSON(http.StatusServiceUnavailable, gin.H{"error": "monitoring unavailable"})
+		return false
+	}
+	snapshot := h.store.Snapshot()
+	if snapshot == nil {
+		c.JSON(http.StatusServiceUnavailable, gin.H{"error": "configuration snapshot unavailable"})
+		return false
+	}
+	org, ok := snapshot.Orgs[c.Param("id")]
+	if !ok || org == nil || org.Warehouse == nil {
+		monitoringWarehouseNotFound(c)
+		return false
+	}
+	return true
+}
+
+func (h *trinoMonitoringHandler) snapshot(c *gin.Context) {
+	if !h.warehouseExists(c) {
+		return
+	}
+	orgID := c.Param("id")
+	response := trinoMonitoringSnapshotResponse{
+		SchemaVersion: trinoMonitoringSchemaVersion,
+		OrgID:         orgID,
+		AsOf:          time.Now().UTC(),
+		Trino:         trinoMonitoringState{State: trinoMonitoringStateNotEnabled},
+		Queries:       []trinoMonitoringQuery{},
+	}
+	if h.trino == nil {
+		c.JSON(http.StatusOK, response)
+		return
+	}
+
+	cell, row, err := h.trino.orgCell(orgID)
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": "monitoring snapshot unavailable"})
+		return
+	}
+	if row == nil || !row.Enabled {
+		c.JSON(http.StatusOK, response)
+		return
+	}
+	response.Trino = trinoMonitoringState{
+		State:    trinoMonitoringLifecycleState(row.State),
+		ReadyAt:  utcTimePointer(row.ReadyAt),
+		FailedAt: utcTimePointer(row.FailedAt),
+	}
+	response.Limits.MaxRunningQueries, response.Limits.MaxQueuedQueries = provisioner.TrinoTierLimits(row.Tier)
+	if cell == nil {
+		c.JSON(http.StatusOK, response)
+		return
+	}
+
+	// Without the principal index no query can be attributed to the org, and
+	// an empty list would read as an idle warehouse. Report it as unavailable.
+	idx, err := cell.index()
+	if err != nil {
+		c.JSON(http.StatusOK, response)
+		return
+	}
+	queries, _, err := cell.liveQueries(c.Request.Context(), idx)
+	if err != nil {
+		c.JSON(http.StatusOK, response)
+		return
+	}
+	response.Available = true
+
+	inFlight := make([]TrinoQuery, 0)
+	for _, q := range queries {
+		if q.Org != orgID {
+			continue
+		}
+		state, ok := TrinoInFlightState(q)
+		if !ok {
+			continue
+		}
+		response.Totals.InFlight++
+		switch state {
+		case TrinoInFlightQueued:
+			response.Totals.Queued++
+		case TrinoInFlightBlocked:
+			response.Totals.Blocked++
+			response.Totals.Running++
+		default:
+			response.Totals.Running++
+		}
+		if q.ElapsedMS > response.Totals.LongestRunningMS {
+			response.Totals.LongestRunningMS = q.ElapsedMS
+		}
+		response.Totals.PhysicalInputBytes += q.PhysicalInputBytes
+		inFlight = append(inFlight, q)
+	}
+
+	sort.SliceStable(inFlight, func(i, j int) bool { return inFlight[i].ElapsedMS > inFlight[j].ElapsedMS })
+	if len(inFlight) > trinoMonitoringQueryLimit {
+		inFlight = inFlight[:trinoMonitoringQueryLimit]
+		response.QueriesTruncated = true
+	}
+	for _, q := range inFlight {
+		owner, _ := idx.orgByPrincipal.Resolve(q.Principal)
+		var createdAt *time.Time
+		if !q.Created.IsZero() {
+			createdAt = utcTimePointer(&q.Created)
+		}
+		response.Queries = append(response.Queries, trinoMonitoringQuery{
+			QueryID:            q.QueryID,
+			State:              strings.ToLower(q.State),
+			User:               owner.Username,
+			Source:             q.Source,
+			Query:              maskTrinoSQLLiterals(q.Query),
+			CreatedAt:          createdAt,
+			ElapsedMS:          q.ElapsedMS,
+			QueuedMS:           q.QueuedMS,
+			CPUMS:              q.CPUMS,
+			PhysicalInputBytes: q.PhysicalInputBytes,
+			PeakMemoryBytes:    q.PeakMemoryBytes,
+			ProcessedInputRows: q.ProcessedInputRows,
+			ProgressPercentage: q.ProgressPercentage,
+			Blocked:            q.State == trinoStateRunning && q.FullyBlocked,
+		})
+	}
+	c.JSON(http.StatusOK, response)
+}
+
+// trinoMonitoringLifecycleState maps the Trino row's state onto the four
+// lifecycle states the contract documents. An empty or unexpected state is
+// pending, which is what the row means before its first reconcile.
+func trinoMonitoringLifecycleState(state configstore.ManagedWarehouseProvisioningState) string {
+	switch state {
+	case configstore.ManagedWarehouseStateProvisioning,
+		configstore.ManagedWarehouseStateReady,
+		configstore.ManagedWarehouseStateFailed:
+		return string(state)
+	default:
+		return string(configstore.ManagedWarehouseStatePending)
+	}
+}
+
+func utcTimePointer(value *time.Time) *time.Time {
+	if value == nil {
+		return nil
+	}
+	utc := value.UTC()
+	return &utc
+}
diff --git a/controlplane/admin/monitoring_trino_test.go b/controlplane/admin/monitoring_trino_test.go
new file mode 100644
index 00000000..7fc4979e
--- /dev/null
+++ b/controlplane/admin/monitoring_trino_test.go
@@ -0,0 +1,263 @@
+//go:build kubernetes
+
+package admin
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"github.com/posthog/duckgres/controlplane/configstore"
+)
+
+const trinoMonitoringTestCell = "registered:cell-a"
+
+func trinoMonitoringRouter(store trinoMonitoringStore, api *TrinoAPI, metrics *MetricsProxy, identitySource string) *gin.Engine {
+	gin.SetMode(gin.TestMode)
+	r := gin.New()
+	r.Use(func(c *gin.Context) {
+		c.Set(ctxIdentityKey, &Identity{Role: RoleAdmin, Source: identitySource})
+		c.Next()
+	})
+	registerTrinoMonitoringAPI(r.Group("/api/v1"), store, api, metrics)
+	return r
+}
+
+func trinoMonitoringFleet(coordinator TrinoCoordinatorClient, orgs TrinoOrgStore) *TrinoAPI {
+	return NewTrinoFleetAPI(
+		[]TrinoCell{{ID: "cell-a", StoredID: trinoMonitoringTestCell}},
+		[]TrinoCoordinatorClient{coordinator},
+		orgs,
+		nil,
+	)
+}
+
+func trinoMonitoringOrgStore(readyAt time.Time) *fakeTrinoOrgStore {
+	return &fakeTrinoOrgStore{
+		orgs: []configstore.TrinoEnabledOrg{
+			{
+				OrgID: "org-a", DatabaseName: "tenantdb", Tier: "growth", CellID: trinoMonitoringTestCell,
+				State: configstore.ManagedWarehouseStateReady, RootPasswordHash: "$2a$10$roothash",
+				Users: []configstore.TrinoOrgUser{{Username: "analyst", PasswordHash: "$2a$10$analysthash"}},
+			},
+			{
+				OrgID: "org-b", DatabaseName: "otherdb", Tier: "free", CellID: trinoMonitoringTestCell,
+				State: configstore.ManagedWarehouseStateReady, RootPasswordHash: "$2a$10$otherhash",
+			},
+		},
+		rows: map[string]*configstore.ManagedWarehouseTrino{
+			"org-a": {
+				OrgID: "org-a", Enabled: true, Tier: "growth", TrinoCellID: trinoMonitoringTestCell,
+				State: configstore.ManagedWarehouseStateReady, ReadyAt: &readyAt,
+				StatusMessage: "sensitive reconcile detail",
+			},
+		},
+	}
+}
+
+func getTrinoMonitoringSnapshot(t *testing.T, r *gin.Engine, orgID string) (trinoMonitoringSnapshotResponse, string) {
+	t.Helper()
+	rec := httptest.NewRecorder()
+	r.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/api/v1/orgs/"+orgID+"/monitoring/trino/snapshot", nil))
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status = %d, want 200: %s", rec.Code, rec.Body.String())
+	}
+	var got trinoMonitoringSnapshotResponse
+	if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil {
+		t.Fatal(err)
+	}
+	return got, rec.Body.String()
+}
+
+func TestTrinoMonitoringSnapshotIsOrgScopedMaskedAndCounted(t *testing.T) {
+	readyAt := time.Date(2026, 9, 1, 12, 0, 0, 0, time.UTC)
+	created := time.Date(2026, 9, 30, 8, 0, 0, 0, time.UTC)
+	half := 50.0
+	coordinator := &fakeTrinoCoordinator{queries: []TrinoQuery{
+		{QueryID: "q-running", State: "RUNNING", Principal: "tenantdb", Source: "dbt", ResourceGroup: "root.tenants.growth.tenantdb",
+			Query: "SELECT * FROM t WHERE email = 'person@example.com'", Created: created, ElapsedMS: 5000, QueuedMS: 20, CPUMS: 900,
+			PhysicalInputBytes: 100, PeakMemoryBytes: 64, ProcessedInputRows: 7, ProgressPercentage: &half},
+		{QueryID: "q-queued", State: "QUEUED", Principal: "tenantdb.analyst", ElapsedMS: 9000},
+		{QueryID: "q-planning", State: "PLANNING", Principal: "tenantdb", ElapsedMS: 100},
+		{QueryID: "q-blocked", State: "RUNNING", FullyBlocked: true, Principal: "tenantdb", ElapsedMS: 7000, PhysicalInputBytes: 50},
+		{QueryID: "q-finished", State: "FINISHED", Principal: "tenantdb", ElapsedMS: 99999},
+		{QueryID: "q-other-org", State: "RUNNING", Principal: "otherdb", Query: "SELECT 'other-org-secret'", ElapsedMS: 99999},
+	}}
+	r := trinoMonitoringRouter(
+		&fakeMonitoringStore{snapshot: monitoringTestSnapshot("org-a")},
+		trinoMonitoringFleet(coordinator, trinoMonitoringOrgStore(readyAt)),
+		nil, "internal-secret",
+	)
+
+	got, body := getTrinoMonitoringSnapshot(t, r, "org-a")
+
+	if got.SchemaVersion != 1 || got.OrgID != "org-a" || !got.Available {
+		t.Fatalf("envelope = %+v", got)
+	}
+	if got.Trino.State != "ready" || got.Trino.ReadyAt == nil || !got.Trino.ReadyAt.Equal(readyAt) {
+		t.Fatalf("trino = %+v, want ready at %s", got.Trino, readyAt)
+	}
+	if got.Limits != (trinoMonitoringLimits{MaxRunningQueries: 10, MaxQueuedQueries: 50}) {
+		t.Fatalf("limits = %+v, want the growth tier's 10 / 50", got.Limits)
+	}
+	wantTotals := trinoMonitoringTotals{InFlight: 4, Running: 3, Queued: 1, Blocked: 1, LongestRunningMS: 9000, PhysicalInputBytes: 150}
+	if got.Totals != wantTotals {
+		t.Fatalf("totals = %+v, want %+v", got.Totals, wantTotals)
+	}
+
+	var order, users []string
+	for _, q := range got.Queries {
+		order = append(order, q.QueryID)
+		users = append(users, q.User)
+	}
+	if strings.Join(order, ",") != "q-queued,q-blocked,q-running,q-planning" {
+		t.Fatalf("query order = %v, want longest-running first and only this org's in-flight queries", order)
+	}
+	if strings.Join(users, ",") != "analyst,root,root,root" {
+		t.Fatalf("users = %v, want duckgres usernames", users)
+	}
+	running := got.Queries[2]
+	if running.State != "running" || running.Query != "SELECT * FROM t WHERE email = ?" || running.Source != "dbt" {
+		t.Fatalf("running row = %+v", running)
+	}
+	if running.CreatedAt == nil || !running.CreatedAt.Equal(created) || running.ProgressPercentage == nil || *running.ProgressPercentage != 50 {
+		t.Fatalf("running row timing = %+v", running)
+	}
+	if !got.Queries[1].Blocked || got.Queries[0].CreatedAt != nil || got.QueriesTruncated {
+		t.Fatalf("blocked flag, null created_at or truncation is wrong: %+v", got)
+	}
+
+	for _, forbidden := range []string{
+		"person@example.com", "other-org-secret", "tenantdb", "otherdb", "cell-a",
+		"root.tenants", "sensitive reconcile detail", "principal", "resource_group", "$2a$10$",
+	} {
+		if strings.Contains(body, forbidden) {
+			t.Errorf("snapshot leaks %q: %s", forbidden, body)
+		}
+	}
+}
+
+func TestTrinoMonitoringSnapshotReportsLifecycleAndAvailability(t *testing.T) {
+	readyAt := time.Date(2026, 9, 1, 12, 0, 0, 0, time.UTC)
+	cases := []struct {
+		name          string
+		api           func() *TrinoAPI
+		wantState     string
+		wantAvailable bool
+		wantLimits    trinoMonitoringLimits
+	}{
+		{"deployment without a Trino cell", func() *TrinoAPI { return nil }, "not_enabled", false, trinoMonitoringLimits{}},
+		{"org without a Trino row", func() *TrinoAPI {
+			store := trinoMonitoringOrgStore(readyAt)
+			delete(store.rows, "org-a")
+			return trinoMonitoringFleet(&fakeTrinoCoordinator{}, store)
+		}, "not_enabled", false, trinoMonitoringLimits{}},
+		{"disabled Trino row", func() *TrinoAPI {
+			store := trinoMonitoringOrgStore(readyAt)
+			store.rows["org-a"].Enabled = false
+			return trinoMonitoringFleet(&fakeTrinoCoordinator{}, store)
+		}, "not_enabled", false, trinoMonitoringLimits{}},
+		{"enabled but not assigned to a configured cell", func() *TrinoAPI {
+			store := trinoMonitoringOrgStore(readyAt)
+			store.rows["org-a"].TrinoCellID = "registered:retired-cell"
+			store.rows["org-a"].State = ""
+			return trinoMonitoringFleet(&fakeTrinoCoordinator{}, store)
+		}, "pending", false, trinoMonitoringLimits{MaxRunningQueries: 10, MaxQueuedQueries: 50}},
+		{"coordinator unreachable", func() *TrinoAPI {
+			return trinoMonitoringFleet(&fakeTrinoCoordinator{queriesErr: errors.New("connection refused")}, trinoMonitoringOrgStore(readyAt))
+		}, "ready", false, trinoMonitoringLimits{MaxRunningQueries: 10, MaxQueuedQueries: 50}},
+		{"org list unreadable", func() *TrinoAPI {
+			store := trinoMonitoringOrgStore(readyAt)
+			store.listErr = errors.New("config store down")
+			return trinoMonitoringFleet(&fakeTrinoCoordinator{queries: []TrinoQuery{{QueryID: "q", State: "RUNNING", Principal: "tenantdb"}}}, store)
+		}, "ready", false, trinoMonitoringLimits{MaxRunningQueries: 10, MaxQueuedQueries: 50}},
+		{"idle and reachable", func() *TrinoAPI {
+			return trinoMonitoringFleet(&fakeTrinoCoordinator{}, trinoMonitoringOrgStore(readyAt))
+		}, "ready", true, trinoMonitoringLimits{MaxRunningQueries: 10, MaxQueuedQueries: 50}},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			r := trinoMonitoringRouter(&fakeMonitoringStore{snapshot: monitoringTestSnapshot("org-a")}, tc.api(), nil, "internal-secret")
+
+			got, body := getTrinoMonitoringSnapshot(t, r, "org-a")
+
+			if got.Trino.State != tc.wantState || got.Available != tc.wantAvailable || got.Limits != tc.wantLimits {
+				t.Fatalf("state/available/limits = %q/%v/%+v, want %q/%v/%+v",
+					got.Trino.State, got.Available, got.Limits, tc.wantState, tc.wantAvailable, tc.wantLimits)
+			}
+			if got.Totals != (trinoMonitoringTotals{}) {
+				t.Fatalf("totals = %+v, want zero", got.Totals)
+			}
+			if !strings.Contains(body, `"queries":[]`) {
+				t.Fatalf("queries must serialize as an empty list, not null: %s", body)
+			}
+		})
+	}
+}
+
+func TestTrinoMonitoringSnapshotCapsTheQueryList(t *testing.T) {
+	queries := make([]TrinoQuery, 0, 205)
+	for i := 0; i < 205; i++ {
+		queries = append(queries, TrinoQuery{QueryID: fmt.Sprintf("q-%03d", i), State: "RUNNING", Principal: "tenantdb", ElapsedMS: int64(i), PhysicalInputBytes: 1})
+	}
+	r := trinoMonitoringRouter(
+		&fakeMonitoringStore{snapshot: monitoringTestSnapshot("org-a")},
+		trinoMonitoringFleet(&fakeTrinoCoordinator{queries: queries}, trinoMonitoringOrgStore(time.Now())),
+		nil, "internal-secret",
+	)
+
+	got, _ := getTrinoMonitoringSnapshot(t, r, "org-a")
+
+	if len(got.Queries) != 200 || !got.QueriesTruncated || got.Queries[0].QueryID != "q-204" {
+		t.Fatalf("list = %d rows, truncated %v, first %q; want 200, true, q-204", len(got.Queries), got.QueriesTruncated, got.Queries[0].QueryID)
+	}
+	if got.Totals.InFlight != 205 || got.Totals.PhysicalInputBytes != 205 || got.Totals.LongestRunningMS != 204 {
+		t.Fatalf("totals = %+v, want them computed over all 205 queries", got.Totals)
+	}
+}
+
+func TestTrinoMonitoringRejectsSSOAndUnknownWarehouses(t *testing.T) {
+	api := trinoMonitoringFleet(&fakeTrinoCoordinator{}, trinoMonitoringOrgStore(time.Now()))
+	store := &fakeMonitoringStore{snapshot: monitoringTestSnapshot("org-a")}
+
+	rec := httptest.NewRecorder()
+	trinoMonitoringRouter(store, api, nil, "sso").ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/api/v1/orgs/org-a/monitoring/trino/snapshot", nil))
+	if rec.Code != http.StatusForbidden {
+		t.Errorf("SSO status = %d, want 403", rec.Code)
+	}
+
+	rec = httptest.NewRecorder()
+	trinoMonitoringRouter(store, api, nil, "internal-secret").ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/api/v1/orgs/org-zzz/monitoring/trino/snapshot", nil))
+	if rec.Code != http.StatusNotFound || !strings.Contains(rec.Body.String(), monitoringWarehouseNotFoundCode) {
+		t.Errorf("unknown warehouse = %d %s, want 404 with the stable code", rec.Code, rec.Body.String())
+	}
+}
+
+func TestTrinoInFlightStatePartitionsActiveQueries(t *testing.T) {
+	cases := []struct {
+		query     TrinoQuery
+		wantState string
+		wantOK    bool
+	}{
+		{TrinoQuery{State: "QUEUED"}, TrinoInFlightQueued, true},
+		{TrinoQuery{State: "RUNNING"}, TrinoInFlightRunning, true},
+		{TrinoQuery{State: "RUNNING", FullyBlocked: true}, TrinoInFlightBlocked, true},
+		{TrinoQuery{State: "PLANNING"}, TrinoInFlightOther, true},
+		{TrinoQuery{State: "WAITING_FOR_RESOURCES"}, TrinoInFlightOther, true},
+		{TrinoQuery{State: "FINISHING", FullyBlocked: true}, TrinoInFlightOther, true},
+		{TrinoQuery{State: "FINISHED"}, "", false},
+		{TrinoQuery{State: "FAILED"}, "", false},
+	}
+	for _, tc := range cases {
+		state, ok := TrinoInFlightState(tc.query)
+		if state != tc.wantState || ok != tc.wantOK {
+			t.Errorf("TrinoInFlightState(%+v) = %q, %v; want %q, %v", tc.query, state, ok, tc.wantState, tc.wantOK)
+		}
+	}
+}
diff --git a/controlplane/provisioner/trino_provisioner.go b/controlplane/provisioner/trino_provisioner.go
index fc7861f0..f5c47d5e 100644
--- a/controlplane/provisioner/trino_provisioner.go
+++ b/controlplane/provisioner/trino_provisioner.go
@@ -2434,6 +2434,14 @@ func tierLimits(tier string) resourceGroupSubGroup {
 	}
 }
 
+// TrinoTierLimits returns the concurrency and queue limits a tenant on tier
+// gets from the generated resource groups. An unknown tier gets the free lane,
+// the same fallback the resource-group selector applies.
+func TrinoTierLimits(tier string) (maxRunning, maxQueued int) {
+	limits := tierLimits(normalizeTier(tier))
+	return limits.HardConcurrencyLimit, limits.MaxQueued
+}
+
 // BuildTrinoResourceGroups renders resource-groups.json. Pure function for
 // unit testing.
 //
diff --git a/controlplane/provisioner/trino_tier_limits_test.go b/controlplane/provisioner/trino_tier_limits_test.go
new file mode 100644
index 00000000..d31ea3fa
--- /dev/null
+++ b/controlplane/provisioner/trino_tier_limits_test.go
@@ -0,0 +1,24 @@
+//go:build kubernetes
+
+package provisioner
+
+import "testing"
+
+func TestTrinoTierLimitsMatchTheResourceGroupLanes(t *testing.T) {
+	cases := []struct {
+		tier                  string
+		maxRunning, maxQueued int
+	}{
+		{"free", 3, 20},
+		{"growth", 10, 50},
+		{"scale", 25, 100},
+		{"", 3, 20},
+		{"unknown-tier", 3, 20},
+	}
+	for _, tc := range cases {
+		maxRunning, maxQueued := TrinoTierLimits(tc.tier)
+		if maxRunning != tc.maxRunning || maxQueued != tc.maxQueued {
+			t.Errorf("TrinoTierLimits(%q) = %d, %d; want %d, %d", tc.tier, maxRunning, maxQueued, tc.maxRunning, tc.maxQueued)
+		}
+	}
+}

From 7708779798a3af89459a0ff1f3365fe41077b672 Mon Sep 17 00:00:00 2001
From: eric <eric@posthog.com>
Date: Thu, 1 Oct 2026 14:31:38 -0400
Subject: [PATCH 3/5] feat: export per-org Trino query metrics from the usage
 collector

---
 controlplane/multitenant.go            |   2 +-
 controlplane/trino_org_metrics.go      | 100 +++++++++++++++
 controlplane/trino_org_metrics_test.go | 164 +++++++++++++++++++++++++
 controlplane/trino_usage_collector.go  |  66 +++++++++-
 docs/metrics.md                        |  26 ++++
 5 files changed, 356 insertions(+), 2 deletions(-)
 create mode 100644 controlplane/trino_org_metrics.go
 create mode 100644 controlplane/trino_org_metrics_test.go

diff --git a/controlplane/multitenant.go b/controlplane/multitenant.go
index 286b3bbe..488c9991 100644
--- a/controlplane/multitenant.go
+++ b/controlplane/multitenant.go
@@ -817,7 +817,7 @@ func SetupMultiTenant(
 		// emits each terminal query, independent of admin-console traffic.
 		for _, wire := range trinoCells {
 			for _, observer := range wire.Observers {
-				janitorLeader.AttachLeaderLoop(newTrinoUsageCollector(observer, store, store.OrgUsageTeamID).Run)
+				janitorLeader.AttachLeaderLoop(newTrinoUsageCollector(observer, store, store.OrgUsageTeamID).withOrgMetrics(trinoOrgMetrics, wire.Cell.ID).Run)
 			}
 		}
 	}
diff --git a/controlplane/trino_org_metrics.go b/controlplane/trino_org_metrics.go
new file mode 100644
index 00000000..5ffedbc3
--- /dev/null
+++ b/controlplane/trino_org_metrics.go
@@ -0,0 +1,100 @@
+//go:build kubernetes
+
+package controlplane
+
+import (
+	"github.com/posthog/duckgres/controlplane/admin"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+var trinoOrgMetrics = newTrinoOrgMetrics(prometheus.DefaultRegisterer)
+
+// trinoOrgMetricSet is the per-org Trino query telemetry behind the product
+// monitoring series. The usage collector is its only writer.
+type trinoOrgMetricSet struct {
+	inFlight     *prometheus.GaugeVec
+	queries      *prometheus.CounterVec
+	duration     *prometheus.HistogramVec
+	queued       *prometheus.HistogramVec
+	scannedBytes *prometheus.CounterVec
+	cpuSeconds   *prometheus.CounterVec
+}
+
+func newTrinoOrgMetrics(reg prometheus.Registerer) *trinoOrgMetricSet {
+	m := &trinoOrgMetricSet{
+		inFlight: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Name: "duckgres_trino_org_queries", Help: "Trino queries in flight for an org at the last successful coordinator poll, by state.",
+		}, []string{"org", "state"}),
+		queries: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "duckgres_trino_org_query_total", Help: "Finished Trino queries observed for an org, by outcome.",
+		}, []string{"org", "status", "error_type"}),
+		duration: prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Name: "duckgres_trino_org_query_duration_seconds", Help: "Elapsed time of finished Trino queries for an org.",
+			Buckets: []float64{0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300, 600, 900, 1800, 3600},
+		}, []string{"org"}),
+		queued: prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Name: "duckgres_trino_org_query_queued_seconds", Help: "Time finished Trino queries for an org spent queued.",
+			Buckets: []float64{0.01, 0.05, 0.1, 0.5, 1, 5, 15, 30, 60, 300, 900},
+		}, []string{"org"}),
+		scannedBytes: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "duckgres_trino_org_query_physical_input_bytes_total", Help: "Physical input bytes read by finished Trino queries for an org.",
+		}, []string{"org"}),
+		cpuSeconds: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "duckgres_trino_org_query_cpu_seconds_total", Help: "CPU seconds used by finished Trino queries for an org.",
+		}, []string{"org"}),
+	}
+	reg.MustRegister(m.inFlight, m.queries, m.duration, m.queued, m.scannedBytes, m.cpuSeconds)
+	return m
+}
+
+func (m *trinoOrgMetricSet) observeFinished(orgID string, query admin.TrinoQuery) {
+	if m == nil {
+		return
+	}
+	status, errorType := "success", "none"
+	if query.State == "FAILED" {
+		status, errorType = "error", trinoErrorTypeLabel(query.ErrorType)
+	}
+	m.queries.WithLabelValues(orgID, status, errorType).Inc()
+	m.duration.WithLabelValues(orgID).Observe(float64(query.ElapsedMS) / 1000)
+	m.queued.WithLabelValues(orgID).Observe(float64(query.QueuedMS) / 1000)
+	m.scannedBytes.WithLabelValues(orgID).Add(float64(query.PhysicalInputBytes))
+	m.cpuSeconds.WithLabelValues(orgID).Add(float64(query.CPUMS) / 1000)
+}
+
+// setInFlight writes every state, so a state with no queries reads zero
+// instead of keeping its previous value.
+func (m *trinoOrgMetricSet) setInFlight(orgID string, counts map[string]int) {
+	if m == nil {
+		return
+	}
+	for _, state := range admin.TrinoInFlightStates {
+		m.inFlight.WithLabelValues(orgID, state).Set(float64(counts[state]))
+	}
+}
+
+func (m *trinoOrgMetricSet) clearInFlight(orgID string) {
+	if m == nil {
+		return
+	}
+	for _, state := range admin.TrinoInFlightStates {
+		m.inFlight.DeleteLabelValues(orgID, state)
+	}
+}
+
+// trinoErrorTypeLabel maps Trino's ErrorType onto a bounded label set, so an
+// unexpected value from a newer Trino cannot add series.
+func trinoErrorTypeLabel(errorType string) string {
+	switch errorType {
+	case "USER_ERROR":
+		return "user"
+	case "INTERNAL_ERROR":
+		return "internal"
+	case "INSUFFICIENT_RESOURCES":
+		return "insufficient_resources"
+	case "EXTERNAL":
+		return "external"
+	default:
+		return "unknown"
+	}
+}
diff --git a/controlplane/trino_org_metrics_test.go b/controlplane/trino_org_metrics_test.go
new file mode 100644
index 00000000..e523e48f
--- /dev/null
+++ b/controlplane/trino_org_metrics_test.go
@@ -0,0 +1,164 @@
+//go:build kubernetes
+
+package controlplane
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/posthog/duckgres/controlplane/admin"
+	"github.com/posthog/duckgres/controlplane/configstore"
+	"github.com/posthog/duckgres/internal/analytics"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+)
+
+const trinoOrgMetricsTestCell = "registered:cell-a"
+
+func trinoOrgMetricsTestOrgs() []configstore.TrinoEnabledOrg {
+	return []configstore.TrinoEnabledOrg{
+		{OrgID: "org-a", DatabaseName: "tenant-a", CellID: trinoOrgMetricsTestCell, RootPasswordHash: "hash"},
+		{OrgID: "org-idle", DatabaseName: "tenant-idle", CellID: trinoOrgMetricsTestCell, RootPasswordHash: "hash"},
+		{OrgID: "org-elsewhere", DatabaseName: "tenant-elsewhere", CellID: "registered:cell-b", RootPasswordHash: "hash"},
+	}
+}
+
+func newTrinoOrgMetricsTestCollector(t *testing.T, coordinator *trinoUsageFakeCoordinator) (*trinoUsageCollector, *trinoOrgMetricSet, *prometheus.Registry) {
+	t.Helper()
+	analytics.SetDefault(&trinoUsageFakeTracker{})
+	t.Cleanup(func() { analytics.SetDefault(nil) })
+	registry := prometheus.NewRegistry()
+	metrics := newTrinoOrgMetrics(registry)
+	collector := newTrinoUsageCollector(coordinator, trinoUsageFakeOrgs{orgs: trinoOrgMetricsTestOrgs()}, nil).
+		withOrgMetrics(metrics, trinoOrgMetricsTestCell)
+	return collector, metrics, registry
+}
+
+func trinoOrgHistogramCount(t *testing.T, registry *prometheus.Registry, name, org string) uint64 {
+	t.Helper()
+	families, err := registry.Gather()
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, family := range families {
+		if family.GetName() != name {
+			continue
+		}
+		for _, metric := range family.GetMetric() {
+			for _, label := range metric.GetLabel() {
+				if label.GetName() == "org" && label.GetValue() == org {
+					return metric.GetHistogram().GetSampleCount()
+				}
+			}
+		}
+	}
+	return 0
+}
+
+func TestTrinoOrgMetricsCountEachFinishedQueryOnce(t *testing.T) {
+	collector, metrics, registry := newTrinoOrgMetricsTestCollector(t, &trinoUsageFakeCoordinator{queries: []admin.TrinoQuery{
+		{QueryID: "ok", State: "FINISHED", Principal: "tenant-a", ElapsedMS: 1500, QueuedMS: 250, CPUMS: 4000, PhysicalInputBytes: 1024},
+		{QueryID: "user-error", State: "FAILED", Principal: "tenant-a", ErrorType: "USER_ERROR", ElapsedMS: 200, PhysicalInputBytes: 6},
+		{QueryID: "new-error-type", State: "FAILED", Principal: "tenant-a", ErrorType: "SOMETHING_NEW"},
+		{QueryID: "operator", State: "FINISHED", Principal: "__observer", PhysicalInputBytes: 999},
+	}})
+
+	collector.collect(context.Background())
+	collector.collect(context.Background())
+
+	for labels, want := range map[[2]string]float64{
+		{"success", "none"}:  1,
+		{"error", "user"}:    1,
+		{"error", "unknown"}: 1,
+	} {
+		if got := testutil.ToFloat64(metrics.queries.WithLabelValues("org-a", labels[0], labels[1])); got != want {
+			t.Errorf("query_total{status=%q,error_type=%q} = %v, want %v", labels[0], labels[1], got, want)
+		}
+	}
+	if got := testutil.ToFloat64(metrics.scannedBytes.WithLabelValues("org-a")); got != 1030 {
+		t.Errorf("scanned bytes = %v, want 1030", got)
+	}
+	if got := testutil.ToFloat64(metrics.cpuSeconds.WithLabelValues("org-a")); got != 4 {
+		t.Errorf("cpu seconds = %v, want 4", got)
+	}
+	for _, name := range []string{"duckgres_trino_org_query_duration_seconds", "duckgres_trino_org_query_queued_seconds"} {
+		if got := trinoOrgHistogramCount(t, registry, name, "org-a"); got != 3 {
+			t.Errorf("%s sample count = %d, want 3", name, got)
+		}
+	}
+	if got := testutil.CollectAndCount(metrics.queries); got != 3 {
+		t.Errorf("query_total series = %d, want 3 (operator queries are not tenant usage)", got)
+	}
+}
+
+func TestTrinoOrgMetricsGaugeInFlightQueriesForThisCellOnly(t *testing.T) {
+	coordinator := &trinoUsageFakeCoordinator{queries: []admin.TrinoQuery{
+		{QueryID: "r1", State: "RUNNING", Principal: "tenant-a"},
+		{QueryID: "r2", State: "RUNNING", Principal: "tenant-a"},
+		{QueryID: "b1", State: "RUNNING", FullyBlocked: true, Principal: "tenant-a"},
+		{QueryID: "q1", State: "QUEUED", Principal: "tenant-a"},
+		{QueryID: "p1", State: "PLANNING", Principal: "tenant-a"},
+		{QueryID: "done", State: "FINISHED", Principal: "tenant-a"},
+	}}
+	collector, metrics, _ := newTrinoOrgMetricsTestCollector(t, coordinator)
+
+	collector.collect(context.Background())
+
+	for state, want := range map[string]float64{"running": 2, "blocked": 1, "queued": 1, "other": 1} {
+		if got := testutil.ToFloat64(metrics.inFlight.WithLabelValues("org-a", state)); got != want {
+			t.Errorf("org-a %s = %v, want %v", state, got, want)
+		}
+	}
+	// 4 states each for org-a and the idle org; none for the org another cell owns.
+	if got := testutil.CollectAndCount(metrics.inFlight); got != 8 {
+		t.Fatalf("in-flight series = %d, want 8", got)
+	}
+
+	coordinator.queries = nil
+	collector.collect(context.Background())
+	if got := testutil.ToFloat64(metrics.inFlight.WithLabelValues("org-a", "running")); got != 0 {
+		t.Errorf("org-a running after the queries finished = %v, want 0", got)
+	}
+}
+
+func TestTrinoOrgMetricsClearGaugesWhenThePollCannotBeTrusted(t *testing.T) {
+	t.Run("coordinator read fails", func(t *testing.T) {
+		coordinator := &trinoUsageFakeCoordinator{queries: []admin.TrinoQuery{{QueryID: "r1", State: "RUNNING", Principal: "tenant-a"}}}
+		collector, metrics, _ := newTrinoOrgMetricsTestCollector(t, coordinator)
+		collector.collect(context.Background())
+
+		coordinator.err = errors.New("coordinator unreachable")
+		collector.collect(context.Background())
+
+		if got := testutil.CollectAndCount(metrics.inFlight); got != 0 {
+			t.Fatalf("in-flight series after a failed poll = %d, want 0", got)
+		}
+	})
+
+	t.Run("leadership ends", func(t *testing.T) {
+		coordinator := &trinoUsageFakeCoordinator{queries: []admin.TrinoQuery{{QueryID: "r1", State: "RUNNING", Principal: "tenant-a"}}}
+		collector, metrics, _ := newTrinoOrgMetricsTestCollector(t, coordinator)
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel()
+
+		collector.Run(ctx)
+
+		if got := testutil.CollectAndCount(metrics.inFlight); got != 0 {
+			t.Fatalf("in-flight series after Run returned = %d, want 0", got)
+		}
+	})
+
+	t.Run("org leaves the cell", func(t *testing.T) {
+		coordinator := &trinoUsageFakeCoordinator{}
+		collector, metrics, _ := newTrinoOrgMetricsTestCollector(t, coordinator)
+		collector.collect(context.Background())
+
+		collector.orgs = trinoUsageFakeOrgs{orgs: trinoOrgMetricsTestOrgs()[:1]}
+		collector.collect(context.Background())
+
+		if got := testutil.CollectAndCount(metrics.inFlight); got != 4 {
+			t.Fatalf("in-flight series = %d, want 4 (only org-a remains)", got)
+		}
+	})
+}
diff --git a/controlplane/trino_usage_collector.go b/controlplane/trino_usage_collector.go
index d87aec49..8f47c408 100644
--- a/controlplane/trino_usage_collector.go
+++ b/controlplane/trino_usage_collector.go
@@ -29,6 +29,18 @@ type trinoUsageCollector struct {
 	seen        map[string]time.Time
 	now         func() time.Time
 	interval    time.Duration
+
+	// Per-org Prometheus metrics. nil leaves the collector emitting only
+	// product-analytics events.
+	metrics *trinoOrgMetricSet
+	// cellID is the stored id of the cell this collector observes. One
+	// collector runs per cell, and each may only write the in-flight gauges
+	// of its own cell's orgs: another cell's collector sees none of their
+	// queries and would overwrite the gauges with zero.
+	cellID string
+	// gauged is the set of orgs whose in-flight gauges this collector set
+	// on its last successful poll.
+	gauged map[string]struct{}
 }
 
 func newTrinoUsageCollector(coordinator admin.TrinoCoordinatorClient, orgs admin.TrinoOrgStore, teamID trinoUsageTeamResolver) *trinoUsageCollector {
@@ -42,6 +54,14 @@ func newTrinoUsageCollector(coordinator admin.TrinoCoordinatorClient, orgs admin
 	}
 }
 
+// withOrgMetrics makes the collector export per-org query metrics for the
+// orgs assigned to cellID.
+func (c *trinoUsageCollector) withOrgMetrics(metrics *trinoOrgMetricSet, cellID string) *trinoUsageCollector {
+	c.metrics = metrics
+	c.cellID = cellID
+	return c
+}
+
 // Run is a leader-only loop. A new leader may see recently completed queries
 // after failover; the query_id property lets downstream consumers deduplicate
 // that bounded overlap without ever capturing customer SQL.
@@ -49,6 +69,9 @@ func (c *trinoUsageCollector) Run(ctx context.Context) {
 	if c == nil || c.coordinator == nil || c.orgs == nil {
 		return
 	}
+	// A gauge is a statement about now. Once this process stops polling it
+	// must stop exporting the last value it saw.
+	defer c.clearInFlight()
 	c.collect(ctx)
 	ticker := time.NewTicker(c.interval)
 	defer ticker.Stop()
@@ -69,12 +92,14 @@ func (c *trinoUsageCollector) collect(ctx context.Context) {
 	orgs, err := c.orgs.ListTrinoEnabledOrgs()
 	if err != nil {
 		slog.Warn("Trino usage collection skipped: list enabled orgs failed", "error", err)
+		c.clearInFlight()
 		return
 	}
 	owners := configstore.NewTrinoPrincipalOwners(orgs)
 	queries, err := c.coordinator.Queries(ctx)
 	if err != nil {
 		slog.Warn("Trino usage collection skipped: list queries failed", "error", err)
+		c.clearInFlight()
 		return
 	}
 	now := c.now()
@@ -83,8 +108,15 @@ func (c *trinoUsageCollector) collect(ctx context.Context) {
 			delete(c.seen, id)
 		}
 	}
+	inFlight := make(map[string]map[string]int)
 	for _, query := range queries {
-		if query.State != "FINISHED" && query.State != "FAILED" {
+		if state, active := admin.TrinoInFlightState(query); active {
+			if owner, ok := owners.Resolve(query.Principal); ok {
+				if inFlight[owner.OrgID] == nil {
+					inFlight[owner.OrgID] = make(map[string]int)
+				}
+				inFlight[owner.OrgID][state]++
+			}
 			continue
 		}
 		if query.QueryID == "" {
@@ -110,8 +142,40 @@ func (c *trinoUsageCollector) collect(ctx context.Context) {
 		} else {
 			analytics.Default().Capture("query_failed", orgID, props)
 		}
+		c.metrics.observeFinished(orgID, query)
 		c.seen[query.QueryID] = now
 	}
+	c.recordInFlight(orgs, inFlight)
+}
+
+func (c *trinoUsageCollector) recordInFlight(orgs []configstore.TrinoEnabledOrg, inFlight map[string]map[string]int) {
+	if c.metrics == nil {
+		return
+	}
+	current := make(map[string]struct{}, len(orgs))
+	for _, org := range orgs {
+		if c.cellID != "" && org.CellID != c.cellID {
+			continue
+		}
+		current[org.OrgID] = struct{}{}
+		c.metrics.setInFlight(org.OrgID, inFlight[org.OrgID])
+	}
+	for orgID := range c.gauged {
+		if _, ok := current[orgID]; !ok {
+			c.metrics.clearInFlight(orgID)
+		}
+	}
+	c.gauged = current
+}
+
+func (c *trinoUsageCollector) clearInFlight() {
+	if c == nil || c.metrics == nil {
+		return
+	}
+	for orgID := range c.gauged {
+		c.metrics.clearInFlight(orgID)
+	}
+	c.gauged = nil
 }
 
 func trinoUsageProperties(query admin.TrinoQuery, teamID int64) map[string]any {
diff --git a/docs/metrics.md b/docs/metrics.md
index 7201b8b0..59dea706 100644
--- a/docs/metrics.md
+++ b/docs/metrics.md
@@ -470,6 +470,32 @@ the current NetworkPolicy. These new lifecycle/logging assertions are covered
 by operator tests; a real blocked-drain deployment remains a separate validation
 step. No network permissions or live deployment behavior change in this work.
 
+## Per-org Trino query metrics
+
+The Trino usage collector exports these metrics while it holds the leader
+lease. They back the product monitoring series route
+(`GET /api/v1/orgs/:id/monitoring/trino/series`).
+
+| Metric | Labels | Meaning |
+|---|---|---|
+| `duckgres_trino_org_queries` | `org`, `state` | Queries in flight at the last successful coordinator poll. `state` is `queued`, `running`, `blocked`, or `other`, and the four values partition the in-flight set. |
+| `duckgres_trino_org_query_total` | `org`, `status`, `error_type` | Finished queries observed. `status` is `success` or `error`. `error_type` is `none`, `user`, `internal`, `insufficient_resources`, `external`, or `unknown`. |
+| `duckgres_trino_org_query_duration_seconds` | `org` | Elapsed time of finished queries. |
+| `duckgres_trino_org_query_queued_seconds` | `org` | Time finished queries spent queued. |
+| `duckgres_trino_org_query_physical_input_bytes_total` | `org` | Physical input bytes read by finished queries. |
+| `duckgres_trino_org_query_cpu_seconds_total` | `org` | CPU seconds used by finished queries. |
+
+One collector runs per Trino cell and writes the gauge only for the orgs
+assigned to its cell. The gauge is removed when a poll fails and when the
+collector stops, so a coordinator outage appears as a gap rather than a frozen
+value. During a leader handover two pods can export the gauge briefly; queries
+must take the maximum across pods, not the sum.
+
+The counters and histograms are best-effort. The coordinator keeps finished
+queries only briefly, so a very busy cell can drop some between polls, and a
+new leader can count a bounded overlap twice because the dedupe set is held in
+memory. They describe operational activity and must not be used for billing.
+
 ## Admission metric migration
 
 The pre-canonical admission family is retired. Existing TSDB history remains,

From 73f1da58a0d587b202923ac7ecec885972004b29 Mon Sep 17 00:00:00 2001
From: eric <eric@posthog.com>
Date: Thu, 1 Oct 2026 14:36:07 -0400
Subject: [PATCH 4/5] feat: add tenant-safe Trino monitoring series route

---
 controlplane/admin/README.md                | 13 +++-
 controlplane/admin/monitoring_trino.go      | 72 +++++++++++++++++++
 controlplane/admin/monitoring_trino_test.go | 77 +++++++++++++++++++++
 3 files changed, 160 insertions(+), 2 deletions(-)

diff --git a/controlplane/admin/README.md b/controlplane/admin/README.md
index edb6cc1d..489777c8 100644
--- a/controlplane/admin/README.md
+++ b/controlplane/admin/README.md
@@ -188,8 +188,8 @@ return 404 with `code: "managed_warehouse_not_found"` before Prometheus is calle
 
 #### Trino monitoring (`monitoring_trino.go`)
 
-The PostHog backend reads `GET /api/v1/orgs/:id/monitoring/trino/snapshot` for
-organizations whose customers query through Trino. Auth and scoping follow the
+The PostHog backend reads `GET /api/v1/orgs/:id/monitoring/trino/snapshot` and
+`/monitoring/trino/series` for organizations whose customers query through Trino. Auth and scoping follow the
 rules above: internal secret only, and the org is fixed by the path.
 
 The snapshot reports the org's Trino lifecycle state (`not_enabled`, `pending`,
@@ -206,6 +206,15 @@ be lexed. The snapshot intentionally omits cell ids, Trino principals,
 resource-group and tier names, status messages, driver counts, error codes, and
 connection details.
 
+The series endpoint accepts only `queries_in_flight`, `query_rate`,
+`error_ratio`, `duration_p50`, `duration_p95`, `queue_time_p95`,
+`scanned_bytes_rate`, `cpu_seconds_rate`, and `storage_bytes`, over the same
+windows as the DuckDB series endpoint. Every query carries an exact `org` label
+selector, and only the `state`, `status`, and `error_type` labels are returned.
+Unknown metrics and windows return 400 and unknown orgs return 404 with
+`code: "managed_warehouse_not_found"`, both before Prometheus is called. The
+metrics are defined in `docs/metrics.md`, "Per-org Trino query metrics".
+
 ### Trino cell views (`trino.go` + `trino_client.go`)
 
 The **Trino cell → Instance recovery** panel lets admins select a shared-pool
diff --git a/controlplane/admin/monitoring_trino.go b/controlplane/admin/monitoring_trino.go
index 1822caeb..878f7572 100644
--- a/controlplane/admin/monitoring_trino.go
+++ b/controlplane/admin/monitoring_trino.go
@@ -51,6 +51,48 @@ func TrinoInFlightState(q TrinoQuery) (state string, ok bool) {
 	}
 }
 
+// trinoMonitoringMetrics is the allow-list behind the Trino series route.
+// Every template carries $ORG, so no series can be read without an exact org
+// selector. The in-flight gauge takes the maximum across pods: the collector
+// runs under a leader lease, and during a handover two pods can export it.
+var trinoMonitoringMetrics = map[string]monitoringMetricSpec{
+	"queries_in_flight": {
+		PromQL: `max by (state) (duckgres_trino_org_queries$ORG)`, Unit: "queries",
+		AllowedLabels: labelSet("state"),
+	},
+	"query_rate": {
+		PromQL: `sum by (status, error_type) (rate(duckgres_trino_org_query_total$ORG[$WIN]))`, Unit: "queries_per_second",
+		AllowedLabels: labelSet("status", "error_type"),
+	},
+	"error_ratio": {
+		PromQL: `(sum(rate(duckgres_trino_org_query_total$ORGERR[$WIN])) or vector(0)) / clamp_min((sum(rate(duckgres_trino_org_query_total$ORG[$WIN])) or vector(0)), 1e-9)`,
+		Unit:   "ratio", AllowedLabels: labelSet(),
+	},
+	"duration_p50": {
+		PromQL: `histogram_quantile(0.50, sum by (le) (rate(duckgres_trino_org_query_duration_seconds_bucket$ORG[$WIN])))`, Unit: "seconds",
+		AllowedLabels: labelSet(),
+	},
+	"duration_p95": {
+		PromQL: `histogram_quantile(0.95, sum by (le) (rate(duckgres_trino_org_query_duration_seconds_bucket$ORG[$WIN])))`, Unit: "seconds",
+		AllowedLabels: labelSet(),
+	},
+	"queue_time_p95": {
+		PromQL: `histogram_quantile(0.95, sum by (le) (rate(duckgres_trino_org_query_queued_seconds_bucket$ORG[$WIN])))`, Unit: "seconds",
+		AllowedLabels: labelSet(),
+	},
+	"scanned_bytes_rate": {
+		PromQL: `(sum(rate(duckgres_trino_org_query_physical_input_bytes_total$ORG[$WIN])) or vector(0))`, Unit: "bytes_per_second",
+		AllowedLabels: labelSet(),
+	},
+	"cpu_seconds_rate": {
+		PromQL: `(sum(rate(duckgres_trino_org_query_cpu_seconds_total$ORG[$WIN])) or vector(0))`, Unit: "cpu_seconds_per_second",
+		AllowedLabels: labelSet(),
+	},
+	"storage_bytes": {
+		PromQL: `max(duckgres_org_storage_tracked_bytes$ORG)`, Unit: "bytes", AllowedLabels: labelSet(),
+	},
+}
+
 type trinoMonitoringStore interface {
 	Snapshot() *configstore.Snapshot
 }
@@ -122,6 +164,7 @@ func registerTrinoMonitoringAPI(r *gin.RouterGroup, store trinoMonitoringStore,
 	h := &trinoMonitoringHandler{store: store, trino: trino, metrics: metrics}
 	group := r.Group("/orgs/:id/monitoring/trino", requireInternalSecret())
 	group.GET("/snapshot", h.snapshot)
+	group.GET("/series", h.series)
 }
 
 // orgCell returns the cell API that owns orgID's Trino assignment, with the
@@ -272,6 +315,35 @@ func (h *trinoMonitoringHandler) snapshot(c *gin.Context) {
 	c.JSON(http.StatusOK, response)
 }
 
+func (h *trinoMonitoringHandler) series(c *gin.Context) {
+	metric := c.Query("metric")
+	spec, ok := trinoMonitoringMetrics[metric]
+	if !ok {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "unknown monitoring metric"})
+		return
+	}
+	window, ok := monitoringWindows[c.DefaultQuery("window", "24h")]
+	if !ok {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "unsupported monitoring window"})
+		return
+	}
+	if !h.warehouseExists(c) {
+		return
+	}
+	if h.metrics == nil || h.metrics.promURL == "" {
+		c.JSON(http.StatusServiceUnavailable, gin.H{"error": "metrics not configured"})
+		return
+	}
+
+	response, err := h.metrics.queryMonitoringRange(c, c.Param("id"), metric, spec, window)
+	if err != nil {
+		c.JSON(http.StatusBadGateway, gin.H{"error": "metrics unavailable"})
+		return
+	}
+	response.SchemaVersion = trinoMonitoringSchemaVersion
+	c.JSON(http.StatusOK, response)
+}
+
 // trinoMonitoringLifecycleState maps the Trino row's state onto the four
 // lifecycle states the contract documents. An empty or unexpected state is
 // pending, which is what the row means before its first reconcile.
diff --git a/controlplane/admin/monitoring_trino_test.go b/controlplane/admin/monitoring_trino_test.go
index 7fc4979e..efca4aeb 100644
--- a/controlplane/admin/monitoring_trino_test.go
+++ b/controlplane/admin/monitoring_trino_test.go
@@ -261,3 +261,80 @@ func TestTrinoInFlightStatePartitionsActiveQueries(t *testing.T) {
 		}
 	}
 }
+
+func TestTrinoMonitoringMetricAllowListAlwaysRequiresOrgSelector(t *testing.T) {
+	for metric, spec := range trinoMonitoringMetrics {
+		if !strings.Contains(spec.PromQL, "$ORG") {
+			t.Errorf("Trino monitoring metric %q does not require an org selector: %s", metric, spec.PromQL)
+		}
+	}
+}
+
+func TestTrinoMonitoringSeriesIsAllowListedOrgScopedAndNormalized(t *testing.T) {
+	var upstreamQuery string
+	prom := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		upstreamQuery = r.URL.Query().Get("query")
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write([]byte(`{"status":"success","data":{"resultType":"matrix","result":[{"metric":{"org":"org-a","state":"running","pod":"must-not-leak"},"values":[[1723456800,"3"],[1723456815,"4"]]}]}}`))
+	}))
+	defer prom.Close()
+	r := trinoMonitoringRouter(&fakeMonitoringStore{snapshot: monitoringTestSnapshot("org-a")}, nil, NewMetricsProxy(prom.URL), "internal-secret")
+
+	rec := httptest.NewRecorder()
+	r.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/api/v1/orgs/org-a/monitoring/trino/series?metric=queries_in_flight&window=6h", nil))
+
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status = %d, want 200: %s", rec.Code, rec.Body.String())
+	}
+	if !strings.Contains(upstreamQuery, `duckgres_trino_org_queries{org="org-a"}`) {
+		t.Fatalf("upstream PromQL is not fixed to org-a's Trino gauge: %s", upstreamQuery)
+	}
+	var got monitoringSeriesResponse
+	if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil {
+		t.Fatal(err)
+	}
+	if got.SchemaVersion != 1 || got.OrgID != "org-a" || got.Metric != "queries_in_flight" || got.Unit != "queries" {
+		t.Fatalf("series identity = %+v", got)
+	}
+	if len(got.Series) != 1 || len(got.Series[0].Points) != 2 || got.Series[0].Points[1].Value != 4 {
+		t.Fatalf("series shape = %+v", got.Series)
+	}
+	if len(got.Series[0].Labels) != 1 || got.Series[0].Labels["state"] != "running" {
+		t.Fatalf("labels = %v, want only state=running", got.Series[0].Labels)
+	}
+}
+
+func TestTrinoMonitoringSeriesRejectsBadRequestsBeforePrometheus(t *testing.T) {
+	var calls int
+	prom := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { calls++ }))
+	defer prom.Close()
+	r := trinoMonitoringRouter(&fakeMonitoringStore{snapshot: monitoringTestSnapshot("org-a")}, nil, NewMetricsProxy(prom.URL), "internal-secret")
+
+	for path, want := range map[string]int{
+		// DuckDB-worker metrics are not part of the Trino contract.
+		"/api/v1/orgs/org-a/monitoring/trino/series?metric=sessions_active&window=1h": http.StatusBadRequest,
+		"/api/v1/orgs/org-a/monitoring/trino/series?metric=acquire_p95&window=1h":     http.StatusBadRequest,
+		"/api/v1/orgs/org-a/monitoring/trino/series?metric=query_rate&window=2h":      http.StatusBadRequest,
+		"/api/v1/orgs/org-b/monitoring/trino/series?metric=query_rate&window=1h":      http.StatusNotFound,
+	} {
+		rec := httptest.NewRecorder()
+		r.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, path, nil))
+		if rec.Code != want {
+			t.Errorf("GET %s status = %d, want %d: %s", path, rec.Code, want, rec.Body.String())
+		}
+	}
+	if calls != 0 {
+		t.Fatalf("Prometheus called %d times for rejected requests, want 0", calls)
+	}
+}
+
+func TestTrinoMonitoringSeriesReportsUnconfiguredMetrics(t *testing.T) {
+	r := trinoMonitoringRouter(&fakeMonitoringStore{snapshot: monitoringTestSnapshot("org-a")}, nil, NewMetricsProxy(""), "internal-secret")
+
+	rec := httptest.NewRecorder()
+	r.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/api/v1/orgs/org-a/monitoring/trino/series?metric=query_rate&window=1h", nil))
+
+	if rec.Code != http.StatusServiceUnavailable {
+		t.Fatalf("status = %d, want 503: %s", rec.Code, rec.Body.String())
+	}
+}

From f88f24af092d5f593a5714fa0452a2a4582a3fc4 Mon Sep 17 00:00:00 2001
From: eric <eric@posthog.com>
Date: Thu, 1 Oct 2026 14:48:10 -0400
Subject: [PATCH 5/5] fix: serialize the Trino usage collector and log
 unavailable snapshots

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
---
 controlplane/admin/README.md           | 12 +++++----
 controlplane/admin/monitoring_trino.go | 11 +++++---
 controlplane/trino_org_metrics_test.go | 36 ++++++++++++++++++++++++++
 controlplane/trino_usage_collector.go  | 20 ++++++++++++--
 docs/metrics.md                        |  4 ++-
 5 files changed, 72 insertions(+), 11 deletions(-)

diff --git a/controlplane/admin/README.md b/controlplane/admin/README.md
index 489777c8..9a10afae 100644
--- a/controlplane/admin/README.md
+++ b/controlplane/admin/README.md
@@ -189,8 +189,9 @@ return 404 with `code: "managed_warehouse_not_found"` before Prometheus is calle
 #### Trino monitoring (`monitoring_trino.go`)
 
 The PostHog backend reads `GET /api/v1/orgs/:id/monitoring/trino/snapshot` and
-`/monitoring/trino/series` for organizations whose customers query through Trino. Auth and scoping follow the
-rules above: internal secret only, and the org is fixed by the path.
+`/monitoring/trino/series` for organizations whose customers query through
+Trino. Auth and scoping follow the rules above: internal secret only, and the
+org is fixed by the path.
 
 The snapshot reports the org's Trino lifecycle state (`not_enabled`, `pending`,
 `provisioning`, `ready`, `failed`), its tier's concurrency and queue limits,
@@ -200,9 +201,10 @@ totals are then zero and must not be shown as an idle warehouse. `in_flight`
 counts every non-terminal query. `running` counts every in-flight query past
 the queue, which is what the concurrency limit counts, so
 `in_flight = running + queued`. `blocked` is the subset of `running` whose
-drivers are all blocked. Query text has literals replaced with `?` and comments
-removed (`trino_sql_mask.go`), and becomes a fixed placeholder when it cannot
-be lexed. The snapshot intentionally omits cell ids, Trino principals,
+drivers are all blocked. `longest_running_ms` is the longest elapsed time since
+submission among all in-flight queries, queued ones included. Query text has
+literals replaced with `?` and comments removed (`trino_sql_mask.go`), and
+becomes a fixed placeholder when it cannot be lexed. The snapshot intentionally omits cell ids, Trino principals,
 resource-group and tier names, status messages, driver counts, error codes, and
 connection details.
 
diff --git a/controlplane/admin/monitoring_trino.go b/controlplane/admin/monitoring_trino.go
index 878f7572..7b1fdb74 100644
--- a/controlplane/admin/monitoring_trino.go
+++ b/controlplane/admin/monitoring_trino.go
@@ -3,6 +3,7 @@
 package admin
 
 import (
+	"log/slog"
 	"net/http"
 	"sort"
 	"strings"
@@ -115,7 +116,8 @@ type trinoMonitoringTotals struct {
 	Running int `json:"running"`
 	Queued  int `json:"queued"`
 	// Blocked is the subset of Running whose drivers are all blocked.
-	Blocked            int   `json:"blocked"`
+	Blocked int `json:"blocked"`
+	// LongestRunningMS is the longest time since submission among all in-flight queries, queued ones included.
 	LongestRunningMS   int64 `json:"longest_running_ms"`
 	PhysicalInputBytes int64 `json:"physical_input_bytes"`
 }
@@ -142,8 +144,8 @@ type trinoMonitoringSnapshotResponse struct {
 	OrgID         string               `json:"org_id"`
 	AsOf          time.Time            `json:"as_of"`
 	Trino         trinoMonitoringState `json:"trino"`
-	// Available is false when the coordinator could not be read. Totals are
-	// then zero, and a consumer must not present them as an idle warehouse.
+	// Available is false when the coordinator or the org index could not be
+	// read, or no configured cell owns the org. Totals are then zero, and a consumer must not present them as an idle warehouse.
 	Available        bool                   `json:"available"`
 	Limits           trinoMonitoringLimits  `json:"limits"`
 	Totals           trinoMonitoringTotals  `json:"totals"`
@@ -179,6 +181,7 @@ func (a *TrinoAPI) orgCell(orgID string) (*TrinoAPI, *configstore.ManagedWarehou
 		return nil, row, nil
 	}
 	if a.fleet == nil {
+		// Production always builds the fleet API, so this branch serves a single-cell API built directly with NewTrinoAPI.
 		return a, row, nil
 	}
 	for _, candidate := range a.fleet {
@@ -248,11 +251,13 @@ func (h *trinoMonitoringHandler) snapshot(c *gin.Context) {
 	// an empty list would read as an idle warehouse. Report it as unavailable.
 	idx, err := cell.index()
 	if err != nil {
+		slog.Warn("admin: trino monitoring snapshot unavailable: org index unreadable", "org", orgID, "error", err)
 		c.JSON(http.StatusOK, response)
 		return
 	}
 	queries, _, err := cell.liveQueries(c.Request.Context(), idx)
 	if err != nil {
+		slog.Warn("admin: trino monitoring snapshot unavailable: coordinator unreadable", "org", orgID, "error", err)
 		c.JSON(http.StatusOK, response)
 		return
 	}
diff --git a/controlplane/trino_org_metrics_test.go b/controlplane/trino_org_metrics_test.go
index e523e48f..e4002bf5 100644
--- a/controlplane/trino_org_metrics_test.go
+++ b/controlplane/trino_org_metrics_test.go
@@ -5,7 +5,9 @@ package controlplane
 import (
 	"context"
 	"errors"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/posthog/duckgres/controlplane/admin"
 	"github.com/posthog/duckgres/controlplane/configstore"
@@ -162,3 +164,37 @@ func TestTrinoOrgMetricsClearGaugesWhenThePollCannotBeTrusted(t *testing.T) {
 		}
 	})
 }
+
+// lockedTrinoUsageTracker lets two collector loops capture events without the
+// shared fake tracker's unguarded slice masking a race inside the collector.
+type lockedTrinoUsageTracker struct{ mu sync.Mutex }
+
+func (f *lockedTrinoUsageTracker) Capture(string, string, map[string]any) { f.mu.Lock(); f.mu.Unlock() }
+
+func (f *lockedTrinoUsageTracker) Close() {}
+
+func TestTrinoUsageCollectorToleratesOverlappingRuns(t *testing.T) {
+	collector, metrics, _ := newTrinoOrgMetricsTestCollector(t, &trinoUsageFakeCoordinator{queries: []admin.TrinoQuery{
+		{QueryID: "r1", State: "RUNNING", Principal: "tenant-a"},
+		{QueryID: "q1", State: "QUEUED", Principal: "tenant-a"},
+		{QueryID: "ok", State: "FINISHED", Principal: "tenant-a", ElapsedMS: 10},
+	}})
+	analytics.SetDefault(&lockedTrinoUsageTracker{})
+	collector.interval = time.Millisecond
+
+	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+	defer cancel()
+	var wg sync.WaitGroup
+	for range 2 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			collector.Run(ctx)
+		}()
+	}
+	wg.Wait()
+
+	if got := testutil.CollectAndCount(metrics.inFlight); got != 0 {
+		t.Fatalf("in-flight series after both loops exited = %d, want 0", got)
+	}
+}
diff --git a/controlplane/trino_usage_collector.go b/controlplane/trino_usage_collector.go
index 8f47c408..b0d67f11 100644
--- a/controlplane/trino_usage_collector.go
+++ b/controlplane/trino_usage_collector.go
@@ -5,6 +5,7 @@ package controlplane
 import (
 	"context"
 	"log/slog"
+	"sync"
 	"time"
 
 	"github.com/posthog/duckgres/controlplane/admin"
@@ -23,6 +24,9 @@ const (
 type trinoUsageTeamResolver func(orgID, username string) int64
 
 type trinoUsageCollector struct {
+	// mu guards seen and gauged: a new leader term can start before the
+	// previous term's loop has returned, so two loops may overlap.
+	mu          sync.Mutex
 	coordinator admin.TrinoCoordinatorClient
 	orgs        admin.TrinoOrgStore
 	teamID      trinoUsageTeamResolver
@@ -89,17 +93,19 @@ func (c *trinoUsageCollector) collect(ctx context.Context) {
 	if c == nil || c.coordinator == nil || c.orgs == nil {
 		return
 	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
 	orgs, err := c.orgs.ListTrinoEnabledOrgs()
 	if err != nil {
 		slog.Warn("Trino usage collection skipped: list enabled orgs failed", "error", err)
-		c.clearInFlight()
+		c.clearInFlightLocked()
 		return
 	}
 	owners := configstore.NewTrinoPrincipalOwners(orgs)
 	queries, err := c.coordinator.Queries(ctx)
 	if err != nil {
 		slog.Warn("Trino usage collection skipped: list queries failed", "error", err)
-		c.clearInFlight()
+		c.clearInFlightLocked()
 		return
 	}
 	now := c.now()
@@ -169,6 +175,16 @@ func (c *trinoUsageCollector) recordInFlight(orgs []configstore.TrinoEnabledOrg,
 }
 
 func (c *trinoUsageCollector) clearInFlight() {
+	if c == nil {
+		return
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.clearInFlightLocked()
+}
+
+// clearInFlightLocked requires c.mu to be held.
+func (c *trinoUsageCollector) clearInFlightLocked() {
 	if c == nil || c.metrics == nil {
 		return
 	}
diff --git a/docs/metrics.md b/docs/metrics.md
index 59dea706..4010f26b 100644
--- a/docs/metrics.md
+++ b/docs/metrics.md
@@ -489,7 +489,9 @@ One collector runs per Trino cell and writes the gauge only for the orgs
 assigned to its cell. The gauge is removed when a poll fails and when the
 collector stops, so a coordinator outage appears as a gap rather than a frozen
 value. During a leader handover two pods can export the gauge briefly; queries
-must take the maximum across pods, not the sum.
+must take the maximum across pods, not the sum. The counters and histograms
+are not removed when an org leaves a cell; their series stay until the process
+restarts.
 
 The counters and histograms are best-effort. The coordinator keeps finished
 queries only briefly, so a very busy cell can drop some between polls, and a