iota_json_rpc/
metrics.rs

1// Copyright (c) Mysten Labs, Inc.
2// Modifications Copyright (c) 2024 IOTA Stiftung
3// SPDX-License-Identifier: Apache-2.0
4
5use std::{collections::HashSet, net::SocketAddr};
6
7use http_body::Body;
8use iota_json_rpc_api::{
9    CLIENT_SDK_TYPE_HEADER, CLIENT_TARGET_API_VERSION_HEADER, TRANSIENT_ERROR_CODE,
10};
11use jsonrpsee::{MethodKind, server::HttpRequest, types::Params};
12use prometheus_filtered::{
13    HistogramVec, IntCounterVec, IntGaugeVec, MetricLevel, register_histogram_vec_with_registry,
14    register_int_counter_vec_with_registry, register_int_gauge_vec_with_registry,
15};
16use tokio::time::Instant;
17
18use crate::logger::{Logger, TransportProtocol};
19
20const SPAM_LABEL: &str = "SPAM";
21const LATENCY_SEC_BUCKETS: &[f64] = &[
22    0.001, 0.005, 0.01, 0.05, 0.1, 0.25, 0.5, 1., 2.5, 5., 10., 20., 30., 60., 90.,
23];
24
25#[derive(Debug, Clone)]
26pub struct Metrics {
27    /// Counter of requests, route is a label (ie separate timeseries per route)
28    requests_by_route: IntCounterVec,
29    /// Gauge of inflight requests, route is a label (ie separate timeseries per
30    /// route)
31    inflight_requests_by_route: IntGaugeVec,
32    /// Request latency, route is a label
33    req_latency_by_route: HistogramVec,
34    /// Failed requests by route
35    errors_by_route: IntCounterVec,
36    server_errors_by_route: IntCounterVec,
37    client_errors_by_route: IntCounterVec,
38    transient_errors_by_route: IntCounterVec,
39    /// Client info
40    client: IntCounterVec,
41    /// Connection count
42    inflight_connection: IntGaugeVec,
43    /// Request size
44    rpc_request_size: HistogramVec,
45    /// Response size
46    rpc_response_size: HistogramVec,
47}
48
49#[derive(Clone)]
50pub struct MetricsLogger {
51    metrics: Metrics,
52    method_whitelist: HashSet<String>,
53}
54
55impl MetricsLogger {
56    fn check_spam<'a>(&'a self, method_name: &'a str) -> &'a str {
57        if self.method_whitelist.contains(method_name) {
58            method_name
59        } else {
60            SPAM_LABEL
61        }
62    }
63
64    pub fn new(registry: &prometheus_filtered::Registry, method_whitelist: &[&str]) -> Self {
65        let metrics = Metrics {
66            requests_by_route: register_int_counter_vec_with_registry!(
67                "rpc_requests_by_route",
68                "Number of requests by route",
69                &["route"],
70                registry;
71                MetricLevel::Warn,
72            )
73            .unwrap(),
74            inflight_requests_by_route: register_int_gauge_vec_with_registry!(
75                "inflight_rpc_requests_by_route",
76                "Number of inflight requests by route",
77                &["route"],
78                registry;
79                MetricLevel::Warn,
80            )
81            .unwrap(),
82            req_latency_by_route: register_histogram_vec_with_registry!(
83                "req_latency_by_route",
84                "Latency of a request by route",
85                &["route"],
86                LATENCY_SEC_BUCKETS.to_vec(),
87                registry;
88                MetricLevel::Warn,
89            )
90            .unwrap(),
91            client_errors_by_route: register_int_counter_vec_with_registry!(
92                "client_errors_by_route",
93                "Number of client errors by route",
94                &["route"],
95                registry,
96            )
97            .unwrap(),
98            server_errors_by_route: register_int_counter_vec_with_registry!(
99                "server_errors_by_route",
100                "Number of server errors by route",
101                &["route"],
102                registry,
103            )
104            .unwrap(),
105            transient_errors_by_route: register_int_counter_vec_with_registry!(
106                "transient_errors_by_route",
107                "Number of transient errors by route",
108                &["route"],
109                registry,
110            )
111            .unwrap(),
112            errors_by_route: register_int_counter_vec_with_registry!(
113                "errors_by_route",
114                "Number of client and server errors by route",
115                &["route"],
116                registry;
117                MetricLevel::Warn,
118            )
119            .unwrap(),
120            client: register_int_counter_vec_with_registry!(
121                "rpc_client",
122                "Connected RPC client's info",
123                &["client_type", "api_version"],
124                registry,
125            )
126            .unwrap(),
127            inflight_connection: register_int_gauge_vec_with_registry!(
128                "rpc_inflight_connection",
129                "Number of inflight RPC connection by protocol",
130                &["protocol"],
131                registry;
132                MetricLevel::Warn,
133            )
134            .unwrap(),
135            rpc_request_size: register_histogram_vec_with_registry!(
136                "rpc_request_size",
137                "Request size of rpc requests",
138                &["protocol"],
139                prometheus_filtered::exponential_buckets(32.0, 2.0, 19)
140                    .unwrap()
141                    .to_vec(),
142                registry,
143            )
144            .unwrap(),
145            rpc_response_size: register_histogram_vec_with_registry!(
146                "rpc_response_size",
147                "Response size of rpc requests",
148                &["protocol"],
149                prometheus_filtered::exponential_buckets(1024.0, 2.0, 20)
150                    .unwrap()
151                    .to_vec(),
152                registry,
153            )
154            .unwrap(),
155        };
156
157        Self {
158            metrics,
159            method_whitelist: method_whitelist.iter().map(|s| (*s).into()).collect(),
160        }
161    }
162}
163
164impl Logger for MetricsLogger {
165    type Instant = Instant;
166
167    fn on_connect(&self, _remote_addr: SocketAddr, request: &HttpRequest, t: TransportProtocol) {
168        let client_type = request
169            .headers()
170            .get(CLIENT_SDK_TYPE_HEADER)
171            .and_then(|v| v.to_str().ok())
172            .unwrap_or("Unknown");
173
174        let api_version = request
175            .headers()
176            .get(CLIENT_TARGET_API_VERSION_HEADER)
177            .and_then(|v| v.to_str().ok())
178            .unwrap_or("Unknown");
179        self.metrics
180            .client
181            .with_label_values(&[client_type, api_version])
182            .inc();
183        self.metrics
184            .inflight_connection
185            .with_label_values(&[&t.to_string()])
186            .inc();
187
188        self.metrics
189            .rpc_request_size
190            .with_label_values(&[&t.to_string()])
191            .observe(
192                request
193                    .size_hint()
194                    .exact()
195                    .unwrap_or_else(|| request.size_hint().lower()) as f64,
196            );
197    }
198
199    fn on_request(&self, _transport: TransportProtocol) -> Self::Instant {
200        Instant::now()
201    }
202
203    fn on_call(
204        &self,
205        method_name: &str,
206        _params: Params,
207        _kind: MethodKind,
208        _transport: TransportProtocol,
209    ) {
210        let method_name = self.check_spam(method_name);
211        self.metrics
212            .inflight_requests_by_route
213            .with_label_values(&[method_name])
214            .inc();
215        self.metrics
216            .requests_by_route
217            .with_label_values(&[method_name])
218            .inc();
219    }
220
221    fn on_result(
222        &self,
223        method_name: &str,
224        _success: bool,
225        error_code: Option<i32>,
226        started_at: Self::Instant,
227        _transport: TransportProtocol,
228    ) {
229        let method_name = self.check_spam(method_name);
230        self.metrics
231            .inflight_requests_by_route
232            .with_label_values(&[method_name])
233            .dec();
234        let req_latency_secs = (Instant::now() - started_at).as_secs_f64();
235        self.metrics
236            .req_latency_by_route
237            .with_label_values(&[method_name])
238            .observe(req_latency_secs);
239
240        if let Some(code) = error_code {
241            if code == jsonrpsee::types::error::CALL_EXECUTION_FAILED_CODE
242                || code == jsonrpsee::types::error::INTERNAL_ERROR_CODE
243            {
244                self.metrics
245                    .server_errors_by_route
246                    .with_label_values(&[method_name])
247                    .inc();
248            } else if code == TRANSIENT_ERROR_CODE {
249                self.metrics
250                    .transient_errors_by_route
251                    .with_label_values(&[method_name])
252                    .inc();
253            } else {
254                self.metrics
255                    .client_errors_by_route
256                    .with_label_values(&[method_name])
257                    .inc();
258            }
259            self.metrics
260                .errors_by_route
261                .with_label_values(&[method_name])
262                .inc();
263        }
264    }
265
266    fn on_response(&self, result: &str, _started_at: Self::Instant, t: TransportProtocol) {
267        self.metrics
268            .rpc_response_size
269            .with_label_values(&[&t.to_string()])
270            .observe(std::mem::size_of_val(result) as f64)
271    }
272
273    fn on_disconnect(&self, _remote_addr: SocketAddr, t: TransportProtocol) {
274        self.metrics
275            .inflight_connection
276            .with_label_values(&[&t.to_string()])
277            .dec();
278    }
279}
iota_json_rpc/metrics.rs

iota_json_rpc/
metrics.rs
