diff --git a/components/src/dynamo/planner/monitoring/traffic_metrics.py b/components/src/dynamo/planner/monitoring/traffic_metrics.py index 7d6ec2fcef28..f59ad6826cef 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,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 (````) while each planner owns a 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 ) 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(): 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";