[Metrics] Add rolling scheduler utilization counters (#37461)

Co-authored-by: Lianmin Zheng <lianminzheng@gmail.com>
Co-authored-by: pranjalssh <pranjalssh@fb.com>

python/sglang/srt/disaggregation/decode.py

@@ -2482,6 +2482,7 @@ class SchedulerDisaggregationDecodeMixin:
             recv_reqs = self.request_receiver.recv_requests()
             self.process_input_requests(recv_reqs)
             if self._engine_paused:
+                self._record_scheduler_state_for_paused_engine()
                 continue
             self.process_decode_queue()
 
@@ -2525,6 +2526,7 @@ class SchedulerDisaggregationDecodeMixin:
             recv_reqs = self.request_receiver.recv_requests()
             self.process_input_requests(recv_reqs)
             if self._engine_paused:
+                self._record_scheduler_state_for_paused_engine()
                 continue
             self.process_decode_queue()
 

python/sglang/srt/disaggregation/prefill.py

@@ -599,6 +599,7 @@ class SchedulerDisaggregationPrefillMixin:
             recv_reqs = self.request_receiver.recv_requests()
             self.process_input_requests(recv_reqs)
             if self._engine_paused:
+                self._record_scheduler_state_for_paused_engine()
                 continue
             self.waiting_queue.extend(
                 self.disagg_prefill_bootstrap_queue.pop_bootstrapped()
@@ -638,6 +639,7 @@ class SchedulerDisaggregationPrefillMixin:
             recv_reqs = self.request_receiver.recv_requests()
             self.process_input_requests(recv_reqs)
             if self._engine_paused:
+                self._record_scheduler_state_for_paused_engine()
                 continue
             self.waiting_queue.extend(
                 self.disagg_prefill_bootstrap_queue.pop_bootstrapped()

python/sglang/srt/hardware_backend/mlx/scheduler_mixin.py

@@ -213,6 +213,7 @@ class SchedulerMlxOverlapMixin:
             recv_reqs = self.request_receiver.recv_requests()
             self.process_input_requests(recv_reqs)
             if self._engine_paused:
+                self._record_scheduler_state_for_paused_engine()
                 continue
 
             # 1. If pending_curr is a pure decode AND no new prefill is waiting,

python/sglang/srt/managers/scheduler.py

@@ -1819,6 +1819,7 @@ class Scheduler(
         if use_mlx():
             # MLX overlap uses mx.async_eval for CPU/GPU overlap,
             # not PyTorch MPS streams.
+            self.metrics_reporter.start_scheduler_time_accounting()
             dispatch_event_loop(self)
             return
 
@@ -1841,6 +1842,7 @@ class Scheduler(
         # on the previous forward's read of the unified memory pool.
         self._war_barrier_enabled = is_cuda() or envs.SGLANG_ENABLE_WAR_BARRIER.get()
         with self.device_module.StreamContext(self.schedule_stream):
+            self.metrics_reporter.start_scheduler_time_accounting()
             dispatch_event_loop(self)
 
     def _apply_war_barrier(self):
@@ -1865,8 +1867,11 @@ class Scheduler(
 
             # Receive requests
             recv_reqs = self.request_receiver.recv_requests()
+            if recv_reqs:
+                self.metrics_reporter.record_scheduler_active()
             self.process_input_requests(recv_reqs)
             if self._engine_paused:
+                self._record_scheduler_state_for_paused_engine()
                 continue
 
             # Get the next batch to run
@@ -1909,8 +1914,11 @@ class Scheduler(
 
             # Receive requests
             recv_reqs = self.request_receiver.recv_requests()
+            if recv_reqs:
+                self.metrics_reporter.record_scheduler_active()
             self.process_input_requests(recv_reqs)
             if self._engine_paused:
+                self._record_scheduler_state_for_paused_engine()
                 continue
 
             # Get the next batch to run
@@ -4121,6 +4129,7 @@ class Scheduler(
         pp_proxy_tensors: Optional[PPProxyTensors] = None,
     ) -> Union[GenerationBatchResult, EmbeddingBatchResult]:
         """Run a batch."""
+        self.metrics_reporter.record_scheduler_active()
         self.forward_ct += 1
         batch.forward_iter = self.forward_ct
         batch.launch_ts = time.monotonic()
@@ -4597,7 +4606,9 @@ class Scheduler(
         # path spins without sleeping, so a wall-clock floor bounds the
         # O(queue) get_loads for both sinks; the fully-idle publish runs
         # post-flush below.
-        if not self.is_fully_idle():
+        fully_idle = self.is_fully_idle()
+        if not fully_idle:
+            self.metrics_reporter.record_scheduler_active()
             now = time.monotonic()
             if now - self._last_stall_publish_ts >= LOAD_STALL_REFRESH_S:
                 self._last_stall_publish_ts = now
@@ -4606,6 +4617,7 @@ class Scheduler(
                     self.load_inquirer.get_loads, force=True, snapshot=snapshot
                 )
             return
+        self.metrics_reporter.record_scheduler_idle()
 
         if self.enable_unified_memory:
             try:
@@ -4658,6 +4670,13 @@ class Scheduler(
 
         # sleep until next event
         self.maybe_sleep_on_idle()
+        self.metrics_reporter.record_scheduler_idle()
+
+    def _record_scheduler_state_for_paused_engine(self) -> None:
+        if self.is_fully_idle():
+            self.metrics_reporter.record_scheduler_idle()
+        else:
+            self.metrics_reporter.record_scheduler_active()
 
     def is_fully_idle(self, for_health_check=False) -> bool:
         # Health check piggybacks on running requests in process_output.

python/sglang/srt/managers/scheduler_components/metrics_reporter.py

@@ -53,6 +53,8 @@ CACHE_HIT_RATE_WINDOW_SECONDS = envs.SGLANG_CACHE_HIT_RATE_WINDOW_SECONDS.get()
 # be re-exported indefinitely. 30s is far above any healthy decode-stats gap,
 # so past it the true recent decode throughput is ~0.
 GEN_THROUGHPUT_STALENESS_SECONDS = 30.0
+# Update scheduler time counters every 1 second.
+_SCHEDULER_TIME_ACCOUNTING_INTERVAL_NS = 1_000_000_000
 
 
 class _CacheHitRateWindow:
@@ -124,6 +126,44 @@ class PrefillStats:
         )
 
 
+@dataclass(slots=True)
+class _SchedulerTimeAccountingSnapshot:
+    start_ns: int
+    last_sample_ns: int
+    start_process_cpu_ns: int
+    accumulate_idle_ns: int
+    # Whether the last sample was idle (for example, no work on the engine).
+    is_idle: bool
+
+    @classmethod
+    def init(
+        cls, now_ns: int, now_process_cpu_ns: int, is_idle: bool
+    ) -> _SchedulerTimeAccountingSnapshot:
+        return cls(
+            start_ns=now_ns,
+            last_sample_ns=now_ns,
+            start_process_cpu_ns=now_process_cpu_ns,
+            accumulate_idle_ns=0,
+            is_idle=is_idle,
+        )
+
+    def sample(self, now_ns: int, is_idle: bool) -> None:
+        if self.is_idle:
+            self.accumulate_idle_ns += now_ns - self.last_sample_ns
+        self.last_sample_ns = now_ns
+        self.is_idle = is_idle
+
+    def should_record(self, now_ns: int) -> bool:
+        return now_ns - self.start_ns >= _SCHEDULER_TIME_ACCOUNTING_INTERVAL_NS
+
+    def reset(self, now_ns: int, now_process_cpu_ns: int, is_idle: bool) -> None:
+        self.start_ns = now_ns
+        self.last_sample_ns = now_ns
+        self.start_process_cpu_ns = now_process_cpu_ns
+        self.accumulate_idle_ns = 0
+        self.is_idle = is_idle
+
+
 @dataclass(kw_only=True)
 class SchedulerMetricsReporter:
     scheduler: Scheduler
@@ -216,6 +256,9 @@ class SchedulerMetricsReporter:
                 self._mfu_log_write_bytes = 0.0
 
         self.fwd_occupancy = float("nan")
+        self._scheduler_time_accounting: Optional[_SchedulerTimeAccountingSnapshot] = (
+            None
+        )
 
         self.forward_pass_device_timer: Optional[DeviceTimer] = None
 
@@ -1196,6 +1239,46 @@ class SchedulerMetricsReporter:
         if self._device_timer_window_batch_count >= self.decode_log_interval:
             self._device_timer_window_batch_count = 0
 
+    def start_scheduler_time_accounting(self) -> None:
+        if not self.enable_metrics:
+            return
+        self._scheduler_time_accounting = _SchedulerTimeAccountingSnapshot.init(
+            time.monotonic_ns(), time.process_time_ns(), True
+        )
+
+    def record_scheduler_active(self) -> None:
+        self._record_scheduler_time(is_idle=False)
+
+    def record_scheduler_idle(self) -> None:
+        self._record_scheduler_time(is_idle=True)
+
+    def _record_scheduler_time(self, is_idle: bool) -> None:
+        if not self.enable_metrics:
+            return
+
+        now_wall_ns = time.monotonic_ns()
+        accounting = self._scheduler_time_accounting
+        if accounting is None:
+            self._scheduler_time_accounting = _SchedulerTimeAccountingSnapshot.init(
+                now_wall_ns, time.process_time_ns(), is_idle
+            )
+            return
+
+        accounting.sample(now_wall_ns, is_idle)
+        if not accounting.should_record(now_wall_ns):
+            return
+
+        now_process_cpu_ns = time.process_time_ns()
+        elapsed_process_cpu_ns = now_process_cpu_ns - accounting.start_process_cpu_ns
+        if accounting.accumulate_idle_ns > 0:
+            self.metrics_collector.increment_scheduler_idle_seconds(
+                accounting.accumulate_idle_ns / 1e9
+            )
+        self.metrics_collector.increment_scheduler_process_cpu_seconds(
+            elapsed_process_cpu_ns / 1e9
+        )
+        accounting.reset(now_wall_ns, now_process_cpu_ns, is_idle)
+
     def _reset_device_timer_window(self):
         """Exclude idle time and invalidate the last forward-occupancy sample."""
         if ENABLE_METRICS_DEVICE_TIMER:

python/sglang/srt/observability/metrics_collector.py

@@ -916,6 +916,22 @@ class SchedulerMetricsCollector(_StatLoggerDIMixin):
             ),
             labelnames=list(labels.keys()) + ["category"],
         )
+        self.scheduler_idle_seconds_total = Counter(
+            name="sglang:scheduler_idle_seconds_total",
+            documentation=(
+                "Total wall time while the scheduler has no runnable or queued work."
+            ),
+            labelnames=labels.keys(),
+        )
+        self.scheduler_process_cpu_seconds_total = Counter(
+            name="sglang:scheduler_process_cpu_seconds_total",
+            documentation=(
+                "Total CPU time consumed by the scheduler process across all threads."
+            ),
+            labelnames=labels.keys(),
+        )
+        self.scheduler_idle_seconds_total.labels(**labels)
+        self.scheduler_process_cpu_seconds_total.labels(**labels)
         self.estimated_flops_per_gpu_total = Counter(
             name="sglang:estimated_flops_per_gpu_total",
             documentation=(
@@ -1302,6 +1318,12 @@ class SchedulerMetricsCollector(_StatLoggerDIMixin):
                 **dp_cooperation_info.to_labels(),
             ).inc(t)
 
+    def increment_scheduler_idle_seconds(self, t: float) -> None:
+        self.scheduler_idle_seconds_total.labels(**self.labels).inc(t)
+
+    def increment_scheduler_process_cpu_seconds(self, t: float) -> None:
+        self.scheduler_process_cpu_seconds_total.labels(**self.labels).inc(t)
+
     def increment_estimated_perf(
         self,
         num_flops_per_gpu: float = 0.0,