Move prometheus endpoint spawner from udp to common crate

crates/common/Cargo.toml

@@ -14,6 +14,7 @@ name = "aquatic_common"
 
 [features]
 rustls = ["dep:rustls", "rustls-pemfile"]
+prometheus = ["dep:metrics", "dep:metrics-util", "dep:metrics-exporter-prometheus", "dep:tokio"]
 
 [dependencies]
 aquatic_toml_config.workspace = true
@@ -34,8 +35,16 @@ serde = { version = "1", features = ["derive"] }
 simple_logger = { version = "4", features = ["stderr"] }
 toml = "0.5"
 
-# Optional
-glommio = { version = "0.8", optional = true }
-hwloc = { version = "0.5", optional = true }
+# rustls feature
 rustls = { version = "0.22", optional = true }
 rustls-pemfile = { version = "2", optional = true }
+
+# prometheus feature
+metrics = { version = "0.22", optional = true }
+metrics-util = { version = "0.16", optional = true }
+metrics-exporter-prometheus = { version = "0.13", optional = true, default-features = false, features = ["http-listener"] }
+tokio = { version = "1", optional = true, features = ["rt", "net", "time"] }
+
+# other optional
+glommio = { version = "0.8", optional = true }
+hwloc = { version = "0.5", optional = true }

crates/common/src/lib.rs

@@ -138,3 +138,63 @@ impl CanonicalSocketAddr {
         self.0.is_ipv4()
     }
 }
+
+#[cfg(feature = "prometheus")]
+pub fn spawn_prometheus_endpoint(
+    addr: SocketAddr,
+    timeout: Option<::std::time::Duration>,
+) -> anyhow::Result<::std::thread::JoinHandle<anyhow::Result<()>>> {
+    use std::thread::Builder;
+    use std::time::Duration;
+
+    use anyhow::Context;
+
+    let handle = Builder::new()
+        .name("prometheus".into())
+        .spawn(move || {
+            #[cfg(feature = "cpu-pinning")]
+            pin_current_if_configured_to(
+                &config.cpu_pinning,
+                config.socket_workers,
+                config.swarm_workers,
+                WorkerIndex::Util,
+            );
+
+            use metrics_exporter_prometheus::PrometheusBuilder;
+            use metrics_util::MetricKindMask;
+
+            let rt = ::tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .context("build prometheus tokio runtime")?;
+
+            rt.block_on(async {
+                let (recorder, exporter) = PrometheusBuilder::new()
+                    .idle_timeout(MetricKindMask::ALL, timeout)
+                    .with_http_listener(addr)
+                    .build()
+                    .context("build prometheus recorder and exporter")?;
+
+                let recorder_handle = recorder.handle();
+
+                ::metrics::set_global_recorder(recorder).context("set global metrics recorder")?;
+
+                ::tokio::spawn(async move {
+                    let mut interval = ::tokio::time::interval(Duration::from_secs(5));
+
+                    loop {
+                        interval.tick().await;
+
+                        // Periodically render metrics to make sure
+                        // idles are cleaned up
+                        recorder_handle.render();
+                    }
+                });
+
+                exporter.await.context("run prometheus exporter")
+            })
+        })
+        .context("spawn prometheus endpoint")?;
+
+    Ok(handle)
+}