#![deny(missing_docs)]
use actix_web::http::Uri;
use log::warn;
use metrics::{describe_gauge, describe_histogram, gauge, histogram, Unit};
use std::collections::{HashMap, HashSet};
use std::future::{ready, Future, Ready};
use std::marker::PhantomData;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::Instant;
use actix_web::{
body::{BodySize, MessageBody},
dev::{self, Service, ServiceRequest, ServiceResponse, Transform},
http::{Method, StatusCode, Version},
web::Bytes,
Error, HttpMessage,
};
use futures_core::ready;
use pin_project_lite::pin_project;
use regex::RegexSet;
use strfmt::strfmt;
#[derive(Debug, Clone)]
pub struct ActixWebMetricsExtension {
pub cardinality_keep_params: Vec<String>,
}
#[derive(Debug)]
pub struct ActixWebMetricsBuilder {
namespace: Option<String>,
const_labels: HashMap<String, String>,
exclude: HashSet<String>,
exclude_regex: RegexSet,
exclude_status: HashSet<StatusCode>,
unmatched_patterns_mask: Option<String>,
metrics_config: ActixWebMetricsConfig,
}
impl ActixWebMetricsBuilder {
pub fn new() -> Self {
Self {
namespace: None,
const_labels: HashMap::new(),
exclude: HashSet::new(),
exclude_regex: RegexSet::empty(),
exclude_status: HashSet::new(),
unmatched_patterns_mask: Some("UNKNOWN".to_string()),
metrics_config: ActixWebMetricsConfig::default(),
}
}
pub fn const_labels(mut self, value: HashMap<String, String>) -> Self {
self.const_labels = value;
self
}
pub fn namespace<T: Into<String>>(mut self, value: T) -> Self {
self.namespace = Some(value.into());
self
}
pub fn exclude<T: Into<String>>(mut self, path: T) -> Self {
self.exclude.insert(path.into());
self
}
pub fn exclude_regex<T: Into<String>>(mut self, path: T) -> Self {
let mut patterns = self.exclude_regex.patterns().to_vec();
patterns.push(path.into());
self.exclude_regex = RegexSet::new(patterns).unwrap();
self
}
pub fn exclude_status<T: Into<StatusCode>>(mut self, status: T) -> Self {
self.exclude_status.insert(status.into());
self
}
pub fn mask_unmatched_patterns<T: Into<String>>(mut self, mask: T) -> Self {
self.unmatched_patterns_mask = Some(mask.into());
self
}
pub fn disable_unmatched_pattern_masking(mut self) -> Self {
self.unmatched_patterns_mask = None;
self
}
pub fn metrics_config(mut self, value: ActixWebMetricsConfig) -> Self {
self.metrics_config = value;
self
}
pub fn build(self) -> ActixWebMetrics {
let namespace_prefix = if let Some(ns) = self.namespace {
format!("{ns}_")
} else {
"".to_string()
};
let http_server_request_duration_name = format!(
"{namespace_prefix}{}",
self.metrics_config.http_server_request_duration_name
);
describe_histogram!(
http_server_request_duration_name.clone(),
Unit::Seconds,
"HTTP request duration in seconds for all requests"
);
let http_server_request_body_size_name = format!(
"{namespace_prefix}{}",
self.metrics_config.http_server_request_body_size_name
);
describe_histogram!(
http_server_request_body_size_name.clone(),
Unit::Bytes,
"HTTP request size in bytes for all requests"
);
let http_server_response_body_size_name = format!(
"{namespace_prefix}{}",
self.metrics_config.http_server_response_body_size_name
);
describe_histogram!(
http_server_response_body_size_name.clone(),
Unit::Bytes,
"HTTP response size in bytes for all requests"
);
let http_server_active_requests_name = format!(
"{namespace_prefix}{}",
self.metrics_config.http_server_active_requests_name
);
describe_gauge!(
http_server_active_requests_name.clone(),
"Number of active HTTP server requests."
);
let mut const_labels: Vec<(&'static str, String)> = self
.const_labels
.iter()
.map(|(k, v)| {
let k: &'static str = Box::leak(Box::new(k.clone()));
(k, v.clone())
})
.collect();
const_labels.sort_by_key(|v| v.0);
ActixWebMetrics {
inner: Arc::new(ActixWebMetricsInner {
exclude: self.exclude,
exclude_regex: self.exclude_regex,
exclude_status: self.exclude_status,
unmatched_patterns_mask: self.unmatched_patterns_mask,
names: MetricsMetadata {
http_server_request_duration: Box::leak(Box::new(
http_server_request_duration_name,
)),
http_server_request_body_size: Box::leak(Box::new(
http_server_request_body_size_name,
)),
http_server_response_body_size: Box::leak(Box::new(
http_server_response_body_size_name,
)),
http_server_active_requests: Box::leak(Box::new(
http_server_active_requests_name,
)),
http_route: Box::leak(Box::new(self.metrics_config.labels.http_route)),
http_request_method: Box::leak(Box::new(
self.metrics_config.labels.http_request_method,
)),
http_response_status_code: Box::leak(Box::new(
self.metrics_config.labels.http_response_status_code,
)),
network_protocol_name: Box::leak(Box::new(
self.metrics_config.labels.network_protocol_name,
)),
network_protocol_version: Box::leak(Box::new(
self.metrics_config.labels.network_protocol_version,
)),
url_scheme: Box::leak(Box::new(self.metrics_config.labels.url_scheme)),
const_labels,
},
}),
}
}
}
impl Default for ActixWebMetricsBuilder {
fn default() -> Self {
Self::new()
}
}
#[derive(Debug, Clone)]
pub struct LabelsConfig {
http_route: String,
http_request_method: String,
http_response_status_code: String,
network_protocol_name: String,
network_protocol_version: String,
url_scheme: String,
}
impl Default for LabelsConfig {
fn default() -> Self {
Self {
http_route: String::from("http.route"),
http_request_method: String::from("http.request.method"),
http_response_status_code: String::from("http.response.status_code"),
network_protocol_name: String::from("network.protocol.name"),
network_protocol_version: String::from("network.protocol.version"),
url_scheme: String::from("url.scheme"),
}
}
}
impl LabelsConfig {
pub fn http_request_method<T: Into<String>>(mut self, name: T) -> Self {
self.http_request_method = name.into();
self
}
pub fn http_route<T: Into<String>>(mut self, name: T) -> Self {
self.http_route = name.into();
self
}
pub fn http_response_status_code<T: Into<String>>(mut self, name: T) -> Self {
self.http_response_status_code = name.into();
self
}
pub fn network_protocol_name<T: Into<String>>(mut self, name: T) -> Self {
self.network_protocol_name = name.into();
self
}
pub fn network_protocol_version<T: Into<String>>(mut self, name: T) -> Self {
self.network_protocol_version = name.into();
self
}
pub fn url_scheme<T: Into<String>>(mut self, name: T) -> Self {
self.url_scheme = name.into();
self
}
}
#[derive(Debug, Clone)]
pub struct ActixWebMetricsConfig {
http_server_request_duration_name: String,
http_server_request_body_size_name: String,
http_server_response_body_size_name: String,
http_server_active_requests_name: String,
labels: LabelsConfig,
}
impl Default for ActixWebMetricsConfig {
fn default() -> Self {
Self {
http_server_request_duration_name: String::from("http.server.request.duration"),
http_server_request_body_size_name: String::from("http.server.request.body.size"),
http_server_response_body_size_name: String::from("http.server.response.body.size"),
http_server_active_requests_name: String::from("http.server.active_requests"),
labels: LabelsConfig::default(),
}
}
}
impl ActixWebMetricsConfig {
pub fn labels(mut self, labels: LabelsConfig) -> Self {
self.labels = labels;
self
}
pub fn http_server_request_duration_name<T: Into<String>>(mut self, name: T) -> Self {
self.http_server_request_duration_name = name.into();
self
}
pub fn http_server_request_body_size_name<T: Into<String>>(mut self, name: T) -> Self {
self.http_server_request_body_size_name = name.into();
self
}
pub fn http_server_response_body_size_name<T: Into<String>>(mut self, name: T) -> Self {
self.http_server_response_body_size_name = name.into();
self
}
pub fn http_server_active_requests_name<T: Into<String>>(mut self, name: T) -> Self {
self.http_server_active_requests_name = name.into();
self
}
}
#[derive(Debug, Clone)]
struct MetricsMetadata {
http_server_request_duration: &'static str,
http_server_request_body_size: &'static str,
http_server_response_body_size: &'static str,
http_server_active_requests: &'static str,
http_route: &'static str,
http_request_method: &'static str,
http_response_status_code: &'static str,
network_protocol_name: &'static str,
network_protocol_version: &'static str,
url_scheme: &'static str,
const_labels: Vec<(&'static str, String)>,
}
#[derive(Clone)]
#[must_use = "must be set up as middleware for actix-web"]
pub struct ActixWebMetrics {
inner: Arc<ActixWebMetricsInner>,
}
struct ActixWebMetricsInner {
pub(crate) names: MetricsMetadata,
pub(crate) exclude: HashSet<String>,
pub(crate) exclude_regex: RegexSet,
pub(crate) exclude_status: HashSet<StatusCode>,
pub(crate) unmatched_patterns_mask: Option<String>,
}
impl ActixWebMetrics {
fn pre_request_update_metrics(&self, req: &ServiceRequest) {
let this = &*self.inner;
let mut labels = Vec::with_capacity(2 + this.names.const_labels.len());
labels.push((
this.names.http_request_method,
req.method().as_str().to_string(),
));
labels.push((this.names.url_scheme, url_scheme(&req.uri()).to_string()));
for (k, v) in &this.names.const_labels {
labels.push((k, v.clone()));
}
gauge!(this.names.http_server_active_requests, &labels).increment(1);
}
#[allow(clippy::too_many_arguments)]
fn post_request_update_metrics(
&self,
http_version: Version,
mixed_pattern: &str,
fallback_pattern: &str,
method: &Method,
status: StatusCode,
scheme: &str,
clock: Instant,
was_path_matched: bool,
request_size: usize,
response_size: usize,
) {
let this = &*self.inner;
{
let mut active_request_labels = Vec::with_capacity(2 + this.names.const_labels.len());
active_request_labels
.push((this.names.http_request_method, method.as_str().to_string()));
active_request_labels.push((this.names.url_scheme, scheme.to_string()));
for (k, v) in &this.names.const_labels {
active_request_labels.push((k, v.clone()));
}
gauge!(
this.names.http_server_active_requests,
&active_request_labels
)
.decrement(1);
}
if this.exclude.contains(mixed_pattern)
|| this.exclude_regex.is_match(mixed_pattern)
|| this.exclude_status.contains(&status)
{
return;
}
let final_pattern = if fallback_pattern != mixed_pattern && (status == 404 || status == 405)
{
fallback_pattern
} else {
mixed_pattern
};
let final_pattern = if was_path_matched {
final_pattern
} else if let Some(mask) = &this.unmatched_patterns_mask {
mask
} else {
final_pattern
};
let mut labels = Vec::with_capacity(5 + this.names.const_labels.len());
labels.push((this.names.http_route, final_pattern.to_string()));
labels.push((this.names.http_request_method, method.as_str().to_string()));
labels.push((
this.names.http_response_status_code,
status.as_str().to_string(),
));
labels.push((this.names.network_protocol_name, "http".to_string()));
if let Some(http_version) = Self::http_version_label(http_version) {
labels.push((
this.names.network_protocol_version,
http_version.to_string(),
));
}
for (k, v) in &this.names.const_labels {
labels.push((k, v.clone()));
}
let elapsed = clock.elapsed();
let duration =
(elapsed.as_secs() as f64) + f64::from(elapsed.subsec_nanos()) / 1_000_000_000_f64;
histogram!(this.names.http_server_request_duration, &labels).record(duration);
histogram!(this.names.http_server_request_body_size, &labels).record(request_size as f64);
histogram!(this.names.http_server_response_body_size, &labels).record(response_size as f64);
}
fn http_version_label(version: Version) -> Option<&'static str> {
let v = match version {
v if v == Version::HTTP_09 => "0.9",
v if v == Version::HTTP_10 => "1.0",
v if v == Version::HTTP_11 => "1.1",
v if v == Version::HTTP_2 => "2",
v if v == Version::HTTP_3 => "3",
_ => return None,
};
Some(v)
}
}
impl<S, B> Transform<S, ServiceRequest> for ActixWebMetrics
where
S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error>,
{
type Response = ServiceResponse<StreamLog<B>>;
type Error = Error;
type InitError = ();
type Transform = MetricsMiddleware<S>;
type Future = Ready<Result<Self::Transform, Self::InitError>>;
fn new_transform(&self, service: S) -> Self::Future {
ready(Ok(MetricsMiddleware {
service,
inner: self.clone(),
}))
}
}
pin_project! {
#[doc(hidden)]
pub struct LoggerResponse<S>
where
S: Service<ServiceRequest>,
{
#[pin]
fut: S::Future,
time: Instant,
inner: ActixWebMetrics,
_t: PhantomData<()>,
}
}
impl<S, B> Future for LoggerResponse<S>
where
S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error>,
{
type Output = Result<ServiceResponse<StreamLog<B>>, Error>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.project();
let res = match ready!(this.fut.poll(cx)) {
Ok(res) => res,
Err(e) => return Poll::Ready(Err(e)),
};
let time = *this.time;
let req = res.request();
let method = req.method().clone();
let version = req.version();
let was_path_matched = req.match_pattern().is_some();
let params_keep_path_cardinality =
match req.extensions_mut().get::<ActixWebMetricsExtension>() {
Some(config) => config.cardinality_keep_params.clone(),
None => vec![],
};
let full_pattern = req.match_pattern();
let path = req.path().to_string();
let fallback_pattern = full_pattern.clone().unwrap_or(path.clone());
let mixed_pattern = match full_pattern {
None => path.clone(),
Some(full_pattern) => {
let mut params: HashMap<String, String> = HashMap::new();
for (key, val) in req.match_info().iter() {
if params_keep_path_cardinality.contains(&key.to_string()) {
params.insert(key.to_string(), val.to_string());
continue;
}
params.insert(key.to_string(), format!("{{{key}}}"));
}
if let Ok(mixed_cardinality_pattern) = strfmt(&full_pattern, ¶ms) {
mixed_cardinality_pattern
} else {
warn!("Cannot build mixed cardinality pattern {full_pattern}, with params {params:?}");
full_pattern
}
}
};
let request_size = req
.headers()
.get("content-length")
.and_then(|v| v.to_str().ok())
.and_then(|v| v.parse::<usize>().ok())
.unwrap_or(0);
let scheme = url_scheme(&req.uri()).to_string();
let inner = this.inner.clone();
Poll::Ready(Ok(res.map_body(move |head, body| StreamLog {
body,
response_size: 0,
request_size,
clock: time,
inner,
status: head.status,
scheme,
mixed_pattern,
fallback_pattern,
method,
version,
was_path_matched,
})))
}
}
#[doc(hidden)]
pub struct MetricsMiddleware<S> {
service: S,
inner: ActixWebMetrics,
}
impl<S, B> Service<ServiceRequest> for MetricsMiddleware<S>
where
S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error>,
{
type Response = ServiceResponse<StreamLog<B>>;
type Error = S::Error;
type Future = LoggerResponse<S>;
dev::forward_ready!(service);
fn call(&self, req: ServiceRequest) -> Self::Future {
self.inner.pre_request_update_metrics(&req);
LoggerResponse {
fut: self.service.call(req),
time: Instant::now(),
inner: self.inner.clone(),
_t: PhantomData,
}
}
}
pin_project! {
#[doc(hidden)]
pub struct StreamLog<B> {
#[pin]
body: B,
response_size: usize,
request_size: usize,
clock: Instant,
inner: ActixWebMetrics,
status: StatusCode,
scheme: String,
mixed_pattern: String,
fallback_pattern: String,
method: Method,
version: Version,
was_path_matched: bool
}
impl<B> PinnedDrop for StreamLog<B> {
fn drop(this: Pin<&mut Self>) {
this.inner
.post_request_update_metrics(this.version, &this.mixed_pattern, &this.fallback_pattern, &this.method, this.status, &this.scheme, this.clock, this.was_path_matched, this.request_size, this.response_size);
}
}
}
impl<B: MessageBody> MessageBody for StreamLog<B> {
type Error = B::Error;
fn size(&self) -> BodySize {
self.body.size()
}
fn poll_next(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Bytes, Self::Error>>> {
let this = self.project();
match ready!(this.body.poll_next(cx)) {
Some(Ok(chunk)) => {
*this.response_size += chunk.len();
Poll::Ready(Some(Ok(chunk)))
}
Some(Err(err)) => Poll::Ready(Some(Err(err))),
None => Poll::Ready(None),
}
}
}
fn url_scheme(uri: &Uri) -> &str {
uri.scheme().map(|s| s.as_str()).unwrap_or("http")
}