frontend: attribute offered load to the selected worker namespace

components/src/dynamo/planner/monitoring/traffic_metrics.py

@@ -92,6 +92,7 @@ def is_valid(self) -> bool:
 class FrontendMetric(BaseModel):
     container: typing.Optional[str] = None
     dynamo_namespace: typing.Optional[str] = None
+    worker_namespace: typing.Optional[str] = None
     endpoint: typing.Optional[str] = None
     instance: typing.Optional[str] = None
     job: typing.Optional[str] = None
@@ -134,6 +135,44 @@ def _frontend_metric_name(self, metric_name: str) -> str:
             return metric_name
         return f"{prometheus_names.name_prefix.FRONTEND}_{metric_name}"
 
+    def _namespace_matches(self, metric_namespace: Optional[str]) -> bool:
+        """Does a frontend series belong to this planner's namespace?
+
+        A frontend serving grouped workers labels its own metrics with the model
+        stem (``<stem>``) while each planner owns a group (``<stem>--<group>``).
+        Those model-wide series carry the only TTFT/ITL data available, so accept
+        a stem that this planner's namespace extends at a ``--`` boundary.
+        Ungrouped deployments have stem == group and match by equality.
+
+        An absent or empty label is rejected rather than treated as a wildcard.
+        Frontends whose series predate the dynamo_namespace relabel carry no
+        namespace at all, and folding those into one planner's reading inflates
+        it; equality rejected them before, so this keeps that behaviour.
+        """
+        if not metric_namespace:
+            return False
+        if metric_namespace == self.dynamo_namespace:
+            return True
+        return self.dynamo_namespace.startswith(f"{metric_namespace}--")
+
+    def _sum_worker_namespace_metric(self, result, model_name: str) -> Optional[float]:
+        """Sum the selection-time counter for exactly this planner's group."""
+        if not result:
+            return None
+
+        total = 0.0
+        matched = False
+        for container in parse_frontend_metric_containers(result):
+            if (
+                container.metric.model
+                and container.metric.model.lower() == model_name.lower()
+                and container.metric.worker_namespace == self.dynamo_namespace
+                and not math.isnan(container.value[1])
+            ):
+                matched = True
+                total += container.value[1]
+        return total if matched else None
+
     def _sum_frontend_metric(self, result, model_name: str) -> Optional[float]:
         if not result:
             return None
@@ -146,7 +185,7 @@ def _sum_frontend_metric(self, result, model_name: str) -> Optional[float]:
             if (
                 container.metric.model
                 and container.metric.model.lower() == model_name.lower()
-                and container.metric.dynamo_namespace == self.dynamo_namespace
+                and self._namespace_matches(container.metric.dynamo_namespace)
                 and not math.isnan(container.value[1])
             ):
                 matched = True
