fix(metrics): make pool_name mean one thing across every gate series

README.md

@@ -314,6 +314,9 @@ For more fine-grained control, configure gates per queue in your configuration f
     The result is used directly as the dispatch budget (no transformation is applied).
   - `fallback` (optional): Fallback budget value (0.0-1.0) returned when the query fails or returns no data.
     Default is `0.0` (fail closed).
+  - `pool` (optional): The InferencePool the query is about. Purely descriptive — it does not
+    affect the query, it only sets the `inference_pool` label on `async_gate_metric_value` and
+    `async_gate_metric_threshold` so you can tell which pool a gauge is reporting on.
 
 - `endpoint-scrape`: Scrapes a raw Prometheus text-format `/metrics` endpoint directly.
   Computes budget as `clamp(1 - saturation - baseline, 0, 1)`.
@@ -572,6 +575,8 @@ The Async Processor exposes Prometheus metrics under the `llm_d_async` subsystem
 | `llm_d_async_async_dispatch_budget` | Gauge | Current dispatch budget [0.0–1.0] returned by the queue's gate; the fraction of system capacity available for new requests (0.0 = gate fully closed). Useful for diagnosing why throughput is throttled. |
 | `llm_d_async_async_pool_worker_limit` | Gauge | Configured worker concurrency limit for a pool (carries only the `pool_name` label). Compare against `llm_d_async_async_inflight_requests` to compute worker utilization. |
 | `llm_d_async_async_gate_decisions_total` | Counter | Count of gate decisions that prevented a message from being dispatched, by `reason`: `gate_closed` (no dispatch budget), `quota_exhausted` (per-attribute quota overflow), `dropped` (gate permanently rejected the request), `error` (gate evaluation failed). |
+| `llm_d_async_async_gate_metric_value` | Gauge | Raw metric value a metric-based gate (`prometheus-saturation`/`-budget`/`-query`) last read — the number compared against the threshold below. For the saturation gate this is `1 - saturation`. |
+| `llm_d_async_async_gate_metric_threshold` | Gauge | Threshold the value above is compared against. The gate closes when `value <= threshold`, which is what drives `async_dispatch_budget` to 0. |
 
 **Labels:**
 
@@ -581,6 +586,14 @@ The Async Processor exposes Prometheus metrics under the `llm_d_async` subsystem
 | `queue_name` | Logical queue name from the queue configuration |
 | `pool_name` | Worker pool the queue routes to (`async_pool_worker_limit` carries only this label) |
 | `reason` | Gate-decision reason (only on `async_gate_decisions_total`): `gate_closed`, `quota_exhausted`, `dropped`, `error` |
+| `inference_pool` | InferencePool a gate queries (only on the two `async_gate_metric_*` gauges), from the gate's `pool` param. Empty when the gate does not name one. |
+
+`pool_name` always names the **async worker pool** that owns the series, never the
+InferencePool a gate happens to query — that is what `inference_pool` is for. Every
+per-queue series therefore carries the same `queue_id`/`queue_name`/`pool_name`
+triple and joins on it, including the two gate gauges. A **pool-level** gate (one
+configured on a worker pool rather than a queue) has no single queue, so its gauges
+carry an empty `queue_id` and `queue_name` and are keyed by `pool_name` alone.
 
 **Example PromQL queries:**
 
