feat(metrics): add gate-decision breakdown counter (llm-d#291)

Adds async_gate_decisions_total{queue_id,queue_name,pool_name,reason}
(llm-d#217), counting messages a gate prevented from being dispatched, by
reason:
- gate_closed:     dispatch budget exhausted (gate returned Refuse)
- quota_exhausted: per-attribute quota overflow (Refuse + overflow class)
- dropped:         gate permanently rejected the request (Drop)
- error:           gate evaluation failed

Emitted at the gate-application branches in both the redis sorted-set
and gcp-pubsub flows. Documented in the README metrics table; unit
tested.

Related: llm-d#217

Signed-off-by: Shimi Bandiel <shimib@google.com>

Author: shimib

README.md

@@ -536,6 +536,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). |
 
 **Labels:**
 
@@ -544,6 +545,7 @@ The Async Processor exposes Prometheus metrics under the `llm_d_async` subsystem
 | `queue_id` | Transport-level queue identifier |
 | `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` |
 
 **Example PromQL queries:**
 

pkg/metrics/metrics.go

@@ -15,6 +15,7 @@ const (
 	LabelQueueID   = "queue_id"
 	LabelQueueName = "queue_name"
 	LabelPoolName  = "pool_name"
+	LabelReason    = "reason"
 )
 
 var queueLabels = []string{LabelQueueID, LabelQueueName, LabelPoolName}
@@ -98,6 +99,18 @@ var (
 		Subsystem: SchedulerSubsystem, Name: "async_pool_worker_limit",
 		Help: "Configured number of concurrent workers (the concurrency limit) for a pool. Compare against async_inflight_requests for worker utilization.",
 	}, []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).",
