Tiny add routing key distribution metrics (#16847)

python/sglang/srt/managers/scheduler_metrics_mixin.py

@@ -13,7 +13,11 @@ from sglang.srt.managers.io_struct import GetLoadReqInput, GetLoadReqOutput
 from sglang.srt.managers.schedule_policy import PrefillAdder
 from sglang.srt.managers.scheduler import Req, ScheduleBatch
 from sglang.srt.managers.utils import GenerationBatchResult
-from sglang.srt.metrics.collector import SchedulerMetricsCollector, SchedulerStats
+from sglang.srt.metrics.collector import (
+    SchedulerMetricsCollector,
+    SchedulerStats,
+    compute_routing_key_stats,
+)
 from sglang.srt.utils import get_bool_env_var
 from sglang.srt.utils.device_timer import DeviceTimer
 
@@ -416,6 +420,16 @@ class SchedulerMetricsMixin:
                     self.disagg_decode_transfer_queue.queue
                 )
 
+            running_routing_keys = [r.routing_key for r in batch.reqs]
+            waiting_routing_keys = [r.routing_key for r in self.waiting_queue]
+            (
+                self.stats.num_unique_running_routing_keys,
+                self.stats.routing_key_running_req_counts,
+            ) = compute_routing_key_stats(running_routing_keys)
+            _, self.stats.routing_key_all_req_counts = compute_routing_key_stats(
+                running_routing_keys + waiting_routing_keys
+            )
+
             # Others
             self.calculate_utilization()
             self.update_lora_metrics()

python/sglang/srt/metrics/collector.py

@@ -25,6 +25,7 @@ from sglang.srt.metrics.utils import exponential_buckets, generate_buckets
 from sglang.srt.model_executor.forward_batch_info import ForwardMode
 from sglang.srt.server_args import ServerArgs
 from sglang.srt.utils import get_bool_env_var
+from sglang.srt.utils.gauge_histogram import GaugeHistogram
 
 SGLANG_TEST_REQUEST_TIME_STATS = get_bool_env_var("SGLANG_TEST_REQUEST_TIME_STATS")
 
@@ -246,6 +247,22 @@ class SchedulerStats:
     lora_pool_slots_total: int = 0
     lora_pool_utilization: float = 0.0
 
+    # Routing key metrics
+    num_unique_running_routing_keys: int = 0
+    routing_key_running_req_counts: List[int] = field(default_factory=list)
+    routing_key_all_req_counts: List[int] = field(default_factory=list)
+
+
+ROUTING_KEY_REQ_COUNT_BUCKET_BOUNDS = [1, 2, 3, 5, 7, 10, 20, 50, 100, 200]
+
+
+def compute_routing_key_stats(routing_keys: List[Optional[str]]) -> tuple:
+    """Returns (num_unique_keys, per_key_counts)."""
+    from collections import Counter
+
+    key_counts = Counter(k for k in routing_keys if k is not None)
+    return len(key_counts), list(key_counts.values())
+
 
 @dataclass
 class DPCooperationInfo:
@@ -721,6 +738,25 @@ class SchedulerMetricsCollector:
                 multiprocess_mode="mostrecent",
             )
 
