metrics: per-org worker-acquire latency by allocation source + axis fix

Two things (per request, same PR):

1. Worker-acquire latency by org and allocation phase. The acquire histograms
   (duckgres_worker_acquire_{total,phase,gate_wait}_seconds) already timed the
   wait but had no `org` label, and the end-to-end total wasn't tagged by HOW
   the worker was obtained. Now:
   - all three carry an `org` label (sliceable per tenant)
   - the total carries a `source` label — idle_reuse | hot_idle_claim | spawn |
     none — so "how long did org X wait, and did it need a cold spawn?" is a
     dashboard query. source is bound to the claim BEFORE completion so a failed
     spawn still attributes its wait to source=spawn (outcome=error).
   org is threaded from p.orgID / assignment.OrgID at every observe site.
   Two allow-listed admin panels expose it: acquire_p95 (p95 by source) and
   acquire_by_source (acquire rate by source), both org-scopable via $ORG.

2. Metrics chart Y-axis fix. The axis had no tickFormatter and a narrow fixed
   width, so large byte-rate values were clipped to a meaningless "00000". Add a
   unit-aware compact formatter (binary bytes for B/s, compact SI otherwise) for
   the tick + tooltip, and widen the axis.

Tests: acquire_metrics_test.go updated for the new labels (+ asserts a cold
spawn records source=spawn end-to-end); metrics_proxy_test validates the new
panels render cleanly; format.test.ts covers the compact/axis/value formatters;
harness asserts the acquire panels are advertised (raw histogram emission is
unit-tested — the :9090 port is NetworkPolicy-blocked in-Job).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NUq2EVxvKQFq3YEDNLF5HP

Author: fuziontech

controlplane/acquire_metrics.go

@@ -34,56 +34,74 @@ const (
 
 	acquireGateOutcomeAcquired = "acquired"
 	acquireGateOutcomeCanceled = "canceled"
+
+	// acquireSource* is how a completed AcquireWorker got its worker — the
+	// "phase that was allocated" dimension on the end-to-end total histogram:
+	// reusing this org's own idle Hot worker (near-instant), claiming a parked
+	// hot-idle worker from the shared pool (fast), or cold-spawning a fresh pod
+	// (slow — EC2 node boot). "none" when no worker was allocated (capacity
+	// backpressure / ctx cancel).
+	acquireSourceIdleReuse    = "idle_reuse"
+	acquireSourceHotIdleClaim = "hot_idle_claim"
+	acquireSourceSpawn        = "spawn"
+	acquireSourceNone         = "none"
 )
 
 // errHotIdleImageMismatch marks a hot-idle claim that yielded a worker of a
 // stale image (retired instead of reused) so the attempt is observed as an
 // error rather than dropped from the phase histogram.
 var errHotIdleImageMismatch = errors.New("hot-idle worker image mismatch")
 
+// All three histograms carry an `org` label so per-org acquire latency is
+// sliceable from a dashboard (which tenant is eating cold-spawn waits). Orgs are
+// bounded managed-warehouse tenants, so the added cardinality is acceptable.
+
 var workerAcquireGateWaitHistogram = promauto.NewHistogramVec(prometheus.HistogramOpts{
 	Name:    "duckgres_worker_acquire_gate_wait_seconds",
-	Help:    "Time a connection spent blocked in the per-org FIFO acquire gate (orgAcquireGate) before owning the slow acquisition path, partitioned by outcome (acquired|canceled).",
+	Help:    "Time a connection spent blocked in the per-org FIFO acquire gate (orgAcquireGate) before owning the slow acquisition path, partitioned by org and outcome (acquired|canceled).",
 	Buckets: acquirePhaseBuckets,
-}, []string{"outcome"})
+}, []string{"org", "outcome"})
 
 var workerAcquirePhaseHistogram = promauto.NewHistogramVec(prometheus.HistogramOpts{
 	Name:    "duckgres_worker_acquire_phase_seconds",
-	Help:    "Duration of individual worker-acquire phases on the remote/k8s backend, partitioned by phase (hot_idle_claim|spawn|activate) and outcome (ok|error).",
+	Help:    "Duration of individual worker-acquire phases on the remote/k8s backend, partitioned by org, phase (hot_idle_claim|spawn|activate) and outcome (ok|error).",
 	Buckets: acquirePhaseBuckets,
-}, []string{"phase", "outcome"})
+}, []string{"org", "phase", "outcome"})
 
 var workerAcquireTotalHistogram = promauto.NewHistogramVec(prometheus.HistogramOpts{
 	Name:    "duckgres_worker_acquire_total_seconds",
-	Help:    "End-to-end OrgReservedPool.AcquireWorker duration, partitioned by outcome (ok|capacity|error|canceled).",
+	Help:    "End-to-end OrgReservedPool.AcquireWorker duration (the time a pending session waits for a worker), partitioned by org, the allocation source (idle_reuse|hot_idle_claim|spawn|none) and outcome (ok|capacity|error|canceled).",
 	Buckets: acquirePhaseBuckets,
-}, []string{"outcome"})
+}, []string{"org", "source", "outcome"})
 
