From 2a0bffe958512eae95f1a8920f7f3ee1c888a953 Mon Sep 17 00:00:00 2001 From: zealsham Date: Fri, 23 Jan 2026 01:14:31 +0100 Subject: [PATCH] Add standalone metrics service This pr addresses #1284 and wil be built on to reflect the discussion on https://github.com/orgs/payjoin/discussions/775 Request passes through the axum middleware and metrics is collected. This currently tracks http request by endpoint type, method and status. and also tracks active and total connections. Remove metrics collection from Payjoin-directory The metrics server is exposed on different interface fix broken test fix broken test fix broken test Fix broken test dude to lock file fix format issue fix format issue Refactor metrics Servie --- Cargo-minimal.lock | 27 +++++-- Cargo-recent.lock | 27 +++++-- payjoin-directory/Cargo.toml | 1 - payjoin-directory/src/cli.rs | 3 - payjoin-directory/src/config.rs | 7 -- payjoin-directory/src/lib.rs | 68 +---------------- payjoin-directory/src/main.rs | 17 +---- payjoin-directory/src/metrics.rs | 60 --------------- payjoin-service/Cargo.toml | 4 + payjoin-service/src/config.rs | 14 ++++ payjoin-service/src/lib.rs | 120 ++++++++++++++++++++++++++---- payjoin-service/src/metrics.rs | 64 ++++++++++++++++ payjoin-service/src/middleware.rs | 32 ++++++++ payjoin-test-utils/src/lib.rs | 16 +++- 14 files changed, 278 insertions(+), 182 deletions(-) delete mode 100644 payjoin-directory/src/metrics.rs create mode 100644 payjoin-service/src/metrics.rs create mode 100644 payjoin-service/src/middleware.rs diff --git a/Cargo-minimal.lock b/Cargo-minimal.lock index fa99ebd80..d5f084e51 100644 --- a/Cargo-minimal.lock +++ b/Cargo-minimal.lock @@ -2637,7 +2637,6 @@ dependencies = [ "hyper-util", "ohttp-relay", "payjoin", - "prometheus", "rand 0.8.5", "serde", "tempfile", @@ -2692,7 +2691,9 @@ dependencies = [ "ohttp-relay", "payjoin-directory", "payjoin-test-utils", + "prometheus", "rand 0.8.5", + "reqwest", "rustls 0.23.31", "serde", "tempfile", @@ -2905,9 +2906,9 @@ dependencies = [ [[package]] name = "prometheus" -version = "0.13.4" +version = "0.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3d33c28a30771f7f96db69893f78b857f7450d7e0237e9c8fc6427a81bae7ed1" +checksum = "3ca5326d8d0b950a9acd87e6a3f94745394f62e4dae1b1ee22b2bc0c394af43a" dependencies = [ "cfg-if", "fnv", @@ -2915,14 +2916,28 @@ dependencies = [ "memchr", "parking_lot 0.12.3", "protobuf", - "thiserror 1.0.63", + "thiserror 2.0.17", ] [[package]] name = "protobuf" -version = "2.28.0" +version = "3.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d65a1d4ddae7d8b5de68153b48f6aa3bba8cb002b243dbdbc55a5afbc98f99f4" +dependencies = [ + "once_cell", + "protobuf-support", + "thiserror 1.0.63", +] + +[[package]] +name = "protobuf-support" +version = "3.7.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "106dd99e98437432fed6519dedecfade6a06a73bb7b2a1e019fdd2bee5778d94" +checksum = "3e36c2f31e0a47f9280fb347ef5e461ffcd2c52dd520d8e216b52f93b0b0d7d6" +dependencies = [ + "thiserror 1.0.63", +] [[package]] name = "quinn" diff --git a/Cargo-recent.lock b/Cargo-recent.lock index fa99ebd80..d5f084e51 100644 --- a/Cargo-recent.lock +++ b/Cargo-recent.lock @@ -2637,7 +2637,6 @@ dependencies = [ "hyper-util", "ohttp-relay", "payjoin", - "prometheus", "rand 0.8.5", "serde", "tempfile", @@ -2692,7 +2691,9 @@ dependencies = [ "ohttp-relay", "payjoin-directory", "payjoin-test-utils", + "prometheus", "rand 0.8.5", + "reqwest", "rustls 0.23.31", "serde", "tempfile", @@ -2905,9 +2906,9 @@ dependencies = [ [[package]] name = "prometheus" -version = "0.13.4" +version = "0.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3d33c28a30771f7f96db69893f78b857f7450d7e0237e9c8fc6427a81bae7ed1" +checksum = "3ca5326d8d0b950a9acd87e6a3f94745394f62e4dae1b1ee22b2bc0c394af43a" dependencies = [ "cfg-if", "fnv", @@ -2915,14 +2916,28 @@ dependencies = [ "memchr", "parking_lot 0.12.3", "protobuf", - "thiserror 1.0.63", + "thiserror 2.0.17", ] [[package]] name = "protobuf" -version = "2.28.0" +version = "3.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d65a1d4ddae7d8b5de68153b48f6aa3bba8cb002b243dbdbc55a5afbc98f99f4" +dependencies = [ + "once_cell", + "protobuf-support", + "thiserror 1.0.63", +] + +[[package]] +name = "protobuf-support" +version = "3.7.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "106dd99e98437432fed6519dedecfade6a06a73bb7b2a1e019fdd2bee5778d94" +checksum = "3e36c2f31e0a47f9280fb347ef5e461ffcd2c52dd520d8e216b52f93b0b0d7d6" +dependencies = [ + "thiserror 1.0.63", +] [[package]] name = "quinn" diff --git a/payjoin-directory/Cargo.toml b/payjoin-directory/Cargo.toml index 9e916db86..09c8674d4 100644 --- a/payjoin-directory/Cargo.toml +++ b/payjoin-directory/Cargo.toml @@ -33,7 +33,6 @@ ohttp-relay = { path = "../ohttp-relay" } payjoin = { version = "1.0.0-rc.1", features = [ "directory", ], default-features = false } -prometheus = "0.13.4" rand = "0.8" serde = { version = "1.0.219", features = ["derive"] } tokio = { version = "1.47.1", features = ["full"] } diff --git a/payjoin-directory/src/cli.rs b/payjoin-directory/src/cli.rs index 548f7e7f0..114baada0 100644 --- a/payjoin-directory/src/cli.rs +++ b/payjoin-directory/src/cli.rs @@ -13,9 +13,6 @@ pub struct Cli { #[arg(long, short = 'p', env = "PJ_DIR_PORT", help = "The port to bind [default: 8080]")] pub port: Option, // TODO tokio_listener::ListenerAddressLFlag - #[arg(long, env = "PJ_METRIC_PORT", help = "The port to bind for prometheus metrics export")] - pub metrics_port: Option, // TODO tokio_listener::ListenerAddressLFlag - #[cfg(feature = "acme")] #[clap(flatten)] pub acme: AcmeCli, diff --git a/payjoin-directory/src/config.rs b/payjoin-directory/src/config.rs index c16c34542..42403def0 100644 --- a/payjoin-directory/src/config.rs +++ b/payjoin-directory/src/config.rs @@ -13,7 +13,6 @@ use crate::cli::Cli; #[derive(Debug, Clone, Deserialize)] pub struct Config { pub listen_addr: String, // TODO tokio_listener::ListenerAddressLFlag - pub metrics_listen_addr: Option, // TODO tokio_listener::ListenerAddressLFlag pub timeout: Duration, pub storage_dir: PathBuf, pub ohttp_keys: PathBuf, // TODO OhttpConfig struct with rotation params, etc @@ -52,7 +51,6 @@ impl Config { Ok(Config { listen_addr: built_config.get("listen_addr")?, - metrics_listen_addr: built_config.get("metrics_listen_addr").ok(), timeout: Duration::from_secs(built_config.get("timeout")?), storage_dir: built_config.get("storage_dir")?, ohttp_keys: built_config.get("ohttp_keys")?, @@ -75,11 +73,6 @@ fn add_defaults(config: Builder, cli: &Cli) -> Result { let config = config .set_default("listen_addr", "[::]:8080")? .set_override_option("listen_addr", cli.port.map(|port| format!("[::]:{}", port)))? - .set_default("metrics_listen_addr", Option::::None)? - .set_override_option( - "metrics_listen_addr", - cli.metrics_port.map(|port| format!("localhost:{}", port)), - )? .set_default("timeout", Some(30))? .set_override_option("timeout", cli.timeout)? .set_default("ohttp_keys", "ohttp_keys")? diff --git a/payjoin-directory/src/lib.rs b/payjoin-directory/src/lib.rs index 4870bf20d..b2aa470fd 100644 --- a/payjoin-directory/src/lib.rs +++ b/payjoin-directory/src/lib.rs @@ -7,7 +7,7 @@ use anyhow::Result; use futures::StreamExt; use http_body_util::combinators::BoxBody; use http_body_util::{BodyExt, Empty, Full}; -use hyper::body::{Body, Bytes, Incoming}; +use hyper::body::{Body, Bytes}; use hyper::header::{HeaderValue, ACCESS_CONTROL_ALLOW_ORIGIN, CONTENT_TYPE}; use hyper::server::conn::http1; use hyper::{Method, Request, Response, StatusCode, Uri}; @@ -27,7 +27,6 @@ pub mod key_config; use ohttp_relay::SentinelTag; pub use crate::key_config::*; -use crate::metrics::Metrics; const CHACHA20_POLY1305_NONCE_LEN: usize = 32; // chacha20poly1305 n_k const POLY1305_TAG_SIZE: usize = 16; @@ -43,7 +42,6 @@ pub(crate) mod db; pub mod cli; pub mod config; -pub mod metrics; pub type BoxError = Box; @@ -69,7 +67,6 @@ fn init_tls_acceptor(cert_key: (Vec, Vec)) -> Result { db: D, ohttp: ohttp::Server, - metrics: Metrics, sentinel_tag: SentinelTag, } @@ -94,8 +91,8 @@ where } impl Service { - pub fn new(db: D, ohttp: ohttp::Server, metrics: Metrics, sentinel_tag: SentinelTag) -> Self { - Self { db, ohttp, metrics, sentinel_tag } + pub fn new(db: D, ohttp: ohttp::Server, sentinel_tag: SentinelTag) -> Self { + Self { db, ohttp, sentinel_tag } } #[cfg(feature = "_manual-tls")] @@ -111,7 +108,6 @@ impl Service { let tls_acceptor = tls_acceptor.clone(); let service = self.clone(); tokio::spawn(async move { - service.metrics.record_connection(); let tls_stream = match tls_acceptor.accept(stream).await { Ok(tls_stream) => tls_stream, Err(e) => { @@ -172,7 +168,6 @@ impl Service { where I: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + Unpin + 'static, { - self.metrics.record_connection(); let hyper_service = TowerToHyperService::new(self.clone()); if let Err(err) = http1::Builder::new() .serve_connection(TokioIo::new(stream), hyper_service) @@ -441,63 +436,6 @@ impl Service { res } - async fn handle_metrics(&self) -> Response> { - match self.metrics.generate_metrics() { - Ok(metrics_data) => { - let mut response = Response::new(full(metrics_data)); - response.headers_mut().insert( - CONTENT_TYPE, - HeaderValue::from_static("text/plain; version=0.0.4; charset=utf-8"), - ); - response - } - Err(e) => { - error!("failed to generate metrics: {}", e); - let mut response = Response::new(full("Error generating metrics")); - *response.status_mut() = StatusCode::INTERNAL_SERVER_ERROR; - response - } - } - } - - pub async fn serve_metrics_tcp( - &self, - listener: tokio::net::TcpListener, - ) -> Result<(), BoxError> { - while let Ok((stream, _)) = listener.accept().await { - let io = TokioIo::new(stream); - let service = self.clone(); - tokio::spawn(async move { - let service = hyper::service::service_fn(move |req| { - let this = service.clone(); - async move { this.serve_metrics_request(req).await } - }); - if let Err(err) = - http1::Builder::new().serve_connection(io, service).with_upgrades().await - { - error!("Error serving connection: {:?}", err); - } - }); - } - - Ok(()) - } - - async fn serve_metrics_request( - &self, - req: Request, - ) -> Result>, anyhow::Error> { - let path = req.uri().path().to_string(); - let (parts, _body) = req.into_parts(); - - let path_segments: Vec<&str> = path.split('/').collect(); - let response = match (parts.method, path_segments.as_slice()) { - (Method::GET, ["", ""]) => self.handle_metrics().await, - _ => not_found(), - }; - - Ok(response) - } } fn handle_peek( diff --git a/payjoin-directory/src/main.rs b/payjoin-directory/src/main.rs index 08da71214..9c68b7831 100644 --- a/payjoin-directory/src/main.rs +++ b/payjoin-directory/src/main.rs @@ -1,6 +1,5 @@ use clap::Parser; use ohttp_relay::SentinelTag; -use payjoin_directory::metrics::Metrics; use payjoin_directory::*; use tokio::net::TcpListener; use tracing_subscriber::filter::LevelFilter; @@ -26,25 +25,11 @@ async fn main() -> Result<(), BoxError> { } }; - let metrics = Metrics::new(); let db = payjoin_directory::FilesDb::init(config.timeout, config.storage_dir) .await .expect("Failed to initialize persistent storage"); - let service = Service::new(db, ohttp.into(), metrics, SentinelTag::new([0u8; 32])); - - // Start metrics server in the background - if let Some(addr) = config.metrics_listen_addr { - let metrics_listener = TcpListener::bind(addr.clone()).await?; - let service_clone = service.clone(); - - tokio::spawn(async move { - if let Err(e) = service_clone.serve_metrics_tcp(metrics_listener).await { - eprintln!("Metrics server error: {e}"); - } - println!("Metrics server started on {}", addr); - }); - } + let service = Service::new(db, ohttp.into(), SentinelTag::new([0u8; 32])); let listener = TcpListener::bind(config.listen_addr).await?; diff --git a/payjoin-directory/src/metrics.rs b/payjoin-directory/src/metrics.rs deleted file mode 100644 index af86e4a93..000000000 --- a/payjoin-directory/src/metrics.rs +++ /dev/null @@ -1,60 +0,0 @@ -use prometheus::{Encoder, IntCounter, Registry, TextEncoder}; - -#[derive(Debug, Clone)] -pub struct Metrics { - /// Total number of connections accepted by the directory - pub connections_total: IntCounter, - registry: Registry, -} - -impl Default for Metrics { - fn default() -> Self { Self::new() } -} - -impl Metrics { - pub fn new() -> Self { - let registry = Registry::new(); - let connections_total = - IntCounter::new("connections_total", "Total number of tcp connections") - .expect("Failed to create connections_total metrics "); - - registry - .register(Box::new(connections_total.clone())) - .expect("Failed to register connections_total"); - - Self { connections_total, registry } - } - - /// Records a new connection - pub fn record_connection(&self) { self.connections_total.inc(); } - - pub fn generate_metrics(&self) -> Result> { - let encoder = TextEncoder::new(); - let all_metrics = self.registry.gather(); - let mut buffer = Vec::new(); - encoder.encode(&all_metrics, &mut buffer)?; - Ok(String::from_utf8(buffer)?) - } -} - -#[cfg(test)] -mod tests { - - use super::*; - - #[test] - fn test_recording_metrics() { - let metrics = Metrics::new(); - metrics.record_connection(); - - let metrics_recorded = metrics.generate_metrics().expect("Failed to generate metrics"); - assert!(metrics_recorded.contains("connections_total")); - } - - #[test] - fn does_not_error_on_empty_metrics() { - let metrics = Metrics::new(); - let metrics_recorded = metrics.generate_metrics().expect("Failed to generate metrics"); - assert!(metrics_recorded.contains("connections_total 0")); - } -} diff --git a/payjoin-service/Cargo.toml b/payjoin-service/Cargo.toml index 71bd27aa4..3668aec47 100644 --- a/payjoin-service/Cargo.toml +++ b/payjoin-service/Cargo.toml @@ -27,6 +27,7 @@ clap = { version = "4.5", features = ["derive", "env"] } config = "0.15" ohttp-relay = { path = "../ohttp-relay", features = ["bootstrap"] } payjoin-directory = { path = "../payjoin-directory" } +prometheus = "0.14.0" rand = "0.8" rustls = { version = "0.23", default-features = false, features = [ "ring", @@ -40,4 +41,7 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] } [dev-dependencies] payjoin-test-utils = { version = "0.0.1" } +reqwest = { version = "0.12", default-features = false, features = [ + "rustls-tls", +] } tempfile = "3.20" diff --git a/payjoin-service/src/config.rs b/payjoin-service/src/config.rs index 96b12e46b..23e30addb 100644 --- a/payjoin-service/src/config.rs +++ b/payjoin-service/src/config.rs @@ -12,6 +12,19 @@ pub struct Config { pub storage_dir: PathBuf, #[serde(deserialize_with = "deserialize_duration_secs")] pub timeout: Duration, + pub metrics: MetricsConfig, +} + +#[derive(Debug, Clone, Deserialize)] +#[serde(default)] +pub struct MetricsConfig { + pub listener: ListenerAddress, +} + +impl Default for MetricsConfig { + fn default() -> Self { + Self { listener: "[::]:9090".parse().expect("valid default metrics listener address") } + } } impl Default for Config { @@ -20,6 +33,7 @@ impl Default for Config { listener: "[::]:8080".parse().expect("valid default listener address"), storage_dir: PathBuf::from("./data"), timeout: Duration::from_secs(30), + metrics: MetricsConfig::default(), } } } diff --git a/payjoin-service/src/lib.rs b/payjoin-service/src/lib.rs index 1dee13f38..7e68426ca 100644 --- a/payjoin-service/src/lib.rs +++ b/payjoin-service/src/lib.rs @@ -1,16 +1,25 @@ +use std::net::SocketAddr; + use axum::extract::State; +use axum::http::header::CONTENT_TYPE; use axum::http::Method; use axum::response::{IntoResponse, Response}; +use axum::routing::get; use axum::Router; use config::Config; use ohttp_relay::SentinelTag; use rand::Rng; use tokio_listener::{Listener, SystemOptions, UserOptions}; -use tower::Service; +use tower::{Service, ServiceBuilder}; use tracing::info; pub mod cli; pub mod config; +pub mod metrics; +pub mod middleware; + +use crate::metrics::MetricsService; +use crate::middleware::{track_connections, track_metrics}; #[derive(Clone)] struct Services { @@ -20,12 +29,15 @@ struct Services { pub async fn serve(config: Config) -> anyhow::Result<()> { let sentinel_tag = generate_sentinel_tag(); + let metrics = MetricsService::new()?; let services = Services { directory: init_directory(&config, sentinel_tag).await?, relay: ohttp_relay::Service::new(sentinel_tag).await, }; - let app = Router::new().fallback(route_request).with_state(services); + + let app = build_app(services, metrics.clone()); + let _ = spawn_metrics_server(config.metrics.listener.clone(), metrics).await?; let listener = Listener::bind(&config.listener, &SystemOptions::default(), &UserOptions::default()) @@ -39,7 +51,7 @@ pub async fn serve(config: Config) -> anyhow::Result<()> { /// Serves payjoin-service with manual TLS configuration. /// /// Binds to `config.listener` (use port 0 to let the OS assign a free port) and returns -/// the actual bound port along with a task handle. +/// the actual bound port, the metrics port, and a task handle. /// /// If `tls_config` is provided, the server will use TLS for incoming connections. /// The `root_store` is used for outgoing relay connections to the gateway. @@ -48,16 +60,18 @@ pub async fn serve_manual_tls( config: Config, tls_config: Option, root_store: rustls::RootCertStore, -) -> anyhow::Result<(u16, tokio::task::JoinHandle>)> { +) -> anyhow::Result<(u16, u16, tokio::task::JoinHandle>)> { use std::net::SocketAddr; let sentinel_tag = generate_sentinel_tag(); + let metrics = MetricsService::new()?; let services = Services { directory: init_directory(&config, sentinel_tag).await?, relay: ohttp_relay::Service::new_with_roots(root_store, sentinel_tag).await, }; - let app = Router::new().fallback(route_request).with_state(services); + let app = build_app(services, metrics.clone()); + let metrics_port = spawn_metrics_server(config.metrics.listener.clone(), metrics).await?; let addr: SocketAddr = config .listener @@ -83,7 +97,7 @@ pub async fn serve_manual_tls( } }; - Ok((port, handle)) + Ok((port, metrics_port, handle)) } /// Generate random sentinel tag at startup. @@ -100,9 +114,8 @@ async fn init_directory( let ohttp_keys_dir = config.storage_dir.join("ohttp-keys"); let ohttp_config = init_ohttp_config(&ohttp_keys_dir)?; - let metrics = payjoin_directory::metrics::Metrics::new(); - Ok(payjoin_directory::Service::new(db, ohttp_config.into(), metrics, sentinel_tag)) + Ok(payjoin_directory::Service::new(db, ohttp_config.into(), sentinel_tag)) } fn init_ohttp_config( @@ -119,6 +132,56 @@ fn init_ohttp_config( } } +fn build_app(services: Services, metrics: MetricsService) -> Router { + Router::new() + .fallback(route_request) + .layer( + ServiceBuilder::new() + .layer(axum::middleware::from_fn_with_state(metrics.clone(), track_metrics)) + .layer(axum::middleware::from_fn_with_state(metrics, track_connections)), + ) + .with_state(services) +} + +fn build_metrics_app(metrics: MetricsService) -> Router { + Router::new().route("/metrics", get(metrics_handler)).with_state(metrics) +} + +async fn metrics_handler(State(metrics): State) -> impl IntoResponse { + match metrics.encode_metrics() { + Ok(body) => ( + axum::http::StatusCode::OK, + [(CONTENT_TYPE, "text/plain; version=0.0.4; charset=utf-8")], + body, + ) + .into_response(), + Err(e) => ( + axum::http::StatusCode::INTERNAL_SERVER_ERROR, + format!("Failed to encode metrics: {}", e), + ) + .into_response(), + } +} + +async fn spawn_metrics_server( + metrics_listener: tokio_listener::ListenerAddress, + metrics: MetricsService, +) -> anyhow::Result { + let addr: SocketAddr = metrics_listener.to_string().parse().map_err(|_| { + anyhow::anyhow!("Metrics listener must be a TCP address (e.g., '[::]:9090')") + })?; + let listener = tokio::net::TcpListener::bind(addr).await?; + let actual_port = listener.local_addr()?.port(); + info!("Metrics server listening on [::]:{actual_port}"); + tokio::spawn(async move { + let app = build_metrics_app(metrics); + if let Err(e) = axum::serve(listener, app).await { + tracing::error!("Metrics server error: {e}"); + } + }); + Ok(actual_port) +} + async fn route_request( State(mut services): State, req: axum::extract::Request, @@ -175,20 +238,24 @@ mod tests { async fn start_service( cert_der: Vec, key_der: Vec, - ) -> (u16, tokio::task::JoinHandle>, tempfile::TempDir) { + ) -> (u16, u16, tokio::task::JoinHandle>, tempfile::TempDir) { let tempdir = tempdir().unwrap(); let config = Config { listener: "[::]:0".parse().expect("valid listener address"), storage_dir: tempdir.path().to_path_buf(), timeout: Duration::from_secs(2), + metrics: config::MetricsConfig { + listener: "[::]:0".parse().expect("valid metrics listener address"), + }, }; let mut root_store = RootCertStore::empty(); root_store.add(CertificateDer::from(cert_der.clone())).unwrap(); let tls_config = RustlsConfig::from_der(vec![cert_der], key_der).await.unwrap(); - let (port, handle) = serve_manual_tls(config, Some(tls_config), root_store).await.unwrap(); - (port, handle, tempdir) + let (port, metrics_port, handle) = + serve_manual_tls(config, Some(tls_config), root_store).await.unwrap(); + (port, metrics_port, handle, tempdir) } #[tokio::test] @@ -197,7 +264,8 @@ mod tests { let cert_der = cert.cert.der().to_vec(); let key_der = cert.signing_key.serialize_der(); - let (port, _handle, _tempdir) = start_service(cert_der.clone(), key_der).await; + let (port, _metrics_port, _handle, _tempdir) = + start_service(cert_der.clone(), key_der).await; let client = Arc::new(http_agent(cert_der.clone()).unwrap()); let base_url = format!("https://localhost:{}", port); @@ -228,9 +296,9 @@ mod tests { let cert_der = cert.cert.der().to_vec(); let key_der = cert.signing_key.serialize_der(); - let (relay_port, _relay_handle, _relay_tempdir) = + let (relay_port, _relay_metrics, _relay_handle, _relay_tempdir) = start_service(cert_der.clone(), key_der.clone()).await; - let (directory_port, _directory_handle, _directory_tempdir) = + let (directory_port, _directory_metrics, _directory_handle, _directory_tempdir) = start_service(cert_der.clone(), key_der).await; let client = Arc::new(http_agent(cert_der).unwrap()); @@ -259,4 +327,28 @@ mod tests { "cross-instance request should not be rejected as forbidden" ); } + + #[tokio::test] + async fn metrics_endpoint_works() { + let cert = local_cert_key(); + let cert_der = cert.cert.der().to_vec(); + let key_der = cert.signing_key.serialize_der(); + + let (port, metrics_port, _handle, _tempdir) = + start_service(cert_der.clone(), key_der).await; + + let client = Arc::new(http_agent(cert_der).unwrap()); + let base_url = format!("https://localhost:{}", port); + wait_for_service_ready(&base_url, client.clone()).await.unwrap(); + + let metrics_url = format!("http://localhost:{}/metrics", metrics_port); + let http_client = reqwest::Client::new(); + let response = + http_client.get(&metrics_url).send().await.expect("metrics request should work"); + + assert_eq!(response.status(), axum::http::StatusCode::OK); + let body = response.text().await.unwrap(); + assert!(body.contains("http_request_total")); + assert!(body.contains("active_connections")); + } } diff --git a/payjoin-service/src/metrics.rs b/payjoin-service/src/metrics.rs new file mode 100644 index 000000000..dc0e34010 --- /dev/null +++ b/payjoin-service/src/metrics.rs @@ -0,0 +1,64 @@ +use std::sync::Arc; + +use prometheus::{Encoder, IntCounter, IntCounterVec, IntGauge, Opts, Registry, TextEncoder}; + +const TOTAL_CONNECTIONS: &str = "total_connections"; +const ACTIVE_CONNECTIONS: &str = "active_connections"; +const HTTP_REQUESTS: &str = "http_request_total"; + +#[derive(Clone)] +pub struct MetricsService { + registry: Arc, + /// Total number of HTTP requests by endpoint type, method, and status code + pub http_requests_total: IntCounterVec, + + /// Total number of Connections + pub total_connections: IntCounter, + /// Total number of active connections right now + pub active_connections: IntGauge, +} + +impl MetricsService { + pub fn new() -> anyhow::Result { + let registry = Registry::new(); + + let http_requests_total = IntCounterVec::new( + Opts::new(HTTP_REQUESTS, "Total number of HTTP requests"), + &["endpoint", "method", "status_code"], + )?; + registry.register(Box::new(http_requests_total.clone()))?; + + let total_connections = IntCounter::new(TOTAL_CONNECTIONS, "Total number of connections")?; + registry.register(Box::new(total_connections.clone()))?; + + let active_connections = IntGauge::new(ACTIVE_CONNECTIONS, "Number of active connections")?; + registry.register(Box::new(active_connections.clone()))?; + + Ok(Self { + registry: Arc::new(registry), + http_requests_total, + total_connections, + active_connections, + }) + } + + pub fn record_http_request(&self, endpoint: &str, method: &str, status_code: u16) { + self.http_requests_total + .with_label_values(&[endpoint, method, &status_code.to_string()]) + .inc(); + } + + pub fn record_connection_open(&self) { + self.total_connections.inc(); + self.active_connections.inc(); + } + pub fn record_connection_close(&self) { self.active_connections.dec(); } + + pub(crate) fn encode_metrics(&self) -> Result, anyhow::Error> { + let encode = TextEncoder::new(); + let all_metrics = self.registry.gather(); + let mut buffer = Vec::new(); + encode.encode(&all_metrics, &mut buffer)?; + Ok(buffer) + } +} diff --git a/payjoin-service/src/middleware.rs b/payjoin-service/src/middleware.rs new file mode 100644 index 000000000..875bc49b4 --- /dev/null +++ b/payjoin-service/src/middleware.rs @@ -0,0 +1,32 @@ +use axum::extract::Request; +use axum::middleware::Next; +use axum::response::Response; + +use crate::metrics::MetricsService; + +pub async fn track_metrics( + metrics: axum::extract::State, + req: Request, + next: Next, +) -> Response { + let method = req.method().to_string(); + let path = req.uri().path().to_string(); + + let response = next.run(req).await; + let status = response.status().as_u16(); + + metrics.record_http_request(&path, &method, status); + + response +} + +pub async fn track_connections( + metrics: axum::extract::State, + req: Request, + next: Next, +) -> Response { + metrics.record_connection_open(); + let response = next.run(req).await; + metrics.record_connection_close(); + response +} diff --git a/payjoin-test-utils/src/lib.rs b/payjoin-test-utils/src/lib.rs index bb62ffe41..7adf7d474 100644 --- a/payjoin-test-utils/src/lib.rs +++ b/payjoin-test-utils/src/lib.rs @@ -13,6 +13,7 @@ use ohttp::{KeyId, SymmetricSuite}; use once_cell::sync::{Lazy, OnceCell}; use payjoin::io::{fetch_ohttp_keys_with_cert, Error as IOError}; use payjoin::OhttpKeys; +use payjoin_service::config::MetricsConfig; use rcgen::Certificate; use reqwest::{Client, ClientBuilder}; use rustls::pki_types::CertificateDer; @@ -121,13 +122,17 @@ pub async fn init_directory( listener: "[::]:0".parse().expect("valid listener address"), // let OS assign a free port storage_dir: tempdir.path().to_path_buf(), timeout: Duration::from_secs(2), + metrics: MetricsConfig { + listener: "[::]:0".parse().expect("valid metrics listener address"), + }, }; let tls_config = RustlsConfig::from_der(vec![local_cert_key.0], local_cert_key.1).await?; - let (port, handle) = payjoin_service::serve_manual_tls(config, Some(tls_config), root_store) - .await - .map_err(|e| e.to_string())?; + let (port, _metrics_port, handle) = + payjoin_service::serve_manual_tls(config, Some(tls_config), root_store) + .await + .map_err(|e| e.to_string())?; let handle = tokio::spawn(async move { let _tempdir = tempdir; // keep the tempdir until the directory shuts down @@ -148,9 +153,12 @@ async fn init_ohttp_relay( listener: "[::]:0".parse().expect("valid listener address"), // let OS assign a free port storage_dir: tempdir.path().to_path_buf(), timeout: Duration::from_secs(2), + metrics: payjoin_service::config::MetricsConfig { + listener: "[::]:0".parse().expect("valid metrics listener address"), + }, }; - let (port, handle) = payjoin_service::serve_manual_tls(config, None, root_store) + let (port, _metrics_port, handle) = payjoin_service::serve_manual_tls(config, None, root_store) .await .map_err(|e| e.to_string())?;