Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 34 additions & 10 deletions lib/llm/src/kv_router/metrics.rs
Original file line number Diff line number Diff line change
Expand Up @@ -843,6 +843,7 @@ impl RoutingOverheadMetrics {
/// distinct `dynamo_component` labels, so pools can be monitored and scaled
/// independently.
pub struct RouterRequestMetrics {
pub requests_started_total: prometheus::IntCounter,
Comment thread
jthomson04 marked this conversation as resolved.
Resolved
pub requests_total: prometheus::IntCounter,
pub time_to_first_token_seconds: prometheus::Histogram,
pub inter_token_latency_seconds: prometheus::Histogram,
Expand All @@ -857,7 +858,6 @@ pub struct RouterRequestMetrics {
}

static ROUTER_REQUEST_METRICS: OnceLock<Arc<RouterRequestMetrics>> = OnceLock::new();
static ROUTER_REQUESTS_STARTED_TOTAL: OnceLock<prometheus::IntCounter> = OnceLock::new();

impl RouterRequestMetrics {
/// Returns the registered metrics if `from_component()` was called earlier.
Expand All @@ -867,9 +867,7 @@ impl RouterRequestMetrics {

/// Total requests admitted by the router scheduler.
pub fn requests_started_total(&self) -> &prometheus::IntCounter {
ROUTER_REQUESTS_STARTED_TOTAL
.get()
.expect("router request metrics must be initialized")
&self.requests_started_total
}
Comment thread
jthomson04 marked this conversation as resolved.
Resolved

/// Create from a Component, memoized in a static OnceLock.
Expand All @@ -893,12 +891,6 @@ impl RouterRequestMetrics {
extra_labels,
)
.expect("failed to create router_requests_started_total");
assert!(
ROUTER_REQUESTS_STARTED_TOTAL
.set(requests_started_total)
.is_ok(),
"router_requests_started_total already initialized"
);
let requests_total = metrics
.create_intcounter(
&router_metric(frontend_service::REQUESTS_TOTAL),
Expand Down Expand Up @@ -990,6 +982,7 @@ impl RouterRequestMetrics {
non_max_overlap_selections_total.with_label_values(&[WORKER_TYPE_PREFILL]);
overlap_blocks_lost.with_label_values(&[WORKER_TYPE_PREFILL]);
Arc::new(Self {
requests_started_total,
requests_total,
time_to_first_token_seconds,
inter_token_latency_seconds,
Expand All @@ -1006,6 +999,37 @@ impl RouterRequestMetrics {
.clone()
}

/// Create unregistered metrics owned by one test router.
#[cfg(test)]
pub(crate) fn for_test() -> Arc<Self> {
fn hist(name: &str) -> prometheus::Histogram {
prometheus::Histogram::with_opts(prometheus::HistogramOpts::new(name, name)).unwrap()
}
fn hist_vec(name: &str) -> prometheus::HistogramVec {
prometheus::HistogramVec::new(prometheus::HistogramOpts::new(name, name), &["reason"])
.unwrap()
}
Arc::new(Self {
requests_started_total: prometheus::IntCounter::new("requests_started_total", "test")
.unwrap(),
requests_total: prometheus::IntCounter::new("requests_total", "test").unwrap(),
time_to_first_token_seconds: hist("ttft_seconds"),
inter_token_latency_seconds: hist("itl_seconds"),
input_sequence_tokens: hist("isl_tokens"),
output_sequence_tokens: hist("osl_tokens"),
kv_hit_rate: hist("kv_hit_rate"),
kv_transfer_estimated_latency_seconds: hist("kv_transfer_seconds"),
shared_cache_hit_rate: hist("shared_cache_hit_rate"),
shared_cache_beyond_blocks: hist("shared_cache_beyond_blocks"),
Comment thread
jthomson04 marked this conversation as resolved.
Resolved
non_max_overlap_selections_total: prometheus::IntCounterVec::new(
prometheus::Opts::new("non_max_overlap_selections_total", "test"),
&["reason"],
Comment thread
jthomson04 marked this conversation as resolved.
Resolved
)
.unwrap(),
overlap_blocks_lost: hist_vec("overlap_blocks_lost"),
})
}

/// Record a selection that sacrificed KV cache overlap.
pub fn observe_non_max_overlap_selection(&self, worker_type: &str, overlap_blocks_lost: f64) {
debug_assert!(overlap_blocks_lost > 0.0);
Expand Down
31 changes: 2 additions & 29 deletions lib/llm/src/kv_router/routing_host/request_guard.rs
Original file line number Diff line number Diff line change
Expand Up @@ -990,38 +990,11 @@ mod prefill_start_tests {
.unwrap()
}

fn test_metrics() -> Arc<RouterRequestMetrics> {
fn hist(name: &str) -> prometheus::Histogram {
prometheus::Histogram::with_opts(prometheus::HistogramOpts::new(name, name)).unwrap()
}
fn hist_vec(name: &str) -> prometheus::HistogramVec {
prometheus::HistogramVec::new(prometheus::HistogramOpts::new(name, name), &["reason"])
.unwrap()
}
Arc::new(RouterRequestMetrics {
requests_total: prometheus::IntCounter::new("requests_total", "test").unwrap(),
time_to_first_token_seconds: hist("ttft_seconds"),
inter_token_latency_seconds: hist("itl_seconds"),
input_sequence_tokens: hist("isl_tokens"),
output_sequence_tokens: hist("osl_tokens"),
kv_hit_rate: hist("kv_hit_rate"),
kv_transfer_estimated_latency_seconds: hist("kv_transfer_seconds"),
shared_cache_hit_rate: hist("shared_cache_hit_rate"),
shared_cache_beyond_blocks: hist("shared_cache_beyond_blocks"),
non_max_overlap_selections_total: prometheus::IntCounterVec::new(
prometheus::Opts::new("non_max_overlap_selections_total", "test"),
&["reason"],
)
.unwrap(),
overlap_blocks_lost: hist_vec("overlap_blocks_lost"),
})
}

async fn dispatch_once(phase: RequestPhase, annotations: Vec<String>) -> Arc<RequestTracker> {
let tracker = Arc::new(RequestTracker::new());
let _permit = tracker.set_phase(phase).await;
let request = test_request(tracker.clone(), annotations);
RequestObservability::new(request.tracker.clone(), test_metrics())
RequestObservability::new(request.tracker.clone(), RouterRequestMetrics::for_test())
.record_prefill_start(&request);
tracker
}
Expand Down Expand Up @@ -1062,7 +1035,7 @@ mod prefill_start_tests {
#[tokio::test]
async fn decode_after_prefill_retains_the_prefill_timestamp() {
let tracker = Arc::new(RequestTracker::new());
let metrics = test_metrics();
let metrics = RouterRequestMetrics::for_test();
let request = test_request(tracker.clone(), Vec::new());

let prefill_permit = tracker.set_phase(RequestPhase::Prefill).await;
Expand Down
25 changes: 18 additions & 7 deletions lib/llm/src/kv_router/routing_host/tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1194,7 +1194,9 @@ async fn router_with_worker_configs(
let inner = PushRouter::from_client(client, RouterMode::KV)
.await
.unwrap();
let router = RoutingHost::new(inner, Arc::new(chooser), session_affinity_ttl).unwrap();
let mut router = RoutingHost::new(inner, Arc::new(chooser), session_affinity_ttl).unwrap();
// Parallel tests must not share the process-global request counters.
router.request_metrics = RouterRequestMetrics::for_test();
Comment thread
jthomson04 marked this conversation as resolved.
Resolved
router
.inner
.client
Expand Down Expand Up @@ -1303,7 +1305,9 @@ async fn router_with_recorded_dispatch_and_affinity(
)
.await
.unwrap();
let router = RoutingHost::new(inner, Arc::new(chooser), session_affinity_ttl).unwrap();
let mut router = RoutingHost::new(inner, Arc::new(chooser), session_affinity_ttl).unwrap();
// Parallel tests must not share the process-global request counters.
router.request_metrics = RouterRequestMetrics::for_test();
(router, dispatch, worker_id, runtime)
}

Expand Down Expand Up @@ -1406,7 +1410,6 @@ async fn track_request(
}

#[tokio::test]
#[serial_test::serial]
async fn route_plan_from_preview_holds_and_releases_the_decode_reservation() {
Comment thread
jthomson04 marked this conversation as resolved.
Resolved
let (router, runtime) = router(None).await;
let request = Context::new(request());
Expand Down Expand Up @@ -1461,7 +1464,6 @@ async fn route_plan_from_preview_holds_and_releases_the_decode_reservation() {
}

#[tokio::test]
#[serial_test::serial]
async fn route_preview_does_not_admit_a_request() {
let (router, runtime) = router(None).await;
let request = Context::new(request());
Expand Down Expand Up @@ -1557,7 +1559,6 @@ async fn route_preview_does_not_acquire_session_affinity() {
}

#[tokio::test]
#[serial_test::serial]
async fn planned_dispatch_transfers_the_reservation_to_request_cleanup() {
let (router, runtime) = router(None).await;
let requests_started_before = router.request_metrics.requests_started_total().get();
Expand Down Expand Up @@ -1588,7 +1589,6 @@ async fn planned_dispatch_transfers_the_reservation_to_request_cleanup() {
}

#[tokio::test]
#[serial_test::serial]
async fn prefill_busy_probe_does_not_admit_a_request() {
let (router, runtime) = router(None).await;
let request = Context::new(request());
Expand Down Expand Up @@ -1653,7 +1653,6 @@ async fn session_affinity_disabled_does_not_create_coordinator() {
}

#[tokio::test]
#[serial_test::serial]
async fn router_request_counters_follow_admission_and_completion_lifecycle() {
let (router, runtime) = router(None).await;
let metrics = router.request_metrics.clone();
Expand Down Expand Up @@ -1751,6 +1750,18 @@ async fn router_request_counters_follow_admission_and_completion_lifecycle() {
drop(builtin_guard);
assert_eq!(metrics.requests_total.get(), completed_before + 1);
Comment thread
jthomson04 marked this conversation as resolved.
Resolved

// Another test router must not change either lifecycle counter.
Comment thread
jthomson04 marked this conversation as resolved.
Resolved
let (other_router, other_runtime) = router_with_workers(None, &[8]).await;
Comment thread
jthomson04 marked this conversation as resolved.
Resolved
let (_, _, mut other_guard) = track_request(&other_router, false).await;
other_guard.start_dispatch("aggregated");
other_guard.mark_dispatched();
other_guard.finish().await;
drop(other_guard);
assert_eq!(metrics.requests_started_total().get(), started_before + 4);
assert_eq!(metrics.requests_total.get(), completed_before + 1);
Comment thread
jthomson04 marked this conversation as resolved.
Resolved
drop(other_router);
other_runtime.shutdown();

drop(router);
runtime.shutdown();
}
Expand Down
Loading