@@ -219,7 +258,7 @@ def _get_average_metric(
                     if (
                         container.metric.model
                         and container.metric.model.lower() == model_name.lower()
-                        and container.metric.dynamo_namespace == self.dynamo_namespace
+                        and self._namespace_matches(container.metric.dynamo_namespace)
                     ):
                         values.append(container.value[1])
                 if not values:
@@ -360,6 +399,21 @@ def get_avg_request_count(self, interval: str, model_name: str):
         # use frontend-started requests so throughput planning sees offered load,
         # not only completed responses.
         try:
+            # Offered load must be attributed to this planner's group. The
+            # model-wide requests_started_total counts every group's traffic, so
+            # each planner would read the whole model's demand and over-scale.
+            worker_ns_metric = self._frontend_metric_name(
+                prometheus_names.frontend_service.WORKER_NAMESPACE_REQUESTS_STARTED_TOTAL
+            )
+            worker_ns_res = self.prom.custom_query(
+                query=f"increase({worker_ns_metric}[{interval}])"
+            )
+            worker_ns_count = self._sum_worker_namespace_metric(worker_ns_res, model_name)
+            if worker_ns_count is not None:
+                return worker_ns_count
+
+            # Frontend predates the per-namespace counter: fall back to the
+            # model-wide one, correct whenever the model has a single group.
             requests_started_metric = self._frontend_metric_name(
                 prometheus_names.frontend_service.REQUESTS_STARTED_TOTAL
             )

components/src/dynamo/planner/tests/unit/test_prometheus.py

@@ -227,6 +227,69 @@ def test_get_average_metric_none_result():
         assert result == 0
 
 
+STEM = "deepseek-ai--DeepSeek-V4-Flash-0731-roce-disagg"
+
+
+@pytest.mark.parametrize(
+    "planner_ns,series_ns,expected",
+    [
+        # A grouped planner must see the model-wide series: TTFT/ITL histograms
+        # only ever carry the frontend's own namespace, which is the stem.
+        (f"{STEM}--g1", STEM, True),
+        (f"{STEM}--g1", f"{STEM}--g1", True),
+        # ...but never a sibling group's.
+        (f"{STEM}--g1", f"{STEM}--g2", False),
+        # An absent or empty label carries no attribution and is not a wildcard.
+        (f"{STEM}--g1", "", False),
+        (f"{STEM}--g1", None, False),
+        # Ungrouped deployments keep the old equality behaviour exactly.
+        (STEM, STEM, True),
+        (STEM, f"{STEM}--g1", False),
+        # A shared prefix that is not a "--" segment boundary is a different model.
+        ("aa--Foo-extra", "aa--Foo", False),
+    ],
+)
+def test_namespace_matches(planner_ns, series_ns, expected):
+    """Frontend series attribution for grouped and ungrouped planners."""
+    client = PrometheusAPIClient("http://localhost:9090", planner_ns)
+    assert client._namespace_matches(series_ns) is expected
+
+
+def test_get_avg_request_count_prefers_worker_namespace_counter():
+    """The per-group counter wins over the model-wide one when it has series."""
+    client = PrometheusAPIClient("http://localhost:9090", f"{STEM}--g1")
+
+    def fake_query(query):
+        if "worker_namespace_requests_started_total" in query:
+            return [
+                {
+                    "metric": {"model": "m", "worker_namespace": f"{STEM}--g1"},
+                    "value": [0, "7"],
+                },
+                {
+                    "metric": {"model": "m", "worker_namespace": f"{STEM}--g2"},
+                    "value": [0, "99"],
+                },
+            ]
+        raise AssertionError("must not fall back when the counter has series")
+
+    with patch.object(client.prom, "custom_query", side_effect=fake_query):
+        assert client.get_avg_request_count("60s", "m") == 7
+
+
+def test_get_avg_request_count_falls_back_without_worker_namespace_counter():
+    """An older frontend has no such series, so the model-wide counter is used."""
+    client = PrometheusAPIClient("http://localhost:9090", STEM)
+
+    def fake_query(query):
+        if "worker_namespace_requests_started_total" in query:
+            return []
+        return [{"metric": {"model": "m", "dynamo_namespace": STEM}, "value": [0, "5"]}]
+
+    with patch.object(client.prom, "custom_query", side_effect=fake_query):
+        assert client.get_avg_request_count("60s", "m") == 5
+
+
 def test_get_average_metric_empty_result():
     """Test _get_average_metric when prometheus returns empty list"""
     client = PrometheusAPIClient("http://localhost:9090", "test_namespace")
@@ -362,9 +425,14 @@ def test_get_avg_request_count_uses_started_requests():
 
     assert result == 150.0
     queries = [call.kwargs["query"] for call in mock_query.call_args_list]
-    assert "dynamo_frontend_requests_started_total" in queries[0]
-    assert "increase(" in queries[0]
-    assert len(queries) == 1
+    # The per-group counter is tried first; these series carry no
+    # worker_namespace label, so it yields nothing and the model-wide
+    # started counter answers.
+    assert "worker_namespace_requests_started_total" in queries[0]
+    assert "dynamo_frontend_requests_started_total" in queries[1]
+    assert "increase(" in queries[1]
+    # Still no query for the completed counter -- started requests answered.
+    assert len(queries) == 2
 
 
 def test_get_avg_request_count_falls_back_to_completed_when_started_missing():
@@ -382,14 +450,15 @@ def test_get_avg_request_count_falls_back_to_completed_when_started_missing():
     ]
 
     with patch.object(client.prom, "custom_query") as mock_query:
-        mock_query.side_effect = [[], completed]
+        mock_query.side_effect = [[], [], completed]
 
         result = client.get_avg_request_count("30s", "target_model")
 
     assert result == 73.0
     queries = [call.kwargs["query"] for call in mock_query.call_args_list]
-    assert "dynamo_frontend_requests_started_total" in queries[0]
-    assert "dynamo_frontend_requests_total" in queries[1]
+    assert "worker_namespace_requests_started_total" in queries[0]
+    assert "dynamo_frontend_requests_started_total" in queries[1]
+    assert "dynamo_frontend_requests_total" in queries[2]
 
 
 def test_vllm_spec_decode_accept_length_query_derives_from_counters():

lib/bindings/python/src/dynamo/prometheus_names.py

@@ -78,6 +78,10 @@ class frontend_service:
     REQUESTS_TOTAL = "requests_total"
     # Total number of LLM requests accepted by the frontend handler
     REQUESTS_STARTED_TOTAL = "requests_started_total"
