From 9622895258419d8c0ed698b13d33f28925f15175 Mon Sep 17 00:00:00 2001 From: Pernekhan Utemuratov Date: Tue, 1 Sep 2026 21:16:28 +0000 Subject: [PATCH 1/2] frontend: attribute offered load to the selected worker namespace A frontend that serves several worker namespaces for one model -- the `--` shape namespace grouping produces -- labels all of its own metrics with its own `dynamo_namespace`, which is the model stem. Every group's planner therefore reads the same model-wide `requests_started_total` and sees the whole model's demand instead of its own, so each one scales for traffic that another group is actually serving. Add `dynamo_frontend_worker_namespace_requests_started_total{model, worker_namespace}`, incremented where the WorkerSet is chosen. That point is before any queue or scheduler work, so it measures offered load: a request later rejected with 529 still counts as demand, which is what a scaling decision needs. Selection is refactored to hand back the chosen WorkerSet alongside the extracted engine so there is exactly one increment site covering every entry path, including the single-set fast path. The router's own `requests_started_total` is not an alternative here: `RouterRequestMetrics::from_component` is a process-global OnceLock, so it cannot carry per-group series, and it increments only after admission. Planner side, `get_avg_request_count` prefers the new counter filtered on its own namespace and falls back to the model-wide one, so a planner running against an older frontend is unaffected. The histogram namespace test becomes a stem-prefix match rather than equality: TTFT/ITL series only exist model-wide, and under equality a grouped planner would read zero for both. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_017rS1uFFKT2xV4x63Amfvtw --- .../planner/monitoring/traffic_metrics.py | 49 ++++++++++++++- .../python/src/dynamo/prometheus_names.py | 4 ++ lib/llm/src/discovery/model.rs | 62 ++++++++++++++++--- lib/llm/src/http/service/metrics.rs | 37 +++++++++++ lib/runtime/src/metrics/prometheus_names.rs | 4 ++ 5 files changed, 147 insertions(+), 9 deletions(-) diff --git a/components/src/dynamo/planner/monitoring/traffic_metrics.py b/components/src/dynamo/planner/monitoring/traffic_metrics.py index 7d6ec2fcef28..3fa5d87db9ee 100644 --- a/components/src/dynamo/planner/monitoring/traffic_metrics.py +++ b/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,35 @@ 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 (````) while each planner owns a group (``--``). + Those model-wide series carry the only TTFT/ITL data available, so accept + a stem prefix rather than requiring equality. Ungrouped deployments have + stem == group and still match. + """ + return self.dynamo_namespace.startswith(metric_namespace or "") + + 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 +176,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 +249,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 +390,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 ) diff --git a/lib/bindings/python/src/dynamo/prometheus_names.py b/lib/bindings/python/src/dynamo/prometheus_names.py index 3d684641835f..86340b42a9d1 100644 --- a/lib/bindings/python/src/dynamo/prometheus_names.py +++ b/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, ...) diff --git a/lib/llm/src/discovery/model.rs b/lib/llm/src/discovery/model.rs index a0565dc52f46..869b51446931 100644 --- a/lib/llm/src/discovery/model.rs +++ b/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(&self, extract: F) -> Option + where + F: Fn(&WorkerSet) -> Option, + { + 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(&self, extract: F) -> Option<(T, Arc)> where F: Fn(&WorkerSet) -> Option, { @@ -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)> = 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 diff --git a/lib/llm/src/http/service/metrics.rs b/lib/llm/src/http/service/metrics.rs index c9de3ff80ab5..7b6f1214fe6a 100644 --- a/lib/llm/src/http/service/metrics.rs +++ b/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 +/// (`--`), 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 = 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 = LazyLock::new(|| { @@ -123,6 +159,7 @@ pub static WORKER_LAST_INTER_TOKEN_LATENCY_GAUGE: LazyLock = 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()))?; diff --git a/lib/runtime/src/metrics/prometheus_names.rs b/lib/runtime/src/metrics/prometheus_names.rs index 42148314deb0..456d1586b998 100644 --- a/lib/runtime/src/metrics/prometheus_names.rs +++ b/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"; From ba2d2edb29df80f57bf3703fc004b28d80c85ed2 Mon Sep 17 00:00:00 2001 From: Pernekhan Utemuratov Date: Tue, 1 Sep 2026 21:35:22 +0000 Subject: [PATCH 2/2] planner: require a namespace segment boundary, reject an empty label The stem match shipped in the previous commit accepted any prefix and treated an absent label as a wildcard. Both are too loose. Prod carries frontend series with no `dynamo_namespace` at all -- on the rig, `deepseek-ai/DeepSeek-V3.2-dynamo` has six such series alongside its labelled ones. A wildcard would sum those into every planner's reading for that model and inflate its offered load. Equality rejected them before, so rejecting an empty label keeps the prior behaviour rather than changing it. Requiring a `--` boundary likewise stops `-extra` from matching ``, which is a different model rather than a group of one. Add the truth table as a unit test, using the rig's real namespace strings, and cover both branches of the new offered-load query: the per-group counter when it has series, and the fall back to the model-wide counter when it does not. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_017rS1uFFKT2xV4x63Amfvtw --- .../planner/monitoring/traffic_metrics.py | 15 +++- .../planner/tests/unit/test_prometheus.py | 81 +++++++++++++++++-- 2 files changed, 87 insertions(+), 9 deletions(-) diff --git a/components/src/dynamo/planner/monitoring/traffic_metrics.py b/components/src/dynamo/planner/monitoring/traffic_metrics.py index 3fa5d87db9ee..f59ad6826cef 100644 --- a/components/src/dynamo/planner/monitoring/traffic_metrics.py +++ b/components/src/dynamo/planner/monitoring/traffic_metrics.py @@ -141,10 +141,19 @@ def _namespace_matches(self, metric_namespace: Optional[str]) -> bool: A frontend serving grouped workers labels its own metrics with the model stem (````) while each planner owns a group (``--``). Those model-wide series carry the only TTFT/ITL data available, so accept - a stem prefix rather than requiring equality. Ungrouped deployments have - stem == group and still match. + 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. """ - return self.dynamo_namespace.startswith(metric_namespace or "") + 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.""" diff --git a/components/src/dynamo/planner/tests/unit/test_prometheus.py b/components/src/dynamo/planner/tests/unit/test_prometheus.py index dbee872ceb47..f0c0c773d600 100644 --- a/components/src/dynamo/planner/tests/unit/test_prometheus.py +++ b/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():