aster_forge_actix_middleware/
metrics.rs

1//! HTTP request metrics middleware for Actix Web services.
2//!
3//! The middleware reads a shared [`aster_forge_metrics::SharedMetricsRecorder`] from Actix app
4//! data, records one HTTP request metric per successful or failed downstream service call, and
5//! groups unmatched routes into stable low-cardinality labels.
6
7use std::rc::Rc;
8use std::time::Instant;
9
10use actix_web::{
11    Error,
12    dev::{Service, ServiceRequest, ServiceResponse, Transform, forward_ready},
13    web,
14};
15use futures::future::{LocalBoxFuture, Ready, ok};
16
17use aster_forge_metrics::{MetricsRecorder, SharedMetricsRecorder};
18
19/// Actix middleware that records request duration and status into the shared metrics recorder.
20pub struct MetricsMiddleware;
21
22impl<S, B> Transform<S, ServiceRequest> for MetricsMiddleware
23where
24    S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error> + 'static,
25    B: 'static,
26{
27    type Response = ServiceResponse<B>;
28    type Error = Error;
29    type InitError = ();
30    type Transform = MetricsService<S>;
31    type Future = Ready<Result<Self::Transform, Self::InitError>>;
32
33    fn new_transform(&self, service: S) -> Self::Future {
34        ok(MetricsService {
35            service: Rc::new(service),
36        })
37    }
38}
39
40/// Service wrapper installed by [`MetricsMiddleware`].
41pub struct MetricsService<S> {
42    service: Rc<S>,
43}
44
45impl<S, B> Service<ServiceRequest> for MetricsService<S>
46where
47    S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error> + 'static,
48    B: 'static,
49{
50    type Response = ServiceResponse<B>;
51    type Error = Error;
52    type Future = LocalBoxFuture<'static, Result<Self::Response, Self::Error>>;
53
54    forward_ready!(service);
55
56    fn call(&self, req: ServiceRequest) -> Self::Future {
57        let svc = self.service.clone();
58        let metrics = request_metrics(&req);
59
60        if !metrics.enabled() {
61            return Box::pin(async move { svc.call(req).await });
62        }
63
64        let started_at = Instant::now();
65        let method = req.method().clone();
66        let route = route_label(&req);
67
68        Box::pin(async move {
69            match svc.call(req).await {
70                Ok(resp) => {
71                    metrics.record_http_request(
72                        method.as_str(),
73                        &route,
74                        resp.status().as_u16(),
75                        started_at.elapsed().as_secs_f64(),
76                    );
77                    Ok(resp)
78                }
79                Err(error) => {
80                    metrics.record_http_request(
81                        method.as_str(),
82                        &route,
83                        error.as_response_error().status_code().as_u16(),
84                        started_at.elapsed().as_secs_f64(),
85                    );
86                    Err(error)
87                }
88            }
89        })
90    }
91}
92
93fn request_metrics(req: &ServiceRequest) -> SharedMetricsRecorder {
94    if let Some(metrics) = req.app_data::<web::Data<dyn MetricsRecorder>>() {
95        return metrics.clone().into_inner();
96    }
97
98    req.app_data::<web::Data<SharedMetricsRecorder>>()
99        .map(|data| data.get_ref().clone())
100        .unwrap_or_else(aster_forge_metrics::NoopMetrics::arc)
101}
102
103fn route_label(req: &ServiceRequest) -> String {
104    req.match_pattern().unwrap_or_else(|| unmatched_route(req))
105}
106
107fn unmatched_route(req: &ServiceRequest) -> String {
108    let path = req.path();
109    if path.starts_with("/api/") {
110        "unmatched_api".to_string()
111    } else if path.starts_with("/health") {
112        "unmatched_health".to_string()
113    } else {
114        "unmatched".to_string()
115    }
116}
117
118#[cfg(test)]
119mod tests {
120    use std::sync::{Arc, Mutex};
121
122    use actix_web::{App, HttpResponse, error, test as actix_test, web};
123    use aster_forge_metrics::{MetricsRecorder, SharedMetricsRecorder};
124
125    use super::{MetricsMiddleware, unmatched_route};
126
127    #[derive(Clone, Debug, PartialEq)]
128    struct HttpMetricRecord {
129        method: String,
130        route: String,
131        status: u16,
132        duration_seconds: f64,
133    }
134
135    struct RecordingMetrics {
136        enabled: bool,
137        records: Mutex<Vec<HttpMetricRecord>>,
138    }
139
140    impl RecordingMetrics {
141        fn enabled() -> Arc<Self> {
142            Arc::new(Self {
143                enabled: true,
144                records: Mutex::new(Vec::new()),
145            })
146        }
147
148        fn disabled() -> Arc<Self> {
149            Arc::new(Self {
150                enabled: false,
151                records: Mutex::new(Vec::new()),
152            })
153        }
154
155        fn shared(self: &Arc<Self>) -> SharedMetricsRecorder {
156            self.clone()
157        }
158
159        fn records(&self) -> Vec<HttpMetricRecord> {
160            self.records.lock().expect("metrics records lock").clone()
161        }
162    }
163
164    impl aster_forge_metrics::DbMetricsRecorder for RecordingMetrics {
165        fn enabled(&self) -> bool {
166            self.enabled
167        }
168
169        fn record_db_query(&self, _metric: &aster_forge_metrics::DbQueryMetric) {}
170    }
171
172    impl MetricsRecorder for RecordingMetrics {
173        fn record_http_request(
174            &self,
175            method: &str,
176            route: &str,
177            status: u16,
178            duration_seconds: f64,
179        ) {
180            self.records
181                .lock()
182                .expect("metrics records lock")
183                .push(HttpMetricRecord {
184                    method: method.to_string(),
185                    route: route.to_string(),
186                    status,
187                    duration_seconds,
188                });
189        }
190    }
191
192    #[test]
193    fn unmatched_route_groups_unknown_paths() {
194        let api = actix_test::TestRequest::get()
195            .uri("/api/v1/missing")
196            .to_srv_request();
197        let health = actix_test::TestRequest::get()
198            .uri("/health/full")
199            .to_srv_request();
200        let other = actix_test::TestRequest::get()
201            .uri("/missing")
202            .to_srv_request();
203
204        assert_eq!(unmatched_route(&api), "unmatched_api");
205        assert_eq!(unmatched_route(&health), "unmatched_health");
206        assert_eq!(unmatched_route(&other), "unmatched");
207    }
208
209    #[actix_web::test]
210    async fn middleware_records_successful_requests_when_metrics_are_enabled() {
211        let metrics = RecordingMetrics::enabled();
212        let app = actix_test::init_service(
213            App::new()
214                .app_data(web::Data::new(metrics.shared()))
215                .wrap(MetricsMiddleware)
216                .route(
217                    "/api/v1/profiles/{id}",
218                    web::get().to(|| async { HttpResponse::Created().finish() }),
219                ),
220        )
221        .await;
222
223        let req = actix_test::TestRequest::get()
224            .uri("/api/v1/profiles/42")
225            .to_request();
226        let resp = actix_test::call_service(&app, req).await;
227        assert_eq!(resp.status(), 201);
228
229        let records = metrics.records();
230        assert_eq!(records.len(), 1);
231        assert_eq!(records[0].method, "GET");
232        assert_eq!(records[0].route, "/api/v1/profiles/{id}");
233        assert_eq!(records[0].status, 201);
234        assert!(records[0].duration_seconds >= 0.0);
235    }
236
237    #[actix_web::test]
238    async fn middleware_accepts_single_arc_trait_object_app_data() {
239        let metrics = RecordingMetrics::enabled();
240        let shared = web::Data::<dyn MetricsRecorder>::from(metrics.shared());
241        let app =
242            actix_test::init_service(App::new().app_data(shared).wrap(MetricsMiddleware).route(
243                "/api/v1/profiles/{id}",
244                web::get().to(|| async { HttpResponse::Accepted().finish() }),
245            ))
246            .await;
247
248        let req = actix_test::TestRequest::get()
249            .uri("/api/v1/profiles/42")
250            .to_request();
251        let resp = actix_test::call_service(&app, req).await;
252        assert_eq!(resp.status(), 202);
253
254        let records = metrics.records();
255        assert_eq!(records.len(), 1);
256        assert_eq!(records[0].route, "/api/v1/profiles/{id}");
257        assert_eq!(records[0].status, 202);
258    }
259
260    #[actix_web::test]
261    async fn middleware_records_error_responses_when_metrics_are_enabled() {
262        let metrics = RecordingMetrics::enabled();
263        let app = actix_test::init_service(
264            App::new()
265                .app_data(web::Data::new(metrics.shared()))
266                .wrap(MetricsMiddleware)
267                .route(
268                    "/api/v1/fails",
269                    web::get().to(|| async {
270                        Err::<HttpResponse, _>(error::ErrorBadRequest("bad request"))
271                    }),
272                ),
273        )
274        .await;
275
276        let req = actix_test::TestRequest::get()
277            .uri("/api/v1/fails")
278            .to_request();
279        let resp = actix_test::call_service(&app, req).await;
280        assert_eq!(resp.status(), 400);
281
282        let records = metrics.records();
283        assert_eq!(records.len(), 1);
284        assert_eq!(records[0].method, "GET");
285        assert_eq!(records[0].route, "/api/v1/fails");
286        assert_eq!(records[0].status, 400);
287    }
288
289    #[actix_web::test]
290    async fn middleware_skips_recording_when_metrics_are_disabled() {
291        let metrics = RecordingMetrics::disabled();
292        let app = actix_test::init_service(
293            App::new()
294                .app_data(web::Data::new(metrics.shared()))
295                .wrap(MetricsMiddleware)
296                .route(
297                    "/health",
298                    web::get().to(|| async { HttpResponse::Ok().finish() }),
299                ),
300        )
301        .await;
302
303        let req = actix_test::TestRequest::get().uri("/health").to_request();
304        assert_eq!(actix_test::call_service(&app, req).await.status(), 200);
305        assert!(metrics.records().is_empty());
306    }
307}
aster_forge_actix_middleware/metrics.rs

aster_forge_actix_middleware/
metrics.rs
