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_or_else(aster_forge_metrics::NoopMetrics::arc, |data| {
100            data.get_ref().clone()
101        })
102}
103
104fn route_label(req: &ServiceRequest) -> String {
105    req.match_pattern().unwrap_or_else(|| unmatched_route(req))
106}
107
108fn unmatched_route(req: &ServiceRequest) -> String {
109    let path = req.path();
110    if path.starts_with("/api/") {
111        "unmatched_api".to_string()
112    } else if path.starts_with("/health") {
113        "unmatched_health".to_string()
114    } else {
115        "unmatched".to_string()
116    }
117}
118
119#[cfg(test)]
120mod tests {
121    use std::sync::{Arc, Mutex};
122
123    use actix_web::{App, HttpResponse, error, test as actix_test, web};
124    use aster_forge_metrics::{MetricsRecorder, SharedMetricsRecorder};
125
126    use super::{MetricsMiddleware, unmatched_route};
127
128    #[derive(Clone, Debug, PartialEq)]
129    struct HttpMetricRecord {
130        method: String,
131        route: String,
132        status: u16,
133        duration_seconds: f64,
134    }
135
136    struct RecordingMetrics {
137        enabled: bool,
138        records: Mutex<Vec<HttpMetricRecord>>,
139    }
140
141    impl RecordingMetrics {
142        fn enabled() -> Arc<Self> {
143            Arc::new(Self {
144                enabled: true,
145                records: Mutex::new(Vec::new()),
146            })
147        }
148
149        fn disabled() -> Arc<Self> {
150            Arc::new(Self {
151                enabled: false,
152                records: Mutex::new(Vec::new()),
153            })
154        }
155
156        fn shared(self: &Arc<Self>) -> SharedMetricsRecorder {
157            self.clone()
158        }
159
160        fn records(&self) -> Vec<HttpMetricRecord> {
161            self.records.lock().expect("metrics records lock").clone()
162        }
163    }
164
165    impl aster_forge_metrics::DbMetricsRecorder for RecordingMetrics {
166        fn enabled(&self) -> bool {
167            self.enabled
168        }
169
170        fn record_db_query(&self, _metric: &aster_forge_metrics::DbQueryMetric) {}
171    }
172
173    impl MetricsRecorder for RecordingMetrics {
174        fn record_http_request(
175            &self,
176            method: &str,
177            route: &str,
178            status: u16,
179            duration_seconds: f64,
180        ) {
181            self.records
182                .lock()
183                .expect("metrics records lock")
184                .push(HttpMetricRecord {
185                    method: method.to_string(),
186                    route: route.to_string(),
187                    status,
188                    duration_seconds,
189                });
190        }
191    }
192
193    #[test]
194    fn unmatched_route_groups_unknown_paths() {
195        let api = actix_test::TestRequest::get()
196            .uri("/api/v1/missing")
197            .to_srv_request();
198        let health = actix_test::TestRequest::get()
199            .uri("/health/full")
200            .to_srv_request();
201        let other = actix_test::TestRequest::get()
202            .uri("/missing")
203            .to_srv_request();
204
205        assert_eq!(unmatched_route(&api), "unmatched_api");
206        assert_eq!(unmatched_route(&health), "unmatched_health");
207        assert_eq!(unmatched_route(&other), "unmatched");
208    }
209
210    #[actix_web::test]
211    async fn middleware_records_successful_requests_when_metrics_are_enabled() {
212        let metrics = RecordingMetrics::enabled();
213        let app = actix_test::init_service(
214            App::new()
215                .app_data(web::Data::new(metrics.shared()))
216                .wrap(MetricsMiddleware)
217                .route(
218                    "/api/v1/profiles/{id}",
219                    web::get().to(|| async { HttpResponse::Created().finish() }),
220                ),
221        )
222        .await;
223
224        let req = actix_test::TestRequest::get()
225            .uri("/api/v1/profiles/42")
226            .to_request();
227        let resp = actix_test::call_service(&app, req).await;
228        assert_eq!(resp.status(), 201);
229
230        let records = metrics.records();
231        assert_eq!(records.len(), 1);
232        assert_eq!(records[0].method, "GET");
233        assert_eq!(records[0].route, "/api/v1/profiles/{id}");
234        assert_eq!(records[0].status, 201);
235        assert!(records[0].duration_seconds >= 0.0);
236    }
237
238    #[actix_web::test]
239    async fn middleware_accepts_single_arc_trait_object_app_data() {
240        let metrics = RecordingMetrics::enabled();
241        let shared = web::Data::<dyn MetricsRecorder>::from(metrics.shared());
242        let app =
243            actix_test::init_service(App::new().app_data(shared).wrap(MetricsMiddleware).route(
244                "/api/v1/profiles/{id}",
245                web::get().to(|| async { HttpResponse::Accepted().finish() }),
246            ))
247            .await;
248
249        let req = actix_test::TestRequest::get()
250            .uri("/api/v1/profiles/42")
251            .to_request();
252        let resp = actix_test::call_service(&app, req).await;
253        assert_eq!(resp.status(), 202);
254
255        let records = metrics.records();
256        assert_eq!(records.len(), 1);
257        assert_eq!(records[0].route, "/api/v1/profiles/{id}");
258        assert_eq!(records[0].status, 202);
259    }
260
261    #[actix_web::test]
262    async fn middleware_records_error_responses_when_metrics_are_enabled() {
263        let metrics = RecordingMetrics::enabled();
264        let app = actix_test::init_service(
265            App::new()
266                .app_data(web::Data::new(metrics.shared()))
267                .wrap(MetricsMiddleware)
268                .route(
269                    "/api/v1/fails",
270                    web::get().to(|| async {
271                        Err::<HttpResponse, _>(error::ErrorBadRequest("bad request"))
272                    }),
273                ),
274        )
275        .await;
276
277        let req = actix_test::TestRequest::get()
278            .uri("/api/v1/fails")
279            .to_request();
280        let resp = actix_test::call_service(&app, req).await;
281        assert_eq!(resp.status(), 400);
282
283        let records = metrics.records();
284        assert_eq!(records.len(), 1);
285        assert_eq!(records[0].method, "GET");
286        assert_eq!(records[0].route, "/api/v1/fails");
287        assert_eq!(records[0].status, 400);
288    }
289
290    #[actix_web::test]
291    async fn middleware_skips_recording_when_metrics_are_disabled() {
292        let metrics = RecordingMetrics::disabled();
293        let app = actix_test::init_service(
294            App::new()
295                .app_data(web::Data::new(metrics.shared()))
296                .wrap(MetricsMiddleware)
297                .route(
298                    "/health",
299                    web::get().to(|| async { HttpResponse::Ok().finish() }),
300                ),
301        )
302        .await;
303
304        let req = actix_test::TestRequest::get().uri("/health").to_request();
305        assert_eq!(actix_test::call_service(&app, req).await.status(), 200);
306        assert!(metrics.records().is_empty());
307    }
308}
aster_forge_actix_middleware/metrics.rs

aster_forge_actix_middleware/
metrics.rs
