From 780d969699008ede2ca7bff4cfd28d7bfb4288c9 Mon Sep 17 00:00:00 2001
From: fzyzcjy <5236035+fzyzcjy@users.noreply.github.com>
Date: Mon, 18 May 2026 18:41:35 +0800
Subject: [PATCH] Add SchedulerMetricsReporter and route metrics state through
 it (#25629)

---
 python/sglang/srt/disaggregation/decode.py    |   6 +-
 .../srt/disaggregation/encode_receiver.py     |   2 +-
 python/sglang/srt/disaggregation/prefill.py   |   9 +-
 python/sglang/srt/dllm/mixin/scheduler.py     |   3 +-
 python/sglang/srt/managers/scheduler.py       |  97 +++--
 .../scheduler_components/metrics_reporter.py  |  47 +++
 .../scheduler_output_processor_mixin.py       |  14 +-
 .../srt/observability/metrics_collector.py    |  66 +++
 .../observability/scheduler_metrics_mixin.py  | 381 ++++++++++--------
 .../test_scheduler_chunked_req_gate.py        |   1 +
 10 files changed, 406 insertions(+), 220 deletions(-)
 create mode 100644 python/sglang/srt/managers/scheduler_components/metrics_reporter.py

diff --git a/python/sglang/srt/disaggregation/decode.py b/python/sglang/srt/disaggregation/decode.py
index 258d411ac..1f0101540 100644
--- a/python/sglang/srt/disaggregation/decode.py
+++ b/python/sglang/srt/disaggregation/decode.py
@@ -637,7 +637,7 @@ class DecodePreallocQueue:
                     error_message,
                     status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
                 )
-                if self.scheduler.enable_metrics:
+                if self.scheduler.metrics_reporter.enable_metrics:
                     self.scheduler.metrics_collector.increment_bootstrap_failed_reqs()
             else:
                 raise ValueError(f"Unexpected poll case: {poll}")
@@ -1516,7 +1516,7 @@ class DecodeTransferQueue:
                 # release pre-allocated kv cache, but don't insert into the tree since it's failed
                 release_kv_cache(decode_req.req, self.tree_cache, is_insert=False)
                 indices_to_remove.add(i)
-                if self.scheduler.enable_metrics:
+                if self.scheduler.metrics_reporter.enable_metrics:
                     self.scheduler.metrics_collector.increment_transfer_failed_reqs()
                 continue
             elif poll == KVPoll.Success:
@@ -1535,7 +1535,7 @@ class DecodeTransferQueue:
                         release_kv_cache(
                             decode_req.req, self.tree_cache, is_insert=False
                         )
-                        if self.scheduler.enable_metrics:
+                        if self.scheduler.metrics_reporter.enable_metrics:
                             self.scheduler.metrics_collector.increment_transfer_failed_reqs()
                     else:
                         transferred_reqs.append(decode_req.req)
diff --git a/python/sglang/srt/disaggregation/encode_receiver.py b/python/sglang/srt/disaggregation/encode_receiver.py
index 63194957e..f5a5b5724 100644
--- a/python/sglang/srt/disaggregation/encode_receiver.py
+++ b/python/sglang/srt/disaggregation/encode_receiver.py
@@ -1031,7 +1031,7 @@ class MMReceiverBase(ABC):
             priority=recv_req.priority,
             metrics_collector=(
                 self.scheduler.metrics_collector
-                if self.scheduler.enable_metrics
+                if self.scheduler.metrics_reporter.enable_metrics
                 else None
             ),
             http_worker_ipc=recv_req.http_worker_ipc,
diff --git a/python/sglang/srt/disaggregation/prefill.py b/python/sglang/srt/disaggregation/prefill.py
index 67ac83bc6..72f0450d8 100644
--- a/python/sglang/srt/disaggregation/prefill.py
+++ b/python/sglang/srt/disaggregation/prefill.py
@@ -312,7 +312,7 @@ class PrefillBootstrapQueue:
                 self.scheduler.stream_output([req], req.return_logprob)
                 indices_to_remove.add(i)
                 failed_reqs.append(req)
-                if self.scheduler.enable_metrics:
+                if self.scheduler.metrics_reporter.enable_metrics:
                     self.scheduler.metrics_collector.increment_bootstrap_failed_reqs()
                 if self.scheduler.enable_hicache_storage:
                     # to release prefetch events associated with the request
@@ -588,6 +588,7 @@ class SchedulerDisaggregationPrefillMixin:
 
         can_run_cuda_graph = getattr(result, "can_run_cuda_graph", False)
         self.report_prefill_stats(
+            self.metrics_reporter,
             batch=batch,
             prefill_stats=batch.prefill_stats,
             can_run_cuda_graph=can_run_cuda_graph,
@@ -659,7 +660,7 @@ class SchedulerDisaggregationPrefillMixin:
                     req, error_message, status_code=HTTPStatus.INTERNAL_SERVER_ERROR
                 )
                 done_reqs.append(req)
