reth_rpc_builder/
metrics.rs

1use jsonrpsee::{
2    core::middleware::{Batch, Notification},
3    server::middleware::rpc::RpcServiceT,
4    types::Request,
5    MethodResponse, RpcModule,
6};
7use reth_metrics::{
8    metrics::{Counter, Histogram},
9    Metrics,
10};
11use reth_primitives_traits::FastInstant as Instant;
12use std::{
13    collections::HashMap,
14    future::Future,
15    pin::Pin,
16    sync::Arc,
17    task::{Context, Poll},
18};
19use tower::Layer;
20
21/// Metrics for the RPC server.
22///
23/// Metrics are divided into two categories:
24/// - Connection metrics: metrics for the connection (e.g. number of connections opened, relevant
25///   for WS and IPC)
26/// - Request metrics: metrics for each RPC method (e.g. number of calls started, time taken to
27///   process a call)
28#[derive(Default, Debug, Clone)]
29pub(crate) struct RpcRequestMetrics {
30    inner: Arc<RpcServerMetricsInner>,
31}
32
33impl RpcRequestMetrics {
34    pub(crate) fn new(module: &RpcModule<()>, transport: RpcTransport) -> Self {
35        Self {
36            inner: Arc::new(RpcServerMetricsInner {
37                connection_metrics: transport.connection_metrics(),
38                call_metrics: module
39                    .method_names()
40                    .map(|method| {
41                        (method, RpcServerCallMetrics::new_with_labels(&[("method", method)]))
42                    })
43                    .collect(),
44            }),
45        }
46    }
47
48    /// Creates a new instance of the metrics layer for HTTP.
49    pub(crate) fn http(module: &RpcModule<()>) -> Self {
50        Self::new(module, RpcTransport::Http)
51    }
52
53    /// Creates a new instance of the metrics layer for same port.
54    ///
55    /// Note: currently it's not possible to track transport specific metrics for a server that runs http and ws on the same port: <https://github.com/paritytech/jsonrpsee/issues/1345> until we have this feature we will use the http metrics for this case.
56    pub(crate) fn same_port(module: &RpcModule<()>) -> Self {
57        Self::http(module)
58    }
59
60    /// Creates a new instance of the metrics layer for Ws.
61    pub(crate) fn ws(module: &RpcModule<()>) -> Self {
62        Self::new(module, RpcTransport::WebSocket)
63    }
64
65    /// Creates a new instance of the metrics layer for Ipc.
66    pub(crate) fn ipc(module: &RpcModule<()>) -> Self {
67        Self::new(module, RpcTransport::Ipc)
68    }
69}
70
71impl<S> Layer<S> for RpcRequestMetrics {
72    type Service = RpcRequestMetricsService<S>;
73
74    fn layer(&self, inner: S) -> Self::Service {
75        RpcRequestMetricsService::new(inner, self.clone())
76    }
77}
78
79/// Metrics for the RPC server
80#[derive(Default, Clone, Debug)]
81struct RpcServerMetricsInner {
82    /// Connection metrics per transport type
83    connection_metrics: RpcServerConnectionMetrics,
84    /// Call metrics per RPC method
85    call_metrics: HashMap<&'static str, RpcServerCallMetrics>,
86}
87
88/// A [`RpcServiceT`] middleware that captures RPC metrics for the server.
89///
90/// This is created per connection and captures metrics for each request.
91#[derive(Clone, Debug)]
92pub struct RpcRequestMetricsService<S> {
93    /// The metrics collector for RPC requests
94    metrics: RpcRequestMetrics,
95    /// The inner service being wrapped
96    inner: S,
97}
98
99impl<S> RpcRequestMetricsService<S> {
100    pub(crate) fn new(service: S, metrics: RpcRequestMetrics) -> Self {
101        // this instance is kept alive for the duration of the connection
102        metrics.inner.connection_metrics.connections_opened_total.increment(1);
103        Self { inner: service, metrics }
104    }
105}
106
107impl<S> RpcServiceT for RpcRequestMetricsService<S>
108where
109    S: RpcServiceT + Send + Sync + Clone + 'static,
110{
111    fn call<'a>(&self, req: Request<'a>) -> impl Future<Output = MethodResponse> + Send + 'a {
112        self.metrics.inner.connection_metrics.requests_started_total.increment(1);
113        let call_metrics = self.metrics.inner.call_metrics.get_key_value(req.method.as_ref());
114        if let Some((_, call_metrics)) = &call_metrics {
115            call_metrics.started_total.increment(1);
116        }
117        MeteredRequestFuture {
118            fut: self.inner.call(req),
119            started_at: Instant::now(),
120            metrics: self.metrics.clone(),
121            method: call_metrics.map(|(method, _)| *method),
122        }
123    }
124
125    fn batch<'a>(&self, req: Batch<'a>) -> impl Future<Output = MethodResponse> + Send + 'a {
126        self.metrics.inner.connection_metrics.batches_started_total.increment(1);
127
128        for batch_entry in req.iter().flatten() {
129            let method_name = batch_entry.method_name();
130            if let Some(call_metrics) = self.metrics.inner.call_metrics.get(method_name) {
131                call_metrics.batched_total.increment(1);
132            }
133        }
134
135        MeteredBatchRequestsFuture {
136            fut: self.inner.batch(req),
137            started_at: Instant::now(),
138            metrics: self.metrics.clone(),
139        }
140    }
141
142    fn notification<'a>(
143        &self,
144        n: Notification<'a>,
145    ) -> impl Future<Output = MethodResponse> + Send + 'a {
146        self.inner.notification(n)
147    }
148}
149
150impl<S> Drop for RpcRequestMetricsService<S> {
151    fn drop(&mut self) {
152        // update connection metrics, connection closed
153        self.metrics.inner.connection_metrics.connections_closed_total.increment(1);
154    }
155}
156
157/// Response future to update the metrics for a single request/response pair.
158#[pin_project::pin_project]
159pub struct MeteredRequestFuture<F> {
160    #[pin]
161    fut: F,
162    /// time when the request started
163    started_at: Instant,
164    /// metrics for the method call
165    metrics: RpcRequestMetrics,
166    /// the method name if known
167    method: Option<&'static str>,
168}
169
170impl<F> std::fmt::Debug for MeteredRequestFuture<F> {
171    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
172        f.write_str("MeteredRequestFuture")
173    }
174}
175
176impl<F: Future<Output = MethodResponse>> Future for MeteredRequestFuture<F> {
177    type Output = F::Output;
178
179    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
180        let this = self.project();
181
182        let res = this.fut.poll(cx);
183        if let Poll::Ready(resp) = &res {
184            let elapsed = this.started_at.elapsed().as_secs_f64();
185
186            // update transport metrics
187            this.metrics.inner.connection_metrics.requests_finished_total.increment(1);
188            this.metrics.inner.connection_metrics.request_time_seconds.record(elapsed);
189
190            // update call metrics
191            if let Some(call_metrics) =
192                this.method.and_then(|method| this.metrics.inner.call_metrics.get(method))
193            {
194                call_metrics.time_seconds.record(elapsed);
195                if resp.is_success() {
196                    call_metrics.successful_total.increment(1);
197                } else {
198                    call_metrics.failed_total.increment(1);
199                }
200            }
201        }
202        res
203    }
204}
205
206/// Response future to update the metrics for a batch of request/response pairs.
207#[pin_project::pin_project]
208pub struct MeteredBatchRequestsFuture<F> {
209    #[pin]
210    fut: F,
211    /// time when the batch request started
212    started_at: Instant,
213    /// metrics for the batch
214    metrics: RpcRequestMetrics,
215}
216
217impl<F> std::fmt::Debug for MeteredBatchRequestsFuture<F> {
218    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
219        f.write_str("MeteredBatchRequestsFuture")
220    }
221}
222
223impl<F> Future for MeteredBatchRequestsFuture<F>
224where
225    F: Future,
226{
227    type Output = F::Output;
228
229    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
230        let this = self.project();
231        let res = this.fut.poll(cx);
232
233        if res.is_ready() {
234            let elapsed = this.started_at.elapsed().as_secs_f64();
235            this.metrics.inner.connection_metrics.batches_finished_total.increment(1);
236            this.metrics.inner.connection_metrics.batch_response_time_seconds.record(elapsed);
237        }
238        res
239    }
240}
241
242/// The transport protocol used for the RPC connection.
243#[derive(Debug, Clone, Copy, Eq, PartialEq)]
244pub(crate) enum RpcTransport {
245    Http,
246    WebSocket,
247    Ipc,
248}
249
250impl RpcTransport {
251    /// Returns the string representation of the transport protocol.
252    pub(crate) const fn as_str(&self) -> &'static str {
253        match self {
254            Self::Http => "http",
255            Self::WebSocket => "ws",
256            Self::Ipc => "ipc",
257        }
258    }
259
260    /// Returns the connection metrics for the transport protocol.
261    fn connection_metrics(&self) -> RpcServerConnectionMetrics {
262        RpcServerConnectionMetrics::new_with_labels(&[("transport", self.as_str())])
263    }
264}
265
266/// Metrics for the RPC connections
267#[derive(Metrics, Clone)]
268#[metrics(scope = "rpc_server.connections")]
269struct RpcServerConnectionMetrics {
270    /// The number of connections opened
271    connections_opened_total: Counter,
272    /// The number of connections closed
273    connections_closed_total: Counter,
274    /// The number of requests started
275    requests_started_total: Counter,
276    /// The number of requests finished
277    requests_finished_total: Counter,
278    /// Response for a single request/response pair
279    request_time_seconds: Histogram,
280    /// The number of batch requests started
281    batches_started_total: Counter,
282    /// The number of batch requests finished
283    batches_finished_total: Counter,
284    /// Response time for a batch request
285    batch_response_time_seconds: Histogram,
286}
287
288/// Metrics for the RPC calls
289#[derive(Metrics, Clone)]
290#[metrics(scope = "rpc_server.calls")]
291struct RpcServerCallMetrics {
292    /// The number of calls started
293    started_total: Counter,
294    /// The number of successful calls
295    successful_total: Counter,
296    /// The number of failed calls
297    failed_total: Counter,
298    /// The number of calls received as batch entries.
299    ///
300    /// jsonrpsee dispatches batch entries internally without invoking this middleware's `call`,
301    /// so only their per-method volume can be tracked; batched calls are excluded from
302    /// `started_total`, `successful_total`, `failed_total` and `time_seconds`.
303    batched_total: Counter,
304    /// Response for a single call
305    time_seconds: Histogram,
306}
reth_rpc_builder/metrics.rs

reth_rpc_builder/
metrics.rs