+    # Offered load attributed to the worker namespace chosen at selection time.
+    # Labels: model, worker_namespace. Lets a per-group planner read its own
+    # demand from a frontend that serves several worker namespaces.
+    WORKER_NAMESPACE_REQUESTS_STARTED_TOTAL = "worker_namespace_requests_started_total"
     # Number of requests waiting in HTTP queue before receiving the first response (gauge)
     QUEUED_REQUESTS = "queued_requests"
     # Number of inflight/concurrent requests going to the engine (vLLM, SGLang, ...)

lib/llm/src/discovery/model.rs

@@ -15,6 +15,7 @@ use serde::Serialize;
 use super::ModelManagerError;
 use super::worker_monitor::LoadThresholdConfig;
 use super::worker_set::WorkerSet;
+use crate::http::service::metrics;
 use crate::protocols::openai::ParsingOptions;
 
 use crate::types::{
@@ -829,6 +830,19 @@ impl Model {
     /// desired engine, or `None` if it doesn't.
     ///
     fn select_worker_set_with<T, F>(&self, extract: F) -> Option<T>
+    where
+        F: Fn(&WorkerSet) -> Option<T>,
+    {
+        let (value, worker_set) = self.select_worker_set_inner(extract)?;
+        metrics::inc_worker_namespace_requests_started(&self.name, worker_set.namespace());
+        Some(value)
+    }
+
+    /// Selection proper. Returns the chosen WorkerSet alongside the extracted
+    /// value so the caller can attribute the request to the namespace it landed
+    /// on -- with namespace grouping the frontend's own namespace label is the
+    /// model stem and cannot identify the group.
+    fn select_worker_set_inner<T, F>(&self, extract: F) -> Option<(T, Arc<WorkerSet>)>
     where
         F: Fn(&WorkerSet) -> Option<T>,
     {
@@ -864,21 +878,21 @@ impl Model {
             if ws.worker_count() == 0 || !ready_namespaces.contains(ws.namespace()) {
                 return None;
             }
-            return extract(ws);
+            return extract(ws).map(|val| (val, ws.clone()));
         }
 
         // Collect eligible sets with their worker counts, skipping sets with no workers or sets in
         // a namespace whose worker set is incomplete.
         // In-process models (no discovery watcher) return count=1, so they always participate.
         // Discovery models with count=0 have no available workers and are skipped.
-        let eligible: Vec<(T, usize)> = snapshot
+        let eligible: Vec<(T, usize, Arc<WorkerSet>)> = snapshot
             .iter()
             .filter_map(|ws| {
                 let count = ws.worker_count();
                 if count == 0 || !ready_namespaces.contains(ws.namespace()) {
                     return None;
                 }
-                extract(ws).map(|val| (val, count))
+                extract(ws).map(|val| (val, count, ws.clone()))
             })
             .collect();
 
@@ -887,15 +901,15 @@ impl Model {
         }
 
         if eligible.len() == 1 {
-            return eligible.into_iter().next().map(|(val, _)| val);
+            return eligible.into_iter().next().map(|(val, _, ws)| (val, ws));
         }
 
         // Weighted random selection proportional to worker count
-        let total_weight: usize = eligible.iter().map(|(_, w)| w).sum();
+        let total_weight: usize = eligible.iter().map(|(_, w, _)| w).sum();
         let mut pick = rand::rng().random_range(0..total_weight);
-        for (val, weight) in eligible {
+        for (val, weight, ws) in eligible {
             if pick < weight {
-                return Some(val);
+                return Some((val, ws));
             }
             pick -= weight;
         }
@@ -1111,6 +1125,40 @@ mod tests {
         assert!(model.get_chat_engine().is_err()); // Still no engines → all filtered out
     }
 
+    #[test]
+    fn test_selection_counts_offered_load_per_worker_namespace() {
+        // Model name is the counter's label, so a name unique to this test keeps
+        // the process-global counter isolated from other tests.
+        let model = Model::new("offered-load-model".to_string());
+        let counted = |ns: &str| {
+            crate::http::service::metrics::WORKER_NAMESPACE_REQUESTS_STARTED
+                .with_label_values(&["offered-load-model", ns])
+                .get()
+        };
+
+        // Nothing selected -> nothing counted.
+        assert!(model.select_worker_set_with(|_| Some(())).is_none());
+        assert_eq!(counted("ns1"), 0);
+
+        // Fast path (single set) must count, not just the weighted path.
+        let (ws1, _tx1) = make_worker_set_with_count("ns1", "abc", vec![1]);
+        model.add_worker_set("ns1".to_string(), ws1);
+        assert!(model.select_worker_set_with(|_| Some(())).is_some());
+        assert_eq!(counted("ns1"), 1);
+
+        // Weighted path: whichever group wins the draw is the one credited.
+        let (ws2, _tx2) = make_worker_set_with_count("ns2", "abc", vec![2]);
+        model.add_worker_set("ns2".to_string(), ws2);
+        for _ in 0..20 {
+            assert!(model.select_worker_set_with(|_| Some(())).is_some());
+        }
+        assert_eq!(counted("ns1") + counted("ns2"), 21);
+
+        // A set that yields no engine is not a selection and must not be counted.
+        assert!(model.select_worker_set_with(|_| None::<()>).is_none());
+        assert_eq!(counted("ns1") + counted("ns2"), 21);
+    }
+
     #[test]
     fn test_total_workers_no_watcher() {
         // In-process WorkerSets (no watcher) default to worker_count=1

lib/llm/src/http/service/metrics.rs

@@ -63,6 +63,42 @@ pub use crate::discovery::{WORKER_TYPE_DECODE, WORKER_TYPE_PREFILL};
 const UNSET_DP_RANK_LABEL: &str = "none";
 const ITL_LOCAL_FLUSH_TOKENS: u64 = 64;
 
+/// Offered load per worker namespace, counted at the moment a WorkerSet is
+/// selected -- before any queue or scheduler work, so a request later rejected
+/// with 529 still counts as demand.
+///
+/// With namespace grouping one frontend serves several worker namespaces
+/// (`<stem>--<group>`), while its own `dynamo_namespace` label is the model
+/// stem. Per-group planners therefore cannot attribute demand from the existing
+/// `requests_started_total`. This counter carries the selected group directly.
+///
+/// Ungrouped models emit the same series shape with their single namespace, so
+/// old and new frontends look alike to the planner.
+///
+/// Labels: model, worker_namespace
+pub static WORKER_NAMESPACE_REQUESTS_STARTED: LazyLock<IntCounterVec> = LazyLock::new(|| {
+    IntCounterVec::new(
+        Opts::new(
+            format!(
+                "{}_{}",
+                name_prefix::FRONTEND,
+                frontend_service::WORKER_NAMESPACE_REQUESTS_STARTED_TOTAL
+            ),
+            "Requests attributed to the selected worker namespace at selection time",
+        ),
+        &["model", "worker_namespace"],
+    )
+    .expect("failed to create worker_namespace_requests_started_total")
+});
+
+/// Increment the per-namespace offered-load counter. Called from the WorkerSet
+/// selection path, which every serving entry point passes through exactly once.
+pub fn inc_worker_namespace_requests_started(model: &str, worker_namespace: &str) {
+    WORKER_NAMESPACE_REQUESTS_STARTED
+        .with_label_values(&[model, worker_namespace])
+        .inc();
+}
+
 /// Global Prometheus gauge for last observed TTFT per worker (in seconds)
 /// Labels: worker_id, dp_rank, worker_type
 pub static WORKER_LAST_TIME_TO_FIRST_TOKEN_GAUGE: LazyLock<GaugeVec> = LazyLock::new(|| {
@@ -123,6 +159,7 @@ pub static WORKER_LAST_INTER_TOKEN_LATENCY_GAUGE: LazyLock<GaugeVec> = LazyLock:
 /// # Errors
 /// Returns an error if the metrics are already registered with the registry.
 pub fn register_worker_timing_metrics(registry: &Registry) -> Result<(), prometheus::Error> {
+    registry.register(Box::new(WORKER_NAMESPACE_REQUESTS_STARTED.clone()))?;
     registry.register(Box::new(WORKER_LAST_TIME_TO_FIRST_TOKEN_GAUGE.clone()))?;
     registry.register(Box::new(WORKER_LAST_INPUT_SEQUENCE_TOKENS_GAUGE.clone()))?;
     registry.register(Box::new(WORKER_LAST_INTER_TOKEN_LATENCY_GAUGE.clone()))?;

lib/runtime/src/metrics/prometheus_names.rs

@@ -177,6 +177,10 @@ pub mod frontend_service {
     /// Total number of LLM requests accepted by the frontend handler
     pub const REQUESTS_STARTED_TOTAL: &str = "requests_started_total";
 
+    /// Offered load per selected worker namespace, counted before queueing
+    pub const WORKER_NAMESPACE_REQUESTS_STARTED_TOTAL: &str =
+        "worker_namespace_requests_started_total";
+
     /// Number of requests waiting in HTTP queue before receiving the first response (gauge)
     pub const QUEUED_REQUESTS: &str = "queued_requests";