prover_telemetry/
lib.rs

1use std::net::SocketAddr;
2
3use axum::{
4    extract::State,
5    http::{Response, StatusCode},
6    routing::get,
7    serve::WithGracefulShutdown,
8    Router,
9};
10use lazy_static::lazy_static;
11use opentelemetry::global;
12use opentelemetry_sdk::metrics::SdkMeterProvider;
13use prometheus::{Encoder as _, Registry, TextEncoder};
14use tokio_util::sync::CancellationToken;
15use tracing::{debug, info};
16
17use crate::{
18    constant::{AGGLAYER_KERNEL_OTEL_SCOPE_NAME, AGGLAYER_RPC_OTEL_SCOPE_NAME},
19    error::MetricsError,
20};
21
22mod constant;
23mod error;
24
25pub use error::Error;
26pub use opentelemetry::KeyValue;
27
28lazy_static! {
29    // Backward compatibility with the old metrics from agglayer go implementation
30    // Those metrics are not linked to any registry
31    pub static ref SEND_TX: opentelemetry::metrics::Counter<u64> = global::meter(AGGLAYER_RPC_OTEL_SCOPE_NAME)
32        .u64_counter("send_tx")
33        .with_description("Number of transactions received on the RPC")
34        .build();
35
36    pub static ref VERIFY_ZKP: opentelemetry::metrics::Counter<u64> = global::meter(AGGLAYER_KERNEL_OTEL_SCOPE_NAME)
37        .u64_counter("verify_zkp")
38        .with_description("Number of ZKP verifications")
39        .build();
40
41    pub static ref VERIFY_SIGNATURE: opentelemetry::metrics::Counter<u64> = global::meter(AGGLAYER_KERNEL_OTEL_SCOPE_NAME)
42        .u64_counter("verify_signature")
43        .with_description("Number of signature verifications")
44        .build();
45
46    pub static ref CHECK_TX: opentelemetry::metrics::Counter<u64> = global::meter(AGGLAYER_KERNEL_OTEL_SCOPE_NAME)
47        .u64_counter("check_tx")
48        .with_description("Number of transactions checked")
49        .build();
50
51    pub static ref EXECUTE: opentelemetry::metrics::Counter<u64> = global::meter(AGGLAYER_KERNEL_OTEL_SCOPE_NAME)
52        .u64_counter("execute")
53        .with_description("Number of transactions executed")
54        .build();
55
56    pub static ref SETTLE: opentelemetry::metrics::Counter<u64> = global::meter(AGGLAYER_KERNEL_OTEL_SCOPE_NAME)
57        .u64_counter("settle")
58        .with_description("Number of transactions settled")
59        .build();
60}
61
62pub mod prover {
63    use lazy_static::lazy_static;
64    use opentelemetry::global;
65
66    use crate::constant::AGGLAYER_PROVER_RPC_OTEL_SCOPE_NAME;
67
68    lazy_static! {
69        pub static ref PROVING_REQUEST_RECV: opentelemetry::metrics::Counter<u64> =
70            global::meter(AGGLAYER_PROVER_RPC_OTEL_SCOPE_NAME)
71                .u64_counter("proving_request_recv")
72                .with_description("Number of proving request received")
73                .build();
74        pub static ref PROVING_REQUEST_SUCCEEDED: opentelemetry::metrics::Counter<u64> =
75            global::meter(AGGLAYER_PROVER_RPC_OTEL_SCOPE_NAME)
76                .u64_counter("proving_request_succeeded")
77                .with_description("Number of proving request that succeeded")
78                .build();
79        pub static ref PROVING_REQUEST_FAILED: opentelemetry::metrics::Counter<u64> =
80            global::meter(AGGLAYER_PROVER_RPC_OTEL_SCOPE_NAME)
81                .u64_counter("proving_request_failed")
82                .with_description("Number of proving request that failed")
83                .build();
84        pub static ref PROVING_FALLBACK_TRIGGERED: opentelemetry::metrics::Counter<u64> =
85            global::meter(AGGLAYER_PROVER_RPC_OTEL_SCOPE_NAME)
86                .u64_counter("proving_fallback_triggered")
87                .with_description("Number of proving fallback triggered")
88                .build();
89    }
90}
91
92pub struct ServerBuilder {}
93
94#[buildstructor::buildstructor]
95impl ServerBuilder {
96    /// Function that builds a new Metrics server and returns a
97    /// [`WithGracefulShutdown`] instance ready to be spawn.
98    ///
99    /// The available methods are:
100    ///
101    /// - `builder`: Creates a new builder instance.
102    /// - `addr`: Sets the [`SocketAddr`] to bind the metrics server to.
103    /// - `registry`: Sets the [`Registry`] to use for metrics. (optional)
104    /// - `build`: Builds the metrics server and returns a
105    ///   [`WithGracefulShutdown`] instance.
106    ///
107    /// # Examples
108    /// ```
109    /// # use std::sync::Arc;
110    /// # use agglayer_telemetry::ServerBuilder;
111    /// # use agglayer_telemetry::Error;
112    /// # use tokio_util::sync::CancellationToken;
113    /// # use std::net::SocketAddr;
114    /// #
115    ///
116    /// async fn build_metrics() -> Result<(), Error> {
117    ///     ServerBuilder::builder()
118    ///         .addr("127.0.0.1".parse::<SocketAddr>().unwrap())
119    ///         .cancellation_token(CancellationToken::new())
120    ///         .build()
121    ///         .await?;
122    ///
123    ///     Ok(())
124    /// }
125    /// ```
126    ///
127    ///
128    /// # Panics
129    ///
130    /// Panics on failure of the gather_metrics internal methods (unlikely)
131    ///
132    /// # Errors
133    ///
134    /// This function will return an error if the provided addr is invalid
135    #[builder(entry = "builder", exit = "build", visibility = "pub")]
136    pub async fn serve(
137        addr: SocketAddr,
138        registry: Option<Registry>,
139        cancellation_token: CancellationToken,
140    ) -> Result<
141        WithGracefulShutdown<
142            axum::routing::IntoMakeService<Router>,
143            axum::Router,
144            impl futures::Future<Output = ()>,
145        >,
146        Error,
147    > {
148        let registry = registry.unwrap_or_default();
149        let _ = Self::init_meter_provider(&registry);
150
151        let app = Router::new()
152            .route(
153                "/metrics",
154                get(|State(registry): State<prometheus::Registry>| async move {
155                    match Self::gather_metrics(&registry) {
156                        Ok(metrics) => Response::new(metrics),
157                        Err(error) => Response::builder()
158                            .status(StatusCode::INTERNAL_SERVER_ERROR)
159                            .body(error.to_string())
160                            .unwrap(),
161                    }
162                }),
163            )
164            .with_state(registry);
165
166        info!("Starting metrics server on {}", addr);
167
168        let listener = tokio::net::TcpListener::bind(addr).await?;
169
170        Ok(axum::serve(listener, app.into_make_service())
171            .with_graceful_shutdown(shutdown_signal(cancellation_token)))
172    }
173
174    fn init_meter_provider(registry: &Registry) -> Result<(), MetricsError> {
175        // configure OpenTelemetry to use the registry
176        let exporter = opentelemetry_prometheus::exporter()
177            .with_registry(registry.clone())
178            .build()
179            .unwrap();
180
181        // set up a meter meter to create instruments
182        let provider = SdkMeterProvider::builder().with_reader(exporter).build();
183
184        global::set_meter_provider(provider);
185        Ok(())
186    }
187
188    fn gather_metrics(registry: &prometheus::Registry) -> Result<String, MetricsError> {
189        // Encode data as text or protobuf
190        let encoder = TextEncoder::new();
191        let metric_families = registry.gather();
192        let mut result = Vec::new();
193        encoder.encode(&metric_families, &mut result)?;
194
195        Ok(String::from_utf8(result)?)
196    }
197}
198
199async fn shutdown_signal(cancellation: CancellationToken) {
200    tokio::select! {
201        _ = cancellation.cancelled() => {
202            debug!("Shutting down metrics server...");
203        },
204    }
205}
prover_telemetry/lib.rs

prover_telemetry/
lib.rs
