8151c6e
use std::future::{Future, Ready, ready};
use std::pin::Pin;
use std::rc::Rc;
use std::sync::OnceLock;
use std::time::Instant;
use actix_web::HttpMessage;
use actix_web::body::MessageBody;
use actix_web::dev::{Service, ServiceRequest, ServiceResponse, Transform, forward_ready};
use opentelemetry::metrics::{Counter, Histogram};
use opentelemetry::trace::Status;
use opentelemetry::{KeyValue, global};
use tracing::Instrument;
use tracing_opentelemetry::OpenTelemetrySpanExt;
use crate::{TRACEPARENT, set_parent_from_traceparent};
pub struct HttpTelemetry {
scope: &'static str,
}
impl HttpTelemetry {
pub fn new(service: impl AsRef<str>) -> Self {
Self {
scope: Box::leak(format!("{}.http", service.as_ref()).into_boxed_str()),
}
}
}
impl<S, B> Transform<S, ServiceRequest> for HttpTelemetry
where
S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = actix_web::Error> + 'static,
S::Future: 'static,
B: MessageBody + 'static,
{
type Response = ServiceResponse<B>;
type Error = actix_web::Error;
type Transform = HttpTelemetryMiddleware<S>;
type InitError = ();
type Future = Ready<Result<Self::Transform, Self::InitError>>;
fn new_transform(&self, service: S) -> Self::Future {
ready(Ok(HttpTelemetryMiddleware {
service: Rc::new(service),
metrics: metrics(self.scope),
}))
}
}
pub struct HttpTelemetryMiddleware<S> {
service: Rc<S>,
metrics: &'static HttpMetrics,
}
impl<S, B> Service<ServiceRequest> for HttpTelemetryMiddleware<S>
where
S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = actix_web::Error> + 'static,
S::Future: 'static,
B: MessageBody + 'static,
{
type Response = ServiceResponse<B>;
type Error = actix_web::Error;
#[allow(clippy::type_complexity)]
type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>>>>;
forward_ready!(service);
fn call(&self, request: ServiceRequest) -> Self::Future {
let service = Rc::clone(&self.service);
let metrics = self.metrics;
let method = request.method().as_str().to_owned();
let path = request.path().to_owned();
let request_id = envelope::context::request_id()
.or_else(|| {
request
.extensions()
.get::<session::RequestId>()
.map(|value| value.0.clone())
})
.filter(|value| !value.is_empty())
.unwrap_or_default();
let traceparent = request
.headers()
.get(TRACEPARENT)
.and_then(|value| value.to_str().ok())
.map(str::to_owned);
let span = tracing::info_span!(
"http.server.request",
"otel.kind" = "server",
"http.request.method" = %method,
"url.path" = %path,
"http.response.status_code" = tracing::field::Empty,
request_id = %request_id,
);
set_parent_from_traceparent(&span, traceparent.as_deref());
let instrumented = span.clone();
let future = async move {
let started = Instant::now();
let result = service.call(request).await;
let elapsed = started.elapsed().as_secs_f64();
let status = match &result {
Ok(response) => response.status().as_u16(),
Err(error) => error.as_response_error().status_code().as_u16(),
};
span.record("http.response.status_code", status);
if status >= 500 {
span.set_status(Status::error(format!("http {status}")));
}
log(method.as_str(), path.as_str(), status, elapsed, &request_id);
let attributes = [
KeyValue::new("http.request.method", method),
KeyValue::new("http.response.status_code", i64::from(status)),
];
metrics.requests.add(1, &attributes);
metrics.duration.record(elapsed, &attributes);
result
};
Box::pin(future.instrument(instrumented))
}
}
struct HttpMetrics {
requests: Counter<u64>,
duration: Histogram<f64>,
}
/// 静态资源单独降级：一次页面加载会带出几十个 chunk，按 info 打印会淹没请求日志。
/// 探针路径：编排器高频轮询，降到 DEBUG 以免淹没业务日志。
const QUIET_PREFIX: &str = "/assets/";
const PROBE_PATHS: [&str; 2] = ["/healthz", "/readyz"];
fn log(method: &str, path: &str, status: u16, elapsed: f64, request_id: &str) {
let elapsed_ms = ((elapsed * 1_000.0) * 1_000.0).round() / 1_000.0;
let request_id = if request_id.is_empty() {
"-"
} else {
request_id
};
if path.starts_with(QUIET_PREFIX) || PROBE_PATHS.contains(&path) {
tracing::debug!(
"http.request.method" = method,
"url.path" = path,
"http.response.status_code" = status,
"duration_ms" = elapsed_ms,
"request_id" = request_id,
"http request"
);
} else {
tracing::info!(
"http.request.method" = method,
"url.path" = path,
"http.response.status_code" = status,
"duration_ms" = elapsed_ms,
"request_id" = request_id,
"http request"
);
}
}
fn metrics(scope: &'static str) -> &'static HttpMetrics {
static METRICS: OnceLock<HttpMetrics> = OnceLock::new();
METRICS.get_or_init(|| {
let meter = global::meter(scope);
HttpMetrics {
requests: meter.u64_counter("http.server.request.count").build(),
duration: meter
.f64_histogram("http.server.request.duration")
.with_unit("s")
.build(),
}
})
}