kora-lib 2.0.5

Core library for Kora gasless relayer
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146

use crate::rpc_server::middleware_utils::{extract_parts_and_body_bytes, get_jsonrpc_method};
use http::{Request, Response};
use jsonrpsee::server::logger::Body;
use prometheus::{CounterVec, HistogramVec, Opts};
use std::{sync::OnceLock, time::Instant};
use tower::Layer;

static HTTP_METRICS: OnceLock<HttpMetrics> = OnceLock::new();

const UNKNOWN_METHOD: &str = "unknown";
const ERROR_STATUS: &str = "error";

pub struct HttpMetrics {
    pub requests_total: CounterVec,
    pub request_duration_seconds: HistogramVec,
}

impl HttpMetrics {
    fn new() -> Self {
        let requests_total = CounterVec::new(
            Opts::new("http_requests_total", "Total number of HTTP requests").namespace("kora"),
            &["method", "status"],
        )
        .unwrap_or_else(|e| {
            log::error!("Failed to create http_requests_total metric: {e:?}");
            panic!("Metrics initialization failed - cannot continue")
        });

        let request_duration_seconds = HistogramVec::new(
            prometheus::HistogramOpts::new(
                "http_request_duration_seconds",
                "HTTP request duration in seconds",
            )
            .namespace("kora")
            .buckets(vec![0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 5.0, 10.0]),
            &["method"],
        )
        .unwrap_or_else(|e| {
            log::error!("Failed to create http_request_duration_seconds metric: {e:?}");
            panic!("Metrics initialization failed - cannot continue")
        });

        prometheus::register(Box::new(requests_total.clone())).unwrap_or_else(|e| {
            log::error!("Failed to register http_requests_total metric: {e:?}");
            panic!("Metrics initialization failed - cannot continue")
        });
        prometheus::register(Box::new(request_duration_seconds.clone())).unwrap_or_else(|e| {
            log::error!("Failed to register http_request_duration_seconds metric: {e:?}");
            panic!("Metrics initialization failed - cannot continue")
        });

        Self { requests_total, request_duration_seconds }
    }

    pub fn get() -> &'static HttpMetrics {
        HTTP_METRICS.get_or_init(HttpMetrics::new)
    }
}
/// Tower layer for collecting HTTP metrics
#[derive(Clone)]
pub struct HttpMetricsLayer;

impl HttpMetricsLayer {
    pub fn new() -> Self {
        Self
    }
}

impl Default for HttpMetricsLayer {
    fn default() -> Self {
        Self::new()
    }
}

impl<S> Layer<S> for HttpMetricsLayer {
    type Service = HttpMetricsService<S>;

    fn layer(&self, service: S) -> Self::Service {
        HttpMetricsService { inner: service }
    }
}

/// Tower service for collecting HTTP metrics
#[derive(Clone)]
pub struct HttpMetricsService<S> {
    inner: S,
}

impl<S> tower::Service<Request<Body>> for HttpMetricsService<S>
where
    S: tower::Service<Request<Body>, Response = Response<Body>> + Clone + Send + 'static,
    S::Future: Send + 'static,
{
    type Response = S::Response;
    type Error = S::Error;
    type Future = std::pin::Pin<
        Box<dyn std::future::Future<Output = Result<Self::Response, Self::Error>> + Send>,
    >;

    fn poll_ready(
        &mut self,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, request: Request<Body>) -> Self::Future {
        let start = Instant::now();
        let mut inner = self.inner.clone();

        Box::pin(async move {
            let (parts, body_bytes) = extract_parts_and_body_bytes(request).await;
            let method = get_jsonrpc_method(&body_bytes).unwrap_or(UNKNOWN_METHOD.to_string());

            // Reconstruct the request with the consumed body
            let new_body = Body::from(body_bytes);
            let new_request = Request::from_parts(parts, new_body);

            // Call the inner service
            let result = inner.call(new_request).await;

            // Record metrics
            let metrics = HttpMetrics::get();
            let duration = start.elapsed();

            match &result {
                Ok(response) => {
                    let status = response.status().as_u16().to_string();
                    metrics.requests_total.with_label_values(&[&method, &status]).inc();
                    metrics
                        .request_duration_seconds
                        .with_label_values(&[&method])
                        .observe(duration.as_secs_f64());
                }
                Err(_) => {
                    metrics
                        .requests_total
                        .with_label_values(&[&method, &ERROR_STATUS.to_string()])
                        .inc();
                }
            }

            result
        })
    }
}