+        self.num_unique_running_routing_keys = Gauge(
+            name="sglang:num_unique_running_routing_keys",
+            documentation="Number of unique routing keys in running batch.",
+            labelnames=labels.keys(),
+            multiprocess_mode="mostrecent",
+        )
+        self.routing_key_running_req_count = GaugeHistogram(
+            name="sglang:routing_key_running_req_count",
+            documentation="Distribution of routing keys by running request count (gt < count <= le).",
+            labelnames=list(labels.keys()),
+            bucket_bounds=ROUTING_KEY_REQ_COUNT_BUCKET_BOUNDS,
+        )
+        self.routing_key_all_req_count = GaugeHistogram(
+            name="sglang:routing_key_all_req_count",
+            documentation="Distribution of routing keys by running+waiting request count (gt < count <= le).",
+            labelnames=list(labels.keys()),
+            bucket_bounds=ROUTING_KEY_REQ_COUNT_BUCKET_BOUNDS,
+        )
+
         self.new_token_ratio = Gauge(
             name="sglang:new_token_ratio",
             documentation="The new token ratio.",
@@ -981,6 +1017,16 @@ class SchedulerMetricsCollector:
             self._log_gauge(self.lora_pool_slots_total, stats.lora_pool_slots_total)
             self._log_gauge(self.lora_pool_utilization, stats.lora_pool_utilization)
 
+        self._log_gauge(
+            self.num_unique_running_routing_keys, stats.num_unique_running_routing_keys
+        )
+        self.routing_key_running_req_count.set_by_current_observations(
+            self.labels, stats.routing_key_running_req_counts
+        )
+        self.routing_key_all_req_count.set_by_current_observations(
+            self.labels, stats.routing_key_all_req_counts
+        )
+
         self.last_log_time = time.perf_counter()
 
     def log_grammar_stats(self, grammar_stats) -> None:

test/registered/metrics/test_metrics.py

@@ -11,6 +11,10 @@ from prometheus_client.parser import text_string_to_metric_families
 from prometheus_client.samples import Sample
 
 from sglang.srt.environ import envs
+from sglang.srt.metrics.collector import (
+    ROUTING_KEY_REQ_COUNT_BUCKET_BOUNDS,
+    compute_routing_key_stats,
+)
 from sglang.srt.utils import kill_process_tree
 from sglang.test.test_utils import (
     DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
@@ -144,10 +148,28 @@ class TestEnableMetrics(CustomTestCase):
             "sglang:e2e_request_latency_seconds",
             "sglang:http_requests_active",
             "sglang:routing_keys_active",
+            "sglang:num_unique_running_routing_keys",
+            "sglang:routing_key_running_req_count",
+            "sglang:routing_key_all_req_count",
         ]
         for metric in essential_metrics:
             self.assertIn(metric, metrics_text, f"Missing metric: {metric}")
 
+        # Verify routing key GaugeHistogram buckets
+        expected_buckets = len(ROUTING_KEY_REQ_COUNT_BUCKET_BOUNDS) + 1
+        for metric_name in [
+            "sglang:routing_key_running_req_count",
+            "sglang:routing_key_all_req_count",
+        ]:
+            gt_le_pairs = set()
+            for sample in metrics.get(metric_name, []):
+                gt_le_pairs.add((sample.labels.get("gt"), sample.labels.get("le")))
+            self.assertEqual(
+                len(gt_le_pairs),
+                expected_buckets,
+                f"{metric_name}: Expected {expected_buckets} buckets, got {len(gt_le_pairs)}",
+            )
+
         self.assertIn(f'model_name="{_MODEL_NAME}"', metrics_text)
         self.assertIn("_sum{", metrics_text)
         self.assertIn("_count{", metrics_text)
@@ -185,5 +207,33 @@ def _check_metrics_positive(test_case, metrics, metrics_to_check):
         test_case.assertGreater(value, 0, f"{metric_name} {labels}")
 
 
+class TestComputeRoutingKeyStats(unittest.TestCase):
+    def test_empty(self):
+        num_unique, req_counts = compute_routing_key_stats([])
+        self.assertEqual(num_unique, 0)
+        self.assertEqual(req_counts, [])
+
+    def test_all_none(self):
+        num_unique, req_counts = compute_routing_key_stats([None, None, None])
+        self.assertEqual(num_unique, 0)
+        self.assertEqual(req_counts, [])
+
+    def test_with_none(self):
+        num_unique, req_counts = compute_routing_key_stats([None, "key1", None])
+        self.assertEqual(num_unique, 1)
+        self.assertEqual(req_counts, [1])
+
+    def test_single_key_multiple_reqs(self):
+        num_unique, req_counts = compute_routing_key_stats(["key1"] * 5)
+        self.assertEqual(num_unique, 1)
+        self.assertEqual(req_counts, [5])
+
+    def test_distribution(self):
+        routing_keys = ["key1"] * 5 + ["key2"] * 1 + ["key3"] * 15 + ["key4"] * 250
+        num_unique, req_counts = compute_routing_key_stats(routing_keys)
+        self.assertEqual(num_unique, 4)
+        self.assertEqual(sorted(req_counts), [1, 5, 15, 250])
+
+
 if __name__ == "__main__":
     unittest.main()