[Fix] Include disagg prefill waiting queue in FPM (#32122)

Signed-off-by: hongkuanz <hongkuanz@nvidia.com>

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

@@ -304,6 +304,8 @@ class SchedulerMetricsReporter:
         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))
+            for req in self.scheduler.waiting_queue:
+                prefill_q.add(len(req.origin_input_ids))
         elif self.scheduler.disaggregation_mode == DisaggregationMode.DECODE:
             for req in self.scheduler.disagg_decode_prealloc_queue.queue:
                 decode_q.add(req.seqlen)

test/registered/unit/observability/test_forward_pass_metrics.py

@@ -232,11 +232,12 @@ class TestForwardPassMetrics(unittest.TestCase):
             self.scheduler._fpm_publisher.metrics[0].wall_time, 0.035, places=4
         )
 
-    def test_disagg_prefill_queued_metrics(self):
+    def test_disagg_prefill_queued_metrics_include_compute_waiting_queue(self):
         self.scheduler.disaggregation_mode = DisaggregationMode.PREFILL
         self.scheduler.disagg_prefill_bootstrap_queue = types.SimpleNamespace(
-            queue=[_FakeReq(100), _FakeReq(200), _FakeReq(50)],
+            queue=[_FakeReq(100)],
         )
+        self.scheduler.waiting_queue = [_FakeReq(200), _FakeReq(50)]
         batch = self._make_batch()
 
         with patch(