diff --git a/README.md b/README.md index 4ddac994..7db25490 100644 --- a/README.md +++ b/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 diff --git a/pipeline/gate.go b/pipeline/gate.go index ea2474f3..0e21287b 100644 --- a/pipeline/gate.go +++ b/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. diff --git a/pkg/async/inference/flowcontrol/gate_factory.go b/pkg/async/inference/flowcontrol/gate_factory.go index 1ba857b4..066dfcb2 100644 --- a/pkg/async/inference/flowcontrol/gate_factory.go +++ b/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", "") diff --git a/pkg/async/inference/flowcontrol/gate_factory_test.go b/pkg/async/inference/flowcontrol/gate_factory_test.go index 58c7f9af..a847f0af 100644 --- a/pkg/async/inference/flowcontrol/gate_factory_test.go +++ b/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) +} diff --git a/pkg/async/inference/flowcontrol/metric_dispatch_gate.go b/pkg/async/inference/flowcontrol/metric_dispatch_gate.go index 8c4ba33d..13e26042 100644 --- a/pkg/async/inference/flowcontrol/metric_dispatch_gate.go +++ b/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 diff --git a/pkg/async/inference/flowcontrol/metric_dispatch_gate_test.go b/pkg/async/inference/flowcontrol/metric_dispatch_gate_test.go index 08f6d5a2..2d961aff 100644 --- a/pkg/async/inference/flowcontrol/metric_dispatch_gate_test.go +++ b/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. diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index 515f92aa..467031a1 100644 --- a/pkg/metrics/metrics.go +++ b/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. diff --git a/pkg/pubsub/pubsubimpl.go b/pkg/pubsub/pubsubimpl.go index d85db969..cca8f096 100644 --- a/pkg/pubsub/pubsubimpl.go +++ b/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) } diff --git a/pkg/redis/sortedset_impl.go b/pkg/redis/sortedset_impl.go index d34c472f..058c51f4 100644 --- a/pkg/redis/sortedset_impl.go +++ b/pkg/redis/sortedset_impl.go @@ -171,9 +171,24 @@ func NewRedisSortedSetFlow(flowOpts SortedSetFlowOptions, connOpts ConnectionOpt r.configMap = make(map[string]queueConfig, len(configs)) for _, cfg := range configs { + // Normalize before anything reads it: configMap is the source of the + // pool_name label on this queue's metrics, and an unset WorkerPoolID + // there would label them "" while the request channel below — and every + // pool-keyed series — says "default". + if cfg.WorkerPoolID == "" { + cfg.WorkerPoolID = "default" + } + workerPoolID := cfg.WorkerPoolID + var gate pipeline.Gate if r.gateFactory != nil && cfg.GateType != "" { - gate, err = r.gateFactory.CreateGate(cfg.GateConfig) + gateCfg := cfg.GateConfig + gateCfg.Owner = pipeline.GateOwner{ + QueueID: cfg.ID, + QueueName: cfg.QueueName, + WorkerPoolID: workerPoolID, + } + gate, err = r.gateFactory.CreateGate(gateCfg) if err != nil { return nil, fmt.Errorf("failed to create gate for queue %q (gate_type=%q): %w", cfg.QueueName, cfg.GateType, err) } @@ -183,11 +198,6 @@ func NewRedisSortedSetFlow(flowOpts SortedSetFlowOptions, connOpts ConnectionOpt gate = pipeline.ConstOpenGate() } - workerPoolID := cfg.WorkerPoolID - if workerPoolID == "" { - workerPoolID = "default" - } - found := false for _, pool := range r.workerPools { if pool.ID == workerPoolID { diff --git a/pkg/redis/sortedset_impl_test.go b/pkg/redis/sortedset_impl_test.go index e6f45a4e..dd4f66d7 100644 --- a/pkg/redis/sortedset_impl_test.go +++ b/pkg/redis/sortedset_impl_test.go @@ -1607,6 +1607,39 @@ func TestNewRedisSortedSetFlow_PoolRequiredAndValidation(t *testing.T) { } } +// TestNewRedisSortedSetFlow_DefaultsWorkerPoolIDInConfigMap checks that a queue +// config with no worker_pool_id is normalized before it is stored: configMap is +// the source of the pool_name label on async_dispatch_budget and +// async_gate_decisions_total, and it used to keep the raw "" while the request +// channel — and every pool-keyed series — said "default" (issue #369). +func TestNewRedisSortedSetFlow_DefaultsWorkerPoolIDInConfigMap(t *testing.T) { + s := miniredis.RunT(t) + defer s.Close() + connOpts := ConnectionOptions{URL: "redis://" + s.Addr()} + opts := SortedSetFlowOptions{ + PollIntervalMs: 1000, + BatchSize: 10, + GateParamsJSON: "{}", + QueuesConfig: `[{"id":"q1","queue_name":"test-queue","inference_objective":"obj","igw_base_url":"http://gw"}]`, + } + + flow, err := NewRedisSortedSetFlow(opts, connOpts, WithSortedSetWorkerPools([]pipeline.WorkerPoolConfig{{ID: "default", Workers: 1}})) + if err != nil { + t.Fatalf("Unexpected error creating flow: %v", err) + } + + cfg, ok := flow.configMap["q1"] + if !ok { + t.Fatal("Expected queue q1 in configMap") + } + if cfg.WorkerPoolID != "default" { + t.Errorf("configMap worker pool = %q, want %q", cfg.WorkerPoolID, "default") + } + if got := flow.requestChannels[0].channel.WorkerPoolID; got != cfg.WorkerPoolID { + t.Errorf("request channel worker pool = %q, configMap says %q; the two must agree or the metrics do not join", got, cfg.WorkerPoolID) + } +} + func TestQueueBacklog(t *testing.T) { _, rdb, ctx, cancel := setupTest(t) defer rdb.Close() // nolint:errcheck diff --git a/pkg/server/runner.go b/pkg/server/runner.go index 94de8296..d02498a6 100644 --- a/pkg/server/runner.go +++ b/pkg/server/runner.go @@ -137,7 +137,13 @@ func (r *Runner) Run(ctx context.Context) (err error) { poolGates := make(map[string]pipeline.Gate) for poolID, pool := range poolsMap { if pool.GateType != "" { - gate, err := gateFactory.CreateGate(pipeline.GateConfig{GateType: pool.GateType, GateParams: pool.GateParams}) + gate, err := gateFactory.CreateGate(pipeline.GateConfig{ + GateType: pool.GateType, + GateParams: pool.GateParams, + // A pool gate covers every queue merged into the pool, so it has + // no single queue to name; pool_name alone identifies it. + Owner: pipeline.GateOwner{WorkerPoolID: poolID}, + }) if err != nil { setupLog.Error(err, "Failed to create pool gate", "poolID", poolID, "gateType", pool.GateType) os.Exit(1) diff --git a/release-notes.d/unreleased/372.md b/release-notes.d/unreleased/372.md new file mode 100644 index 00000000..237d7bbe --- /dev/null +++ b/release-notes.d/unreleased/372.md @@ -0,0 +1,7 @@ +--- +pr: 372 +url: https://github.com/llm-d/llm-d-async/pull/372 +author: shimib +date: 2026-07-29 +--- +Fix `pool_name`, which carried three different meanings across the five series an operator needs to diagnose a throttled queue, so none of them joined. `async_gate_metric_value` and `async_gate_metric_threshold` labeled it with the gate's `pool` param — the InferencePool being queried — and carried no queue labels at all; `async_dispatch_budget` and `async_gate_decisions_total` read it from the raw queue config, so a queue that omitted `worker_pool_id` reported `pool_name=""` while its own `async_broker_backlog` and `async_pool_worker_limit` reported `"default"`. `pool_name` now always names the async worker pool that owns the series, the queue config is normalized before it is stored, and the two gate gauges carry the full `queue_id`/`queue_name`/`pool_name` triple so they join with the rest of the queue's metrics. The InferencePool a gate queries moves to its own `inference_pool` label, which also makes two worker pools gating on one InferencePool distinguishable. **Breaking for dashboards on the two gauges:** select `{inference_pool="..."}` where you previously selected `{pool_name="..."}`.