feat: expose prometheus metrics

Author: fbozic

Cargo.lock

@@ -3,22 +3,24 @@ members = [".", "examples/dcutr", "examples/chat", "examples/autonat"]
 
 [package]
 name = "boot-node"
-version = "0.5.2"
+version = "0.6.0"
 authors = ["Calimero Limited <[email protected]>"]
 edition = "2021"
 repository = "https://github.com/calimero-network/boot-node"
 license = "MIT OR Apache-2.0"
 
 [dependencies]
+axum = "0.7"
 camino = "1.1.6"
 clap = { version = "4.5.4", features = ["derive", "env"] }
 eyre = "0.6.12"
 futures-util = "0.3.30"
-libp2p = { version = "0.53.2", features = [
+libp2p = { version = "0.56.0", features = [
     "autonat",
     "identify",
     "kad",
     "macros",
+    "metrics",
     "noise",
     "ping",
     "quic",
@@ -29,7 +31,9 @@ libp2p = { version = "0.53.2", features = [
     "tls",
     "yamux",
 ] }
+libp2p-metrics = "0.17.0"
 multiaddr = "0.18.1"
+prometheus-client = "0.23"
 serde = "1.0.196"
 serde_json = "1.0.113"
 tokio = { version = "1.35.1", features = ["macros", "rt", "rt-multi-thread"] }

Cargo.toml

@@ -4,7 +4,7 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
-libp2p = { version = "0.55.0", features = [
+libp2p = { version = "0.56.0", features = [
     "dns",
     "rsa",
     "rendezvous",

examples/autonat/Cargo.toml

@@ -11,7 +11,7 @@ bytes = "1.6.0"
 clap = { version = "4.5.4", features = ["derive", "env"] }
 eyre = "0.6.12"
 futures-util = { version = "0.3.30" }
-libp2p = { version = "0.53.2", features = [
+libp2p = { version = "0.56.0", features = [
     "dcutr",
     "dns",
     "gossipsub",

examples/chat/Cargo.toml

@@ -10,7 +10,7 @@ license = "MIT OR Apache-2.0"
 camino = "1.1.6"
 clap = { version = "4.5.4", features = ["derive", "env"] }
 eyre = "0.6.12"
-libp2p = { version = "0.53.2", features = [
+libp2p = { version = "0.56.0", features = [
     "dcutr",
     "dns",
     "identify",

examples/dcutr/Cargo.toml

@@ -0,0 +1,9 @@
+[toolchain]
+channel = "1.89.0"
+components = ["cargo", "clippy"]
+
+[profile.dev]
+opt-level = 0
+
+[profile.release]
+opt-level = 3

rust-toolchain.toml

@@ -0,0 +1,55 @@
+use std::{
+    net::SocketAddr,
+    sync::{Arc, Mutex},
+};
+
+use axum::{extract::State, http::StatusCode, response::IntoResponse, routing::get, Router};
+use prometheus_client::{encoding::text::encode, registry::Registry};
+use tokio::net::TcpListener;
+
+const METRICS_CONTENT_TYPE: &str = "application/openmetrics-text;charset=utf-8;version=1.0.0";
+
+pub(crate) async fn metrics_server(registry: Registry) -> Result<(), std::io::Error> {
+    // Serve on localhost.
+    let addr: SocketAddr = ([127, 0, 0, 1], 0).into();
+    let service = MetricService::new(registry);
+    let server = Router::new()
+        .route("/metrics", get(respond_with_metrics))
+        .with_state(service);
+    let tcp_listener = TcpListener::bind(addr).await?;
+    let local_addr = tcp_listener.local_addr()?;
+    tracing::info!(metrics_server=%format!("http://{}/metrics", local_addr));
+    axum::serve(tcp_listener, server.into_make_service()).await?;
+    Ok(())
+}
+
+#[derive(Clone)]
+pub(crate) struct MetricService {
+    reg: Arc<Mutex<Registry>>,
+}
+
+async fn respond_with_metrics(state: State<MetricService>) -> impl IntoResponse {
+    let mut sink = String::new();
+    let reg = state.get_reg();
+    encode(&mut sink, &reg.lock().unwrap()).unwrap();
+
+    (
+        StatusCode::OK,
+        [(axum::http::header::CONTENT_TYPE, METRICS_CONTENT_TYPE)],
+        sink,
+    )
+}
+
+type SharedRegistry = Arc<Mutex<Registry>>;
+
+impl MetricService {
+    fn new(registry: Registry) -> Self {
+        Self {
+            reg: Arc::new(Mutex::new(registry)),
+        }
+    }
+
+    fn get_reg(&self) -> SharedRegistry {
+        Arc::clone(&self.reg)
+    }
+}

src/http_service.rs

@@ -6,10 +6,13 @@ use libp2p::swarm::{NetworkBehaviour, SwarmEvent};
 use libp2p::{
     autonat, identify, identity, kad, ping, relay, rendezvous, Multiaddr, StreamProtocol, Swarm,
 };
+use libp2p_metrics::{Metrics, Recorder, Registry};
 use tracing::info;
 use tracing_subscriber::prelude::*;
 use tracing_subscriber::EnvFilter;
 
+mod http_service;
+
 const PROTOCOL_VERSION: &str = concat!("/", env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION"));
 const CALIMERO_KAD_PROTO_NAME: StreamProtocol = StreamProtocol::new("/calimero/kad/1.0.0");
 const MAX_RELAY_CIRCUIT_BYTES: u64 = 100 << 20; // 100 MiB
@@ -56,6 +59,8 @@ async fn main() -> eyre::Result<()> {
 
     info!("Peer id: {:?}", peer_id);
 
+    let mut metric_registry = Registry::default();
+
     let mut swarm = libp2p::SwarmBuilder::with_existing_identity(keypair)
         .with_tokio()
         .with_tcp(
@@ -64,15 +69,15 @@ async fn main() -> eyre::Result<()> {
             libp2p::yamux::Config::default,
         )?
         .with_quic()
+        .with_bandwidth_metrics(&mut metric_registry)
         .with_behaviour(|keypair| Behaviour {
             autonat: autonat::Behaviour::new(peer_id.clone(), Default::default()),
             identify: identify::Behaviour::new(identify::Config::new(
                 PROTOCOL_VERSION.to_owned(),
                 keypair.public(),
             )),
             kad: {
-                let mut kademlia_config = kad::Config::default();
-                kademlia_config.set_protocol_names(vec![CALIMERO_KAD_PROTO_NAME]);
+                let mut kademlia_config = kad::Config::new(CALIMERO_KAD_PROTO_NAME);
                 // Instantly remove records and provider records.
                 // TODO: figure out what to do with these values, ref: https://github.com/libp2p/rust-libp2p/blob/1aa016e1c7e3976748a726eab37af44d1c5b7a6e/misc/server/src/behaviour.rs#L38
                 kademlia_config.set_record_ttl(Some(std::time::Duration::from_secs(0)));
@@ -116,16 +121,28 @@ async fn main() -> eyre::Result<()> {
         .with(multiaddr::Protocol::QuicV1);
     swarm.listen_on(listen_addr_quic)?;
 
+    let metrics = Metrics::new(&mut metric_registry);
+    tokio::spawn(http_service::metrics_server(metric_registry));
+
     loop {
         let event = swarm.next().await;
-        handle_swarm_event(&mut swarm, event.expect("Swarm stream to be infinite.")).await;
+        handle_swarm_event(
+            &mut swarm,
+            event.expect("Swarm stream to be infinite."),
+            &metrics,
+        )
+        .await;
     }
 }
 
-async fn handle_swarm_event(swarm: &mut Swarm<Behaviour>, event: SwarmEvent<BehaviourEvent>) {
+async fn handle_swarm_event(
+    swarm: &mut Swarm<Behaviour>,
+    event: SwarmEvent<BehaviourEvent>,
+    metrics: &Metrics,
+) {
     match event {
         SwarmEvent::Behaviour(event) => {
-            handle_swarm_behaviour_event(swarm, event).await;
+            handle_swarm_behaviour_event(swarm, event, metrics).await;
         }
         SwarmEvent::NewListenAddr { address, .. } => {
             info!("Listening on {address:?}");
@@ -134,12 +151,17 @@ async fn handle_swarm_event(swarm: &mut Swarm<Behaviour>, event: SwarmEvent<Beha
     }
 }
 
-async fn handle_swarm_behaviour_event(swarm: &mut Swarm<Behaviour>, event: BehaviourEvent) {
+async fn handle_swarm_behaviour_event(
+    swarm: &mut Swarm<Behaviour>,
+    event: BehaviourEvent,
+    metrics: &Metrics,
+) {
     match event {
         BehaviourEvent::Autonat(event) => {
             info!("AutoNat event: {event:?}");
         }
         BehaviourEvent::Identify(event) => {
+            metrics.record(&event);
             info!("Identify event: {event:?}");
             match event {
                 identify::Event::Received {
@@ -153,9 +175,11 @@ async fn handle_swarm_behaviour_event(swarm: &mut Swarm<Behaviour>, event: Behav
             }
         }
         BehaviourEvent::Kad(event) => {
+            metrics.record(&event);
             info!("Kad event: {event:?}");
         }
         BehaviourEvent::Relay(event) => {
+            metrics.record(&event);
             info!("Relay event: {event:?}");
         }
         BehaviourEvent::Rendezvous(event) => {