#![deny(missing_docs)]
use std::collections::HashMap;
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, EitherBody, MessageBody},
dev::{self, Service, ServiceRequest, ServiceResponse, Transform},
http::{
header::{HeaderValue, CONTENT_TYPE},
Method, StatusCode,
},
web::Bytes,
Error,
};
use futures_core::ready;
use pin_project_lite::pin_project;
use prometheus::{
Encoder, HistogramOpts, HistogramVec, IntCounterVec, Opts, Registry, TextEncoder,
};
#[derive(Debug)]
pub struct PrometheusMetricsBuilder {
namespace: String,
endpoint: Option<String>,
const_labels: HashMap<String, String>,
registry: Registry,
buckets: Vec<f64>,
}
impl PrometheusMetricsBuilder {
pub fn new(namespace: &str) -> Self {
Self {
namespace: namespace.into(),
endpoint: None,
const_labels: HashMap::new(),
registry: Registry::new(),
buckets: prometheus::DEFAULT_BUCKETS.to_vec(),
}
}
pub fn endpoint(mut self, value: &str) -> Self {
self.endpoint = Some(value.into());
self
}
pub fn buckets(mut self, value: &[f64]) -> Self {
self.buckets = value.to_vec();
self
}
pub fn const_labels(mut self, value: HashMap<String, String>) -> Self {
self.const_labels = value;
self
}
pub fn registry(mut self, value: Registry) -> Self {
self.registry = value;
self
}
pub fn build(self) -> Result<PrometheusMetrics, Box<dyn std::error::Error>> {
let http_requests_total_opts =
Opts::new("http_requests_total", "Total number of HTTP requests")
.namespace(&self.namespace)
.const_labels(self.const_labels.clone());
let http_requests_total =
IntCounterVec::new(http_requests_total_opts, &["endpoint", "method", "status"])?;
let http_requests_duration_seconds_opts = HistogramOpts::new(
"http_requests_duration_seconds",
"HTTP request duration in seconds for all requests",
)
.namespace(&self.namespace)
.buckets(self.buckets.to_vec())
.const_labels(self.const_labels.clone());
let http_requests_duration_seconds = HistogramVec::new(
http_requests_duration_seconds_opts,
&["endpoint", "method", "status"],
)?;
self.registry
.register(Box::new(http_requests_total.clone()))?;
self.registry
.register(Box::new(http_requests_duration_seconds.clone()))?;
Ok(PrometheusMetrics {
http_requests_total,
http_requests_duration_seconds,
registry: self.registry,
namespace: self.namespace,
endpoint: self.endpoint,
const_labels: self.const_labels,
})
}
}
#[derive(Clone)]
#[must_use = "must be set up as middleware for actix-web"]
pub struct PrometheusMetrics {
pub(crate) http_requests_total: IntCounterVec,
pub(crate) http_requests_duration_seconds: HistogramVec,
pub registry: Registry,
pub(crate) namespace: String,
pub(crate) endpoint: Option<String>,
pub(crate) const_labels: HashMap<String, String>,
}
impl PrometheusMetrics {
fn metrics(&self) -> String {
let mut buffer = vec![];
TextEncoder::new()
.encode(&self.registry.gather(), &mut buffer)
.unwrap();
String::from_utf8(buffer).unwrap()
}
fn matches(&self, path: &str, method: &Method) -> bool {
if self.endpoint.is_some() {
self.endpoint.as_ref().unwrap() == path && method == Method::GET
} else {
false
}
}
fn update_metrics(&self, path: &str, method: &Method, status: StatusCode, clock: Instant) {
let method = method.to_string();
let status = status.as_u16().to_string();
let elapsed = clock.elapsed();
let duration =
(elapsed.as_secs() as f64) + f64::from(elapsed.subsec_nanos()) / 1_000_000_000_f64;
self.http_requests_duration_seconds
.with_label_values(&[path, &method, &status])
.observe(duration);
self.http_requests_total
.with_label_values(&[path, &method, &status])
.inc();
}
}
impl<S, B> Transform<S, ServiceRequest> for PrometheusMetrics
where
S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error>,
{
type Response = ServiceResponse<EitherBody<StreamLog<B>, StreamLog<String>>>;
type Error = Error;
type InitError = ();
type Transform = PrometheusMetricsMiddleware<S>;
type Future = Ready<Result<Self::Transform, Self::InitError>>;
fn new_transform(&self, service: S) -> Self::Future {
ready(Ok(PrometheusMetricsMiddleware {
service,
inner: Arc::new(self.clone()),
}))
}
}
pin_project! {
#[doc(hidden)]
pub struct LoggerResponse<S>
where
S: Service<ServiceRequest>,
{
#[pin]
fut: S::Future,
time: Instant,
inner: Arc<PrometheusMetrics>,
_t: PhantomData<()>,
}
}
impl<S, B> Future for LoggerResponse<S>
where
S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error>,
{
type Output = Result<ServiceResponse<EitherBody<StreamLog<B>, StreamLog<String>>>, 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 pattern_or_path = req
.match_pattern()
.unwrap_or_else(|| req.path().to_string());
let path = req.path().to_string();
let inner = this.inner.clone();
Poll::Ready(Ok(res.map_body(move |mut head, body| {
if inner.matches(&path, &method) {
head.status = StatusCode::OK;
head.headers.insert(
CONTENT_TYPE,
HeaderValue::from_static("text/plain; version=0.0.4; charset=utf-8"),
);
EitherBody::right(StreamLog {
body: inner.metrics(),
size: 0,
clock: time,
inner,
status: head.status,
path: pattern_or_path,
method,
})
} else {
EitherBody::left(StreamLog {
body,
size: 0,
clock: time,
inner,
status: head.status,
path: pattern_or_path,
method,
})
}
})))
}
}
#[doc(hidden)]
pub struct PrometheusMetricsMiddleware<S> {
service: S,
inner: Arc<PrometheusMetrics>,
}
impl<S, B> Service<ServiceRequest> for PrometheusMetricsMiddleware<S>
where
S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error>,
{
type Response = ServiceResponse<EitherBody<StreamLog<B>, StreamLog<String>>>;
type Error = S::Error;
type Future = LoggerResponse<S>;
dev::forward_ready!(service);
fn call(&self, req: ServiceRequest) -> Self::Future {
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,
size: usize,
clock: Instant,
inner: Arc<PrometheusMetrics>,
status: StatusCode,
path: String,
method: Method,
}
impl<B> PinnedDrop for StreamLog<B> {
fn drop(this: Pin<&mut Self>) {
this.inner
.update_metrics(&this.path, &this.method, this.status, this.clock);
}
}
}
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.size += chunk.len();
Poll::Ready(Some(Ok(chunk)))
}
Some(Err(err)) => Poll::Ready(Some(Err(err))),
None => Poll::Ready(None),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use actix_web::test::{call_and_read_body, call_service, init_service, read_body, TestRequest};
use actix_web::{web, App, HttpResponse, Resource, Scope};
use prometheus::{Counter, Opts};
#[actix_web::test]
async fn middleware_basic() {
let prometheus = PrometheusMetricsBuilder::new("actix_web_prom")
.endpoint("/metrics")
.build()
.unwrap();
let app = init_service(
App::new()
.wrap(prometheus)
.service(web::resource("/health_check").to(HttpResponse::Ok)),
)
.await;
let res = call_service(&app, TestRequest::with_uri("/health_check").to_request()).await;
assert!(res.status().is_success());
assert_eq!(read_body(res).await, "");
let res = call_service(&app, TestRequest::with_uri("/metrics").to_request()).await;
assert_eq!(
res.headers().get(CONTENT_TYPE).unwrap(),
"text/plain; version=0.0.4; charset=utf-8"
);
let body = String::from_utf8(read_body(res).await.to_vec()).unwrap();
assert!(&body.contains(
&String::from_utf8(web::Bytes::from(
"# HELP actix_web_prom_http_requests_duration_seconds HTTP request duration in seconds for all requests
# TYPE actix_web_prom_http_requests_duration_seconds histogram
actix_web_prom_http_requests_duration_seconds_bucket{endpoint=\"/health_check\",method=\"GET\",status=\"200\",le=\"0.005\"} 1
"
).to_vec()).unwrap()));
assert!(body.contains(
&String::from_utf8(
web::Bytes::from(
"# HELP actix_web_prom_http_requests_total Total number of HTTP requests
# TYPE actix_web_prom_http_requests_total counter
actix_web_prom_http_requests_total{endpoint=\"/health_check\",method=\"GET\",status=\"200\"} 1
"
)
.to_vec()
)
.unwrap()
));
}
#[actix_web::test]
async fn middleware_scope() {
let prometheus = PrometheusMetricsBuilder::new("actix_web_prom")
.endpoint("/internal/metrics")
.build()
.unwrap();
let app = init_service(
App::new().service(
web::scope("/internal")
.wrap(prometheus)
.service(web::resource("/health_check").to(HttpResponse::Ok)),
),
)
.await;
let res = call_service(
&app,
TestRequest::with_uri("/internal/health_check").to_request(),
)
.await;
assert!(res.status().is_success());
assert_eq!(read_body(res).await, "");
let res = call_service(
&app,
TestRequest::with_uri("/internal/metrics").to_request(),
)
.await;
assert_eq!(
res.headers().get(CONTENT_TYPE).unwrap(),
"text/plain; version=0.0.4; charset=utf-8"
);
let body = String::from_utf8(read_body(res).await.to_vec()).unwrap();
assert!(&body.contains(
&String::from_utf8(web::Bytes::from(
"# HELP actix_web_prom_http_requests_duration_seconds HTTP request duration in seconds for all requests
# TYPE actix_web_prom_http_requests_duration_seconds histogram
actix_web_prom_http_requests_duration_seconds_bucket{endpoint=\"/internal/health_check\",method=\"GET\",status=\"200\",le=\"0.005\"} 1
"
).to_vec()).unwrap()));
assert!(body.contains(
&String::from_utf8(
web::Bytes::from(
"# HELP actix_web_prom_http_requests_total Total number of HTTP requests
# TYPE actix_web_prom_http_requests_total counter
actix_web_prom_http_requests_total{endpoint=\"/internal/health_check\",method=\"GET\",status=\"200\"} 1
"
)
.to_vec()
)
.unwrap()
));
}
#[actix_web::test]
async fn middleware_match_pattern() {
let prometheus = PrometheusMetricsBuilder::new("actix_web_prom")
.endpoint("/metrics")
.build()
.unwrap();
let app = init_service(
App::new()
.wrap(prometheus)
.service(web::resource("/resource/{id}").to(HttpResponse::Ok)),
)
.await;
let res = call_service(&app, TestRequest::with_uri("/resource/123").to_request()).await;
assert!(res.status().is_success());
assert_eq!(read_body(res).await, "");
let res = call_and_read_body(&app, TestRequest::with_uri("/metrics").to_request()).await;
let body = String::from_utf8(res.to_vec()).unwrap();
assert!(&body.contains(
&String::from_utf8(web::Bytes::from(
"# HELP actix_web_prom_http_requests_duration_seconds HTTP request duration in seconds for all requests
# TYPE actix_web_prom_http_requests_duration_seconds histogram
actix_web_prom_http_requests_duration_seconds_bucket{endpoint=\"/resource/{id}\",method=\"GET\",status=\"200\",le=\"0.005\"} 1
"
).to_vec()).unwrap()));
assert!(body.contains(
&String::from_utf8(
web::Bytes::from(
"# HELP actix_web_prom_http_requests_total Total number of HTTP requests
# TYPE actix_web_prom_http_requests_total counter
actix_web_prom_http_requests_total{endpoint=\"/resource/{id}\",method=\"GET\",status=\"200\"} 1
"
)
.to_vec()
)
.unwrap()
));
}
#[actix_web::test]
async fn middleware_metrics_exposed_with_conflicting_pattern() {
let prometheus = PrometheusMetricsBuilder::new("actix_web_prom")
.endpoint("/metrics")
.build()
.unwrap();
let app = init_service(
App::new()
.wrap(prometheus)
.service(web::resource("/{path}").to(HttpResponse::Ok)),
)
.await;
let res = call_service(&app, TestRequest::with_uri("/something").to_request()).await;
assert!(res.status().is_success());
assert_eq!(read_body(res).await, "");
let res = call_and_read_body(&app, TestRequest::with_uri("/metrics").to_request()).await;
let body = String::from_utf8(res.to_vec()).unwrap();
assert!(&body.contains(
&String::from_utf8(web::Bytes::from(
"# HELP actix_web_prom_http_requests_duration_seconds HTTP request duration in seconds for all requests"
).to_vec()).unwrap()));
}
#[actix_web::test]
async fn middleware_basic_failure() {
let prometheus = PrometheusMetricsBuilder::new("actix_web_prom")
.endpoint("/prometheus")
.build()
.unwrap();
let app = init_service(
App::new()
.wrap(prometheus)
.service(web::resource("/health_check").to(HttpResponse::Ok)),
)
.await;
call_service(&app, TestRequest::with_uri("/health_checkz").to_request()).await;
let res = call_and_read_body(&app, TestRequest::with_uri("/prometheus").to_request()).await;
assert!(String::from_utf8(res.to_vec()).unwrap().contains(
&String::from_utf8(
web::Bytes::from(
"# HELP actix_web_prom_http_requests_total Total number of HTTP requests
# TYPE actix_web_prom_http_requests_total counter
actix_web_prom_http_requests_total{endpoint=\"/health_checkz\",method=\"GET\",status=\"404\"} 1
"
)
.to_vec()
)
.unwrap()
));
}
#[actix_web::test]
async fn middleware_custom_counter() {
let counter_opts = Opts::new("counter", "some random counter").namespace("actix_web_prom");
let counter = IntCounterVec::new(counter_opts, &["endpoint", "method", "status"]).unwrap();
let prometheus = PrometheusMetricsBuilder::new("actix_web_prom")
.endpoint("/metrics")
.build()
.unwrap();
prometheus
.registry
.register(Box::new(counter.clone()))
.unwrap();
let app = init_service(
App::new()
.wrap(prometheus)
.service(web::resource("/health_check").to(HttpResponse::Ok)),
)
.await;
call_service(&app, TestRequest::with_uri("/health_check").to_request()).await;
let res = call_and_read_body(&app, TestRequest::with_uri("/metrics").to_request()).await;
assert!(!String::from_utf8(res.to_vec()).unwrap().contains(
&String::from_utf8(
web::Bytes::from(
"# HELP actix_web_prom_counter some random counter
# TYPE actix_web_prom_counter counter
actix_web_prom_counter{endpoint=\"endpoint\",method=\"method\",status=\"status\"} 1
"
)
.to_vec()
)
.unwrap()
));
counter
.with_label_values(&["endpoint", "method", "status"])
.inc();
counter
.with_label_values(&["endpoint", "method", "status"])
.inc();
call_service(&app, TestRequest::with_uri("/metrics").to_request()).await;
let res = call_and_read_body(&app, TestRequest::with_uri("/metrics").to_request()).await;
assert!(String::from_utf8(res.to_vec()).unwrap().contains(
&String::from_utf8(
web::Bytes::from(
"# HELP actix_web_prom_counter some random counter
# TYPE actix_web_prom_counter counter
actix_web_prom_counter{endpoint=\"endpoint\",method=\"method\",status=\"status\"} 2
"
)
.to_vec()
)
.unwrap()
));
}
#[actix_web::test]
async fn middleware_none_endpoint() {
let prometheus = PrometheusMetricsBuilder::new("actix_web_prom")
.build()
.unwrap();
let app = init_service(App::new().wrap(prometheus.clone()).service(
web::resource("/metrics").to(|| async { HttpResponse::Ok().body("not prometheus") }),
))
.await;
let response =
call_and_read_body(&app, TestRequest::with_uri("/metrics").to_request()).await;
assert_eq!(
String::from_utf8(response.to_vec()).unwrap(),
"not prometheus"
);
let mut buffer = Vec::new();
let encoder = TextEncoder::new();
let metric_families = prometheus.registry.gather();
encoder.encode(&metric_families, &mut buffer).unwrap();
let output = String::from_utf8(buffer).unwrap();
assert!(output.contains(
"actix_web_prom_http_requests_total{endpoint=\"/metrics\",method=\"GET\",status=\"200\"} 1"
));
}
#[actix_web::test]
async fn middleware_custom_registry_works() {
let registry = Registry::new();
let counter_opts = Opts::new("test_counter", "test counter help");
let counter = Counter::with_opts(counter_opts).unwrap();
registry.register(Box::new(counter.clone())).unwrap();
counter.inc_by(10_f64);
let prometheus = PrometheusMetricsBuilder::new("actix_web_prom")
.registry(registry)
.endpoint("/metrics")
.build()
.unwrap();
let app = init_service(
App::new()
.wrap(prometheus.clone())
.service(web::resource("/test").to(|| async { HttpResponse::Ok().finish() })),
)
.await;
let response =
call_and_read_body(&app, TestRequest::with_uri("/metrics").to_request()).await;
let ten_test_counter =
"# HELP test_counter test counter help\n# TYPE test_counter counter\ntest_counter 10\n";
assert_eq!(
String::from_utf8(response.to_vec()).unwrap(),
ten_test_counter
);
let response =
call_and_read_body(&app, TestRequest::with_uri("/metrics").to_request()).await;
let response_string = String::from_utf8(response.to_vec()).unwrap();
let one_http_counters = "# HELP actix_web_prom_http_requests_total Total number of HTTP requests\n# TYPE actix_web_prom_http_requests_total counter\nactix_web_prom_http_requests_total{endpoint=\"/metrics\",method=\"GET\",status=\"200\"} 1";
assert!(response_string.contains(ten_test_counter));
assert!(response_string.contains(one_http_counters));
}
#[actix_web::test]
async fn middleware_const_labels() {
let mut labels = HashMap::new();
labels.insert("label1".to_string(), "value1".to_string());
labels.insert("label2".to_string(), "value2".to_string());
let prometheus = PrometheusMetricsBuilder::new("actix_web_prom")
.endpoint("/metrics")
.const_labels(labels)
.build()
.unwrap();
let app = init_service(
App::new()
.wrap(prometheus)
.service(web::resource("/health_check").to(HttpResponse::Ok)),
)
.await;
let res = call_service(&app, TestRequest::with_uri("/health_check").to_request()).await;
assert!(res.status().is_success());
assert_eq!(read_body(res).await, "");
let res = call_and_read_body(&app, TestRequest::with_uri("/metrics").to_request()).await;
let body = String::from_utf8(res.to_vec()).unwrap();
assert!(&body.contains(
&String::from_utf8(web::Bytes::from(
"# HELP actix_web_prom_http_requests_duration_seconds HTTP request duration in seconds for all requests
# TYPE actix_web_prom_http_requests_duration_seconds histogram
actix_web_prom_http_requests_duration_seconds_bucket{endpoint=\"/health_check\",label1=\"value1\",label2=\"value2\",method=\"GET\",status=\"200\",le=\"0.005\"} 1
"
).to_vec()).unwrap()));
assert!(body.contains(
&String::from_utf8(
web::Bytes::from(
"# HELP actix_web_prom_http_requests_total Total number of HTTP requests
# TYPE actix_web_prom_http_requests_total counter
actix_web_prom_http_requests_total{endpoint=\"/health_check\",label1=\"value1\",label2=\"value2\",method=\"GET\",status=\"200\"} 1
"
)
.to_vec()
)
.unwrap()
));
}
#[test]
fn compat_with_non_boxed_middleware() {
let _app = App::new()
.wrap(PrometheusMetricsBuilder::new("").build().unwrap())
.wrap(actix_web::middleware::Logger::default())
.route("", web::to(|| async { "" }));
let _app = App::new()
.wrap(actix_web::middleware::Logger::default())
.wrap(PrometheusMetricsBuilder::new("").build().unwrap())
.route("", web::to(|| async { "" }));
let _scope = Scope::new("")
.wrap(PrometheusMetricsBuilder::new("").build().unwrap())
.route("", web::to(|| async { "" }));
let _resource = Resource::new("")
.wrap(PrometheusMetricsBuilder::new("").build().unwrap())
.route(web::to(|| async { "" }));
}
}