use axum::extract::MatchedPath;
use http::{HeaderValue, Request, Response};
use pin_project_lite::pin_project;
use postrust_core::trace::{valid_request_id, TraceContext, TraceId};
use std::cell::Cell;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Instant;
use tower::{Layer, Service};
use tracing::field::Empty;
use tracing::Span;
#[cfg(feature = "otel")]
use crate::otel::RequestMeter;
#[cfg(not(feature = "otel"))]
struct RequestMeter;
#[cfg(not(feature = "otel"))]
impl RequestMeter {
#[inline(always)]
fn start(_: &http::Method, _: Option<&str>) -> Self {
RequestMeter
}
#[inline(always)]
fn finish(&self, _: http::StatusCode, _: Instant) {}
}
pub const REQUEST_ID_HEADER: &str = "x-request-id";
#[derive(Clone, Copy, Debug, Default)]
pub struct RequestTraceLayer;
impl<S> Layer<S> for RequestTraceLayer {
type Service = RequestTrace<S>;
fn layer(&self, inner: S) -> Self::Service {
RequestTrace { inner }
}
}
#[derive(Clone, Debug)]
pub struct RequestTrace<S> {
inner: S,
}
impl<S, B, ResBody> Service<Request<B>> for RequestTrace<S>
where
S: Service<Request<B>, Response = Response<ResBody>>,
{
type Response = Response<ResBody>;
type Error = S::Error;
type Future = ResponseFuture<S::Future>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.inner.poll_ready(cx)
}
fn call(&mut self, mut request: Request<B>) -> Self::Future {
let start = Instant::now();
let request_id = request_id(request.headers());
let trace = TraceContext::from_headers(request.headers())
.unwrap_or_else(|| TraceContext::root(new_trace_id()));
let span = tracing::info_span!(
"request",
"http.request.method" = request.method().as_str(),
"url.path" = request.uri().path(),
"http.route" = Empty,
"http.response.status_code" = Empty,
"otel.kind" = "server",
"otel.status_code" = Empty,
"otel.name" = Empty,
request_id = request_id.to_str().unwrap_or_default(),
trace_id = %trace.trace_id,
);
let route = request.extensions().get::<MatchedPath>();
if let Some(route) = route {
span.record("http.route", route.as_str());
}
let meter = RequestMeter::start(request.method(), route.map(MatchedPath::as_str));
#[cfg(feature = "otel")]
{
crate::otel::name_request_span(&span, request.method(), route.map(MatchedPath::as_str));
crate::otel::join_trace(&span, &trace);
}
request.extensions_mut().insert(trace);
let inner = {
let _entered = span.enter();
self.inner.call(request)
};
ResponseFuture {
inner,
span,
request_id: Some(request_id),
start,
meter,
}
}
}
pin_project! {
pub struct ResponseFuture<F> {
#[pin]
inner: F,
span: Span,
request_id: Option<HeaderValue>,
start: Instant,
meter: RequestMeter,
}
}
impl<F, ResBody, E> Future for ResponseFuture<F>
where
F: Future<Output = Result<Response<ResBody>, E>>,
{
type Output = Result<Response<ResBody>, E>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.project();
let _entered = this.span.enter();
let mut response = match this.inner.poll(cx) {
Poll::Pending => return Poll::Pending,
Poll::Ready(Ok(response)) => response,
Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
};
let status = response.status();
this.span
.record("http.response.status_code", status.as_u16());
if status.is_server_error() {
this.span.record("otel.status_code", "ERROR");
}
this.meter.finish(status, *this.start);
tracing::debug!(
status = status.as_u16(),
latency_ms = this.start.elapsed().as_secs_f64() * 1000.0,
"request finished"
);
if let Some(id) = this.request_id.take() {
response
.headers_mut()
.entry(REQUEST_ID_HEADER)
.or_insert(id);
}
Poll::Ready(Ok(response))
}
}
pub fn request_id(headers: &http::HeaderMap) -> HeaderValue {
let mut sent = headers.get_all(REQUEST_ID_HEADER).iter();
match (sent.next(), sent.next()) {
(Some(id), None) if valid_request_id(id.as_bytes()) => id.clone(),
_ => new_request_id(),
}
}
pub fn new_request_id() -> HeaderValue {
let mut bytes = random_u128().to_be_bytes();
bytes[6] = (bytes[6] & 0x0f) | 0x40;
bytes[8] = (bytes[8] & 0x3f) | 0x80;
const HEX: &[u8; 16] = b"0123456789abcdef";
let mut text = [0u8; 36];
let mut at = 0;
for (i, b) in bytes.iter().enumerate() {
if matches!(i, 4 | 6 | 8 | 10) {
text[at] = b'-';
at += 1;
}
text[at] = HEX[(b >> 4) as usize];
text[at + 1] = HEX[(b & 0x0f) as usize];
at += 2;
}
HeaderValue::from_bytes(&text).expect("hex digits and hyphens are a valid header value")
}
pub fn new_trace_id() -> TraceId {
loop {
if let Some(id) = TraceId::from_bytes(random_u128().to_be_bytes()) {
return id;
}
}
}
thread_local! {
static RNG_STATE: Cell<u64> = Cell::new(seed());
}
fn seed() -> u64 {
use std::hash::BuildHasher;
std::collections::hash_map::RandomState::new().hash_one(0x5eed_u64)
}
fn random_u128() -> u128 {
fn next(state: &Cell<u64>) -> u64 {
let s = state.get().wrapping_add(0x9e37_79b9_7f4a_7c15);
state.set(s);
let mut z = s;
z = (z ^ (z >> 30)).wrapping_mul(0xbf58_476d_1ce4_e5b9);
z = (z ^ (z >> 27)).wrapping_mul(0x94d0_49bb_1331_11eb);
z ^ (z >> 31)
}
RNG_STATE.with(|state| ((next(state) as u128) << 64) | next(state) as u128)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_generated_request_id_is_a_v4_uuid() {
let id = new_request_id();
let text = id.to_str().unwrap();
assert_eq!(text.len(), 36);
let parts: Vec<&str> = text.split('-').collect();
assert_eq!(
parts.iter().map(|p| p.len()).collect::<Vec<_>>(),
[8, 4, 4, 4, 12]
);
assert!(parts[2].starts_with('4'));
assert!(matches!(parts[3].as_bytes()[0], b'8' | b'9' | b'a' | b'b'));
assert!(valid_request_id(id.as_bytes()));
assert_ne!(new_request_id(), id);
}
#[test]
fn a_usable_request_id_is_kept() {
let mut headers = http::HeaderMap::new();
headers.insert(REQUEST_ID_HEADER, "abc-123".parse().unwrap());
assert_eq!(request_id(&headers), "abc-123");
}
#[test]
fn an_unusable_request_id_is_replaced() {
for bad in ["has space", &"x".repeat(129)] {
let mut headers = http::HeaderMap::new();
headers.insert(REQUEST_ID_HEADER, bad.parse().unwrap());
let id = request_id(&headers);
assert_ne!(id, bad);
assert_eq!(id.len(), 36);
}
let mut headers = http::HeaderMap::new();
headers.append(REQUEST_ID_HEADER, "one".parse().unwrap());
headers.append(REQUEST_ID_HEADER, "two".parse().unwrap());
assert_eq!(request_id(&headers).len(), 36);
}
#[test]
fn trace_ids_differ() {
assert_ne!(new_trace_id(), new_trace_id());
}
}