@@ -599,6 +612,18 @@ histogram_quantile(0.95, sum by (queue_name, le) (rate(llm_d_async_async_inferen
 
 # p95 queue residence time by queue (async delay, excluding model time)
 histogram_quantile(0.95, sum by (queue_name, le) (rate(llm_d_async_async_queue_residence_time_millis_bucket[5m])))
+
+# Why is a queue's gate closed? The gauges join on the queue triple, so you can
+# put the budget, the value it came from, and the threshold on one panel.
+llm_d_async_async_dispatch_budget
+llm_d_async_async_gate_metric_value
+llm_d_async_async_gate_metric_threshold
+
+# How much headroom does each queue's gate have?
+llm_d_async_async_gate_metric_value - on(queue_id, queue_name, pool_name) llm_d_async_async_gate_metric_threshold
+
+# Throttling rate against the pool it is throttling
+sum by (pool_name) (rate(llm_d_async_async_gate_decisions_total{reason="gate_closed"}[5m]))
 ```
 
 ## Implementations

pipeline/gate.go

@@ -95,10 +95,24 @@ func ApplyChain(ctx context.Context, msg *api.InternalRequest, gates []Gate, rel
 	return Continue(), nil
 }
 
+// GateOwner identifies the queue or worker pool a gate was created for. It is
+// stamped by the message queue implementation at construction time and is never
+// read from user config, so a gate's own metrics can carry the same
+// queue_id/queue_name/pool_name labels as the owner's other series. A pool-level
+// gate leaves the queue fields empty.
+type GateOwner struct {
+	QueueID      string
+	QueueName    string
+	WorkerPoolID string
+}
+
 // GateConfig holds the configuration for a single gate instance.
 type GateConfig struct {
 	GateType   string         `json:"gate_type,omitempty"`
 	GateParams map[string]any `json:"gate_params,omitempty"`
+
+	// Owner is set by the caller of CreateGate, not deserialized from config.
+	Owner GateOwner `json:"-"`
 }
 
 // GateFactory defines the interface for creating Gate instances.

pkg/async/inference/flowcontrol/gate_factory.go

@@ -109,6 +109,7 @@ func (f *GateFactory) CreateGate(cfg pipeline.GateConfig) (pipeline.Gate, error)
 
 		var innerGates []pipeline.Gate
 		for _, innerCfg := range configs {
+			innerCfg.Owner = cfg.Owner
 			gate, err := f.CreateGate(innerCfg)
 			if err != nil {
 				return nil, fmt.Errorf("composite gate failed to create inner gate %q: %w", innerCfg.GateType, err)
@@ -124,6 +125,7 @@ func (f *GateFactory) CreateGate(cfg pipeline.GateConfig) (pipeline.Gate, error)
 			return nil, err
 		}
 
+		innerCfg.Owner = cfg.Owner
 		innerGate, err := f.CreateGate(innerCfg)
 		if err != nil {
 			return nil, fmt.Errorf("wait-on-refuse gate failed to create inner gate %q: %w", innerCfg.GateType, err)
@@ -144,6 +146,7 @@ func (f *GateFactory) CreateGate(cfg pipeline.GateConfig) (pipeline.Gate, error)
 		satGate, err := f.CreateGate(pipeline.GateConfig{
 			GateType:   satGateType,
 			GateParams: satGateParams,
+			Owner:      cfg.Owner,
 		})
 		if err != nil {
 			return nil, fmt.Errorf("tier-priority-admission gate failed to create saturation gate %q: %w", satGateType, err)
@@ -230,7 +233,8 @@ func (f *GateFactory) CreateGate(cfg pipeline.GateConfig) (pipeline.Gate, error)
 			ms = NewCachedMetricSource(source, f.cacheTTL)
 		}
 		return NewSaturationDispatchGate(ms, threshold, fallback).
-			WithPoolLabel(paramString(params, "pool", "")), nil
+			WithOwner(cfg.Owner).
+			WithInferencePool(paramString(params, "pool", "")), nil
 
 	case "prometheus-budget":
 		if f.prometheusURL == "" {
@@ -276,7 +280,9 @@ func (f *GateFactory) CreateGate(cfg pipeline.GateConfig) (pipeline.Gate, error)
 			cachedSource(primary, f.cacheTTL),
 			cachedSource(secondary, f.cacheTTL),
 		)
-		return NewBudgetDispatchGate(ms, baseline, fallback).WithPoolLabel(pool), nil
+		return NewBudgetDispatchGate(ms, baseline, fallback).
+			WithOwner(cfg.Owner).
+			WithInferencePool(pool), nil
 
 	case "prometheus-query":
 		if f.prometheusURL == "" {
@@ -298,10 +304,12 @@ func (f *GateFactory) CreateGate(cfg pipeline.GateConfig) (pipeline.Gate, error)
 		if err != nil {
 			return nil, err
 		}
-		// Optional 'pool' param labels the async_gate_metric_value gauge; set it to
-		// match the inference pool referenced in the query.
+		// Optional 'pool' param records which InferencePool the query is about, as
+		// the inference_pool label on async_gate_metric_value; it does not affect
+		// the query. pool_name comes from the queue or pool that owns the gate.
 		return NewMetricDispatchGate(cachedSource(source, f.cacheTTL), 0.0, fallback).
-			WithPoolLabel(paramString(params, "pool", "")), nil
+			WithOwner(cfg.Owner).
+			WithInferencePool(paramString(params, "pool", "")), nil
 
 	case "endpoint-scrape":
 		url := paramString(params, "url", "")

pkg/async/inference/flowcontrol/gate_factory_test.go

@@ -298,3 +298,82 @@ func TestGateFactory_PrometheusQueryGateWithAllParams(t *testing.T) {
 	assert.NoError(t, err, "should create gate with all params specified")
 	assert.NotNil(t, gate)
 }
+
+// TestGateFactory_StampsOwnerOnMetricGate checks that the owning queue and worker
+// pool reach the gauges a metric gate records, and that the 'pool' param lands on
+// inference_pool instead of overwriting pool_name (issue #369).
+func TestGateFactory_StampsOwnerOnMetricGate(t *testing.T) {
+	owner := pipeline.GateOwner{QueueID: "team-a-premium", QueueName: "queue:a", WorkerPoolID: "model-a-pool"}
+	factory := NewGateFactory("http://localhost:9090")
+
+	gate, err := factory.CreateGate(pipeline.GateConfig{
+		GateType:   "prometheus-query",
+		GateParams: map[string]any{"query": "up", "pool": "optimized-baseline"},
+		Owner:      owner,
+	})
+	assert.NoError(t, err)
+
+	metricGate, ok := gate.(*MetricDispatchGate)
+	assert.True(t, ok, "prometheus-query should produce a MetricDispatchGate")
+	assert.Equal(t, owner, metricGate.owner)
+	assert.Equal(t, "optimized-baseline", metricGate.inferencePool)
+}
+
+// TestGateFactory_PropagatesOwnerThroughWrappers checks that the owner survives the
+// factory's recursive gate types — a gate nested inside wait-on-refuse inside
+// composite still labels its metrics with the queue that owns it (issue #369).
+func TestGateFactory_PropagatesOwnerThroughWrappers(t *testing.T) {
+	owner := pipeline.GateOwner{QueueID: "team-b-standard", QueueName: "queue:b", WorkerPoolID: "model-b-pool"}
+	factory := NewGateFactory("http://localhost:9090")
+
+	gate, err := factory.CreateGate(pipeline.GateConfig{
+		GateType: "composite",
+		GateParams: map[string]any{
+			"gates": []any{
+				map[string]any{
+					"gate_type": "wait-on-refuse",
+					"gate_params": map[string]any{
+						"gate": map[string]any{
+							"gate_type":   "prometheus-query",
+							"gate_params": map[string]any{"query": "up"},
+						},
+					},
+				},
+			},
+		},
+		Owner: owner,
+	})
+	assert.NoError(t, err)
+
+	composite, ok := gate.(*CompositeGate)
+	assert.True(t, ok)
+	assert.Len(t, composite.gates, 1)
+	waiter, ok := composite.gates[0].(*WaitOnRefuseGate)
+	assert.True(t, ok)
+	metricGate, ok := waiter.inner.(*MetricDispatchGate)
+	assert.True(t, ok)
+	assert.Equal(t, owner, metricGate.owner)
+}
+
+// TestGateFactory_PropagatesOwnerToSaturationGate covers the third recursion site,
+// tier-priority-admission's inner saturation gate (issue #369).
+func TestGateFactory_PropagatesOwnerToSaturationGate(t *testing.T) {
+	owner := pipeline.GateOwner{QueueID: "team-c-batch", QueueName: "queue:c", WorkerPoolID: "model-c-pool"}
+	factory := NewGateFactory("http://localhost:9090")
+
+	gate, err := factory.CreateGate(pipeline.GateConfig{
+		GateType: "tier-priority-admission",
+		GateParams: map[string]any{
+			"saturation_gate":        "prometheus-query",
+			"saturation_gate_params": map[string]any{"query": "up"},
+		},
+		Owner: owner,
+	})
+	assert.NoError(t, err)
+
+	tierGate, ok := gate.(*TierPriorityAdmissionGate)
+	assert.True(t, ok)
+	metricGate, ok := tierGate.saturationGate.(*MetricDispatchGate)
+	assert.True(t, ok)
+	assert.Equal(t, owner, metricGate.owner)
+}

pkg/async/inference/flowcontrol/metric_dispatch_gate.go

@@ -36,16 +36,27 @@ var _ pipeline.Gate = (*MetricDispatchGate)(nil)
 // formula N = max_SYS × (D − B) when threshold is set to the reserved baseline B.
 // On error or missing/invalid data, the gate returns the configured fallback budget.
 type MetricDispatchGate struct {
-	source    MetricSource
-	threshold float64
-	fallback  float64
-	poolLabel string
+	source        MetricSource
+	threshold     float64
+	fallback      float64
+	owner         pipeline.GateOwner
+	inferencePool string
 }
 
-// WithPoolLabel sets the pool name used to label the async_gate_metric_value and
-// async_gate_metric_threshold gauges this gate records on each evaluation.
-func (g *MetricDispatchGate) WithPoolLabel(pool string) *MetricDispatchGate {
-	g.poolLabel = pool
+// WithOwner sets the queue or worker pool this gate belongs to. Its
+// queue_id/queue_name/pool_name label the async_gate_metric_value and
+// async_gate_metric_threshold gauges, so they join with the owner's other
+// series (async_dispatch_budget, async_gate_decisions_total, ...).
+func (g *MetricDispatchGate) WithOwner(owner pipeline.GateOwner) *MetricDispatchGate {
+	g.owner = owner
+	return g
+}
+
+// WithInferencePool sets the InferencePool this gate queries, exposed as the
+// inference_pool label on the gauges above. It is what the gate measures, not
+// who it throttles — pool_name is the latter.
+func (g *MetricDispatchGate) WithInferencePool(pool string) *MetricDispatchGate {
+	g.inferencePool = pool
 	return g
 }
 
@@ -102,7 +113,7 @@ func (g *MetricDispatchGate) Budget(ctx context.Context) float64 {
 
 	// Expose the raw value and threshold so operators can see why the gate is
 	// open/closed (it closes when value <= threshold).
-	metrics.SetGateMetricValue(value, g.threshold, g.poolLabel)
+	metrics.SetGateMetricValue(value, g.threshold, g.owner.QueueID, g.owner.QueueName, g.owner.WorkerPoolID, g.inferencePool)
 
 	if value <= g.threshold {
 		return 0.0

pkg/async/inference/flowcontrol/metric_dispatch_gate_test.go

@@ -23,6 +23,7 @@ import (
 	"sync"
 	"testing"
 
+	"github.com/llm-d/llm-d-async/pipeline"
 	"github.com/llm-d/llm-d-async/pkg/metrics"
 	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/stretchr/testify/require"
@@ -118,21 +119,24 @@ func TestBudgetDispatchGate(t *testing.T) {
 
 // TestMetricDispatchGate_RecordsGateMetricValue verifies that a metric gate
 // records the raw value it read and the threshold it compared against, labeled by
-// pool (issue #217, "Saturation Metric Value").
+// its owning queue and worker pool plus the InferencePool it queries
+// (issues #217, "Saturation Metric Value", and #369, label collision).
 func TestMetricDispatchGate_RecordsGateMetricValue(t *testing.T) {
-	const pool = "test-pool-217"
-	metrics.GateMetricValue.DeleteLabelValues(pool)
-	metrics.GateMetricThreshold.DeleteLabelValues(pool)
+	owner := pipeline.GateOwner{QueueID: "q-217", QueueName: "queue-217", WorkerPoolID: "test-pool-217"}
+	const inferencePool = "optimized-baseline"
+	labels := []string{owner.QueueID, owner.QueueName, owner.WorkerPoolID, inferencePool}
+	metrics.GateMetricValue.DeleteLabelValues(labels...)
+	metrics.GateMetricThreshold.DeleteLabelValues(labels...)
 
 	// Saturation gate: source returns D = 1 - saturation; the gate's threshold is
 	// 1 - satThreshold. With saturation 0.3 -> D=0.7, satThreshold 0.8 -> threshold=0.2.
 	source := &mockMetricSource{samples: []Sample{{Value: 0.7}}}
-	gate := NewSaturationDispatchGate(source, 0.8, 0.0).WithPoolLabel(pool)
+	gate := NewSaturationDispatchGate(source, 0.8, 0.0).WithOwner(owner).WithInferencePool(inferencePool)
 
 	_ = gate.Budget(context.Background())
 
-	require.InDelta(t, 0.7, testutil.ToFloat64(metrics.GateMetricValue.WithLabelValues(pool)), 1e-9)
-	require.InDelta(t, 0.2, testutil.ToFloat64(metrics.GateMetricThreshold.WithLabelValues(pool)), 1e-9)
+	require.InDelta(t, 0.7, testutil.ToFloat64(metrics.GateMetricValue.WithLabelValues(labels...)), 1e-9)
+	require.InDelta(t, 0.2, testutil.ToFloat64(metrics.GateMetricThreshold.WithLabelValues(labels...)), 1e-9)
 }
 
 // switchableMetricSource allows changing what Query returns between calls.

pkg/metrics/metrics.go

@@ -16,10 +16,21 @@ const (
 	LabelQueueName = "queue_name"
 	LabelPoolName  = "pool_name"
 	LabelReason    = "reason"
+
+	// LabelInferencePool names the InferencePool a gate queries. It is distinct
+	// from pool_name, which always names the async worker pool that owns the
+	// series — several worker pools may gate on one InferencePool, and one
+	// worker pool may serve several.
+	LabelInferencePool = "inference_pool"
 )
 
 var queueLabels = []string{LabelQueueID, LabelQueueName, LabelPoolName}
 
+// gateLabels is queueLabels plus the queried InferencePool. Gate gauges carry
+// the full queue triple so they join with the queue's other series; a
+// pool-level gate leaves queue_id and queue_name empty.
+var gateLabels = []string{LabelQueueID, LabelQueueName, LabelPoolName, LabelInferencePool}
+
 var (
 	Retries = prometheus.NewCounterVec(prometheus.CounterOpts{
 		Subsystem: SchedulerSubsystem, Name: "async_request_retries_total",
@@ -105,12 +116,12 @@ var (
 	}, []string{LabelQueueID, LabelQueueName, LabelPoolName, LabelReason})
 	GateMetricValue = prometheus.NewGaugeVec(prometheus.GaugeOpts{
 		Subsystem: SchedulerSubsystem, Name: "async_gate_metric_value",
-		Help: "Raw metric value last read by a metric-based dispatch gate (prometheus-saturation/-budget/-query), i.e. the value compared against async_gate_metric_threshold to decide the gate. The gate closes when value <= threshold. For the saturation gate the value is 1 - saturation.",
-	}, []string{LabelPoolName})
+		Help: "Raw metric value last read by a metric-based dispatch gate (prometheus-saturation/-budget/-query), i.e. the value compared against async_gate_metric_threshold to decide the gate. The gate closes when value <= threshold. For the saturation gate the value is 1 - saturation. Labeled by the queue or worker pool that owns the gate; inference_pool names the InferencePool the gate queries.",
+	}, gateLabels)
 	GateMetricThreshold = prometheus.NewGaugeVec(prometheus.GaugeOpts{
 		Subsystem: SchedulerSubsystem, Name: "async_gate_metric_threshold",
 		Help: "Threshold a metric-based dispatch gate compares async_gate_metric_value against; the gate closes when value <= this threshold.",
-	}, []string{LabelPoolName})
+	}, gateLabels)
 )
 
 // Gate decision reason label values for async_gate_decisions_total.
@@ -202,11 +213,14 @@ func RecordGateDecision(reason, queueID, queueName, poolName string) {
 }
 
 // SetGateMetricValue records the raw metric value a metric-based dispatch gate
-// last read and the threshold it is compared against, for the given pool. Helps
-// answer "why is the gate closed?" (value <= threshold).
-func SetGateMetricValue(value, threshold float64, poolName string) {
-	GateMetricValue.WithLabelValues(poolName).Set(value)
-	GateMetricThreshold.WithLabelValues(poolName).Set(threshold)
+// last read and the threshold it is compared against. Helps answer "why is the
+// gate closed?" (value <= threshold). queueID/queueName/poolName identify the
+// gate's owner and match the labels on that queue's other series; inferencePool
+// is the InferencePool the gate queries, and is empty when the gate does not
+// name one.
+func SetGateMetricValue(value, threshold float64, queueID, queueName, poolName, inferencePool string) {
+	GateMetricValue.WithLabelValues(queueID, queueName, poolName, inferencePool).Set(value)
+	GateMetricThreshold.WithLabelValues(queueID, queueName, poolName, inferencePool).Set(threshold)
 }
 
 // GetCollectors returns all custom collectors for the async processor.

pkg/pubsub/pubsubimpl.go

@@ -181,8 +181,15 @@ func NewGCPPubSubMQFlow(pubsubOpts Options, fns ...PubSubOption) (*PubSubMQFlow,
 		// Determine gate for this topic
 		var gate pipeline.Gate
 		if p.gateFactory != nil && cfg.GateType != "" {
-			// Use factory to create per-topic gate
-			gate, err = p.gateFactory.CreateGate(cfg.GateConfig)
+			// Use factory to create per-topic gate. The subscriber ID is what the
+			// rest of this backend's metrics use as queue_name (there is no
+			// separate queue ID), so the gate's own gauges join with them.
+			gateCfg := cfg.GateConfig
+			gateCfg.Owner = pipeline.GateOwner{
+				QueueName:    cfg.SubscriberID,
+				WorkerPoolID: workerPoolID,
+			}
+			gate, err = p.gateFactory.CreateGate(gateCfg)
 			if err != nil {
 				return nil, fmt.Errorf("failed to create gate for topic subscriber %q (gate_type=%q): %w", cfg.SubscriberID, cfg.GateType, err)
 			}