Add Prometheus metrics

Author: GamePad64

.cargo/config.toml

@@ -0,0 +1,5 @@
+[env]
+AXUM_HTTP_REQUESTS_TOTAL = "notifico_http_requests_total"
+AXUM_HTTP_REQUESTS_DURATION_SECONDS = "notifico_http_requests_duration_seconds"
+AXUM_HTTP_REQUESTS_PENDING = "notifico_http_requests_pending"
+AXUM_HTTP_RESPONSE_BODY_SIZE = "notifico_http_response_body_size"

Cargo.lock

@@ -16,6 +16,7 @@ anyhow = "1.0.95"
 async-trait = "0.1.84"
 axum = { workspace = true }
 axum-extra = { workspace = true }
+axum-prometheus = "0.8.0"
 backoff = { version = "0.4.0", features = ["tokio"] }
 chrono = "0.4.39"
 clap = { workspace = true }

notifico-app/Cargo.toml

@@ -36,9 +36,9 @@ pub struct ApiKeyController {
 impl ApiKeyController {
     pub fn new(db: DatabaseConnection) -> Self {
         let authorization_cache_capacity = 100;
-        gauge!("ingest_api_key_cache_capacity").set(authorization_cache_capacity as f64);
+        gauge!("notifico_ingest_api_key_cache_capacity").set(authorization_cache_capacity as f64);
 
-        let authorization_cache_gauge = gauge!("ingest_api_key_cache_total");
+        let authorization_cache_gauge = gauge!("notifico_ingest_api_key_cache_total");
         let authorization_cache_gauge_for_fut = authorization_cache_gauge.clone();
 
         let authorization_cache = Cache::builder()
@@ -54,9 +54,9 @@ impl ApiKeyController {
             db,
             authorization_cache,
             authorization_cache_gauge,
-            authorization_cache_hit: counter!("ingest_api_key_cache_hit"),
-            authorization_cache_miss: counter!("ingest_api_key_cache_miss"),
-            authorization_invalid_key: counter!("ingest_api_key_invalid"),
+            authorization_cache_hit: counter!("notifico_ingest_api_key_cache_hit"),
+            authorization_cache_miss: counter!("notifico_ingest_api_key_cache_miss"),
+            authorization_invalid_key: counter!("notifico_ingest_api_key_invalid"),
         }
     }
 }

notifico-app/src/controllers/api_key.rs

@@ -5,6 +5,7 @@ use axum::{Extension, Json};
 use axum_extra::headers::authorization::Bearer;
 use axum_extra::headers::Authorization;
 use axum_extra::TypedHeader;
+use axum_prometheus::PrometheusMetricLayer;
 use notifico_core::pipeline::context::EventContext;
 use notifico_core::pipeline::event::{ProcessEventRequest, RecipientSelector};
 use notifico_core::queue::SenderChannel;
@@ -40,6 +41,7 @@ pub async fn start(serviceapi_bind: SocketAddr, ext: HttpIngestExtensions) {
     let app = OpenApiRouter::with_openapi(ApiDoc::openapi())
         .routes(routes!(trigger))
         .routes(routes!(trigger_webhook))
+        .layer(PrometheusMetricLayer::new())
         .layer(Extension(ext));
 
     let (mut app, api) = app.split_for_parts();

notifico-app/src/http/ingest.rs

@@ -0,0 +1,18 @@
+use axum::routing::get;
+use axum::Router;
+use axum_prometheus::metrics_exporter_prometheus::PrometheusHandle;
+use axum_prometheus::PrometheusMetricLayer;
+use std::net::SocketAddr;
+use tokio::net::TcpListener;
+
+pub async fn start(bind: SocketAddr, handle: PrometheusHandle) {
+    // Bind everything now to catch any errors before spinning up the coroutines
+    let listener = TcpListener::bind(bind).await.unwrap();
+
+    // API
+    let app = Router::new()
+        .route("/metrics", get(|| async move { handle.render() }))
+        .layer(PrometheusMetricLayer::new());
+
+    tokio::spawn(async { axum::serve(listener, app).await.unwrap() });
+}

notifico-app/src/http/metrics.rs

@@ -1,4 +1,5 @@
 pub mod auth;
 pub mod ingest;
+pub mod metrics;
 pub mod public;
 pub mod ui;

notifico-app/src/http/mod.rs

@@ -7,6 +7,7 @@ use axum::{Extension, Json};
 use axum_extra::headers::authorization::Bearer;
 use axum_extra::headers::Authorization;
 use axum_extra::TypedHeader;
+use axum_prometheus::PrometheusMetricLayer;
 use jsonwebtoken::{DecodingKey, Validation};
 use notifico_core::http::SecretKey;
 use serde::Deserialize;
@@ -48,6 +49,7 @@ pub(crate) async fn start(bind: SocketAddr, ext: HttpPublicExtensions) {
     let app = OpenApiRouter::with_openapi(openapi)
         .routes(routes!(list_unsubscribe))
         .routes(routes!(subscription_parameters))
+        .layer(PrometheusMetricLayer::new())
         .layer(Extension(ext.secret_key.clone()))
         .layer(Extension(ext.subscription_controller.clone()));
 

notifico-app/src/http/public.rs

@@ -12,6 +12,7 @@ use axum::http::header::CONTENT_TYPE;
 use axum::http::{StatusCode, Uri};
 use axum::response::{Html, IntoResponse, Response};
 use axum::Router;
+use axum_prometheus::PrometheusMetricLayer;
 use notifico_core::credentials::env::EnvCredentialStorage;
 use notifico_core::http::SecretKey;
 use notifico_core::transport::TransportRegistry;
@@ -44,7 +45,9 @@ pub(crate) async fn start(bind: SocketAddr, ext: HttpUiExtensions) {
     let service_listener = TcpListener::bind(bind).await.unwrap();
 
     // Service API
-    let app = Router::new().nest("/api", api::get_router(ext.clone()));
+    let app = Router::new()
+        .nest("/api", api::get_router(ext.clone()))
+        .layer(PrometheusMetricLayer::new());
     let app = app.fallback(static_handler);
 
     tokio::spawn(async { axum::serve(service_listener, app).await.unwrap() });

notifico-app/src/http/ui/mod.rs

@@ -19,6 +19,7 @@ use crate::http::ingest::HttpIngestExtensions;
 use crate::http::public::HttpPublicExtensions;
 use crate::http::ui::HttpUiExtensions;
 use crate::plugin::SubscriptionPlugin;
+use axum_prometheus::{Handle, MakeDefaultHandle};
 use clap::{Parser, Subcommand};
 use migration::{Migrator, MigratorTrait};
 use notifico_attachment::AttachmentPlugin;
@@ -70,6 +71,9 @@ struct Args {
     #[clap(long, env = "NOTIFICO_PUBLIC_BIND", default_value = "[::]:8002")]
     public: SocketAddr,
 
+    #[clap(long, env = "NOTIFICO_METRICS_BIND")]
+    metrics: Option<SocketAddr>,
+
     #[command(subcommand)]
     command: Commands,
 }
@@ -94,6 +98,8 @@ async fn main() {
         .with(EnvFilter::from_default_env())
         .init();
 
+    let prometheus_handle = Handle::make_default_handle(Handle::default()); // Registers Prometheus as default metrics recorder
+
     debug!("Config: {:#?}", args);
 
     match args.command {
@@ -200,6 +206,10 @@ async fn main() {
             let transport_registry = Arc::new(transport_registry);
 
             // Spawn HTTP servers
+            if let Some(metrics_bind) = args.metrics {
+                info!("Starting HTTP metrics server on {}", metrics_bind);
+                http::metrics::start(metrics_bind, prometheus_handle).await;
+            }
             if components.is_empty() || components.contains(COMPONENT_INGEST) {
                 info!("Starting HTTP ingest server on {}", args.ingest);
                 let ext = HttpIngestExtensions {