+	}, []string{LabelQueueID, LabelQueueName, LabelPoolName, LabelReason})
+)
+
+// Gate decision reason label values for async_gate_decisions_total.
+const (
+	ReasonGateClosed     = "gate_closed"
+	ReasonQuotaExhausted = "quota_exhausted"
+	ReasonDropped        = "dropped"
+	ReasonError          = "error"
 )
 
 func RecordRetry(queueID, queueName, poolName string) {
@@ -174,12 +187,18 @@ func SetPoolWorkerLimit(poolName string, n float64) {
 	PoolWorkerLimit.WithLabelValues(poolName).Set(n)
 }
 
+// RecordGateDecision increments the count of gate decisions that prevented a
+// message from being dispatched, labeled by reason.
+func RecordGateDecision(reason, queueID, queueName, poolName string) {
+	GateDecisions.WithLabelValues(queueID, queueName, poolName, reason).Inc()
+}
+
 // GetCollectors returns all custom collectors for the async processor.
 func GetAsyncProcessorCollectors(supportsMessageLatency bool) []prometheus.Collector {
 	collectors := []prometheus.Collector{
 		Retries, AsyncReqs, ExceededDeadlineReqs, FailedReqs, SuccessfulReqs, SheddedRequests,
 		QueueDepth, InflightRequests, BrokerBacklog, InferenceLatencyTime, QueueResidenceTime,
-		DispatchBudget, PoolWorkerLimit,
+		DispatchBudget, PoolWorkerLimit, GateDecisions,
 	}
 	if supportsMessageLatency {
 		collectors = append(collectors, MessageLatencyTime)

pkg/metrics/metrics_test.go

@@ -76,6 +76,28 @@ func TestSetPoolWorkerLimit(t *testing.T) {
 	}
 }
 
+func TestRecordGateDecision(t *testing.T) {
+	RecordGateDecision(ReasonQuotaExhausted, "q9", "queue-9", "pool-z")
+	RecordGateDecision(ReasonQuotaExhausted, "q9", "queue-9", "pool-z")
+	RecordGateDecision(ReasonGateClosed, "q9", "queue-9", "pool-z")
+
+	if got := testutil.ToFloat64(GateDecisions.WithLabelValues("q9", "queue-9", "pool-z", ReasonQuotaExhausted)); got != 2 {
+		t.Errorf("GateDecisions[quota_exhausted] = %v, want 2", got)
+	}
+	if got := testutil.ToFloat64(GateDecisions.WithLabelValues("q9", "queue-9", "pool-z", ReasonGateClosed)); got != 1 {
+		t.Errorf("GateDecisions[gate_closed] = %v, want 1", got)
+	}
+}
+
+func TestGetAsyncProcessorCollectors_includesGateDecisions(t *testing.T) {
+	for _, withLatency := range []bool{false, true} {
+		collectors := GetAsyncProcessorCollectors(withLatency)
+		if !containsCollector(collectors, GateDecisions) {
+			t.Errorf("expected GateDecisions to be present (supportsMessageLatency=%v)", withLatency)
+		}
+	}
+}
+
 func containsCollector(collectors []prometheus.Collector, target prometheus.Collector) bool {
 	for _, c := range collectors {
 		if c == target {

pkg/pubsub/pubsubimpl.go

@@ -498,11 +498,17 @@ func (r *PubSubMQFlow) processMessages(ctx context.Context, receive receiveFunc,
 		verdict, err := gate.Apply(ctx, ir, &releases)
 		if err != nil {
 			logger.V(logutil.DEFAULT).Error(err, "Gating failed")
+			metrics.RecordGateDecision(metrics.ReasonError, "", subscriberID, poolID)
 			msg.Nack()
 			return
 		}
 
 		if verdict.Action == pipeline.ActionRefuse {
+			reason := metrics.ReasonGateClosed
+			if ir.GetClassification() == api.ClassificationOverflow {
+				reason = metrics.ReasonQuotaExhausted
+			}
+			metrics.RecordGateDecision(reason, "", subscriberID, poolID)
 			logger.V(logutil.DEBUG).Info("Quota exceeded or capacity full, delaying Nack", "msgID", msg.ID, "delay", quotaExceededNackDelay)
 			go func() {
 				select {
@@ -516,6 +522,7 @@ func (r *PubSubMQFlow) processMessages(ctx context.Context, receive receiveFunc,
 		}
 
 		if verdict.Action == pipeline.ActionDrop {
+			metrics.RecordGateDecision(metrics.ReasonDropped, "", subscriberID, poolID)
 			var resultMsg api.ResultMessage
 			if verdict.Result != nil {
 				resultMsg = *verdict.Result

pkg/redis/sortedset_impl.go

@@ -459,20 +459,27 @@ func (r *RedisSortedSetFlow) processMessages(ctx context.Context, msgChannel cha
 		verdict, err := gate.Apply(ctx, ir, &releases)
 		if err != nil {
 			logger.V(logutil.DEFAULT).Error(err, "Gating failed")
+			metrics.RecordGateDecision(metrics.ReasonError, queueID, queueName, poolName)
 			// Re-enqueue the message on gating failure
 			member, _ := json.Marshal(ir)
 			r.rdb.ZAdd(ctx, queueName, redis.Z{Score: deadline, Member: string(member)})
 			continue
 		}
 
 		if verdict.Action == pipeline.ActionRefuse {
+			reason := metrics.ReasonGateClosed
+			if ir.GetClassification() == api.ClassificationOverflow {
+				reason = metrics.ReasonQuotaExhausted
+			}
+			metrics.RecordGateDecision(reason, queueID, queueName, poolName)
 			// Re-enqueue the message (wait for capacity or quota)
 			member, _ := json.Marshal(ir)
 			r.rdb.ZAdd(ctx, queueName, redis.Z{Score: deadline, Member: string(member)})
 			continue
 		}
 
 		if verdict.Action == pipeline.ActionDrop {
+			metrics.RecordGateDecision(metrics.ReasonDropped, queueID, queueName, poolName)
 			if verdict.Result != nil {
 				r.resultChannel <- *verdict.Result
 			} else {