forest/metrics/
mod.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3
4pub mod db;
5
6use crate::{db::DBStatistics, networks::ChainConfig, shim::clock::ChainEpoch};
7use axum::{Router, http::StatusCode, response::IntoResponse, routing::get};
8use parking_lot::{RwLock, RwLockWriteGuard};
9use prometheus_client::{
10    collector::Collector,
11    encoding::EncodeLabelSet,
12    metrics::{
13        counter::Counter,
14        family::Family,
15        gauge::Gauge,
16        histogram::{Histogram, exponential_buckets},
17    },
18};
19use std::sync::{Arc, LazyLock};
20use std::{path::PathBuf, time::Instant};
21use tokio::net::TcpListener;
22use tower_http::compression::CompressionLayer;
23use tracing::warn;
24
25static DEFAULT_REGISTRY: LazyLock<RwLock<prometheus_client::registry::Registry>> =
26    LazyLock::new(Default::default);
27
28static COLLECTOR_REGISTRY: LazyLock<RwLock<prometheus_client::registry::Registry>> =
29    LazyLock::new(Default::default);
30
31pub fn default_registry<'a>() -> RwLockWriteGuard<'a, prometheus_client::registry::Registry> {
32    DEFAULT_REGISTRY.write()
33}
34
35pub fn collector_registry<'a>() -> RwLockWriteGuard<'a, prometheus_client::registry::Registry> {
36    COLLECTOR_REGISTRY.write()
37}
38
39pub fn register_collector(collector: Box<dyn Collector>) {
40    #[allow(clippy::disallowed_methods)]
41    collector_registry().register_collector(collector)
42}
43
44pub fn reset_collector_registry() {
45    *collector_registry() = Default::default();
46}
47
48pub static RPC_METHOD_FAILURE: LazyLock<Family<RpcMethodLabel, Counter>> = LazyLock::new(|| {
49    let metric = Family::default();
50    DEFAULT_REGISTRY.write().register(
51        "rpc_method_failure",
52        "Number of failed RPC calls",
53        metric.clone(),
54    );
55    metric
56});
57
58pub static RPC_IN_FLIGHT: LazyLock<Gauge> = LazyLock::new(|| {
59    let metric = Gauge::default();
60    DEFAULT_REGISTRY.write().register(
61        "rpc_in_flight_requests",
62        "Number of RPC requests currently being processed",
63        metric.clone(),
64    );
65    metric
66});
67
68pub static RPC_METHOD_TIME: LazyLock<Family<RpcMethodLabel, Histogram>> = LazyLock::new(|| {
69    let metric = Family::<RpcMethodLabel, Histogram>::new_with_constructor(|| {
70        // Histogram with 5 buckets starting from 0.1ms going to 1s, each bucket 10 times as big as the last.
71        Histogram::new(exponential_buckets(0.1, 10., 5))
72    });
73    crate::metrics::default_registry().register(
74        "rpc_processing_time",
75        "Duration of RPC method call in milliseconds",
76        metric.clone(),
77    );
78    metric
79});
80
81pub async fn init_prometheus<DB>(
82    prometheus_listener: TcpListener,
83    db_directory: PathBuf,
84    db: Arc<DB>,
85    chain_config: Arc<ChainConfig>,
86    get_chain_head_height: Arc<impl Fn() -> ChainEpoch + Send + Sync + 'static>,
87    get_chain_head_actor_version: Arc<impl Fn() -> u64 + Send + Sync + 'static>,
88) -> anyhow::Result<()>
89where
90    DB: DBStatistics + Send + Sync + 'static,
91{
92    // Add the process collector to the registry
93    if let Err(err) = kubert_prometheus_process::register(
94        collector_registry().sub_registry_with_prefix("process"),
95    ) {
96        warn!("Failed to register process metrics: {err}");
97    }
98
99    register_collector(Box::new(
100        crate::utils::version::ForestVersionCollector::new(),
101    ));
102    register_collector(Box::new(crate::metrics::db::DBCollector::new(db_directory)));
103    register_collector(Box::new(
104        crate::networks::metrics::NetworkVersionCollector::new(
105            chain_config,
106            get_chain_head_height,
107            get_chain_head_actor_version,
108        ),
109    ));
110
111    // Create an configure HTTP server
112    let app = Router::new()
113        .route("/metrics", get(collect_prometheus_metrics))
114        .route("/stats/db", get(collect_db_metrics::<DB>))
115        .layer(CompressionLayer::new())
116        .with_state(db);
117
118    // Wait for server to exit
119    Ok(axum::serve(prometheus_listener, app.into_make_service()).await?)
120}
121
122async fn collect_prometheus_metrics() -> impl IntoResponse {
123    let mut metrics = String::new();
124    if let Err(e) =
125        prometheus_client::encoding::text::encode_registry(&mut metrics, &DEFAULT_REGISTRY.read())
126    {
127        warn!("failed to encode the default metrics registry: {e}");
128    };
129    if let Err(e) =
130        prometheus_client::encoding::text::encode_registry(&mut metrics, &COLLECTOR_REGISTRY.read())
131    {
132        warn!("failed to encode the collector metrics registry: {e}");
133    };
134    if let Err(e) = prometheus_client::encoding::text::encode_eof(&mut metrics) {
135        warn!("failed to encode metrics eof {e}");
136    };
137    (
138        StatusCode::OK,
139        [("content-type", "text/plain; charset=utf-8")],
140        metrics,
141    )
142}
143
144async fn collect_db_metrics<DB>(
145    axum::extract::State(db): axum::extract::State<Arc<DB>>,
146) -> impl IntoResponse
147where
148    DB: DBStatistics,
149{
150    let mut metrics = "# DB statistics:\n".to_owned();
151    if let Some(db_stats) = db.get_statistics() {
152        metrics.push_str(&db_stats);
153    } else {
154        metrics.push_str("Not enabled. Set enable_statistics to true in config and restart daemon");
155    }
156    (
157        StatusCode::OK,
158        [("content-type", "text/plain; charset=utf-8")],
159        metrics,
160    )
161}
162
163#[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet)]
164pub struct RpcMethodLabel {
165    pub method: &'static str,
166}
167
168#[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet, derive_more::Constructor)]
169pub struct KindLabel {
170    kind: &'static str,
171}
172
173pub fn default_histogram() -> Histogram {
174    // Default values from go client(https://github.com/prometheus/client_golang/blob/5d584e2717ef525673736d72cd1d12e304f243d7/prometheus/histogram.go#L68)
175    Histogram::new([
176        0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0,
177    ])
178}
179
180pub struct HistogramTimer<'a> {
181    histogram: &'a Histogram,
182    start: Instant,
183}
184
185impl Drop for HistogramTimer<'_> {
186    fn drop(&mut self) {
187        let duration = Instant::now() - self.start;
188        self.histogram.observe(duration.as_secs_f64());
189    }
190}
191
192pub trait HistogramTimerExt {
193    fn start_timer(&self) -> HistogramTimer<'_>;
194}
195
196impl HistogramTimerExt for Histogram {
197    fn start_timer(&self) -> HistogramTimer<'_> {
198        HistogramTimer {
199            histogram: self,
200            start: Instant::now(),
201        }
202    }
203}
204
205/// RAII guard that increments a gauge on creation and decrements it on drop,
206/// keeping the count balanced even if the guarded scope ends via cancellation
207/// (its future is dropped) or panic-unwind. Created via [`GaugeGuardExt::inc_guard`].
208pub struct GaugeGuard<'a> {
209    gauge: &'a Gauge,
210}
211
212impl Drop for GaugeGuard<'_> {
213    fn drop(&mut self) {
214        self.gauge.dec();
215    }
216}
217
218pub trait GaugeGuardExt {
219    /// Increments the gauge and returns a [`GaugeGuard`] that decrements it on drop.
220    #[must_use = "dropping the guard immediately makes the increment a no-op; bind it to a variable"]
221    fn inc_guard(&self) -> GaugeGuard<'_>;
222}
223
224impl GaugeGuardExt for Gauge {
225    fn inc_guard(&self) -> GaugeGuard<'_> {
226        self.inc();
227        GaugeGuard { gauge: self }
228    }
229}
forest/metrics/mod.rs

forest/metrics/
mod.rs
