[metrics] Add more useful metrics

python/sglang/srt/disaggregation/decode.py

@@ -836,6 +836,7 @@ def event_loop_normal_disagg_decode(self: Scheduler):
 
         while True:
             # Receive requests
+            self.iter_start_time = time.perf_counter()
             recv_reqs = self.recv_requests()
             self.process_input_requests(recv_reqs)
             # polling and allocating kv cache
@@ -863,6 +864,7 @@ def event_loop_overlap_disagg_decode(self: Scheduler):
 
         while True:
             # Receive requests
+            self.iter_start_time = time.perf_counter()
             recv_reqs = self.recv_requests()
             self.process_input_requests(recv_reqs)
             # polling and allocating kv cache
@@ -980,6 +982,10 @@ def get_new_prebuilt_batch(self: Scheduler) -> Optional[ScheduleBatch]:
 
         for req in can_run_list:
             req.time_stats.forward_entry_time = time.perf_counter()
+            if self.enable_metrics:
+                self.metrics_collector.observe_request_waiting_time(
+                    req.time_stats.get_request_waiting_time(),
+                )
 
         # construct a schedule batch with those requests and mark as decode
         new_batch = ScheduleBatch.init_new(

python/sglang/srt/disaggregation/prefill.py

@@ -439,6 +439,15 @@ def process_batch_result_disagg_prefill(
                     logits_output.input_token_logprobs.tolist()
                 )
 
+        if self.enable_metrics:
+            self.iter_forward_finish_time = time.time()
+            run_batch_time = (
+                self.iter_forward_finish_time - self.iter_forward_start_time
+            )
+            self.stats.run_batch_time = run_batch_time
+            self.metrics_collector.log_stats(self.stats)
+
+        hidden_state_offset = 0
         for i, (req, next_token_id) in enumerate(
             zip(batch.reqs, next_token_ids, strict=True)
         ):
@@ -519,6 +528,9 @@ def process_batch_result_disagg_prefill(
                     RequestStage.PREFILL_CHUNKED_FORWARD, req.rid, auto_next_anon=True
                 )
 
+        # Log DP-level prefill load-balancing metrics
+        if self.current_scheduler_metrics_enabled:
+            self.log_prefill_dp_balance_stats(batch)
         self.maybe_send_health_check_signal()
 
     def process_disagg_prefill_inflight_queue(
@@ -577,6 +589,10 @@ def process_disagg_prefill_inflight_queue(
 
         for req in done_reqs:
             req.time_stats.completion_time = time.perf_counter()
+            if self.enable_metrics:
+                self.metrics_collector.observe_request_first_token_forward_time(
+                    req.time_stats.get_request_first_token_forward_time()
+                )
 
         # Stream requests which have finished transfer
         self.stream_output(

python/sglang/srt/entrypoints/http_server.py

@@ -1206,8 +1206,10 @@ async def continue_generation(obj: ContinueGenerationReqInput, request: Request)
 @app.post("/v1/completions", dependencies=[Depends(validate_json_request)])
 async def openai_v1_completions(request: CompletionRequest, raw_request: Request):
     """OpenAI-compatible text completion endpoint."""
+    # Timestamp when the HTTP request is received and handed off to the tokenizer
+    tokenizer_rev_request_time = time.time()
     return await raw_request.app.state.openai_serving_completion.handle_request(
-        request, raw_request
+        request, raw_request, tokenizer_rev_request_time
     )
 
 
@@ -1216,8 +1218,10 @@ async def openai_v1_chat_completions(
     request: ChatCompletionRequest, raw_request: Request
 ):
     """OpenAI-compatible chat completion endpoint."""
+    # Timestamp when the HTTP request is received and handed off to the tokenizer
+    tokenizer_rev_request_time = time.time()
     return await raw_request.app.state.openai_serving_chat.handle_request(
-        request, raw_request
+        request, raw_request, tokenizer_rev_request_time
     )
 
 

python/sglang/srt/entrypoints/openai/serving_base.py

@@ -84,7 +84,10 @@ def _validate_lora_enabled(self, adapter_name: str) -> None:
             )
 
     async def handle_request(
-        self, request: OpenAIServingRequest, raw_request: Request
+        self,
+        request: OpenAIServingRequest,
+        raw_request: Request,
+        tokenizer_rev_request_time: Optional[float] = None,
     ) -> Union[Any, StreamingResponse, ErrorResponse]:
         """Handle the specific request type with common pattern
         If you want to override this method, you should be careful to record the validation time.
@@ -114,11 +117,17 @@ async def handle_request(
             # Note(Xinyuan): raw_request below is only used for detecting the connection of the client
             if hasattr(request, "stream") and request.stream:
                 return await self._handle_streaming_request(
-                    adapted_request, processed_request, raw_request
+                    adapted_request,
+                    processed_request,
+                    raw_request,
+                    tokenizer_rev_request_time,
                 )
             else:
                 return await self._handle_non_streaming_request(
-                    adapted_request, processed_request, raw_request
+                    adapted_request,
+                    processed_request,
+                    raw_request,
+                    tokenizer_rev_request_time,
                 )
         except HTTPException as e:
             return self.create_error_response(

python/sglang/srt/entrypoints/openai/serving_chat.py

@@ -517,10 +517,13 @@ async def _handle_streaming_request(
         adapted_request: GenerateReqInput,
         request: ChatCompletionRequest,
         raw_request: Request,
+        tokenizer_rev_request_time: Optional[float] = None,
     ) -> StreamingResponse:
         """Handle streaming chat completion request"""
         return StreamingResponse(
-            self._generate_chat_stream(adapted_request, request, raw_request),
+            self._generate_chat_stream(
+                adapted_request, request, raw_request, tokenizer_rev_request_time
+            ),
             media_type="text/event-stream",
             background=self.tokenizer_manager.create_abort_task(adapted_request),
         )
@@ -530,6 +533,7 @@ async def _generate_chat_stream(
         adapted_request: GenerateReqInput,
         request: ChatCompletionRequest,
         raw_request: Request,
+        tokenizer_rev_request_time: Optional[float] = None,
     ) -> AsyncGenerator[str, None]:
         """Generate streaming chat completion response"""
         # Parsers for tool calls and reasoning
@@ -551,7 +555,7 @@ async def _generate_chat_stream(
 
         try:
             async for content in self.tokenizer_manager.generate_request(
-                adapted_request, raw_request
+                adapted_request, raw_request, tokenizer_rev_request_time
             ):
                 index = content.get("index", 0)
 
@@ -769,11 +773,12 @@ async def _handle_non_streaming_request(
         adapted_request: GenerateReqInput,
         request: ChatCompletionRequest,
         raw_request: Request,
+        tokenizer_rev_request_time: Optional[float] = None,
     ) -> Union[ChatCompletionResponse, ErrorResponse, ORJSONResponse]:
         """Handle non-streaming chat completion request"""
         try:
             ret = await self.tokenizer_manager.generate_request(
-                adapted_request, raw_request
+                adapted_request, raw_request, tokenizer_rev_request_time
             ).__anext__()
         except ValueError as e:
             return self.create_error_response(str(e))

python/sglang/srt/entrypoints/openai/serving_completions.py

@@ -177,10 +177,13 @@ async def _handle_streaming_request(
         adapted_request: GenerateReqInput,
         request: CompletionRequest,
         raw_request: Request,
+        tokenizer_rev_request_time: Optional[float] = None,
     ) -> StreamingResponse:
         """Handle streaming completion request"""
         return StreamingResponse(
-            self._generate_completion_stream(adapted_request, request, raw_request),
+            self._generate_completion_stream(
+                adapted_request, request, raw_request, tokenizer_rev_request_time
+            ),
             media_type="text/event-stream",
             background=self.tokenizer_manager.create_abort_task(adapted_request),
         )
@@ -190,6 +193,7 @@ async def _generate_completion_stream(
         adapted_request: GenerateReqInput,
         request: CompletionRequest,
         raw_request: Request,
+        tokenizer_rev_request_time: Optional[float] = None,
     ) -> AsyncGenerator[str, None]:
         """Generate streaming completion response"""
         created = int(time.time())
@@ -206,7 +210,7 @@ async def _generate_completion_stream(
 
         try:
             async for content in self.tokenizer_manager.generate_request(
-                adapted_request, raw_request
+                adapted_request, raw_request, tokenizer_rev_request_time
             ):
                 index = content.get("index", 0)
 
@@ -341,11 +345,12 @@ async def _handle_non_streaming_request(
         adapted_request: GenerateReqInput,
         request: CompletionRequest,
         raw_request: Request,
+        tokenizer_rev_request_time: Optional[float] = None,
     ) -> Union[CompletionResponse, ErrorResponse, ORJSONResponse]:
         """Handle non-streaming completion request"""
         try:
             generator = self.tokenizer_manager.generate_request(
-                adapted_request, raw_request
+                adapted_request, raw_request, tokenizer_rev_request_time
             )
             ret = await generator.__anext__()
         except ValueError as e:

python/sglang/srt/managers/io_struct.py

@@ -716,6 +716,9 @@ class TokenizedGenerateReqInput(BaseReq):
     # Session info for continual prompting
     session_params: Optional[SessionParams] = None
 
+    # Timestamp when tokenizer dispatches the request to the scheduler
+    dispatch_to_scheduler_time: Optional[float] = None
+
     # LoRA related
     lora_id: Optional[str] = None  # None means just use the base model
 
@@ -924,6 +927,8 @@ class TokenizedEmbeddingReqInput(BaseReq):
     priority: Optional[int] = None
     # The number of dimensions the resulting output embeddings should have. It is applicable for Matryoshka Embeddings.
     dimensions: Optional[int] = None
+    # Timestamp when tokenizer dispatches the request to the scheduler
+    dispatch_to_scheduler_time: Optional[float] = None
 
 
 @dataclass

python/sglang/srt/managers/schedule_batch.py

@@ -512,6 +512,7 @@ def __init__(
         return_hidden_states: bool = False,
         return_routed_experts: bool = False,
         eos_token_ids: Optional[Set[int]] = None,
+        dispatch_to_scheduler_time: Optional[float] = None,
         bootstrap_host: Optional[str] = None,
         bootstrap_port: Optional[int] = None,
         bootstrap_room: Optional[int] = None,
@@ -745,6 +746,12 @@ def __init__(
         self.has_log_time_stats: bool = False
         self.last_tic = time.monotonic()
 
+        # Timestamp when tokenizer dispatches the request to the scheduler
+        self.dispatch_to_scheduler_time = dispatch_to_scheduler_time
+        # TODO (suhang): Move the dispatch_to_scheduler_time synchronization into Req’s own initializer:
+        # Once dispatch_to_scheduler_time is passed into Req
+        # TimeStats can synchronize it automatically, so the scheduler no longer needs that extra getattr check.
+
         # For disaggregation
         self.bootstrap_host: str = bootstrap_host
         self.bootstrap_port: Optional[int] = bootstrap_port
@@ -1214,11 +1221,15 @@ class ScheduleBatch(ScheduleBatchDisaggregationDecodeMixin):
     inner_idle_batch: Optional[ScheduleBatch] = None
     global_num_tokens: Optional[List[int]] = None
     global_num_tokens_for_logprob: Optional[List[int]] = None
+    dp_global_num_tokens_for_metric: Optional[List[int]] = None
     is_extend_in_batch: bool = False
     can_run_dp_cuda_graph: bool = False
     tbo_split_seq_index: Optional[int] = None
     global_forward_mode: Optional[ForwardMode] = None
 
+    # DP all_gather latency for this batch
+    all_gather_latency: float = 0.0
+
     # For processing logprobs
     return_logprob: bool = False
     top_logprobs_nums: Optional[List[int]] = None
@@ -2195,6 +2206,8 @@ def copy(self):
             spec_algorithm=self.spec_algorithm,
             global_num_tokens=self.global_num_tokens,
             global_num_tokens_for_logprob=self.global_num_tokens_for_logprob,
+            dp_global_num_tokens_for_metric=self.dp_global_num_tokens_for_metric,
+            all_gather_latency=self.all_gather_latency,
             can_run_dp_cuda_graph=self.can_run_dp_cuda_graph,
             is_extend_in_batch=self.is_extend_in_batch,
             is_prefill_only=self.is_prefill_only,

python/sglang/srt/managers/scheduler.py

@@ -319,12 +319,6 @@ def __init__(
         # Init model configs
         self.init_model_config()
 
-        # Init metrics stats
-        self.init_metrics(tp_rank, pp_rank, dp_rank)
-
-        # Init inter-process communication
-        self.init_ipc_channels(port_args)
-
         # Init PD-multiplexing context
         if self.enable_pdmux:
             self.init_pdmux()
@@ -338,6 +332,12 @@ def __init__(
         # Launch a model worker and draft model worker if using speculative decoding
         self.init_model_worker()
 
+        # Init metrics stats
+        self.init_metrics(tp_rank, pp_rank, dp_rank)
+
+        # Init inter-process communication
+        self.init_ipc_channels(port_args)
+
         if (t := envs.SGLANG_TEST_STUCK_SCHEDULER_INIT.get()) > 0:
             time.sleep(t)
 
@@ -1056,6 +1056,7 @@ def event_loop_normal(self):
         """A normal scheduler loop."""
         while True:
             # Receive requests
+            self.iter_start_time = time.perf_counter()
             recv_reqs = self.recv_requests()
             self.process_input_requests(recv_reqs)
             if self._engine_paused:
@@ -1092,6 +1093,7 @@ def pop_and_process():
 
         while True:
             # Receive requests
+            self.iter_start_time = time.perf_counter()
             recv_reqs = self.recv_requests()
             self.process_input_requests(recv_reqs)
             if self._engine_paused:
@@ -1429,6 +1431,7 @@ def handle_generate_request(
                 return_hidden_states=recv_req.return_hidden_states,
                 return_routed_experts=recv_req.return_routed_experts,
                 eos_token_ids=self.model_config.hf_eos_token_id,
+                dispatch_to_scheduler_time=recv_req.dispatch_to_scheduler_time,
                 bootstrap_host=recv_req.bootstrap_host,
                 bootstrap_port=recv_req.bootstrap_port,
                 bootstrap_room=recv_req.bootstrap_room,
@@ -1444,6 +1447,12 @@ def handle_generate_request(
                 dllm_config=self.dllm_config,
             )
             req.tokenizer = self.tokenizer
+            if getattr(recv_req, "dispatch_to_scheduler_time", 0.0):
+                # Keep dispatch timestamp only when present, clamp to zero to avoid negative values
+                req.time_stats.dispatch_to_scheduler_time = max(
+                    0.0, recv_req.dispatch_to_scheduler_time
+                )
+            req.time_stats.arrive_scheduler_time = time.perf_counter()
 
             if self.disaggregation_mode != DisaggregationMode.NULL:
                 # Invalid request for disaggregated mode
@@ -1674,6 +1683,7 @@ def handle_embedding_request(
             recv_req.input_text,
             recv_req.input_ids,
             recv_req.sampling_params,
+            dispatch_to_scheduler_time=recv_req.dispatch_to_scheduler_time,
             token_type_ids=recv_req.token_type_ids,
             priority=recv_req.priority,
             dimensions=recv_req.dimensions,
@@ -2001,9 +2011,15 @@ def _get_new_batch_prefill_raw(
             if req.time_stats.forward_entry_time == 0:
                 req.time_stats.forward_entry_time = time.perf_counter()
                 if self.enable_metrics:
+                    self.metrics_collector.observe_request_zmq_time(
+                        req.time_stats.get_request_zmq_time(),
+                    )
                     self.metrics_collector.observe_queue_time(
                         req.time_stats.get_queueing_time(),
                     )
+                    self.metrics_collector.observe_request_waiting_time(
+                        req.time_stats.get_request_waiting_time(),
+                    )
 
         # Create a new batch
         new_batch = ScheduleBatch.init_new(
@@ -2141,6 +2157,7 @@ def run_batch(
     ) -> Union[GenerationBatchResult, EmbeddingBatchResult]:
         """Run a batch."""
         self.forward_ct += 1
+        self.iter_forward_start_time = time.time()
 
         # Whether to run the profiler
         self._profile_batch_predicate(batch)