logo
  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

use std::time::Instant;

use libopentelemetry::{
    global,
    metrics::{Counter, Unit, ValueRecorder},
};
use opentelemetry_semantic_conventions::trace;

use crate::{Endpoint, IntoResponse, Middleware, Request, Response, Result};

/// Middleware for metrics with OpenTelemetry.
#[cfg_attr(docsrs, doc(cfg(feature = "opentelemetry")))]
pub struct OpenTelemetryMetrics {
    request_count: Counter<u64>,
    error_count: Counter<u64>,
    duration: ValueRecorder<f64>,
}

impl Default for OpenTelemetryMetrics {
    fn default() -> Self {
        Self::new()
    }
}

impl OpenTelemetryMetrics {
    /// Create `OpenTelemetryMetrics` middleware with `meter`.
    pub fn new() -> Self {
        let meter = global::meter("poem");
        Self {
            request_count: meter
                .u64_counter("poem_requests_count")
                .with_description("total request count (since start of service)")
                .init(),
            error_count: meter
                .u64_counter("poem_errors_count")
                .with_description("failed request count (since start of service)")
                .init(),
            duration: meter
                .f64_value_recorder("poem_request_duration_ms")
                .with_unit(Unit::new("milliseconds"))
                .with_description(
                    "request duration histogram (in milliseconds, since start of service)",
                )
                .init(),
        }
    }
}

impl<E: Endpoint> Middleware<E> for OpenTelemetryMetrics {
    type Output = OpenTelemetryMetricsEndpoint<E>;

    fn transform(&self, ep: E) -> Self::Output {
        OpenTelemetryMetricsEndpoint {
            request_count: self.request_count.clone(),
            error_count: self.error_count.clone(),
            duration: self.duration.clone(),
            inner: ep,
        }
    }
}

/// Endpoint for OpenTelemetryMetrics middleware.
#[cfg_attr(docsrs, doc(cfg(feature = "opentelemetry")))]
pub struct OpenTelemetryMetricsEndpoint<E> {
    request_count: Counter<u64>,
    error_count: Counter<u64>,
    duration: ValueRecorder<f64>,
    inner: E,
}

#[async_trait::async_trait]
impl<E: Endpoint> Endpoint for OpenTelemetryMetricsEndpoint<E> {
    type Output = Response;

    async fn call(&self, req: Request) -> Result<Self::Output> {
        let mut labels = Vec::with_capacity(3);
        labels.push(trace::HTTP_METHOD.string(req.method().to_string()));
        labels.push(trace::HTTP_TARGET.string(req.uri().path().to_string()));

        let s = Instant::now();
        let res = self.inner.call(req).await.map(IntoResponse::into_response);
        let elapsed = s.elapsed();

        match &res {
            Ok(resp) => {
                labels.push(trace::HTTP_STATUS_CODE.i64(resp.status().as_u16() as i64));
            }
            Err(err) => {
                self.error_count.add(1, &labels);
                labels.push(trace::EXCEPTION_MESSAGE.string(err.to_string()));
            }
        }

        self.request_count.add(1, &labels);
        self.duration
            .record(elapsed.as_secs_f64() / 1000.0, &labels);

        res
    }
}