diff --git a/python/sglang/srt/managers/scheduler_metrics_mixin.py b/python/sglang/srt/managers/scheduler_metrics_mixin.py index d8a9f8c02..8f5c02670 100644 --- a/python/sglang/srt/managers/scheduler_metrics_mixin.py +++ b/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() diff --git a/python/sglang/srt/metrics/collector.py b/python/sglang/srt/metrics/collector.py index 0137c961a..316cc177a 100644 --- a/python/sglang/srt/metrics/collector.py +++ b/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: diff --git a/test/registered/metrics/test_metrics.py b/test/registered/metrics/test_metrics.py index f7c18ea2d..31a38b8d0 100644 --- a/test/registered/metrics/test_metrics.py +++ b/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()