-func observeAcquireGateWait(d time.Duration, outcome string) {
+func observeAcquireGateWait(d time.Duration, org, outcome string) {
 	if d < 0 {
 		d = 0
 	}
-	workerAcquireGateWaitHistogram.WithLabelValues(outcome).Observe(d.Seconds())
+	workerAcquireGateWaitHistogram.WithLabelValues(org, outcome).Observe(d.Seconds())
 }
 
 // observeAcquirePhase records one attempt of a single acquire phase. err==nil
 // observes outcome=ok, otherwise outcome=error.
-func observeAcquirePhase(phase string, d time.Duration, err error) {
+func observeAcquirePhase(phase, org string, d time.Duration, err error) {
 	if d < 0 {
 		d = 0
 	}
 	outcome := acquireOutcomeOK
 	if err != nil {
 		outcome = acquireOutcomeError
 	}
-	workerAcquirePhaseHistogram.WithLabelValues(phase, outcome).Observe(d.Seconds())
+	workerAcquirePhaseHistogram.WithLabelValues(org, phase, outcome).Observe(d.Seconds())
 }
 
-func observeAcquireTotal(d time.Duration, outcome string) {
+// observeAcquireTotal records the end-to-end acquire latency for one session,
+// tagged by the org, how the worker was ultimately obtained (source), and the
+// outcome.
+func observeAcquireTotal(d time.Duration, org, source, outcome string) {
 	if d < 0 {
 		d = 0
 	}
-	workerAcquireTotalHistogram.WithLabelValues(outcome).Observe(d.Seconds())
+	workerAcquireTotalHistogram.WithLabelValues(org, source, outcome).Observe(d.Seconds())
 }
 
 // acquireTotalOutcome classifies an AcquireWorker error for the end-to-end

controlplane/acquire_metrics_test.go

@@ -14,31 +14,34 @@ import (
 // label combination they can emit) so a registration conflict or label-arity
 // mistake panics here instead of in production.
 func TestAcquireMetricsObserveHelpers(t *testing.T) {
-	gateBefore := histogramVecLabelSampleCount(t, workerAcquireGateWaitHistogram, acquireGateOutcomeAcquired)
-	observeAcquireGateWait(125*time.Millisecond, acquireGateOutcomeAcquired)
-	observeAcquireGateWait(-1*time.Second, acquireGateOutcomeCanceled) // negative durations clamp to 0
-	if got := histogramVecLabelSampleCount(t, workerAcquireGateWaitHistogram, acquireGateOutcomeAcquired); got != gateBefore+1 {
+	const org = "metrics-test-org"
+	gateBefore := histogramVecLabelSampleCount(t, workerAcquireGateWaitHistogram, org, acquireGateOutcomeAcquired)
+	observeAcquireGateWait(125*time.Millisecond, org, acquireGateOutcomeAcquired)
+	observeAcquireGateWait(-1*time.Second, org, acquireGateOutcomeCanceled) // negative durations clamp to 0
+	if got := histogramVecLabelSampleCount(t, workerAcquireGateWaitHistogram, org, acquireGateOutcomeAcquired); got != gateBefore+1 {
 		t.Fatalf("gate wait acquired samples = %d, want %d", got, gateBefore+1)
 	}
 
 	for _, phase := range []string{acquirePhaseHotIdleClaim, acquirePhaseSpawn, acquirePhaseActivate} {
-		okBefore := histogramVecLabelSampleCount(t, workerAcquirePhaseHistogram, phase, acquireOutcomeOK)
-		errBefore := histogramVecLabelSampleCount(t, workerAcquirePhaseHistogram, phase, acquireOutcomeError)
-		observeAcquirePhase(phase, 50*time.Millisecond, nil)
-		observeAcquirePhase(phase, 50*time.Millisecond, errors.New("boom"))
-		if got := histogramVecLabelSampleCount(t, workerAcquirePhaseHistogram, phase, acquireOutcomeOK); got != okBefore+1 {
+		okBefore := histogramVecLabelSampleCount(t, workerAcquirePhaseHistogram, org, phase, acquireOutcomeOK)
+		errBefore := histogramVecLabelSampleCount(t, workerAcquirePhaseHistogram, org, phase, acquireOutcomeError)
+		observeAcquirePhase(phase, org, 50*time.Millisecond, nil)
+		observeAcquirePhase(phase, org, 50*time.Millisecond, errors.New("boom"))
+		if got := histogramVecLabelSampleCount(t, workerAcquirePhaseHistogram, org, phase, acquireOutcomeOK); got != okBefore+1 {
 			t.Fatalf("phase %q ok samples = %d, want %d", phase, got, okBefore+1)
 		}
-		if got := histogramVecLabelSampleCount(t, workerAcquirePhaseHistogram, phase, acquireOutcomeError); got != errBefore+1 {
+		if got := histogramVecLabelSampleCount(t, workerAcquirePhaseHistogram, org, phase, acquireOutcomeError); got != errBefore+1 {
 			t.Fatalf("phase %q error samples = %d, want %d", phase, got, errBefore+1)
 		}
 	}
 
-	for _, outcome := range []string{acquireOutcomeOK, acquireOutcomeCapacity, acquireOutcomeError, acquireOutcomeCanceled} {
-		before := histogramVecLabelSampleCount(t, workerAcquireTotalHistogram, outcome)
-		observeAcquireTotal(time.Second, outcome)
-		if got := histogramVecLabelSampleCount(t, workerAcquireTotalHistogram, outcome); got != before+1 {
-			t.Fatalf("total outcome %q samples = %d, want %d", outcome, got, before+1)
+	for _, source := range []string{acquireSourceIdleReuse, acquireSourceHotIdleClaim, acquireSourceSpawn, acquireSourceNone} {
+		for _, outcome := range []string{acquireOutcomeOK, acquireOutcomeCapacity, acquireOutcomeError, acquireOutcomeCanceled} {
+			before := histogramVecLabelSampleCount(t, workerAcquireTotalHistogram, org, source, outcome)
+			observeAcquireTotal(time.Second, org, source, outcome)
+			if got := histogramVecLabelSampleCount(t, workerAcquireTotalHistogram, org, source, outcome); got != before+1 {
+				t.Fatalf("total source=%q outcome=%q samples = %d, want %d", source, outcome, got, before+1)
+			}
 		}
 	}
 }
@@ -67,10 +70,13 @@ func TestAcquireTotalOutcomeClassification(t *testing.T) {
 // acquisition (no idle worker → gate → spawn → activate) records one sample in
 // each phase histogram plus the gate-wait and end-to-end histograms.
 func TestOrgReservedPoolAcquireObservesPhaseMetrics(t *testing.T) {
-	gateBefore := histogramVecLabelSampleCount(t, workerAcquireGateWaitHistogram, acquireGateOutcomeAcquired)
-	spawnBefore := histogramVecLabelSampleCount(t, workerAcquirePhaseHistogram, acquirePhaseSpawn, acquireOutcomeOK)
-	activateBefore := histogramVecLabelSampleCount(t, workerAcquirePhaseHistogram, acquirePhaseActivate, acquireOutcomeOK)
-	totalBefore := histogramVecLabelSampleCount(t, workerAcquireTotalHistogram, acquireOutcomeOK)
+	const org = "analytics"
+	gateBefore := histogramVecLabelSampleCount(t, workerAcquireGateWaitHistogram, org, acquireGateOutcomeAcquired)
+	spawnBefore := histogramVecLabelSampleCount(t, workerAcquirePhaseHistogram, org, acquirePhaseSpawn, acquireOutcomeOK)
+	activateBefore := histogramVecLabelSampleCount(t, workerAcquirePhaseHistogram, org, acquirePhaseActivate, acquireOutcomeOK)
+	// A no-idle-worker acquire cold-spawns, so the end-to-end total lands under
+	// source=spawn (proving the source label tracks the allocation path).
+	totalBefore := histogramVecLabelSampleCount(t, workerAcquireTotalHistogram, org, acquireSourceSpawn, acquireOutcomeOK)
 
 	shared, _ := newTestK8sPool(t, 5)
 	shared.healthCheckFunc = func(ctx context.Context, worker *ManagedWorker) error {
@@ -83,7 +89,7 @@ func TestOrgReservedPoolAcquireObservesPhaseMetrics(t *testing.T) {
 		return nil
 	}
 
-	pool := NewOrgReservedPool(shared, "analytics", 2, shared.workerImage, nil)
+	pool := NewOrgReservedPool(shared, org, 2, shared.workerImage, nil)
 	pool.activateReservedWorker = func(ctx context.Context, worker *ManagedWorker) error {
 		return nil
 	}
@@ -94,16 +100,16 @@ func TestOrgReservedPoolAcquireObservesPhaseMetrics(t *testing.T) {
 		t.Fatalf("AcquireWorker: %v", err)
 	}
 
-	if got := histogramVecLabelSampleCount(t, workerAcquireGateWaitHistogram, acquireGateOutcomeAcquired); got != gateBefore+1 {
+	if got := histogramVecLabelSampleCount(t, workerAcquireGateWaitHistogram, org, acquireGateOutcomeAcquired); got != gateBefore+1 {
 		t.Errorf("gate wait acquired samples = %d, want %d", got, gateBefore+1)
 	}
-	if got := histogramVecLabelSampleCount(t, workerAcquirePhaseHistogram, acquirePhaseSpawn, acquireOutcomeOK); got != spawnBefore+1 {
+	if got := histogramVecLabelSampleCount(t, workerAcquirePhaseHistogram, org, acquirePhaseSpawn, acquireOutcomeOK); got != spawnBefore+1 {
 		t.Errorf("spawn ok samples = %d, want %d", got, spawnBefore+1)
 	}
-	if got := histogramVecLabelSampleCount(t, workerAcquirePhaseHistogram, acquirePhaseActivate, acquireOutcomeOK); got != activateBefore+1 {
+	if got := histogramVecLabelSampleCount(t, workerAcquirePhaseHistogram, org, acquirePhaseActivate, acquireOutcomeOK); got != activateBefore+1 {
 		t.Errorf("activate ok samples = %d, want %d", got, activateBefore+1)
 	}
-	if got := histogramVecLabelSampleCount(t, workerAcquireTotalHistogram, acquireOutcomeOK); got != totalBefore+1 {
-		t.Errorf("total ok samples = %d, want %d", got, totalBefore+1)
+	if got := histogramVecLabelSampleCount(t, workerAcquireTotalHistogram, org, acquireSourceSpawn, acquireOutcomeOK); got != totalBefore+1 {
+		t.Errorf("total spawn/ok samples = %d, want %d", got, totalBefore+1)
 	}
 }

controlplane/admin/metrics_proxy.go

@@ -48,6 +48,12 @@ var rangePanels = map[string]string{
 	"s3_bytes_rate":   `sum(rate(duckgres_s3_bytes_read_total$ORG[$WIN]))`,
 	"worker_states":   `sum by (state) (duckgres_worker_lifecycle_count)`,
 	"queue_depth":     `sum(duckgres_control_plane_worker_queue_depth)`,
+	// Worker-acquire latency: how long a pending session waits for a worker,
+	// split by the allocation source (idle_reuse|hot_idle_claim|spawn). p95 for
+	// the tail, plus the rate of acquisitions per source so cold-spawn frequency
+	// is visible. $ORG scopes both to a single org.
+	"acquire_p95":       `histogram_quantile(0.95, sum by (le, source) (rate(duckgres_worker_acquire_total_seconds_bucket$ORG[$WIN])))`,
+	"acquire_by_source": `sum by (source) (rate(duckgres_worker_acquire_total_seconds_count$ORG[$WIN]))`,
 }
 
 // renderPanel substitutes the named tokens into a panel template. $ORGERR is

controlplane/admin/ui/src/lib/format.test.ts

@@ -0,0 +1,46 @@
+import { describe, expect, it } from "vitest";
+import { fmtCompact, fmtMetricAxis, fmtMetricValue } from "./format";
+
+describe("fmtCompact", () => {
+  it("renders compact SI so large numbers don't overflow an axis", () => {
+    expect(fmtCompact(20_000_000)).toBe("20M");
+    expect(fmtCompact(1500)).toBe("1.5K");
+    expect(fmtCompact(0)).toBe("0");
+    expect(fmtCompact(0.42)).toBe("0.4");
+  });
+  it("handles nullish", () => {
+    expect(fmtCompact(null)).toBe("—");
+    expect(fmtCompact(undefined)).toBe("—");
+    expect(fmtCompact(NaN)).toBe("—");
+  });
+});
+
+describe("fmtMetricAxis", () => {
+  it("formats byte units with binary prefixes (the S3 bytes-rate axis)", () => {
+    // 20,000,000 B ≈ 19.1 MiB — the bug was this rendering as a clipped '00000'.
+    expect(fmtMetricAxis(20_000_000, "B/s")).toBe("19.1 MB");
+    expect(fmtMetricAxis(1024, "B")).toBe("1.0 KB");
+  });
+  it("formats non-byte units compactly", () => {
+    expect(fmtMetricAxis(20_000_000, "ops/s")).toBe("20M");
+    expect(fmtMetricAxis(0.5, "")).toBe("0.5");
+  });
+  it("returns empty string for nullish (recharts skips the tick)", () => {
+    expect(fmtMetricAxis(null, "B/s")).toBe("");
+    expect(fmtMetricAxis(NaN)).toBe("");
+  });
+});
+
+describe("fmtMetricValue", () => {
+  it("byte rate gets a /s suffix; plain bytes don't", () => {
+    expect(fmtMetricValue(20_000_000, "B/s")).toBe("19.1 MB/s");
+    expect(fmtMetricValue(1024, "B")).toBe("1.0 KB");
+  });
+  it("non-byte value keeps its unit label", () => {
+    expect(fmtMetricValue(1500, "ops/s")).toBe("1.5K ops/s");
+    expect(fmtMetricValue(0.42, "")).toBe("0.4");
+  });
+  it("nullish renders a dash", () => {
+    expect(fmtMetricValue(null, "B/s")).toBe("—");
+  });
+});

controlplane/admin/ui/src/lib/format.ts

@@ -48,6 +48,33 @@ export function fmtBytes(n: number | null | undefined): string {
   return `${v.toFixed(v >= 100 || i === 0 ? 0 : 1)} ${units[i]}`;
 }
 
+// fmtCompact renders a number in compact SI notation (1.5K, 20M, 3.2B). Keeps a
+// wide-ranging metric axis readable and NARROW — a raw 20000000 would be clipped
+// to "00000" by a fixed-width axis.
+export function fmtCompact(n: number | null | undefined): string {
+  if (n == null || Number.isNaN(n)) return "—";
+  return new Intl.NumberFormat("en-US", { notation: "compact", maximumFractionDigits: 1 }).format(n);
+}
+
+// fmtMetricAxis compacts a metric value for a chart Y-axis tick. Byte units use
+// binary prefixes (19 MB); everything else uses compact SI (1.5K, 20M). Returns
+// "" for nullish so recharts skips the tick rather than drawing a dash.
+export function fmtMetricAxis(v: number | null | undefined, unit?: string): string {
+  if (v == null || Number.isNaN(v)) return "";
+  if (unit === "B/s" || unit === "B") return fmtBytes(v);
+  return fmtCompact(v);
+}
+
+// fmtMetricValue is the full "value + unit" string for a metric tooltip: byte
+// units render as bytes (with a /s rate suffix when applicable), everything else
+// as a compact number followed by its unit label.
+export function fmtMetricValue(v: number | null | undefined, unit?: string): string {
+  if (v == null || Number.isNaN(v)) return "—";
+  if (unit === "B/s") return `${fmtBytes(v)}/s`;
+  if (unit === "B") return fmtBytes(v);
+  return `${fmtCompact(v)}${unit ? ` ${unit}` : ""}`;
+}
+
 // fmtDurationMs renders a MILLISECOND duration as a compact human string
 // (250ms, 3.4s, 2m 5s, 1h 3m). <=0 / nullish → "—". Used for running-query
 // elapsed time in the Live view + detail dialog. (fmtDuration below takes

controlplane/admin/ui/src/pages/Metrics.tsx

@@ -14,7 +14,7 @@ import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card";
 import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select";
 import { LoadingState } from "@/components/states";
 import { useMetricRange, useMetricsPanels, useOrgs } from "@/hooks/useApi";
-import { promToSeries } from "@/lib/format";
+import { fmtMetricAxis, fmtMetricValue, promToSeries } from "@/lib/format";
 
 // Window selector. The backend caps the step at ~250 points per window.
 const WINDOWS = ["15m", "1h", "6h", "24h"];
@@ -30,6 +30,8 @@ const PANELS: { key: string; title: string; unit: string }[] = [
   { key: "s3_bytes_rate", title: "S3 read bytes rate", unit: "B/s" },
   { key: "worker_states", title: "Workers by state", unit: "" },
   { key: "queue_depth", title: "Connection queue depth", unit: "" },
+  { key: "acquire_p95", title: "Worker acquire p95 by source", unit: "s" },
+  { key: "acquire_by_source", title: "Worker acquire rate by source", unit: "ops/s" },
 ];
 
 export function Metrics() {
@@ -165,7 +167,12 @@ function MetricCard({
                 stroke="hsl(var(--muted-foreground))"
                 fontSize={10}
               />
-              <YAxis stroke="hsl(var(--muted-foreground))" fontSize={10} width={48} />
+              <YAxis
+                stroke="hsl(var(--muted-foreground))"
+                fontSize={10}
+                width={56}
+                tickFormatter={(v: number) => fmtMetricAxis(v, unit)}
+              />
               <RTooltip
                 contentStyle={{
                   background: "hsl(var(--popover))",
@@ -174,7 +181,7 @@ function MetricCard({
                   fontSize: 12,
                 }}
                 labelFormatter={(t) => new Date(t as number).toLocaleString()}
-                formatter={(v: number, name) => [`${v}${unit ? ` ${unit}` : ""}`, name]}
+                formatter={(v: number, name) => [fmtMetricValue(v, unit), name]}
               />
               {keys.map((k, i) => (
                 <Line

controlplane/k8s_pool_acquire.go

@@ -427,7 +427,7 @@ func (p *K8sWorkerPool) completeSharedWorkerReservation(ctx context.Context, cla
 		// attempt.
 		hotClaimStart := time.Now()
 		worker, reserveErr := p.reserveClaimedWorker(ctx, claim.hotClaimed, assignment)
-		observeAcquirePhase("hot_idle_claim", time.Since(hotClaimStart), reserveErr)
+		observeAcquirePhase("hot_idle_claim", assignment.OrgID, time.Since(hotClaimStart), reserveErr)
 		if reserveErr == nil {
 			worker.hotIdleReclaimed = true
 			return worker, false, nil

controlplane/k8s_pool_spawn.go

@@ -558,7 +558,7 @@ func (p *K8sWorkerPool) spawnReservedWorkerForSlot(ctx context.Context, id int,
 	// connect → reserve. Named-return defer so every failure stage observes.
 	spawnStart := time.Now()
 	defer func() {
-		observeAcquirePhase("spawn", time.Since(spawnStart), err)
+		observeAcquirePhase("spawn", assignment.OrgID, time.Since(spawnStart), err)
 	}()
 	// A default (nil profile) request spawns a default-sized worker: the zero
 	// profile makes workerResourcesForProfile fall back to the pool-global request.