iota_metrics/
runtime_metrics.rs

1// Copyright (c) 2026 IOTA Stiftung
2// SPDX-License-Identifier: Apache-2.0
3
4//! Per-runtime observability for the tokio runtimes the node runs on.
5//!
6//! The node splits work across separate tokio runtimes (notably a node-core
7//! runtime and a client-facing serving runtime). These metrics make it possible
8//! to tell whether one runtime is starving another for worker threads: the
9//! `scheduler_lag_seconds` heartbeat and a non-zero `global_queue_depth` are
10//! the direct signals of a runtime whose workers cannot keep up with ready
11//! tasks.
12
13use std::{
14    sync::Arc,
15    time::{Duration, Instant},
16};
17
18use prometheus_filtered::{
19    HistogramVec, IntGaugeVec, MetricLevel, Registry, register_histogram_vec_with_registry,
20    register_int_gauge_vec_with_registry,
21};
22use tokio::runtime::Handle;
23
24/// How often the tokio runtime counters are sampled.
25const SAMPLE_INTERVAL: Duration = Duration::from_secs(5);
26/// How often the scheduler-lag heartbeat fires.
27const HEARTBEAT_INTERVAL: Duration = Duration::from_millis(100);
28
29pub struct RuntimeMonitorMetrics {
30    workers: IntGaugeVec,
31    alive_tasks: IntGaugeVec,
32    global_queue_depth: IntGaugeVec,
33    scheduler_lag_seconds: HistogramVec,
34}
35
36impl RuntimeMonitorMetrics {
37    pub fn new(registry: &Registry) -> Arc<Self> {
38        Arc::new(Self {
39            workers: register_int_gauge_vec_with_registry!(
40                "tokio_runtime_workers",
41                "Number of worker threads in the tokio runtime.",
42                &["runtime"],
43                registry;
44                MetricLevel::Warn,
45            )
46            .unwrap(),
47            alive_tasks: register_int_gauge_vec_with_registry!(
48                "tokio_runtime_alive_tasks",
49                "Number of alive (spawned, not yet completed) tasks in the tokio runtime.",
50                &["runtime"],
51                registry;
52                MetricLevel::Warn,
53            )
54            .unwrap(),
55            global_queue_depth: register_int_gauge_vec_with_registry!(
56                "tokio_runtime_global_queue_depth",
57                "Tasks waiting in the runtime's global injection queue. A persistently \
58                 non-zero value means the workers cannot keep up with ready tasks.",
59                &["runtime"],
60                registry;
61                MetricLevel::Warn,
62            )
63            .unwrap(),
64            scheduler_lag_seconds: register_histogram_vec_with_registry!(
65                "tokio_runtime_scheduler_lag_seconds",
66                "Delay between when a fixed-interval heartbeat task was scheduled to wake \
67                 and when it actually ran. High values indicate worker-thread starvation.",
68                &["runtime"],
69                vec![
70                    0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0
71                ],
72                registry;
73                MetricLevel::Warn,
74            )
75            .unwrap(),
76        })
77    }
78}
79
80/// Starts background monitoring of the tokio runtime identified by `handle`,
81/// labelling all metrics with `runtime`.
82///
83/// Both the counter sampler and the scheduler-lag heartbeat run *on the
84/// monitored runtime* so that the heartbeat observes that runtime's scheduling
85/// latency directly.
86pub fn start_runtime_monitor(
87    runtime: &'static str,
88    handle: &Handle,
89    metrics: Arc<RuntimeMonitorMetrics>,
90) {
91    // Sampler: periodically read the stable RuntimeMetrics counters.
92    {
93        let runtime_metrics = handle.metrics();
94        let workers = metrics.workers.with_label_values(&[runtime]);
95        let alive_tasks = metrics.alive_tasks.with_label_values(&[runtime]);
96        let global_queue_depth = metrics.global_queue_depth.with_label_values(&[runtime]);
97        handle.spawn(async move {
98            loop {
99                workers.set(runtime_metrics.num_workers() as i64);
100                alive_tasks.set(runtime_metrics.num_alive_tasks() as i64);
101                global_queue_depth.set(runtime_metrics.global_queue_depth() as i64);
102                tokio::time::sleep(SAMPLE_INTERVAL).await;
103            }
104        });
105    }
106
107    // Heartbeat: the excess of the actual sleep duration over the intended
108    // interval is the time the task spent waiting for a free worker thread.
109    {
110        let lag = metrics.scheduler_lag_seconds.with_label_values(&[runtime]);
111        handle.spawn(async move {
112            loop {
113                let start = Instant::now();
114                tokio::time::sleep(HEARTBEAT_INTERVAL).await;
115                let lag_secs = start
116                    .elapsed()
117                    .saturating_sub(HEARTBEAT_INTERVAL)
118                    .as_secs_f64();
119                lag.observe(lag_secs);
120            }
121        });
122    }
123}
iota_metrics/runtime_metrics.rs

iota_metrics/
runtime_metrics.rs
