diff --git a/docs_new/docs/references/production_metrics.mdx b/docs_new/docs/references/production_metrics.mdx index f15f51c9f67c..10c325c0bd99 100644 --- a/docs_new/docs/references/production_metrics.mdx +++ b/docs_new/docs/references/production_metrics.mdx @@ -7,6 +7,8 @@ SGLang exposes the following metrics via Prometheus. You can enable it by adding An example of the monitoring dashboard is available in [examples/monitoring/grafana.json](https://github.com/sgl-project/sglang/blob/main/examples/monitoring/grafana/dashboards/json/sglang-dashboard.json). +The `sglang:per_stage_req_latency_seconds` histogram includes scheduler-side prefill queue stages. `prefill_queue_idle` measures time from queue entry until the scheduler first attempts prefill admission. `prefill_budget_wait` measures time from that first attempt until the request enters forward. Use these stages to separate backlog from token budget or chunking pressure when TTFT rises. + Here is an example of the metrics: ```text Output diff --git a/docs_new/docs/references/production_request_trace.mdx b/docs_new/docs/references/production_request_trace.mdx index 071a8c620938..b1aafe45107d 100644 --- a/docs_new/docs/references/production_request_trace.mdx +++ b/docs_new/docs/references/production_request_trace.mdx @@ -7,6 +7,8 @@ SGLang exports request trace data based on the OpenTelemetry Collector. You can You can find example screenshots of the visualization in https://github.com/sgl-project/sglang/issues/8965. +Scheduler traces include `prefill_queue_idle` and `prefill_budget_wait` slices for prefill requests. The first slice shows time spent waiting before the scheduler first tries to admit the request. The second slice shows time spent after that first attempt before forward starts, which usually points to token budget, chunking, or retry pressure. + ## Setup Guide This section explains how to configure the request tracing and export the trace data. 1. Install the required packages and tools diff --git a/python/sglang/srt/managers/schedule_policy.py b/python/sglang/srt/managers/schedule_policy.py index 1b34f0f54b6d..9e6ed819b34e 100644 --- a/python/sglang/srt/managers/schedule_policy.py +++ b/python/sglang/srt/managers/schedule_policy.py @@ -858,6 +858,8 @@ def add_req_state(r, insert_sort=False): def add_one_req( self, req: Req, has_chunked_req: bool, truncation_align_size: Optional[int] ): + req.time_stats.set_first_prefill_attempt_time() + if (self.prefill_delayer_single_pass is not None) and ( not self.prefill_delayer_single_pass.negotiate_should_allow_prefill( local_prefillable=True, diff --git a/python/sglang/srt/observability/req_time_stats.py b/python/sglang/srt/observability/req_time_stats.py index 0c654795d8af..8feaac289493 100644 --- a/python/sglang/srt/observability/req_time_stats.py +++ b/python/sglang/srt/observability/req_time_stats.py @@ -126,6 +126,16 @@ class RequestStage: # equal to "observe_queue_time" metrics_is_observed=False, ) + PREFILL_QUEUE_IDLE = RequestStageConfig( + "prefill_queue_idle", + level=1, + metrics_is_observed=True, + ) + PREFILL_BUDGET_WAIT = RequestStageConfig( + "prefill_budget_wait", + level=1, + metrics_is_observed=True, + ) DECODE_FORWARD = RequestStageConfig( "decode_forward", level=1, @@ -578,6 +588,7 @@ class SchedulerReqTimeStats(ReqTimeStatsBase): # common, get by time.perf_counter() wait_queue_entry_time: float = 0.0 + first_prefill_attempt_time: float = 0.0 forward_entry_time: float = 0.0 prefill_finished_time: float = 0.0 completion_time: float = 0.0 @@ -623,6 +634,7 @@ def __getstate__(self) -> object: state = { "wait_queue_entry_time": self.wait_queue_entry_time, + "first_prefill_attempt_time": self.first_prefill_attempt_time, "forward_entry_time": self.forward_entry_time, "prefill_finished_time": self.prefill_finished_time, "diff_realtime_monotonic": global_diff_realtime_monotonic, @@ -693,12 +705,14 @@ def set_retract_time(self, ts=None): self.last_chunked_prefill_finish_time = 0.0 self.last_decode_finish_time = 0.0 self.last_decode_scheduled_time = 0.0 + self.first_prefill_attempt_time = 0.0 if self.trace_ctx.tracing_enable: self.trace_ctx.trace_event("retract", 1, convert_time_to_realtime_ns(ts)) def reset_prefill_retry_time(self): self.wait_queue_entry_time = 0.0 + self.first_prefill_attempt_time = 0.0 self.forward_entry_time = 0.0 self.prefill_finished_time = 0.0 self.completion_time = 0.0 @@ -729,6 +743,10 @@ def set_wait_queue_entry_time(self, ts=None): self.wait_queue_entry_time = ts + def set_first_prefill_attempt_time(self, ts=None): + if self.first_prefill_attempt_time == 0.0: + self.first_prefill_attempt_time = ts or time.perf_counter() + def set_forward_entry_time(self, ts=None): ts = ts or time.perf_counter() if self.forward_entry_time == 0.0: @@ -747,6 +765,8 @@ def set_forward_entry_time(self, ts=None): self.observe_per_stage_req_latency(stage, ts - slice_start_time) self.trace_slice(stage, slice_start_time, ts) + if self.disagg_mode != DisaggregationMode.DECODE: + self.observe_prefill_queue_breakdown(ts) if self.disagg_mode == DisaggregationMode.DECODE: self.trace_ctx.trace_slice_start( @@ -1020,6 +1040,43 @@ def set_decode_prebuilt_finish_time(self, ts=None): def get_queueing_time(self) -> float: return self.forward_entry_time - self.wait_queue_entry_time + def get_idle_in_queue_time(self) -> float: + return self.duration_between( + self.wait_queue_entry_time, self.first_prefill_attempt_time + ) + + def get_prefill_budget_wait_time(self) -> float: + return self.duration_between( + self.first_prefill_attempt_time, self.forward_entry_time + ) + + def observe_prefill_queue_breakdown(self, forward_entry_time: float): + if self.first_prefill_attempt_time <= 0.0: + return + + idle_in_queue_time = self.get_idle_in_queue_time() + prefill_budget_wait_time = self.duration_between( + self.first_prefill_attempt_time, forward_entry_time + ) + + self.observe_per_stage_req_latency( + RequestStage.PREFILL_QUEUE_IDLE, idle_in_queue_time + ) + self.trace_slice( + RequestStage.PREFILL_QUEUE_IDLE, + self.wait_queue_entry_time, + self.first_prefill_attempt_time, + ) + + self.observe_per_stage_req_latency( + RequestStage.PREFILL_BUDGET_WAIT, prefill_budget_wait_time + ) + self.trace_slice( + RequestStage.PREFILL_BUDGET_WAIT, + self.first_prefill_attempt_time, + forward_entry_time, + ) + def convert_to_duration(self) -> str: if self.disagg_mode == DisaggregationMode.NULL: queue_duration = self.duration_between( @@ -1034,7 +1091,12 @@ def convert_to_duration(self) -> str: queue_duration >= 0 and forward_duration >= 0 ), f"queue_duration={queue_duration} < 0 or forward_duration={forward_duration} < 0" - return f"queue_duration={self.format_duration(queue_duration)}, forward_duration={self.format_duration(forward_duration)}, entry_time={self.format_wallclock(self.wait_queue_entry_time)}" + return ( + f"{self.format_prefill_queue_breakdown()}" + f"queue_duration={self.format_duration(queue_duration)}, " + f"forward_duration={self.format_duration(forward_duration)}, " + f"entry_time={self.format_wallclock(self.wait_queue_entry_time)}" + ) elif self.disagg_mode == DisaggregationMode.PREFILL: bootstrap_queue_duration = self.duration_between( self.prefill_bootstrap_queue_entry_time, self.wait_queue_entry_time @@ -1075,6 +1137,7 @@ def convert_to_duration(self) -> str: return ( f"{bootstrap_fields}" + f"{self.format_prefill_queue_breakdown()}" f"queue_duration={self.format_duration(queue_duration)}, " f"forward_duration={self.format_duration(forward_duration)}, " f"entry_time={self.format_wallclock(self.prefill_bootstrap_queue_entry_time)}, " @@ -1153,8 +1216,25 @@ def convert_to_output_meta_info(self): "queue_time": self.get_queueing_time(), } ) + if self.first_prefill_attempt_time > 0.0: + meta_data.update( + { + "idle_in_queue_time": self.get_idle_in_queue_time(), + "prefill_budget_wait_time": self.get_prefill_budget_wait_time(), + } + ) return meta_data + def format_prefill_queue_breakdown(self) -> str: + if self.first_prefill_attempt_time <= 0.0: + return "" + return ( + f"idle_in_queue_duration=" + f"{self.format_duration(self.get_idle_in_queue_time())}, " + f"prefill_budget_wait_duration=" + f"{self.format_duration(self.get_prefill_budget_wait_time())}, " + ) + def format_duration(self, duration: float) -> str: return f"{duration * 1e3:.2f}ms" diff --git a/test/registered/unit/managers/test_prefill_adder.py b/test/registered/unit/managers/test_prefill_adder.py index ded85b1fb1bb..81ab6664d6c3 100644 --- a/test/registered/unit/managers/test_prefill_adder.py +++ b/test/registered/unit/managers/test_prefill_adder.py @@ -80,7 +80,10 @@ def create_mock_req(self, rid, priority, max_new_tokens, output_len=0, wait_time req.extend_logprob_start_len = 0 req.output_ids = [0] * output_len req.sampling_params = SimpleNamespace(max_new_tokens=max_new_tokens) - req.time_stats = SimpleNamespace(wait_queue_entry_time=wait_time) + req.time_stats = SimpleNamespace( + wait_queue_entry_time=wait_time, + set_first_prefill_attempt_time=MagicMock(), + ) req.retracted_stain = False req.finished.return_value = False req.needs_host_load_back.return_value = False diff --git a/test/registered/unit/observability/test_req_time_stats.py b/test/registered/unit/observability/test_req_time_stats.py new file mode 100644 index 000000000000..f037377b557c --- /dev/null +++ b/test/registered/unit/observability/test_req_time_stats.py @@ -0,0 +1,65 @@ +import unittest +from unittest.mock import MagicMock + +from sglang.srt.observability.req_time_stats import ( + RequestStage, + SchedulerReqTimeStats, +) +from sglang.test.ci.ci_register import register_cpu_ci + +register_cpu_ci(est_time=1, suite="base-b-test-cpu") + + +class TestSchedulerReqTimeStats(unittest.TestCase): + def test_prefill_queue_breakdown_from_first_attempt(self): + collector = MagicMock() + stats = SchedulerReqTimeStats() + stats.set_metrics_collector(collector) + + stats.set_wait_queue_entry_time(ts=10.0) + stats.set_first_prefill_attempt_time(ts=13.0) + stats.set_forward_entry_time(ts=21.0) + + self.assertEqual(stats.get_queueing_time(), 11.0) + self.assertEqual(stats.get_idle_in_queue_time(), 3.0) + self.assertEqual(stats.get_prefill_budget_wait_time(), 8.0) + + collector.observe_queue_time.assert_called_once_with(11.0) + collector.observe_per_stage_req_latency.assert_any_call( + RequestStage.PREFILL_QUEUE_IDLE.stage_name, 3.0 + ) + collector.observe_per_stage_req_latency.assert_any_call( + RequestStage.PREFILL_BUDGET_WAIT.stage_name, 8.0 + ) + + meta = stats.convert_to_output_meta_info() + self.assertEqual(meta["queue_time"], 11.0) + self.assertEqual(meta["idle_in_queue_time"], 3.0) + self.assertEqual(meta["prefill_budget_wait_time"], 8.0) + + duration = stats.convert_to_duration() + self.assertIn("idle_in_queue_duration=3000.00ms", duration) + self.assertIn("prefill_budget_wait_duration=8000.00ms", duration) + + def test_first_prefill_attempt_is_set_once_and_serialized(self): + stats = SchedulerReqTimeStats() + + stats.set_first_prefill_attempt_time(ts=13.0) + stats.set_first_prefill_attempt_time(ts=14.0) + + self.assertEqual(stats.first_prefill_attempt_time, 13.0) + self.assertEqual(stats.__getstate__()["first_prefill_attempt_time"], 13.0) + + def test_retry_reset_clears_first_attempt(self): + stats = SchedulerReqTimeStats() + + stats.set_wait_queue_entry_time(ts=10.0) + stats.set_first_prefill_attempt_time(ts=13.0) + stats.reset_prefill_retry_time() + + self.assertEqual(stats.wait_queue_entry_time, 0.0) + self.assertEqual(stats.first_prefill_attempt_time, 0.0) + + +if __name__ == "__main__": + unittest.main()