feat: add router-level Prometheus metrics and centralize request tracking

lib/bindings/python/rust/llm/kv.rs

@@ -947,6 +947,7 @@ impl KvPushRouter {
             tokio::spawn(async move {
                 let mut stream = stream;
                 let mut first_item = true;
+                let mut first_token_gauges_observed = false;
 
                 while let Some(mut response) = stream.next().await {
                     // Inject worker_id into first response if tracker is available
@@ -957,6 +958,21 @@ impl KvPushRouter {
                         }
                     }
 
+                    // Observe per-worker TTFT/ISL gauges on first response with actual tokens
+                    if !first_token_gauges_observed {
+                        let has_tokens = response
+                            .data
+                            .as_ref()
+                            .map(|d| !d.token_ids.is_empty())
+                            .unwrap_or(false);
+                        if has_tokens {
+                            if let Some(ref tracker) = tracker {
+                                tracker.observe_first_token_gauges();
+                            }
+                            first_token_gauges_observed = true;
+                        }
+                    }
+
                     // Convert LLMEngineOutput to PyObject
                     let py_response = Python::with_gil(|py| {
                         pythonize(py, &response.data)
@@ -976,6 +992,11 @@ impl KvPushRouter {
                         }
                     }
                 }
+
+                // Observe per-worker ITL gauge at stream end
+                if let Some(ref tracker) = tracker {
+                    tracker.observe_finish_gauges();
+                }
             });
 
             // Return a Python async generator wrapper

lib/llm/src/discovery.rs

@@ -12,7 +12,5 @@ pub use watcher::{ModelUpdate, ModelWatcher};
 
 mod worker_monitor;
 pub use worker_monitor::{
-    KvWorkerMonitor, LoadThresholdConfig, WORKER_ACTIVE_DECODE_BLOCKS_GAUGE,
-    WORKER_ACTIVE_PREFILL_TOKENS_GAUGE, WORKER_TYPE_DECODE, WORKER_TYPE_PREFILL, WorkerLoadState,
-    register_worker_load_metrics,
+    KvWorkerMonitor, LoadThresholdConfig, WORKER_TYPE_DECODE, WORKER_TYPE_PREFILL, WorkerLoadState,
 };

lib/llm/src/discovery/worker_monitor.rs

@@ -3,88 +3,43 @@
 
 use std::collections::HashMap;
 use std::sync::Arc;
+use std::sync::RwLock;
 use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
-use std::sync::{LazyLock, RwLock};
 
 use tokio::sync::Notify;
 
 use dashmap::DashMap;
-use prometheus::{IntGaugeVec, Opts, Registry};
 use serde::{Deserialize, Serialize};
 
 use crate::http::service::metrics::{
     WORKER_LAST_INPUT_SEQUENCE_TOKENS_GAUGE, WORKER_LAST_INTER_TOKEN_LATENCY_GAUGE,
     WORKER_LAST_TIME_TO_FIRST_TOKEN_GAUGE,
 };
 use crate::kv_router::KV_METRICS_SUBJECT;
+use crate::kv_router::metrics::WORKER_LOAD_METRICS;
 use crate::kv_router::protocols::ActiveLoad;
 use crate::model_card::ModelDeploymentCard;
 use dynamo_runtime::component::Client;
 use dynamo_runtime::discovery::{DiscoveryQuery, watch_and_extract_field};
-use dynamo_runtime::metrics::prometheus_names::frontend_service;
 use dynamo_runtime::pipeline::{WorkerLoadMonitor, async_trait};
 use dynamo_runtime::traits::DistributedRuntimeProvider;
 use dynamo_runtime::transports::event_plane::EventSubscriber;
 
 // Re-export worker type constants from timing.rs (single source of truth)
 pub use crate::protocols::common::timing::{WORKER_TYPE_DECODE, WORKER_TYPE_PREFILL};
 
-/// Global Prometheus gauge for active decode blocks per worker (labels: worker_id, dp_rank, worker_type)
-/// This is shared across all KvWorkerMonitor instances.
-pub static WORKER_ACTIVE_DECODE_BLOCKS_GAUGE: LazyLock<IntGaugeVec> = LazyLock::new(|| {
-    IntGaugeVec::new(
-        Opts::new(
-            format!(
-                "dynamo_frontend_{}",
-                frontend_service::WORKER_ACTIVE_DECODE_BLOCKS
-            ),
-            "Active KV cache decode blocks per worker",
-        ),
-        &["worker_id", "dp_rank", "worker_type"],
-    )
-    .expect("Failed to create worker_active_decode_blocks gauge")
-});
-
-/// Global Prometheus gauge for active prefill tokens per worker (labels: worker_id, dp_rank, worker_type)
-/// This is shared across all KvWorkerMonitor instances.
-pub static WORKER_ACTIVE_PREFILL_TOKENS_GAUGE: LazyLock<IntGaugeVec> = LazyLock::new(|| {
-    IntGaugeVec::new(
-        Opts::new(
-            format!(
-                "dynamo_frontend_{}",
-                frontend_service::WORKER_ACTIVE_PREFILL_TOKENS
-            ),
-            "Active prefill tokens queued per worker",
-        ),
-        &["worker_id", "dp_rank", "worker_type"],
-    )
-    .expect("Failed to create worker_active_prefill_tokens gauge")
-});
-
-/// Register the global worker load Prometheus metrics with the given registry.
-///
-/// This should be called once during HTTP service setup to expose the worker load
-/// metrics via the `/metrics` endpoint.
-///
-/// # Errors
-/// Returns an error if the metrics are already registered with the registry.
-pub fn register_worker_load_metrics(registry: &Registry) -> Result<(), prometheus::Error> {
-    registry.register(Box::new(WORKER_ACTIVE_DECODE_BLOCKS_GAUGE.clone()))?;
-    registry.register(Box::new(WORKER_ACTIVE_PREFILL_TOKENS_GAUGE.clone()))?;
-    Ok(())
-}
-
 /// Clean up all Prometheus metrics for a worker across the specified dp_ranks.
 ///
 /// This removes metrics with the given worker_id, dp_rank, and worker_type label combination.
 /// Called when workers are removed to prevent stale metrics from accumulating.
 fn cleanup_worker_metrics(worker_id: u64, dp_ranks: &[u32], worker_type: &str) {
     let worker_id_str = worker_id.to_string();
+    let m = &*WORKER_LOAD_METRICS;
     for dp_rank in dp_ranks {
         let dp_rank_str = dp_rank.to_string();
         let labels = &[worker_id_str.as_str(), dp_rank_str.as_str(), worker_type];
-        let _ = WORKER_ACTIVE_DECODE_BLOCKS_GAUGE.remove_label_values(labels);
-        let _ = WORKER_ACTIVE_PREFILL_TOKENS_GAUGE.remove_label_values(labels);
+        let _ = m.active_decode_blocks.remove_label_values(labels);
+        let _ = m.active_prefill_tokens.remove_label_values(labels);
         let _ = WORKER_LAST_TIME_TO_FIRST_TOKEN_GAUGE.remove_label_values(labels);
         let _ = WORKER_LAST_INPUT_SEQUENCE_TOKENS_GAUGE.remove_label_values(labels);
         let _ = WORKER_LAST_INTER_TOKEN_LATENCY_GAUGE.remove_label_values(labels);
@@ -215,9 +170,9 @@ impl WorkerLoadState {
 /// Cloning shares state via internal Arc-wrapped fields. This allows multiple pipelines
 /// (e.g., chat and completions) to share the same monitor instance.
 ///
-/// Prometheus metrics are exposed via the global gauges [`WORKER_ACTIVE_DECODE_BLOCKS_GAUGE`]
-/// and [`WORKER_ACTIVE_PREFILL_TOKENS_GAUGE`], which should be registered with the HTTP
-/// service's Prometheus registry using [`register_worker_load_metrics`].
+/// Prometheus metrics are exposed via [`WORKER_LOAD_METRICS`] (defined in `kv_router::sequence`),
+/// which should be registered with the HTTP service's Prometheus registry using
+/// [`register_worker_load_metrics`](crate::kv_router::metrics::register_worker_load_metrics).
 ///
 /// In disaggregated mode, use `set_prefill_client` to register the prefill endpoint for
 /// proper TTFT metric cleanup when prefill workers are removed.
@@ -251,8 +206,9 @@ impl KvWorkerMonitor {
     /// - `active_prefill_tokens_threshold`: DEFAULT_MAX_TOKENS (effectively disabled)
     /// - `active_prefill_tokens_threshold_frac`: 1.5 (effectively disabled)
     ///
-    /// Prometheus metrics are exposed via the global gauges and should be registered
-    /// using [`register_worker_load_metrics`] during HTTP service setup.
+    /// Prometheus metrics are exposed via [`WORKER_LOAD_METRICS`] and should be registered
+    /// using [`register_worker_load_metrics`](crate::kv_router::metrics::register_worker_load_metrics)
+    /// during HTTP service setup.
     ///
     /// For disaggregated mode, call `set_prefill_client` after creation to enable
     /// proper TTFT metric cleanup when prefill workers are removed.

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

@@ -111,7 +111,7 @@ pub fn register_worker_timing_metrics(registry: &Registry) -> Result<(), prometh
 /// # Note
 /// With 2 significant figures, there are roughly 90 unique values per order of magnitude.
 /// Requesting more buckets than can be uniquely represented will result in deduplication.
-fn generate_log_buckets(min: f64, max: f64, count: usize) -> Vec<f64> {
+pub fn generate_log_buckets(min: f64, max: f64, count: usize) -> Vec<f64> {
     if count == 0 {
         return vec![];
     }
@@ -153,7 +153,7 @@ fn generate_log_buckets(min: f64, max: f64, count: usize) -> Vec<f64> {
 }
 
 /// Round a number to a specified number of significant figures
-fn round_to_sig_figs(value: f64, sig_figs: u32) -> f64 {
+pub fn round_to_sig_figs(value: f64, sig_figs: u32) -> f64 {
     if value == 0.0 {
         return 0.0;
     }

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

@@ -16,8 +16,9 @@ use super::Metrics;
 use super::RouteDoc;
 use super::metrics;
 use super::metrics::register_worker_timing_metrics;
-use crate::discovery::{ModelManager, register_worker_load_metrics};
+use crate::discovery::ModelManager;
 use crate::endpoint_type::EndpointType;
+use crate::kv_router::metrics::{register_routing_overhead_metrics, register_worker_load_metrics};
 use crate::request_template::RequestTemplate;
 use anyhow::Result;
 use axum_server::tls_rustls::RustlsConfig;
@@ -391,6 +392,12 @@ impl HttpServiceConfigBuilder {
             tracing::warn!("Failed to register worker timing metrics: {}", e);
         }
 
+        // Register routing overhead metrics (block hashing, find matches, scheduling latencies)
+        // These are updated by KvRouter::find_best_match on every routing decision
+        if let Err(e) = register_routing_overhead_metrics(&registry) {
+            tracing::warn!("Failed to register routing overhead metrics: {}", e);
+        }
+
         let mut router = axum::Router::new();
 
         let mut all_docs = Vec::new();

lib/llm/src/kv_router.rs

@@ -2,9 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use std::collections::HashMap;
-use std::time::Duration;
-#[cfg(feature = "bench")]
-use std::time::Instant;
+use std::time::{Duration, Instant};
 
 use anyhow::Result;
 use dynamo_runtime::{
@@ -27,6 +25,7 @@ pub use dynamo_kv_router::indexer;
 pub use dynamo_kv_router::protocols;
 
 pub mod config;
+pub mod metrics;
 pub mod prefill_router;
 pub mod publisher;
 pub mod push_router;
@@ -63,7 +62,6 @@ pub const KV_METRICS_ENDPOINT: &str = "load_metrics";
 
 // for metric publishing (push-based)
 pub const KV_EVENT_SUBJECT: &str = "kv-events";
-pub const KV_HIT_RATE_SUBJECT: &str = "kv-hit-rate";
 pub const KV_METRICS_SUBJECT: &str = "kv_metrics";
 
 // for inter-router comms
@@ -316,7 +314,6 @@ impl KvRouter {
         update_states: bool,
         lora_name: Option<String>,
     ) -> anyhow::Result<(WorkerWithDpRank, u32)> {
-        #[cfg(feature = "bench")]
         let start = Instant::now();
 
         if update_states && context_id.is_none() {
@@ -326,16 +323,16 @@ impl KvRouter {
         let isl_tokens = tokens.len();
 
         let block_hashes = compute_block_hash_for_seq(tokens, self.block_size, None);
-        #[cfg(feature = "bench")]
         let hash_elapsed = start.elapsed();
+
         let overlap_scores = self.indexer.find_matches(block_hashes).await?;
-        #[cfg(feature = "bench")]
         let find_matches_elapsed = start.elapsed();
 
         // Compute seq_hashes only if scheduler needs it for active blocks tracking
         let maybe_seq_hashes = self
             .kv_router_config
             .compute_seq_hashes_for_tracking(tokens, self.block_size);
+        let seq_hash_elapsed = start.elapsed();
 
         let best_worker = self
             .scheduler
@@ -349,19 +346,25 @@ impl KvRouter {
                 lora_name,
             )
             .await?;
+        let total_elapsed = start.elapsed();
+
+        metrics::ROUTING_OVERHEAD_METRICS.observe(
+            hash_elapsed,
+            find_matches_elapsed,
+            seq_hash_elapsed,
+            total_elapsed,
+        );
 
         #[cfg(feature = "bench")]
-        {
-            let total_elapsed = start.elapsed();
-            tracing::info!(
-                isl_tokens,
-                hash_us = hash_elapsed.as_micros() as u64,
-                find_matches_us = (find_matches_elapsed - hash_elapsed).as_micros() as u64,
-                schedule_us = (total_elapsed - find_matches_elapsed).as_micros() as u64,
-                total_us = total_elapsed.as_micros() as u64,
-                "find_best_match completed"
-            );
-        }
+        tracing::info!(
+            isl_tokens,
+            hash_us = hash_elapsed.as_micros() as u64,
+            find_matches_us = (find_matches_elapsed - hash_elapsed).as_micros() as u64,
+            seq_hash_us = (seq_hash_elapsed - find_matches_elapsed).as_micros() as u64,
+            schedule_us = (total_elapsed - seq_hash_elapsed).as_micros() as u64,
+            total_us = total_elapsed.as_micros() as u64,
+            "find_best_match completed"
+        );
 
         // Note: Routing decision recording (for approximate mode) is now handled
         // by KvPushRouter::generate after select_worker returns.