-                if self.enable_metrics:
+                if self.metrics_reporter.enable_metrics:
                     self.metrics_collector.increment_transfer_failed_reqs()
             else:
                 logger.warning_once(
@@ -685,9 +686,9 @@ class SchedulerDisaggregationPrefillMixin:
             if metrics:
                 # Update last-value for REST API
                 if "latency_ms" in metrics:
-                    self.kv_transfer_latency_ms = metrics["latency_ms"]
+                    self.metrics_reporter.kv_transfer_latency_ms = metrics["latency_ms"]
                 if "speed_gb_s" in metrics:
-                    self.kv_transfer_speed_gb_s = metrics["speed_gb_s"]
+                    self.metrics_reporter.kv_transfer_speed_gb_s = metrics["speed_gb_s"]
 
         # Stream requests which have finished transfer
         self.stream_output(
diff --git a/python/sglang/srt/dllm/mixin/scheduler.py b/python/sglang/srt/dllm/mixin/scheduler.py
index 157ab2192..8179674bb 100644
--- a/python/sglang/srt/dllm/mixin/scheduler.py
+++ b/python/sglang/srt/dllm/mixin/scheduler.py
@@ -80,7 +80,7 @@ class SchedulerDllmMixin:
                     continue
 
                 req.fill_ids[-new_tokens:] = next_token_ids[:]
-                self.num_generated_tokens += new_tokens
+                self.metrics_reporter.num_generated_tokens += new_tokens
 
                 req.output_ids.extend(next_token_ids)
                 req.check_finished(new_accepted_len=new_tokens)
@@ -94,6 +94,7 @@ class SchedulerDllmMixin:
 
         can_run_cuda_graph = getattr(result, "can_run_cuda_graph", False)
         self.report_prefill_stats(
+            self.metrics_reporter,
             batch=batch,
             prefill_stats=batch.prefill_stats,
             can_run_cuda_graph=can_run_cuda_graph,
diff --git a/python/sglang/srt/managers/scheduler.py b/python/sglang/srt/managers/scheduler.py
index c2c5b3036..420885ae0 100644
--- a/python/sglang/srt/managers/scheduler.py
+++ b/python/sglang/srt/managers/scheduler.py
@@ -176,6 +176,9 @@ from sglang.srt.managers.scheduler_components.kv_events_publisher import (
 from sglang.srt.managers.scheduler_components.load_inquirer import (
     SchedulerLoadInquirer,
 )
+from sglang.srt.managers.scheduler_components.metrics_reporter import (
+    SchedulerMetricsReporter,
+)
 from sglang.srt.managers.scheduler_components.pool_stats_observer import (
     SchedulerPoolStatsObserver,
 )
@@ -203,6 +206,7 @@ from sglang.srt.mem_cache.common import maybe_cache_unfinished_req, release_kv_c
 from sglang.srt.model_executor.forward_batch_info import ForwardMode, PPProxyTensors
 from sglang.srt.model_loader.utils import get_resolved_model_impl
 from sglang.srt.multiplex.multiplexing_mixin import SchedulerMultiplexMixin
+from sglang.srt.observability.metrics_collector import SchedulerMetricsCollector
 from sglang.srt.observability.req_time_stats import (
     real_time,
     set_schedule_time_batch,
@@ -452,7 +456,17 @@ class Scheduler(
         self.init_model_config()
 
         # Init metrics stats
-        self.init_metrics(tp_rank, pp_rank, dp_rank)
+        self.metrics_collector_context = SchedulerMetricsCollector.init_new(
+            server_args=self.server_args,
+            ps=self.ps,
+            tp_rank=tp_rank,
+            pp_rank=pp_rank,
+            dp_rank=dp_rank,
+            enable_priority_scheduling=self.enable_priority_scheduling,
+            enable_lora=self.enable_lora,
+            enable_hierarchical_cache=self.enable_hierarchical_cache,
+        )
+        self.metrics_collector = self.metrics_collector_context.collector
 
         # Init inter-process communication
         self.init_ipc_channels(port_args)
@@ -481,7 +495,6 @@ class Scheduler(
 
         # Launch a model worker and draft model worker if using speculative decoding
         self.init_model_worker()
-        self.install_device_timer_on_runners()
 
         if (t := envs.SGLANG_TEST_STUCK_SCHEDULER_INIT.get()) > 0:
             time.sleep(t)
@@ -496,7 +509,7 @@ class Scheduler(
             attn_tp_cpu_group=self.attn_tp_cpu_group,
             tp_cpu_group=self.tp_cpu_group,
             attn_cp_cpu_group=self.attn_cp_cpu_group,
-            enable_metrics=self.enable_metrics,
+            enable_metrics=self.server_args.enable_metrics,
             enable_kv_cache_events=bool(
                 self.server_args.kv_events_config
                 and self.ps.attn_tp_rank == 0
@@ -559,6 +572,15 @@ class Scheduler(
         # Init diffusion LLM
         self.init_diffusion_llm()
 
+        self.metrics_reporter = SchedulerMetricsReporter(
+            scheduler=self,
+            tp_rank=tp_rank,
+            pp_rank=pp_rank,
+            dp_rank=dp_rank,
+            metrics_collector_context=self.metrics_collector_context,
+            metrics_collector=self.metrics_collector,
+        )
+
         # Init schedule policy and new token estimation
         self.init_schedule_policy()
 
@@ -695,7 +717,7 @@ class Scheduler(
             send_metrics_from_scheduler=self.send_metrics_from_scheduler,
             max_running_requests=self.max_running_requests,
             max_total_num_tokens=self.max_total_num_tokens,
-            get_stats=lambda: self.stats,
+            get_stats=lambda: self.metrics_reporter.stats,
         )
 
         self.load_inquirer = SchedulerLoadInquirer(
@@ -710,14 +732,14 @@ class Scheduler(
             spec_algorithm=self.spec_algorithm,
             get_running_batch=lambda: self.running_batch,
             get_waiting_queue=lambda: self.waiting_queue,
-            get_stats=lambda: self.stats,
+            get_stats=lambda: self.metrics_reporter.stats,
             get_chunked_req=lambda: self.chunked_req,
             get_disagg_prefill_bootstrap_queue=lambda: self.disagg_prefill_bootstrap_queue,
             get_disagg_prefill_inflight_queue=lambda: self.disagg_prefill_inflight_queue,
             get_disagg_decode_prealloc_queue=lambda: self.disagg_decode_prealloc_queue,
             get_disagg_decode_transfer_queue=lambda: self.disagg_decode_transfer_queue,
-            get_spec_total_num_accept_tokens=lambda: self.spec_total_num_accept_tokens,
-            get_spec_total_num_forward_ct=lambda: self.spec_total_num_forward_ct,
+            get_spec_total_num_accept_tokens=lambda: self.metrics_reporter.spec_total_num_accept_tokens,
+            get_spec_total_num_forward_ct=lambda: self.metrics_reporter.spec_total_num_forward_ct,
         )
 
         self.is_initializing = False
@@ -800,7 +822,10 @@ class Scheduler(
             self.send_to_tokenizer = SenderWrapper(None)
             self.send_to_detokenizer = SenderWrapper(None)
 
-        if self.current_scheduler_metrics_enabled:
+        if self.server_args.enable_metrics and (
+            self.ps.attn_tp_rank == 0
+            or self.server_args.enable_metrics_for_all_schedulers
+        ):
             self.send_metrics_from_scheduler = get_zmq_socket(
                 context, zmq.PUSH, port_args.metrics_ipc_name, False
             )
@@ -1010,7 +1035,7 @@ class Scheduler(
                 f"{'available_cpu_mem' if self.device == 'cpu' else 'available_gpu_mem'}={avail_mem:.2f} GB"
             )
 
-        if self.enable_metrics:
+        if self.server_args.enable_metrics:
             self.metrics_collector.emit_constants(
                 max_total_num_tokens=self.max_total_num_tokens,
                 # TODO: max_running_requests_under_SLO has no setter — dead chain.
@@ -1036,8 +1061,6 @@ class Scheduler(
         self.forward_ct = 0
         self.return_health_check_ipcs: Deque[Optional[str]] = deque()
         self._pending_flush: Optional[Tuple[FlushCacheReqInput, float]] = None
-        self.num_retracted_reqs: int = 0
-        self.num_paused_reqs: int = 0
         self.session_controller = SessionController(self.tree_cache)
         self.forward_sleep_time = None
         self._engine_paused = False
@@ -1114,7 +1137,9 @@ class Scheduler(
                     device_group=self.tp_group.device_group,
                     server_args=self.server_args,
                     metrics_collector=(
-                        self.metrics_collector if self.enable_metrics else None
+                        self.metrics_collector
+                        if self.metrics_reporter.enable_metrics
+                        else None
                     ),
                     max_delay_passes=self.server_args.prefill_delayer_max_delay_passes,
                     token_usage_low_watermark=self.server_args.prefill_delayer_token_usage_low_watermark,
@@ -1860,7 +1885,9 @@ class Scheduler(
                 vocab_size=self.model_config.vocab_size,
                 priority=recv_req.priority,
                 metrics_collector=(
-                    self.metrics_collector if self.enable_metrics else None
+                    self.metrics_collector
+                    if self.metrics_reporter.enable_metrics
+                    else None
                 ),
                 routing_key=recv_req.routing_key,
                 extra_key=recv_req.extra_key,
@@ -1902,7 +1929,7 @@ class Scheduler(
                 eos_token_ids=self.model_config.hf_eos_token_id,
             )
             # TODO: set trace context
-            if self.enable_metrics:
+            if self.metrics_reporter.enable_metrics:
                 req.time_stats.set_metrics_collector(self.metrics_collector)
             if isinstance(req.finished_reason, FINISH_ABORT):
                 self.init_req_max_new_tokens(req)
@@ -2752,9 +2779,9 @@ class Scheduler(
                 else None
             )
 
-            self.num_retracted_reqs = len(retracted_reqs)
-            if self.enable_metrics and len(retracted_reqs) > 0:
-                self.metrics_collector.increment_retracted_reqs(
+            self.metrics_reporter.num_retracted_reqs = len(retracted_reqs)
+            if self.metrics_reporter.enable_metrics and len(retracted_reqs) > 0:
+                self.metrics_reporter.metrics_collector.increment_retracted_reqs(
                     num_retracted_reqs=len(retracted_reqs),
                     num_retracted_input_tokens=sum(
                         len(r.origin_input_ids) for r in retracted_reqs
@@ -3040,15 +3067,15 @@ class Scheduler(
         elif batch.forward_mode.is_idle():
             self.process_batch_result_idle(batch, result)
 
-        self.log_batch_result_stats(batch, result)
+        self.log_batch_result_stats(self.metrics_reporter, batch, result)
 
         # Emit forward pass metrics (every iteration when enabled)
         if self.enable_fpm:
-            self._emit_forward_pass_metrics(batch, result)
+            self._emit_forward_pass_metrics(self.metrics_reporter, batch, result)
 
         self._maybe_clear_mm_inputs(batch)
         self.maybe_send_health_check_signal()
-        self.update_device_timer()
+        self.update_device_timer(self.metrics_reporter)
 
     def maybe_send_health_check_signal(self):
         if self.return_health_check_ipcs:
@@ -3174,7 +3201,7 @@ class Scheduler(
         self.new_token_ratio = self.init_new_token_ratio
 
         # reset device timer window so idle time isn't counted
-        self.reset_device_timer_window()
+        self.reset_device_timer_window(self.metrics_reporter)
 
         # sleep until next event
         self.maybe_sleep_on_idle()
@@ -3335,7 +3362,7 @@ class Scheduler(
             self.req_to_token_pool.clear()
             self.token_to_kv_pool_allocator.clear()
             self.grammar_manager.clear()
-            self.reset_metrics()
+            self.reset_metrics(self.metrics_reporter)
 
             if self.draft_worker:
                 self.draft_worker.clear_cache_pool()
@@ -3358,7 +3385,7 @@ class Scheduler(
 
     def get_internal_state(self, recv_req: GetInternalStateReq):
         ret = dict(vars(get_global_server_args()))  # vars returns a ref to obj.__dict__
-        ret["last_gen_throughput"] = self.last_gen_throughput
+        ret["last_gen_throughput"] = self.metrics_reporter.last_gen_throughput
         ret["memory_usage"] = {
             "weight": round(self.tp_worker.model_runner.weight_load_mem_usage, 2),
             "kvcache": round(
@@ -3369,13 +3396,17 @@ class Scheduler(
         }
         ret["effective_max_running_requests_per_dp"] = self.max_running_requests
 
-        if not self.spec_algorithm.is_none() and self.spec_total_num_forward_ct > 0:
+        if (
+            not self.spec_algorithm.is_none()
+            and self.metrics_reporter.spec_total_num_forward_ct > 0
+        ):
             ret["avg_spec_accept_length"] = (
-                self.spec_total_num_accept_tokens / self.spec_total_num_forward_ct
+                self.metrics_reporter.spec_total_num_accept_tokens
+                / self.metrics_reporter.spec_total_num_forward_ct
             )
 
         if RECORD_STEP_TIME:
-            ret["step_time_dict"] = self.step_time_dict
+            ret["step_time_dict"] = self.metrics_reporter.step_time_dict
 
         # This field is not serializable.
         ret.pop("model_config", None)
@@ -3408,12 +3439,18 @@ class Scheduler(
                 break
 
         if if_success:
-            if not self.spec_algorithm.is_none() and self.spec_total_num_forward_ct > 0:
+            if (
+                not self.spec_algorithm.is_none()
+                and self.metrics_reporter.spec_total_num_forward_ct > 0
+            ):
                 avg_spec_accept_length = (
-                    self.spec_total_num_accept_tokens / self.spec_total_num_forward_ct
+                    self.metrics_reporter.spec_total_num_accept_tokens
+                    / self.metrics_reporter.spec_total_num_forward_ct
                 )
                 logger.info(f"{avg_spec_accept_length=}")
-            self.spec_total_num_accept_tokens = self.spec_total_num_forward_ct = 0
+            self.metrics_reporter.spec_total_num_accept_tokens = (
+                self.metrics_reporter.spec_total_num_forward_ct
+            ) = 0
             for k, v in server_args_dict.items():
                 setattr(get_global_server_args(), k, v)
             logger.info(f"Global server args updated! {get_global_server_args()=}")
@@ -3944,4 +3981,4 @@ def run_scheduler_process(
         if scheduler is not None:
             # FPM has a background ZMQ publisher thread that needs explicit
             # teardown to flush queued metrics and close the socket cleanly.
-            scheduler._shutdown_fpm()
+            scheduler._shutdown_fpm(scheduler.metrics_reporter)
diff --git a/python/sglang/srt/managers/scheduler_components/metrics_reporter.py b/python/sglang/srt/managers/scheduler_components/metrics_reporter.py
new file mode 100644
index 000000000..bbe17d101
--- /dev/null
+++ b/python/sglang/srt/managers/scheduler_components/metrics_reporter.py
@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Optional
+
+from sglang.srt.observability.metrics_collector import (
+    SchedulerMetricsCollector,
+    SchedulerMetricsCollectorContext,
+)
+from sglang.srt.observability.scheduler_metrics_mixin import (
+    SchedulerMetricsMixin,
+)
+
+if TYPE_CHECKING:
+    from sglang.srt.managers.scheduler import Scheduler
+
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(kw_only=True)
+class SchedulerMetricsReporter:
+    scheduler: "Scheduler"
+    tp_rank: int
+    pp_rank: int
+    dp_rank: Optional[int]
+    metrics_collector_context: SchedulerMetricsCollectorContext
+    metrics_collector: Optional[SchedulerMetricsCollector]
+    num_retracted_reqs: int = 0
+    num_paused_reqs: int = 0
+
+    def __post_init__(self) -> None:
+        self.enable_metrics = self.metrics_collector_context.enable_metrics
+        self.is_stats_logging_rank = (
+            self.metrics_collector_context.is_stats_logging_rank
+        )
+        self.current_scheduler_metrics_enabled = (
+            self.metrics_collector_context.current_scheduler_metrics_enabled
+        )
+        self.enable_kv_cache_events = (
+            self.metrics_collector_context.enable_kv_cache_events
+        )
+        SchedulerMetricsMixin._init_metrics(
+            self, self.tp_rank, self.pp_rank, self.dp_rank
+        )
+        SchedulerMetricsMixin._install_device_timer_on_runners(self)
diff --git a/python/sglang/srt/managers/scheduler_output_processor_mixin.py b/python/sglang/srt/managers/scheduler_output_processor_mixin.py
index e7ad362c7..b52f742ef 100644
--- a/python/sglang/srt/managers/scheduler_output_processor_mixin.py
+++ b/python/sglang/srt/managers/scheduler_output_processor_mixin.py
@@ -400,6 +400,7 @@ class SchedulerOutputProcessorMixin:
 
         can_run_cuda_graph = getattr(result, "can_run_cuda_graph", False)
         self.report_prefill_stats(
+            self.metrics_reporter,
             batch=batch,
             prefill_stats=batch.prefill_stats,
             can_run_cuda_graph=can_run_cuda_graph,
@@ -511,10 +512,12 @@ class SchedulerOutputProcessorMixin:
         # else: Spec V1 — output_ids, check_finished, grammar, and reasoning tokens
         # are already handled in the verify phase (eagle_info.py / ngram_info.py).
 
-        self.num_generated_tokens += len(batch.reqs)
+        self.metrics_reporter.num_generated_tokens += len(batch.reqs)
         if not batch.spec_algorithm.is_none():
-            self.update_spec_metrics(batch.batch_size(), result.num_correct_drafts)
-        if self.enable_metrics:
+            self.update_spec_metrics(
+                self.metrics_reporter, batch.batch_size(), result.num_correct_drafts
+            )
+        if self.metrics_reporter.enable_metrics:
             self.metrics_collector.increment_decode_cuda_graph_pass(
                 value=can_run_cuda_graph
             )
@@ -624,8 +627,11 @@ class SchedulerOutputProcessorMixin:
         self.stream_output(batch.reqs, batch.return_logprob)
         self.token_to_kv_pool_allocator.free_group_end()
 
-        self.forward_ct_decode = (self.forward_ct_decode + 1) % (1 << 30)
+        self.metrics_reporter.forward_ct_decode = (
+            self.metrics_reporter.forward_ct_decode + 1
+        ) % (1 << 30)
         self.report_decode_stats(
+            self.metrics_reporter,
             can_run_cuda_graph,
             running_batch=batch,
             num_correct_drafts=result.num_correct_drafts,
diff --git a/python/sglang/srt/observability/metrics_collector.py b/python/sglang/srt/observability/metrics_collector.py
index 4fe3b6f37..f71fe1d99 100644
--- a/python/sglang/srt/observability/metrics_collector.py
+++ b/python/sglang/srt/observability/metrics_collector.py
@@ -23,6 +23,7 @@ from collections import Counter
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Union
 
+from sglang.srt.disaggregation.utils import DisaggregationMode
 from sglang.srt.environ import envs
 from sglang.srt.model_executor.forward_batch_info import ForwardMode
 from sglang.srt.observability.utils import exponential_buckets, generate_buckets
@@ -182,6 +183,15 @@ class DPCooperationInfo:
         return dataclasses.asdict(self)
 
 
+@dataclass(kw_only=True, frozen=True, slots=True)
+class SchedulerMetricsCollectorContext:
+    enable_metrics: bool
+    is_stats_logging_rank: bool
+    current_scheduler_metrics_enabled: bool
+    enable_kv_cache_events: bool
+    collector: Optional["SchedulerMetricsCollector"]
+
+
 class SchedulerMetricsCollector:
 
     def __init__(
@@ -938,6 +948,62 @@ class SchedulerMetricsCollector:
             multiprocess_mode="mostrecent",
         )
 
+    @classmethod
+    def init_new(
+        cls,
+        *,
+        server_args: "ServerArgs",
+        ps: Any,
+        tp_rank: int,
+        pp_rank: int,
+        dp_rank: Optional[int],
+        enable_priority_scheduling: bool,
+        enable_lora: bool,
+        enable_hierarchical_cache: bool,
+    ) -> "SchedulerMetricsCollectorContext":
+        enable_metrics = server_args.enable_metrics
+        is_stats_logging_rank = ps.attn_tp_rank == 0
+        current_scheduler_metrics_enabled = enable_metrics and (
+            is_stats_logging_rank or server_args.enable_metrics_for_all_schedulers
+        )
+        enable_kv_cache_events = bool(
+            server_args.kv_events_config
+            and ps.attn_tp_rank == 0
+            and ps.attn_cp_rank == 0
+        )
+        collector: Optional["SchedulerMetricsCollector"] = None
+        if enable_metrics:
+            engine_type = DisaggregationMode.to_engine_type(
+                server_args.disaggregation_mode
+            )
+            labels = {
+                "model_name": server_args.served_model_name,
+                "engine_type": engine_type,
+                "tp_rank": tp_rank,
+                "pp_rank": pp_rank,
+                "moe_ep_rank": ps.moe_ep_rank,
+            }
+            if enable_priority_scheduling:
+                labels["priority"] = ""
+            if dp_rank is not None:
+                labels["dp_rank"] = dp_rank
+            if server_args.extra_metric_labels:
+                labels.update(server_args.extra_metric_labels)
+            collector = cls(
+                labels=labels,
+                enable_lora=enable_lora,
+                enable_hierarchical_cache=enable_hierarchical_cache,
+                enable_streaming_session=server_args.enable_streaming_session,
+                server_args=server_args,
+            )
+        return SchedulerMetricsCollectorContext(
+            enable_metrics=enable_metrics,
+            is_stats_logging_rank=is_stats_logging_rank,
+            current_scheduler_metrics_enabled=current_scheduler_metrics_enabled,
+            enable_kv_cache_events=enable_kv_cache_events,
+            collector=collector,
+        )
+
     def _log_gauge(self, gauge: Gauge, data: Union[int, float]) -> None:
         # Convenience function for logging a scalar to gauge.
         gauge.labels(**self.labels).set(data)
diff --git a/python/sglang/srt/observability/scheduler_metrics_mixin.py b/python/sglang/srt/observability/scheduler_metrics_mixin.py
index 8a1afc243..81ab6514f 100644
--- a/python/sglang/srt/observability/scheduler_metrics_mixin.py
+++ b/python/sglang/srt/observability/scheduler_metrics_mixin.py
@@ -14,7 +14,6 @@ from sglang.srt.managers.utils import GenerationBatchResult
 from sglang.srt.observability.metrics_collector import (
     DPCooperationInfo,
     QueueCount,
-    SchedulerMetricsCollector,
     SchedulerStats,
     compute_routing_key_stats,
 )
@@ -24,7 +23,7 @@ from sglang.srt.utils.scheduler_status_logger import SchedulerStatusLogger
 if TYPE_CHECKING:
     from sglang.srt.managers.schedule_batch import Req
     from sglang.srt.managers.schedule_policy import PrefillAdder
-    from sglang.srt.managers.scheduler import EmbeddingBatchResult, Scheduler
+    from sglang.srt.managers.scheduler import EmbeddingBatchResult
 
 logger = logging.getLogger(__name__)
 
@@ -67,8 +66,12 @@ class PrefillStats:
 class SchedulerMetricsMixin:
     enable_fpm: bool = False
 
-    def init_metrics(
-        self: Scheduler, tp_rank: int, pp_rank: int, dp_rank: Optional[int]
+    @staticmethod
+    def _init_metrics(
+        self: "SchedulerMetricsReporter",
+        tp_rank: int,
+        pp_rank: int,
+        dp_rank: Optional[int],
     ):
         # Basic stats
         self.forward_ct_decode = 0
@@ -83,7 +86,7 @@ class SchedulerMetricsMixin:
             "cpu": "cpu graph",
             "npu": "npu graph",
             "musa": "musa graph",
-        }.get(getattr(self, "device", ""), "cuda graph")
+        }.get(getattr(self.scheduler, "device", ""), "cuda graph")
 
         # Cumulative spec-decoding counters (reset every decode_log_interval).
         # Each update adds (num_correct_drafts + bs, bs).
@@ -97,43 +100,12 @@ class SchedulerMetricsMixin:
         self.kv_transfer_speed_gb_s: float = 0.0
         self.kv_transfer_latency_ms: float = 0.0
 
-        # Metrics
-        self.enable_metrics = self.server_args.enable_metrics
-        self.is_stats_logging_rank = self.ps.attn_tp_rank == 0
-        self.current_scheduler_metrics_enabled = self.enable_metrics and (
-            self.is_stats_logging_rank
-            or self.server_args.enable_metrics_for_all_schedulers
-        )
         self.enable_mfu_metrics = False
 
         if self.enable_metrics:
-            engine_type = DisaggregationMode.to_engine_type(
-                self.server_args.disaggregation_mode
-            )
-
-            labels = {
-                "model_name": self.server_args.served_model_name,
-                "engine_type": engine_type,
-                "tp_rank": tp_rank,
-                "pp_rank": pp_rank,
-                "moe_ep_rank": self.ps.moe_ep_rank,
-            }
-            if self.enable_priority_scheduling:
-                labels["priority"] = ""
-            if dp_rank is not None:
-                labels["dp_rank"] = dp_rank
-            if self.server_args.extra_metric_labels:
-                labels.update(self.server_args.extra_metric_labels)
-            self.metrics_collector = SchedulerMetricsCollector(
-                labels=labels,
-                enable_lora=self.enable_lora,
-                enable_hierarchical_cache=self.enable_hierarchical_cache,
-                enable_streaming_session=self.server_args.enable_streaming_session,
-                server_args=self.server_args,
-            )
-            self.enable_mfu_metrics = self.server_args.enable_mfu_metrics
+            self.enable_mfu_metrics = self.scheduler.server_args.enable_mfu_metrics
             if self.enable_mfu_metrics:
-                self._init_estimated_perf_constants()
+                SchedulerMetricsMixin._init_estimated_perf_constants(self)
                 self._mfu_log_flops = 0.0
                 self._mfu_log_read_bytes = 0.0
                 self._mfu_log_write_bytes = 0.0
@@ -156,54 +128,62 @@ class SchedulerMetricsMixin:
                 reporter=_wrap_execution_reporter,
             )
 
-        self._init_fpm()
+        SchedulerMetricsMixin._init_fpm(self)
 
         self.scheduler_status_logger = SchedulerStatusLogger.maybe_create(
             enable_metrics=self.enable_metrics
         )
 
-    def install_device_timer_on_runners(self: Scheduler):
+    @staticmethod
+    def _install_device_timer_on_runners(self: "SchedulerMetricsReporter"):
         if self.forward_pass_device_timer is None:
             return
         timer = self.forward_pass_device_timer
-        self.tp_worker.model_runner.device_timer = timer
-        if self.draft_worker is not None:
-            dw = getattr(self.draft_worker, "draft_worker", None)
+        self.scheduler.tp_worker.model_runner.device_timer = timer
+        if self.scheduler.draft_worker is not None:
+            dw = getattr(self.scheduler.draft_worker, "draft_worker", None)
             if dw is not None:
                 if hasattr(dw, "draft_runner"):
                     dw.draft_runner.device_timer = timer
                 for r in getattr(dw, "draft_runner_list", []):
                     r.device_timer = timer
 
-    def _init_fpm(self: Scheduler):
+    @staticmethod
+    def _init_fpm(self: "SchedulerMetricsReporter"):
         """Initialize Forward Pass Metrics (FPM) publisher if configured."""
-        self.enable_fpm = False
+        self.scheduler.enable_fpm = False
         if (
-            self.server_args.enable_forward_pass_metrics
-            and self.ps.attn_tp_rank == 0
-            and self.ps.pp_rank == self.ps.pp_size - 1
+            self.scheduler.server_args.enable_forward_pass_metrics
+            and self.scheduler.ps.attn_tp_rank == 0
+            and self.scheduler.ps.pp_rank == self.scheduler.ps.pp_size - 1
         ):
             from sglang.srt.observability.forward_pass_metrics import (
                 _FpmPublisherThread,
             )
 
-            self._fpm_dp_rank = self.ps.dp_rank if self.ps.dp_rank is not None else 0
-            self._fpm_worker_id = self.server_args.forward_pass_metrics_worker_id
-            base_endpoint = self.server_args.forward_pass_metrics_ipc_name
+            self.scheduler._fpm_dp_rank = (
+                self.scheduler.ps.dp_rank
+                if self.scheduler.ps.dp_rank is not None
+                else 0
+            )
+            self.scheduler._fpm_worker_id = (
+                self.scheduler.server_args.forward_pass_metrics_worker_id
+            )
+            base_endpoint = self.scheduler.server_args.forward_pass_metrics_ipc_name
             if base_endpoint is None:
                 ipc_path = tempfile.NamedTemporaryFile(delete=False).name
                 base_endpoint = f"ipc://{ipc_path}"
-                self.server_args.forward_pass_metrics_ipc_name = base_endpoint
-            endpoint = f"{base_endpoint}.{self._fpm_dp_rank}"
-            self._fpm_publisher = _FpmPublisherThread(
+                self.scheduler.server_args.forward_pass_metrics_ipc_name = base_endpoint
+            endpoint = f"{base_endpoint}.{self.scheduler._fpm_dp_rank}"
+            self.scheduler._fpm_publisher = _FpmPublisherThread(
                 endpoint,
-                worker_id=self._fpm_worker_id,
-                dp_rank=self._fpm_dp_rank,
+                worker_id=self.scheduler._fpm_worker_id,
+                dp_rank=self.scheduler._fpm_dp_rank,
             )
-            self._fpm_gpu_time_acc = 0.0
+            self.scheduler._fpm_gpu_time_acc = 0.0
 
             def _fpm_device_timer_reporter(t, **_kwargs):
-                self._fpm_gpu_time_acc += t
+                self.scheduler._fpm_gpu_time_acc += t
 
             if self.forward_pass_device_timer is not None:
                 self.forward_pass_device_timer.add_reporter(_fpm_device_timer_reporter)
@@ -211,16 +191,19 @@ class SchedulerMetricsMixin:
                 self.forward_pass_device_timer = DeviceTimer(
                     reporter=_fpm_device_timer_reporter,
                 )
-            self._fpm_uses_device_timer = True
-            self.enable_fpm = True
+            self.scheduler._fpm_uses_device_timer = True
+            self.scheduler.enable_fpm = True
             logger.info(
                 "FPM: ZMQ PUB bound on %s (dp_rank=%d, device_timer=%s)",
                 endpoint,
-                self._fpm_dp_rank,
-                self._fpm_uses_device_timer,
+                self.scheduler._fpm_dp_rank,
+                self.scheduler._fpm_uses_device_timer,
             )
 
-    def _build_scheduled_request_metrics(self: Scheduler, batch: ScheduleBatch):
+    @staticmethod
+    def _build_scheduled_request_metrics(
+        self: "SchedulerMetricsReporter", batch: ScheduleBatch
+    ):
         from sglang.srt.observability.forward_pass_metrics import (
             ScheduledRequestMetrics,
             WelfordAccumulator,
@@ -265,7 +248,8 @@ class SchedulerMetricsMixin:
             var_decode_kv_tokens=decode_kv.variance(),
         )
 
-    def _build_queued_request_metrics(self: Scheduler):
+    @staticmethod
+    def _build_queued_request_metrics(self: "SchedulerMetricsReporter"):
         from sglang.srt.observability.forward_pass_metrics import (
             QueuedRequestMetrics,
             WelfordAccumulator,
@@ -273,16 +257,16 @@ class SchedulerMetricsMixin:
 
         prefill_q = WelfordAccumulator()
         decode_q = WelfordAccumulator()
-        if self.disaggregation_mode == DisaggregationMode.PREFILL:
-            for req in self.disagg_prefill_bootstrap_queue.queue:
+        if self.scheduler.disaggregation_mode == DisaggregationMode.PREFILL:
+            for req in self.scheduler.disagg_prefill_bootstrap_queue.queue:
                 prefill_q.add(len(req.origin_input_ids))
-        elif self.disaggregation_mode == DisaggregationMode.DECODE:
-            for req in self.disagg_decode_prealloc_queue.queue:
+        elif self.scheduler.disaggregation_mode == DisaggregationMode.DECODE:
+            for req in self.scheduler.disagg_decode_prealloc_queue.queue:
                 decode_q.add(req.seqlen)
-            for req in self.disagg_decode_transfer_queue.queue:
+            for req in self.scheduler.disagg_decode_transfer_queue.queue:
                 decode_q.add(req.seqlen)
         else:
-            for req in self.waiting_queue:
+            for req in self.scheduler.waiting_queue:
                 if len(req.output_ids) > 0:
                     decode_q.add(req.seqlen)
                 else:
@@ -297,22 +281,28 @@ class SchedulerMetricsMixin:
             var_decode_kv_tokens=decode_q.variance(),
         )
 
-    def update_spec_metrics(self: Scheduler, bs: int, num_correct_drafts: int):
+    @staticmethod
+    def update_spec_metrics(
+        self: "SchedulerMetricsReporter", bs: int, num_correct_drafts: int
+    ):
         self.spec_num_accept_tokens += num_correct_drafts + bs
         self.spec_num_forward_ct += bs
 
         # Bonus tokens updated elsewhere
         self.num_generated_tokens += num_correct_drafts
 
-    def _init_estimated_perf_constants(self: Scheduler) -> None:
-        model_config = self.model_config
+    @staticmethod
+    def _init_estimated_perf_constants(self: "SchedulerMetricsReporter") -> None:
+        model_config = self.scheduler.model_config
         hf_text_config = model_config.hf_text_config
 
         hidden_size = float(model_config.hidden_size)
         num_layers = float(getattr(model_config, "num_attention_layers", 0))
         head_dim = float(getattr(model_config, "head_dim", 0))
-        num_attn_heads = float(model_config.get_num_attention_heads(self.ps.tp_size))
-        num_kv_heads = float(model_config.get_num_kv_heads(self.ps.tp_size))
+        num_attn_heads = float(
+            model_config.get_num_attention_heads(self.scheduler.ps.tp_size)
+        )
+        num_kv_heads = float(model_config.get_num_kv_heads(self.scheduler.ps.tp_size))
         intermediate_size = getattr(hf_text_config, "intermediate_size", None)
         if intermediate_size is None:
             intermediate_size = getattr(hf_text_config, "ffn_hidden_size", 0)
@@ -386,8 +376,9 @@ class SchedulerMetricsMixin:
             num_attn_heads * head_dim * act_bytes * num_layers
         )
 
+    @staticmethod
     def _estimate_prefill_perf(
-        self: Scheduler, num_tokens: int
+        self: "SchedulerMetricsReporter", num_tokens: int
     ) -> Tuple[float, float, float]:
         tokens = max(0, int(num_tokens))
         if tokens == 0:
@@ -412,8 +403,9 @@ class SchedulerMetricsMixin:
         )
         return flops, read_bytes, write_bytes
 
+    @staticmethod
     def _estimate_decode_perf(
-        self: Scheduler, batch: ScheduleBatch, num_tokens: int
+        self: "SchedulerMetricsReporter", batch: ScheduleBatch, num_tokens: int
     ) -> Tuple[float, float, float]:
         tokens = max(0, int(num_tokens))
         if tokens == 0:
@@ -437,7 +429,8 @@ class SchedulerMetricsMixin:
         )
         return flops, read_bytes, write_bytes
 
-    def reset_metrics(self: Scheduler):
+    @staticmethod
+    def reset_metrics(self: "SchedulerMetricsReporter"):
         self.forward_ct_decode = 0
         self.num_generated_tokens = 0
         self.spec_num_accept_tokens = 0
@@ -445,8 +438,9 @@ class SchedulerMetricsMixin:
         self.spec_total_num_accept_tokens = 0
         self.spec_total_num_forward_ct = 0
 
+    @staticmethod
     def report_prefill_stats(
-        self: Scheduler,
+        self: "SchedulerMetricsReporter",
         batch: Optional[ScheduleBatch],
         prefill_stats: PrefillStats,
         can_run_cuda_graph: bool,
@@ -465,14 +459,14 @@ class SchedulerMetricsMixin:
             prefill_stats.log_input_tokens / gap_latency if gap_latency > 0 else 0.0
         )
 
-        pool_stats = self.pool_stats_observer.get_pool_stats()
+        pool_stats = self.scheduler.pool_stats_observer.get_pool_stats()
         token_usage_msg = ", ".join(pool_stats.get_prefill_usage_msg_parts()) + ", "
 
         self.stats.new_token_ratio = prefill_stats.new_token_ratio
         batch_iter = (
             batch.forward_iter
             if batch is not None and batch.forward_iter is not None
-            else self.forward_ct
+            else self.scheduler.forward_ct
         )
         iter_msg = f" [{batch_iter}]" if LOG_FORWARD_ITERS else ""
 
@@ -483,25 +477,32 @@ class SchedulerMetricsMixin:
             f"#cached-token: {prefill_stats.log_hit_tokens}, "
             f"{token_usage_msg}"
             f"#running-req: {prefill_stats.num_running_reqs.total}, "
-            f"#queue-req: {len(self.waiting_queue)}, "
+            f"#queue-req: {len(self.scheduler.waiting_queue)}, "
             f"#pending-token: {prefill_stats.num_pending_tokens}, "
         )
 
-        if self.disaggregation_mode == DisaggregationMode.PREFILL:
-            msg += f"#bootstrap-req: {len(self.disagg_prefill_bootstrap_queue.queue)}, "
-            msg += f"#inflight-req: {len(self.disagg_prefill_inflight_queue)}, "
+        if self.scheduler.disaggregation_mode == DisaggregationMode.PREFILL:
+            msg += f"#bootstrap-req: {len(self.scheduler.disagg_prefill_bootstrap_queue.queue)}, "
+            msg += (
+                f"#inflight-req: {len(self.scheduler.disagg_prefill_inflight_queue)}, "
+            )
 
         if (
-            self.server_args.language_only
-            and self.server_args.encoder_transfer_backend == "zmq_to_scheduler"
+            self.scheduler.server_args.language_only
+            and self.scheduler.server_args.encoder_transfer_backend
+            == "zmq_to_scheduler"
         ):
-            msg += f"waiting-image-req: {len(self.mm_receiver.waiting_list)}, "
+            msg += (
+                f"waiting-image-req: {len(self.scheduler.mm_receiver.waiting_list)}, "
+            )
 
         msg += f"{self._graph_backend_label}: {can_run_cuda_graph}, "
         msg += f"input throughput (token/s): {self.last_input_throughput:.2f}"
 
         if self.enable_mfu_metrics and gap_latency > 0:
-            flops, _, _ = self._estimate_prefill_perf(prefill_stats.log_input_tokens)
+            flops, _, _ = SchedulerMetricsMixin._estimate_prefill_perf(
+                self, prefill_stats.log_input_tokens
+            )
             tflops_per_s = flops / gap_latency / 1e12
             msg += f", est. prefill TFLOPS/s (per GPU): {tflops_per_s:.2f}"
 
@@ -520,8 +521,10 @@ class SchedulerMetricsMixin:
                 dp_cooperation_info=dp_cooperation_info,
             )
             if self.enable_mfu_metrics:
-                flops, read_bytes, write_bytes = self._estimate_prefill_perf(
-                    prefill_stats.log_input_tokens
+                flops, read_bytes, write_bytes = (
+                    SchedulerMetricsMixin._estimate_prefill_perf(
+                        self, prefill_stats.log_input_tokens
+                    )
                 )
                 self.metrics_collector.increment_estimated_perf(
                     num_flops_per_gpu=flops,
@@ -529,7 +532,7 @@ class SchedulerMetricsMixin:
                     num_write_bytes_per_gpu=write_bytes,
                 )
 
-            priority_enabled = self.enable_priority_scheduling
+            priority_enabled = self.scheduler.enable_priority_scheduling
             total_tokens = prefill_stats.log_input_tokens + prefill_stats.log_hit_tokens
             cache_hit_rate = (
                 prefill_stats.log_hit_tokens / total_tokens if total_tokens > 0 else 0.0
@@ -538,9 +541,9 @@ class SchedulerMetricsMixin:
             # Basics
             self.stats.num_running_reqs = prefill_stats.num_running_reqs
             self.stats.num_queue_reqs = QueueCount.from_reqs(
-                self.waiting_queue, priority_enabled
+                self.scheduler.waiting_queue, priority_enabled
             )
-            self.stats.num_grammar_queue_reqs = len(self.grammar_manager)
+            self.stats.num_grammar_queue_reqs = len(self.scheduler.grammar_manager)
             self.stats.cache_hit_rate = cache_hit_rate
 
             # Memory pool usage ratios / Absolute token counts
@@ -552,39 +555,41 @@ class SchedulerMetricsMixin:
             self.num_retracted_reqs = self.num_paused_reqs = 0
 
             # PD disaggregation
-            if self.disaggregation_mode == DisaggregationMode.PREFILL:
+            if self.scheduler.disaggregation_mode == DisaggregationMode.PREFILL:
                 self.stats.num_prefill_bootstrap_queue_reqs = QueueCount.from_reqs(
-                    self.disagg_prefill_bootstrap_queue.queue, priority_enabled
+                    self.scheduler.disagg_prefill_bootstrap_queue.queue,
+                    priority_enabled,
                 )
                 self.stats.num_prefill_inflight_queue_reqs = QueueCount.from_reqs(
-                    self.disagg_prefill_inflight_queue, priority_enabled
+                    self.scheduler.disagg_prefill_inflight_queue, priority_enabled
                 )
                 self.stats.kv_transfer_speed_gb_s = self.kv_transfer_speed_gb_s
                 self.stats.kv_transfer_latency_ms = self.kv_transfer_latency_ms
-            elif self.disaggregation_mode == DisaggregationMode.DECODE:
+            elif self.scheduler.disaggregation_mode == DisaggregationMode.DECODE:
                 self.stats.num_decode_prealloc_queue_reqs = QueueCount.from_reqs(
-                    self.disagg_decode_prealloc_queue.queue, priority_enabled
+                    self.scheduler.disagg_decode_prealloc_queue.queue, priority_enabled
                 )
                 self.stats.num_decode_transfer_queue_reqs = QueueCount.from_reqs(
-                    self.disagg_decode_transfer_queue.queue, priority_enabled
+                    self.scheduler.disagg_decode_transfer_queue.queue, priority_enabled
                 )
 
             # Utilization / LoRA / HiCache
-            self.calculate_utilization()
+            SchedulerMetricsMixin._calculate_utilization(self)
             self.stats.fwd_occupancy = self.fwd_occupancy
-            self.update_lora_metrics()
-            self._log_hicache_stats()
+            SchedulerMetricsMixin._update_lora_metrics(self)
+            SchedulerMetricsMixin._log_hicache_stats(self)
             self.metrics_collector.log_stats(self.stats)
-            self.kv_events_publisher.emit_kv_metrics()
-        self.kv_events_publisher.publish_kv_events()
+            self.scheduler.kv_events_publisher.emit_kv_metrics()
+        self.scheduler.kv_events_publisher.publish_kv_events()
 
+    @staticmethod
     def report_decode_stats(
-        self: Scheduler,
+        self: "SchedulerMetricsReporter",
         can_run_cuda_graph: bool,
         running_batch: ScheduleBatch = None,
         num_correct_drafts: int = 0,
     ):
-        batch = running_batch or self.running_batch
+        batch = running_batch or self.scheduler.running_batch
 
         # Every-iteration work: realtime token counting + status logger
         if self.current_scheduler_metrics_enabled:
@@ -595,8 +600,10 @@ class SchedulerMetricsMixin:
                 dp_cooperation_info=batch.dp_cooperation_info,
             )
             if self.enable_mfu_metrics:
-                flops, read_bytes, write_bytes = self._estimate_decode_perf(
-                    batch, decode_tokens
+                flops, read_bytes, write_bytes = (
+                    SchedulerMetricsMixin._estimate_decode_perf(
+                        self, batch, decode_tokens
+                    )
                 )
                 self.metrics_collector.increment_estimated_perf(
                     num_flops_per_gpu=flops,
@@ -608,10 +615,10 @@ class SchedulerMetricsMixin:
                 self._mfu_log_write_bytes += write_bytes
 
             if x := self.scheduler_status_logger:
-                x.maybe_dump(batch, self.waiting_queue)
+                x.maybe_dump(batch, self.scheduler.waiting_queue)
 
         # Periodic work: log + heavy metrics at decode_log_interval
-        if self.forward_ct_decode % self.server_args.decode_log_interval != 0:
+        if self.forward_ct_decode % self.scheduler.server_args.decode_log_interval != 0:
             return
         if (
             not self.is_stats_logging_rank
@@ -626,32 +633,34 @@ class SchedulerMetricsMixin:
         self.num_generated_tokens = 0
         num_running_reqs = len(batch.reqs)
 
-        pool_stats = self.pool_stats_observer.get_pool_stats()
+        pool_stats = self.scheduler.pool_stats_observer.get_pool_stats()
         token_usage_msg = ", ".join(pool_stats.get_decode_usage_msg_parts()) + ", "
 
         if RECORD_STEP_TIME:
             self.step_time_dict[num_running_reqs].append(
-                gap_latency / self.server_args.decode_log_interval
+                gap_latency / self.scheduler.server_args.decode_log_interval
             )
 
         batch_iter = (
             batch.forward_iter
             if batch is not None and batch.forward_iter is not None
-            else self.forward_ct
+            else self.scheduler.forward_ct
         )
         iter_msg = f" [{batch_iter}]" if LOG_FORWARD_ITERS else ""
         msg = f"Decode batch{iter_msg}, #running-req: {num_running_reqs}, {token_usage_msg}"
 
-        if self.spec_algorithm.is_none():
+        if self.scheduler.spec_algorithm.is_none():
             spec_accept_length = 0
             spec_accept_rate = 0
         else:
             spec_accept_length = self.spec_num_accept_tokens / self.spec_num_forward_ct
             num_correct_drafts = self.spec_num_accept_tokens - self.spec_num_forward_ct
-            if self.server_args.speculative_num_draft_tokens:
-                draft_per_round = self.server_args.speculative_num_draft_tokens - 1
+            if self.scheduler.server_args.speculative_num_draft_tokens:
+                draft_per_round = (
+                    self.scheduler.server_args.speculative_num_draft_tokens - 1
+                )
             else:
-                draft_per_round = self.server_args.speculative_num_steps or 0
+                draft_per_round = self.scheduler.server_args.speculative_num_steps or 0
             total_draft_tokens = self.spec_num_forward_ct * draft_per_round
             spec_accept_rate = (
                 num_correct_drafts / total_draft_tokens if total_draft_tokens > 0 else 0
@@ -662,22 +671,25 @@ class SchedulerMetricsMixin:
             msg += f"accept len: {spec_accept_length:.2f}, accept rate: {spec_accept_rate:.2f}, "
         cache_hit_rate = 0.0
 
-        if self.disaggregation_mode == DisaggregationMode.DECODE:
-            msg += f"pre-allocated usage: {self.disagg_decode_prealloc_queue.num_tokens_pre_allocated / self.max_total_num_tokens:.2f}, "
-            msg += f"#prealloc-req: {len(self.disagg_decode_prealloc_queue.queue)}, "
-            msg += f"#transfer-req: {len(self.disagg_decode_transfer_queue.queue)}, "
-            msg += f"#retracted-req: {len(self.disagg_decode_prealloc_queue.retracted_queue)}, "
+        if self.scheduler.disaggregation_mode == DisaggregationMode.DECODE:
+            msg += f"pre-allocated usage: {self.scheduler.disagg_decode_prealloc_queue.num_tokens_pre_allocated / self.scheduler.max_total_num_tokens:.2f}, "
+            msg += f"#prealloc-req: {len(self.scheduler.disagg_decode_prealloc_queue.queue)}, "
+            msg += f"#transfer-req: {len(self.scheduler.disagg_decode_transfer_queue.queue)}, "
+            msg += f"#retracted-req: {len(self.scheduler.disagg_decode_prealloc_queue.retracted_queue)}, "
 
         if (
-            self.server_args.language_only
-            and self.server_args.encoder_transfer_backend == "zmq_to_scheduler"
+            self.scheduler.server_args.language_only
+            and self.scheduler.server_args.encoder_transfer_backend
+            == "zmq_to_scheduler"
         ):
-            msg += f"waiting-image-req: {len(self.mm_receiver.waiting_list)}, "
+            msg += (
+                f"waiting-image-req: {len(self.scheduler.mm_receiver.waiting_list)}, "
+            )
 
         msg += (
             f"{self._graph_backend_label}: {can_run_cuda_graph}, "
             f"gen throughput (token/s): {self.last_gen_throughput:.2f}, "
-            f"#queue-req: {len(self.waiting_queue)}"
+            f"#queue-req: {len(self.scheduler.waiting_queue)}"
         )
 
         if self.enable_mfu_metrics and gap_latency > 0:
@@ -702,16 +714,16 @@ class SchedulerMetricsMixin:
         if self.is_stats_logging_rank:
             logger.info(msg)
         if self.current_scheduler_metrics_enabled:
-            priority_enabled = self.enable_priority_scheduling
+            priority_enabled = self.scheduler.enable_priority_scheduling
 
             # Basics
             self.stats.num_running_reqs = QueueCount.from_reqs(
                 batch.reqs, priority_enabled
             )
             self.stats.num_queue_reqs = QueueCount.from_reqs(
-                self.waiting_queue, priority_enabled
+                self.scheduler.waiting_queue, priority_enabled
             )
-            self.stats.num_grammar_queue_reqs = len(self.grammar_manager)
+            self.stats.num_grammar_queue_reqs = len(self.scheduler.grammar_manager)
             self.stats.gen_throughput = self.last_gen_throughput
             self.stats.cache_hit_rate = cache_hit_rate
             self.stats.decode_sum_seq_lens = batch.seq_lens_cpu.sum().item()
@@ -729,34 +741,37 @@ class SchedulerMetricsMixin:
             self.num_retracted_reqs = self.num_paused_reqs = 0
 
             # PD disaggregation
-            if self.disaggregation_mode == DisaggregationMode.PREFILL:
+            if self.scheduler.disaggregation_mode == DisaggregationMode.PREFILL:
                 self.stats.num_prefill_bootstrap_queue_reqs = QueueCount.from_reqs(
-                    self.disagg_prefill_bootstrap_queue.queue, priority_enabled
+                    self.scheduler.disagg_prefill_bootstrap_queue.queue,
+                    priority_enabled,
                 )
                 self.stats.num_prefill_inflight_queue_reqs = QueueCount.from_reqs(
-                    self.disagg_prefill_inflight_queue, priority_enabled
+                    self.scheduler.disagg_prefill_inflight_queue, priority_enabled
                 )
-            elif self.disaggregation_mode == DisaggregationMode.DECODE:
+            elif self.scheduler.disaggregation_mode == DisaggregationMode.DECODE:
                 self.stats.num_decode_prealloc_queue_reqs = QueueCount.from_reqs(
-                    self.disagg_decode_prealloc_queue.queue, priority_enabled
+                    self.scheduler.disagg_decode_prealloc_queue.queue, priority_enabled
                 )
                 self.stats.num_decode_transfer_queue_reqs = QueueCount.from_reqs(
-                    self.disagg_decode_transfer_queue.queue, priority_enabled
+                    self.scheduler.disagg_decode_transfer_queue.queue, priority_enabled
                 )
 
             # Streaming session metrics
             self.stats.num_streaming_sessions = (
-                self.pool_stats_observer.streaming_session_count()
+                self.scheduler.pool_stats_observer.streaming_session_count()
             )
             self.stats.streaming_session_held_tokens = (
-                self.pool_stats_observer.session_held_tokens()
+                self.scheduler.pool_stats_observer.session_held_tokens()
             )
 
             # Routing key metrics
             # (to reduce the overhead, we only compute this when all requests have routing_key)
             if all(r.routing_key is not None for r in batch.reqs):
                 running_routing_keys = [r.routing_key for r in batch.reqs]
-                waiting_routing_keys = [r.routing_key for r in self.waiting_queue]
+                waiting_routing_keys = [
+                    r.routing_key for r in self.scheduler.waiting_queue
+                ]
                 (
                     self.stats.num_unique_running_routing_keys,
                     self.stats.routing_key_running_req_counts,
@@ -766,16 +781,17 @@ class SchedulerMetricsMixin:
                 )
 
             # Utilization / LoRA / HiCache
-            self.calculate_utilization()
+            SchedulerMetricsMixin._calculate_utilization(self)
             self.stats.fwd_occupancy = self.fwd_occupancy
-            self.update_lora_metrics()
-            self._log_hicache_stats()
+            SchedulerMetricsMixin._update_lora_metrics(self)
+            SchedulerMetricsMixin._log_hicache_stats(self)
             self.metrics_collector.log_stats(self.stats)
-            self.kv_events_publisher.emit_kv_metrics()
-        self.kv_events_publisher.publish_kv_events()
+            self.scheduler.kv_events_publisher.emit_kv_metrics()
+        self.scheduler.kv_events_publisher.publish_kv_events()
 
+    @staticmethod
     def log_batch_result_stats(
-        self: Scheduler,
+        self: "SchedulerMetricsReporter",
         batch: ScheduleBatch,
         result: Union[GenerationBatchResult, EmbeddingBatchResult],
     ):
@@ -790,8 +806,9 @@ class SchedulerMetricsMixin:
                 balancedness=m.eplb_balancedness.item(),
             )
 
+    @staticmethod
     def _emit_forward_pass_metrics(
-        self: Scheduler,
+        self: "SchedulerMetricsReporter",
         batch: ScheduleBatch,
         result=None,
     ):
@@ -801,61 +818,66 @@ class SchedulerMetricsMixin:
         model_runner.forward / cuda_graph.replay via PR #24197).
         Falls back to monotonic clock when DeviceTimer is not enabled.
         """
-        if not self.enable_fpm:
+        if not self.scheduler.enable_fpm:
             return
 
         from sglang.srt.observability.forward_pass_metrics import (
             ForwardPassMetrics,
         )
 
-        if self._fpm_uses_device_timer:
+        if self.scheduler._fpm_uses_device_timer:
             self.forward_pass_device_timer._report()
-            wall_time = self._fpm_gpu_time_acc
-            self._fpm_gpu_time_acc = 0.0
+            wall_time = self.scheduler._fpm_gpu_time_acc
+            self.scheduler._fpm_gpu_time_acc = 0.0
             if wall_time == 0.0:
                 return
         else:
             wall_time = max(0.0, time.monotonic() - batch.fpm_start_time)
 
         fpm = ForwardPassMetrics(
-            worker_id=self._fpm_worker_id,
-            dp_rank=self._fpm_dp_rank,
+            worker_id=self.scheduler._fpm_worker_id,
+            dp_rank=self.scheduler._fpm_dp_rank,
             wall_time=wall_time,
-            scheduled_requests=self._build_scheduled_request_metrics(batch),
-            queued_requests=self._build_queued_request_metrics(),
+            scheduled_requests=SchedulerMetricsMixin._build_scheduled_request_metrics(
+                self, batch
+            ),
+            queued_requests=SchedulerMetricsMixin._build_queued_request_metrics(self),
         )
-        self._fpm_publisher.publish(fpm)
+        self.scheduler._fpm_publisher.publish(fpm)
 
-    def _shutdown_fpm(self: Scheduler):
+    @staticmethod
+    def _shutdown_fpm(self: "SchedulerMetricsReporter"):
         """Shut down the FPM publisher thread."""
-        if self.enable_fpm:
-            self._fpm_publisher.shutdown()
+        if self.scheduler.enable_fpm:
+            self.scheduler._fpm_publisher.shutdown()
 
-    def _log_hicache_stats(self: Scheduler):
+    @staticmethod
+    def _log_hicache_stats(self: "SchedulerMetricsReporter"):
         """Populate HiCache host-tier stats on self.stats.
 
         These are pushed to Prometheus by SchedulerMetricsCollector.log_stats().
         """
-        if not self.enable_hierarchical_cache:
+        if not self.scheduler.enable_hierarchical_cache:
             return
 
-        host_pool = getattr(self.tree_cache, "token_to_kv_pool_host", None) or getattr(
-            self.tree_cache, "full_kv_pool_host", None
-        )
+        host_pool = getattr(
+            self.scheduler.tree_cache, "token_to_kv_pool_host", None
+        ) or getattr(self.scheduler.tree_cache, "full_kv_pool_host", None)
         assert host_pool is not None, "Host pool not found"
         self.stats.hicache_host_used_tokens = (
             host_pool.size - host_pool.available_size()
         )
         self.stats.hicache_host_total_tokens = host_pool.size
 
-    def update_lora_metrics(self: Scheduler):
+    @staticmethod
+    def _update_lora_metrics(self: "SchedulerMetricsReporter"):
         """Update LoRA pool metrics for monitoring and autoscaling."""
-        if not self.enable_lora:
+        if not self.scheduler.enable_lora:
             return
 
         try:
             # Get LoRA memory pool stats
-            lora_manager = self.tp_worker.model_runner.lora_manager
+            lora_manager = self.scheduler.tp_worker.model_runner.lora_manager
             if lora_manager is None or lora_manager.memory_pool is None:
                 return
 
@@ -867,16 +889,16 @@ class SchedulerMetricsMixin:
             active_lora_ids = set()
 
             # For PP mode, check all running micro batches
-            if self.server_args.pp_size > 1:
-                for batch in self.running_mbs:
+            if self.scheduler.server_args.pp_size > 1:
+                for batch in self.scheduler.running_mbs:
                     if batch and hasattr(batch, "reqs"):
                         for req in batch.reqs:
                             if hasattr(req, "lora_id") and req.lora_id is not None:
                                 active_lora_ids.add(req.lora_id)
             # For normal mode, check running_batch
-            elif self.running_batch:
-                if hasattr(self.running_batch, "reqs"):
-                    for req in self.running_batch.reqs:
+            elif self.scheduler.running_batch:
+                if hasattr(self.scheduler.running_batch, "reqs"):
+                    for req in self.scheduler.running_batch.reqs:
                         if hasattr(req, "lora_id") and req.lora_id is not None:
                             active_lora_ids.add(req.lora_id)
 
@@ -892,19 +914,23 @@ class SchedulerMetricsMixin:
         except Exception as e:
             logger.warning(f"Failed to update LoRA metrics: {e}")
 
-    def calculate_utilization(self: Scheduler):
-        if self.disaggregation_mode == DisaggregationMode.PREFILL:
+    @staticmethod
+    def _calculate_utilization(self: "SchedulerMetricsReporter"):
+        if self.scheduler.disaggregation_mode == DisaggregationMode.PREFILL:
             self.stats.utilization = -1
         else:
             # TODO: max_running_requests_under_SLO has no setter — sglang:utilization stuck at 0 (regressed #22713).
-            max_under_slo = getattr(self, "max_running_requests_under_SLO", None)
+            max_under_slo = getattr(
+                self.scheduler, "max_running_requests_under_SLO", None
+            )
             if max_under_slo is not None and max_under_slo > 0:
                 self.stats.utilization = max(
                     self.stats.num_running_reqs.total / max_under_slo,
                     self.stats.token_usage / 0.9,
                 )
 
-    def update_device_timer(self: Scheduler):
+    @staticmethod
+    def update_device_timer(self: "SchedulerMetricsReporter"):
         if not ENABLE_METRICS_DEVICE_TIMER:
             return
         self.forward_pass_device_timer._report()
@@ -924,11 +950,12 @@ class SchedulerMetricsMixin:
         self._device_timer_window_batch_count += 1
         if (
             self._device_timer_window_batch_count
-            >= self.server_args.decode_log_interval
+            >= self.scheduler.server_args.decode_log_interval
         ):
             self._device_timer_window_batch_count = 0
 
-    def reset_device_timer_window(self: Scheduler):
+    @staticmethod
+    def reset_device_timer_window(self: "SchedulerMetricsReporter"):
         if ENABLE_METRICS_DEVICE_TIMER:
             self._device_timer_window_batch_count = 0
             self.fwd_occupancy = float("nan")
diff --git a/test/registered/unit/managers/test_scheduler_chunked_req_gate.py b/test/registered/unit/managers/test_scheduler_chunked_req_gate.py
index 4ec25c836..d4a92ba82 100644
--- a/test/registered/unit/managers/test_scheduler_chunked_req_gate.py
+++ b/test/registered/unit/managers/test_scheduler_chunked_req_gate.py
@@ -77,6 +77,7 @@ def _scheduler_for_get_next_batch(*, tree_cache, chunked_req) -> Scheduler:
     s.dllm_config = None
     s.dllm_manager = None
     s.enable_hisparse = False
+    s.enable_fpm = False
     s.last_batch = None
     s.require_mlp_sync = False
     s.spec_algorithm = MagicMock()