dynamo_runtime/metrics/
work_handler_pool.rs

1// SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Worker-pool saturation metrics for the shared TCP server (backend side).
5//!
6//! These metrics expose queue buildup between `work_tx.send()` and dispatcher
7//! pickup, and permit starvation in the bounded worker pool.
8
9use once_cell::sync::{Lazy, OnceCell};
10use prometheus::{Histogram, HistogramOpts, IntCounter, IntGauge};
11
12use super::prometheus_names::{name_prefix, work_handler};
13use crate::MetricsRegistry;
14
15fn work_handler_metric_name(suffix: &str) -> String {
16    format!("{}_{}", name_prefix::WORK_HANDLER, suffix)
17}
18
19/// Requests shed because the worker was at capacity: all engine in-flight slots
20/// (`--engine-request-limit`) held AND the overflow queue
21/// (`--dynamo-request-queue-limit`) full.
22pub static REJECTION_REQUEST_TOTAL: Lazy<IntCounter> = Lazy::new(|| {
23    IntCounter::new(
24        format!("{}_request_total", name_prefix::REJECTION),
25        "Requests rejected because the worker is at capacity (engine in-flight limit and Dynamo queue both full)",
26    )
27    .expect("rejection_request_total counter")
28});
29
30/// Requests currently in the engine (worker-pool permits in use). Driven from
31/// `ActiveTaskGuard`, alongside `WORK_HANDLER_POOL_ACTIVE_TASKS`.
32pub static ENGINE_REQUEST_GAUGE: Lazy<IntGauge> = Lazy::new(|| {
33    IntGauge::new(
34        "dynamo_engine_request",
35        "Current number of requests being handled by the engine (--engine-request-limit)",
36    )
37    .expect("dynamo_engine_request gauge")
38});
39
40/// Requests queued in Dynamo but not yet in the engine. Driven alongside
41/// `WORK_HANDLER_QUEUE_DEPTH` (inc on enqueue, dec on dispatcher recv).
42pub static REQUEST_QUEUE_GAUGE: Lazy<IntGauge> = Lazy::new(|| {
43    IntGauge::new(
44        "dynamo_request_queue",
45        "Current number of requests queued in Dynamo not yet in the engine (--dynamo-request-queue-limit)",
46    )
47    .expect("dynamo_request_queue gauge")
48});
49
50/// Current items sitting in the bounded mpsc work queue awaiting dispatcher
51/// pickup. Incremented on successful `work_tx.send()` and decremented immediately
52/// after `work_rx.recv()`. Permit-acquire wait is NOT counted here — see
53/// `WORK_HANDLER_PERMIT_WAIT_SECONDS`.
54pub static WORK_HANDLER_QUEUE_DEPTH: Lazy<IntGauge> = Lazy::new(|| {
55    IntGauge::new(
56        work_handler_metric_name(work_handler::QUEUE_DEPTH),
57        "Current items in the bounded work queue awaiting dispatcher pickup",
58    )
59    .expect("work_handler_queue_depth gauge")
60});
61
62/// Configured capacity of the bounded work queue. Static; set once at server init.
63pub static WORK_HANDLER_QUEUE_CAPACITY: Lazy<IntGauge> = Lazy::new(|| {
64    IntGauge::new(
65        work_handler_metric_name(work_handler::QUEUE_CAPACITY),
66        "Configured capacity of the bounded work queue",
67    )
68    .expect("work_handler_queue_capacity gauge")
69});
70
71/// Times enqueuing work failed because the dispatcher channel was closed
72/// (receiver gone). A full queue is a rejection, counted by
73/// `REJECTION_REQUEST_TOTAL`, not here.
74pub static WORK_HANDLER_ENQUEUE_REJECTED_TOTAL: Lazy<IntCounter> = Lazy::new(|| {
75    IntCounter::new(
76        work_handler_metric_name(work_handler::ENQUEUE_REJECTED_TOTAL),
77        "Times enqueuing work failed because the dispatcher channel was closed",
78    )
79    .expect("work_handler_enqueue_rejected_total counter")
80});
81
82/// Time spent waiting to acquire a worker-pool permit. Normal operation is
83/// sub-millisecond; saturation pushes p99 into seconds.
84pub static WORK_HANDLER_PERMIT_WAIT_SECONDS: Lazy<Histogram> = Lazy::new(|| {
85    Histogram::with_opts(
86        HistogramOpts::new(
87            work_handler_metric_name(work_handler::PERMIT_WAIT_SECONDS),
88            "Time spent waiting for a worker-pool permit (seconds)",
89        )
90        .buckets(vec![
91            0.0001, 0.001, 0.01, 0.05, 0.1, 0.5, 1.0, 5.0, 10.0, 30.0, 60.0,
92        ]),
93    )
94    .expect("work_handler_permit_wait_seconds histogram")
95});
96
97/// Current number of active worker-pool tasks (permits in use).
98pub static WORK_HANDLER_POOL_ACTIVE_TASKS: Lazy<IntGauge> = Lazy::new(|| {
99    IntGauge::new(
100        work_handler_metric_name(work_handler::POOL_ACTIVE_TASKS),
101        "Current number of active worker-pool tasks (permits in use)",
102    )
103    .expect("work_handler_pool_active_tasks gauge")
104});
105
106/// Configured worker-pool capacity (total permits). Static; set once at server init.
107pub static WORK_HANDLER_POOL_CAPACITY: Lazy<IntGauge> = Lazy::new(|| {
108    IntGauge::new(
109        work_handler_metric_name(work_handler::POOL_CAPACITY),
110        "Configured worker-pool capacity (total permits)",
111    )
112    .expect("work_handler_pool_capacity gauge")
113});
114
115/// Guards idempotency for the `MetricsRegistry` registration path.
116static METRICS_REGISTERED: OnceCell<()> = OnceCell::new();
117
118/// Register worker-pool saturation metrics with the given registry. Idempotent.
119pub fn ensure_work_handler_pool_metrics_registered(registry: &MetricsRegistry) {
120    let _ = METRICS_REGISTERED.get_or_init(|| {
121        registry.add_metric_or_warn(
122            Box::new(WORK_HANDLER_QUEUE_DEPTH.clone()),
123            "work_handler_queue_depth",
124        );
125        registry.add_metric_or_warn(
126            Box::new(WORK_HANDLER_QUEUE_CAPACITY.clone()),
127            "work_handler_queue_capacity",
128        );
129        registry.add_metric_or_warn(
130            Box::new(WORK_HANDLER_ENQUEUE_REJECTED_TOTAL.clone()),
131            "work_handler_enqueue_rejected_total",
132        );
133        registry.add_metric_or_warn(
134            Box::new(WORK_HANDLER_PERMIT_WAIT_SECONDS.clone()),
135            "work_handler_permit_wait_seconds",
136        );
137        registry.add_metric_or_warn(
138            Box::new(WORK_HANDLER_POOL_ACTIVE_TASKS.clone()),
139            "work_handler_pool_active_tasks",
140        );
141        registry.add_metric_or_warn(
142            Box::new(WORK_HANDLER_POOL_CAPACITY.clone()),
143            "work_handler_pool_capacity",
144        );
145        registry.add_metric_or_warn(
146            Box::new(REJECTION_REQUEST_TOTAL.clone()),
147            "rejection_request_total",
148        );
149        registry.add_metric_or_warn(
150            Box::new(ENGINE_REQUEST_GAUGE.clone()),
151            "dynamo_engine_request",
152        );
153        registry.add_metric_or_warn(
154            Box::new(REQUEST_QUEUE_GAUGE.clone()),
155            "dynamo_request_queue",
156        );
157    });
158}
dynamo_runtime/metrics/work_handler_pool.rs

dynamo_runtime/metrics/
work_handler_pool.rs
