diff --git a/README.md b/README.md index 7db25490..de519b82 100644 --- a/README.md +++ b/README.md @@ -574,7 +574,7 @@ The Async Processor exposes Prometheus metrics under the `llm_d_async` subsystem | `llm_d_async_async_queue_residence_time_millis` | Histogram | Time in milliseconds a message spent buffered in-process, from broker ingestion until a worker pulled it for processing. Measures the async delay introduced by the system (queue time). Always registered. | | `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_decisions_total` | Counter | Count of gate decisions that prevented dispatch, by `reason`: `gate_closed` (no dispatch budget), `quota_exhausted` (per-attribute quota overflow), `dropped` (gate permanently rejected the request), `error` (gate evaluation failed). `quota_exhausted`, `dropped` and `error` count individual messages refused after being dequeued. `gate_closed` counts those plus every dequeue round in which the budget shrank the batch to zero — the way budget-based gates (`prometheus-budget`/`-saturation`/`-query`) shed work *before* a message is dequeued — so its rate reflects throttled dispatch opportunities, not messages. All four `reason` series are created at 0 when a queue starts, so a query returns 0 rather than an empty vector. | | `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. | diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index 467031a1..72852e79 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -112,7 +112,7 @@ var ( }, []string{LabelPoolName}) GateDecisions = prometheus.NewCounterVec(prometheus.CounterOpts{ Subsystem: SchedulerSubsystem, Name: "async_gate_decisions_total", - Help: "Count of gate decisions that prevented a message from being dispatched, by reason (gate_closed, quota_exhausted, dropped, error).", + Help: "Count of gate decisions that prevented dispatch, by reason (gate_closed, quota_exhausted, dropped, error). quota_exhausted/dropped/error count individual messages refused after they were dequeued; gate_closed counts those plus each dequeue round in which the gate's budget shrank the batch to zero, which is how budget-based gates (prometheus-budget/-saturation/-query) shed work before any message is dequeued.", }, []string{LabelQueueID, LabelQueueName, LabelPoolName, LabelReason}) GateMetricValue = prometheus.NewGaugeVec(prometheus.GaugeOpts{ Subsystem: SchedulerSubsystem, Name: "async_gate_metric_value", @@ -212,6 +212,17 @@ func RecordGateDecision(reason, queueID, queueName, poolName string) { GateDecisions.WithLabelValues(queueID, queueName, poolName, reason).Inc() } +// InitGateDecisions pre-creates a queue's async_gate_decisions_total series with +// every reason at 0. A CounterVec label set that has never been incremented is +// absent from /metrics entirely, so querying a reason that has not fired yet +// yields an empty vector rather than 0 — indistinguishable from a queue that was +// never configured or a scrape that never landed. +func InitGateDecisions(queueID, queueName, poolName string) { + for _, reason := range []string{ReasonGateClosed, ReasonQuotaExhausted, ReasonDropped, ReasonError} { + GateDecisions.WithLabelValues(queueID, queueName, poolName, reason) + } +} + // SetGateMetricValue records the raw metric value a metric-based dispatch gate // last read and the threshold it is compared against. Helps answer "why is the // gate closed?" (value <= threshold). queueID/queueName/poolName identify the diff --git a/pkg/metrics/metrics_test.go b/pkg/metrics/metrics_test.go index 18252029..96427720 100644 --- a/pkg/metrics/metrics_test.go +++ b/pkg/metrics/metrics_test.go @@ -89,6 +89,21 @@ func TestRecordGateDecision(t *testing.T) { } } +func TestInitGateDecisions(t *testing.T) { + InitGateDecisions("q10", "queue-10", "pool-y") + + for _, reason := range []string{ReasonGateClosed, ReasonQuotaExhausted, ReasonDropped, ReasonError} { + // A never-incremented CounterVec label set is absent from /metrics, so + // the point of the pre-creation is that these series exist and read 0. + if got := testutil.ToFloat64(GateDecisions.WithLabelValues("q10", "queue-10", "pool-y", reason)); got != 0 { + t.Errorf("GateDecisions[%s] = %v, want 0", reason, got) + } + } + if got := testutil.CollectAndCount(GateDecisions, "llm_d_async_async_gate_decisions_total"); got < 4 { + t.Errorf("GateDecisions series count = %d, want at least 4", got) + } +} + func TestGetAsyncProcessorCollectors_includesGateDecisions(t *testing.T) { for _, withLatency := range []bool{false, true} { collectors := GetAsyncProcessorCollectors(withLatency) diff --git a/pkg/pubsub/pubsubimpl.go b/pkg/pubsub/pubsubimpl.go index cca8f096..f45bcce1 100644 --- a/pkg/pubsub/pubsubimpl.go +++ b/pkg/pubsub/pubsubimpl.go @@ -523,6 +523,8 @@ func (r *PubSubMQFlow) requestWorker(ctx context.Context, pubSubClient *pubsub.C sub := pubSubClient.Subscriber(subscriberID) + metrics.InitGateDecisions("", subscriberID, poolID) + for ctx.Err() == nil { receiveCtx, cancel := context.WithCancel(ctx) budget := gate.Budget(ctx) @@ -549,6 +551,13 @@ func (r *PubSubMQFlow) requestWorker(ctx context.Context, pubSubClient *pubsub.C sub.ReceiveSettings.MaxOutstandingMessages = currBatchSize sub.ReceiveSettings.NumGoroutines = 1 if currBatchSize <= 0 { + // Same pre-dequeue back-pressure as the sorted-set path: with no + // outstanding-message slots the receive callback never runs, so + // gate.Apply never records the refusal. Count the throttled receive + // window instead (#368). Unlike Redis there is no cheap depth probe + // — the subscription backlog comes from Cloud Monitoring — so this + // counts the window whether or not messages happen to be waiting. + metrics.RecordGateDecision(metrics.ReasonGateClosed, "", subscriberID, poolID) <-receiveCtx.Done() cancel() continue diff --git a/pkg/redis/sortedset_impl.go b/pkg/redis/sortedset_impl.go index 058c51f4..24b28c97 100644 --- a/pkg/redis/sortedset_impl.go +++ b/pkg/redis/sortedset_impl.go @@ -418,6 +418,8 @@ func (r *RedisSortedSetFlow) requestWorker(ctx context.Context, msgChannel chan gate = r.gate } + metrics.InitGateDecisions(queueID, queueName, r.poolNameFor(queueID)) + for { select { case <-ctx.Done(): @@ -428,16 +430,36 @@ func (r *RedisSortedSetFlow) requestWorker(ctx context.Context, msgChannel chan } } +// poolNameFor returns the worker pool a queue routes to, or "" when the queue has +// no config entry (the metric label is then empty rather than wrong). +func (r *RedisSortedSetFlow) poolNameFor(queueID string) string { + if cfg, ok := r.configMap[queueID]; ok { + return cfg.WorkerPoolID + } + return "" +} + func (r *RedisSortedSetFlow) processMessages(ctx context.Context, msgChannel chan *api.InternalRequest, queueName string, queueID string, gate pipeline.Gate, logger logr.Logger) { currentTime := float64(time.Now().Unix()) budget := gate.Budget(ctx) - poolName := "" - if cfg, ok := r.configMap[queueID]; ok { - poolName = cfg.WorkerPoolID - } + poolName := r.poolNameFor(queueID) metrics.SetDispatchBudget(budget, queueID, queueName, poolName) batchSize := int(math.Floor(float64(r.batchSize) * budget)) + if batchSize <= 0 { + // Back-pressure here is applied pre-dequeue: the budget shrank the batch + // to zero, so no message reaches gate.Apply below — the only other site + // that records a refusal. Count the throttled poll itself, or gate_closed + // stays silent exactly while the gate is doing its job (#368). Only count + // when work is actually waiting; an idle queue was not held back. + depth, err := r.rdb.ZCard(ctx, queueName).Result() + if err != nil { + logger.V(logutil.DEFAULT).Error(err, "Failed to read queue depth for a closed gate", "queue", queueName) + } else if depth > 0 { + metrics.RecordGateDecision(metrics.ReasonGateClosed, queueID, queueName, poolName) + } + return + } for i := 0; i < batchSize; i++ { results, err := r.rdb.ZPopMin(ctx, queueName, 1).Result() diff --git a/pkg/redis/sortedset_impl_test.go b/pkg/redis/sortedset_impl_test.go index dd4f66d7..f7e8cb47 100644 --- a/pkg/redis/sortedset_impl_test.go +++ b/pkg/redis/sortedset_impl_test.go @@ -12,8 +12,11 @@ import ( "time" "github.com/alicebob/miniredis/v2" + "github.com/go-logr/logr" "github.com/llm-d/llm-d-async/api" "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/redis/go-redis/v9" ) @@ -1037,6 +1040,55 @@ func TestSortedSetFlow_ZeroBudget(t *testing.T) { } } +func TestSortedSetFlow_ClosedGateRecordsGateClosedDecision(t *testing.T) { + s, rdb, ctx, cancel := setupTest(t) + defer s.Close() + defer rdb.Close() // nolint:errcheck + defer cancel() + + queue := "gate-closed-metric-queue" + queueID := "gate-closed-metric-id" + gateClosed := func() float64 { + return testutil.ToFloat64(metrics.GateDecisions.WithLabelValues(queueID, queue, "", metrics.ReasonGateClosed)) + } + + flow := &RedisSortedSetFlow{ + rdb: rdb, + pollInterval: 50 * time.Millisecond, + batchSize: 10, + gate: pipeline.DispatchGateFunc(func(ctx context.Context) float64 { return 0.0 }), + } + msgChannel := make(chan *api.InternalRequest, 1) + + // Nothing queued: the gate held nothing back, so nothing is counted. + flow.processMessages(ctx, msgChannel, queue, queueID, flow.gate, logr.Discard()) + if got := gateClosed(); got != 0 { + t.Fatalf("gate_closed on an empty queue = %v, want 0", got) + } + + msg := api.RequestMessage{ + ID: "gate-closed-1", + Created: time.Now().Unix(), + Deadline: 9999999999, + Payload: map[string]any{"test": "data"}, + } + rdb.ZAdd(ctx, queue, redis.Z{Score: float64(time.Now().Unix()), Member: envelopeJSON(msg)}) + + // Work is waiting and the budget zeroes the batch: each throttled poll is a + // gate_closed decision, even though no message is ever dequeued (#368). + flow.processMessages(ctx, msgChannel, queue, queueID, flow.gate, logr.Discard()) + flow.processMessages(ctx, msgChannel, queue, queueID, flow.gate, logr.Discard()) + if got := gateClosed(); got != 2 { + t.Fatalf("gate_closed with a backlogged queue = %v, want 2", got) + } + if len(msgChannel) != 0 { + t.Fatalf("message dispatched while the gate was closed") + } + if count, _ := rdb.ZCard(ctx, queue).Result(); count != 1 { + t.Fatalf("queue depth = %d, want the message left in place", count) + } +} + func TestSortedSetFlow_ResultRetryAfterFailure(t *testing.T) { s, rdb, ctx, cancel := setupTest(t) defer s.Close() diff --git a/release-notes.d/unreleased/371.md b/release-notes.d/unreleased/371.md new file mode 100644 index 00000000..565d2574 --- /dev/null +++ b/release-notes.d/unreleased/371.md @@ -0,0 +1,7 @@ +--- +pr: 371 +url: https://github.com/llm-d/llm-d-async/pull/371 +author: shimib +date: 2026-07-28 +--- +Fix `async_gate_decisions_total{reason="gate_closed"}`, which could never increment for a budget-based gate: back-pressure is applied pre-dequeue by shrinking the dispatch batch to zero, while the counter was only incremented post-dequeue inside the loop that batch size had just emptied — and a counter series that never increments is absent from `/metrics` entirely, so the query returned an empty vector rather than 0. The decision is now recorded where it is made: the sorted-set path counts each throttled poll whose batch the budget zeroed while the queue is non-empty, and the Pub/Sub path counts each receive window it skips for the same reason. Both backends also pre-create all four `reason` series at 0 when a queue starts. The counter's unit is decisions, not messages; the help text and README metric table now say so.