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())?;