allstak 0.2.2

Native Rust SDK for AllStak: error monitoring, performance tracing, structured logging, and release-health sessions over the AllStak ingest protocol.
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
147
148
149
150

//! actix-web integration.
//!
//! [`Allstak`] is a wrap-able middleware transform. For each request it binds a
//! fresh per-request [`Hub`] (cloned from the top of the stack), continues an
//! incoming distributed trace, records the request to
//! `/ingest/v1/http-requests` (using the matched route pattern as the path),
//! and marks the session crashed on a 5xx response.

use std::future::{ready, Future, Ready};
use std::pin::Pin;
use std::rc::Rc;
use std::task::{Context as TaskContext, Poll};
use std::time::Instant;

use actix_web::dev::{Service, ServiceRequest, ServiceResponse, Transform};
use actix_web::Error;

use crate::hub::Hub;
use crate::propagation;
use crate::protocol::HttpRequestRecord;
use crate::util;

/// The actix-web middleware transform. Add with `.wrap(Allstak::new())`.
#[derive(Clone, Default)]
pub struct Allstak {
    capture_server_errors: bool,
}

impl Allstak {
    /// New middleware that captures 5xx responses.
    pub fn new() -> Self {
        Allstak {
            capture_server_errors: true,
        }
    }

    /// Toggle whether 5xx responses mark the session crashed.
    pub fn capture_server_errors(mut self, enable: bool) -> Self {
        self.capture_server_errors = enable;
        self
    }
}

impl<S, B> Transform<S, ServiceRequest> for Allstak
where
    S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error> + 'static,
    B: 'static,
{
    type Response = ServiceResponse<B>;
    type Error = Error;
    type Transform = AllstakMiddleware<S>;
    type InitError = ();
    type Future = Ready<Result<Self::Transform, Self::InitError>>;

    fn new_transform(&self, service: S) -> Self::Future {
        ready(Ok(AllstakMiddleware {
            service: Rc::new(service),
            capture_server_errors: self.capture_server_errors,
        }))
    }
}

/// The middleware service produced by [`Allstak`].
pub struct AllstakMiddleware<S> {
    service: Rc<S>,
    capture_server_errors: bool,
}

impl<S, B> Service<ServiceRequest> for AllstakMiddleware<S>
where
    S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error> + 'static,
    B: 'static,
{
    type Response = ServiceResponse<B>;
    type Error = Error;
    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>>>>;

    fn poll_ready(&self, cx: &mut TaskContext<'_>) -> Poll<Result<(), Self::Error>> {
        self.service.poll_ready(cx)
    }

    fn call(&self, req: ServiceRequest) -> Self::Future {
        let hub = Hub::new_from_top(&Hub::current());

        let method = req.method().to_string();
        let host = req.connection_info().host().to_string();
        let path = req
            .match_pattern()
            .unwrap_or_else(|| req.path().to_string());
        let req_id = req
            .headers()
            .get("x-request-id")
            .or_else(|| req.headers().get("x-allstak-request-id"))
            .and_then(|v| v.to_str().ok())
            .map(|s| s.to_string());

        let trace = {
            let headers = req.headers();
            propagation::extract(|name| headers.get(name).and_then(|v| v.to_str().ok()))
        };
        if let Some(tid) = &trace.trace_id {
            hub.configure_scope(|scope| scope.set_trace_id(Some(tid.clone())));
        }
        if let Some(rid) = &req_id {
            hub.configure_scope(|scope| scope.set_request_id(Some(rid.clone())));
        }

        let started = Instant::now();
        let service = self.service.clone();
        let capture_errors = self.capture_server_errors;
        let hub2 = hub.clone();
        let trace_id = trace.trace_id.clone();

        Box::pin(async move {
            let fut = service.call(req);
            let res = fut.await;

            let status_code = match &res {
                Ok(r) => r.status().as_u16(),
                Err(_) => 500,
            };
            let duration_ms = started.elapsed().as_millis() as u64;

            if capture_errors && status_code >= 500 {
                hub2.mark_session_crashed();
            }

            let record = HttpRequestRecord {
                trace_id,
                request_id: req_id,
                direction: "inbound".to_string(),
                method,
                host,
                path,
                status_code,
                duration_ms,
                request_size: None,
                response_size: None,
                user_id: None,
                error_fingerprint: None,
                timestamp: util::now_iso8601(),
            };
            if let Some(client) = hub2.client() {
                client.capture_http_request(record);
            }

            res
        })
    }
}