Add unix domain socket for telemetry server

Author: gpgabriel

examples/http_server/example_conf.yaml

@@ -57,7 +57,11 @@ telemetry:
     # Enables telemetry server
     enabled: true
     # Telemetry server address.
+    # Can be either a TCP socket address (e.g., "127.0.0.1:8080") 
+    # or a Unix domain socket path (e.g., "/tmp/telemetry.sock") on Unix systems.
     addr: "127.0.0.1:0"
+    # Example Unix socket configuration (uncomment to use):
+    # addr: "/tmp/telemetry.sock"
 # HTTP endpoints configuration.
 endpoints:
   Example endpoint:

foundations/src/telemetry/driver.rs

@@ -9,7 +9,6 @@ use std::task::{Context, Poll};
 
 feature_use!(cfg(feature = "telemetry-server"), {
     use super::server::TelemetryServerFuture;
-    use std::net::SocketAddr;
 });
 
 /// A future that drives async telemetry functionality and that is returned
@@ -21,7 +20,7 @@ feature_use!(cfg(feature = "telemetry-server"), {
 /// [security syscall-related]: `crate::security`
 pub struct TelemetryDriver {
     #[cfg(feature = "telemetry-server")]
-    server_addr: Option<SocketAddr>,
+    server_addr: Option<String>,
 
     #[cfg(feature = "telemetry-server")]
     server_fut: Option<TelemetryServerFuture>,
@@ -36,7 +35,7 @@ impl TelemetryDriver {
     ) -> Self {
         Self {
             #[cfg(feature = "telemetry-server")]
-            server_addr: server_fut.as_ref().map(|fut| fut.local_addr()),
+            server_addr: server_fut.as_ref().and_then(|fut| fut.local_addr().ok()),
 
             #[cfg(feature = "telemetry-server")]
             server_fut,
@@ -49,8 +48,8 @@ impl TelemetryDriver {
     ///
     /// Returns `None` if the server wasn't spawned.
     #[cfg(feature = "telemetry-server")]
-    pub fn server_addr(&self) -> Option<SocketAddr> {
-        self.server_addr
+    pub fn server_addr(&self) -> Option<String> {
+        self.server_addr.clone()
     }
 
     /// Instructs the telemetry driver and server to perform an orderly shutdown when the given

foundations/src/telemetry/server/mod.rs

@@ -1,6 +1,6 @@
 #[cfg(feature = "metrics")]
 use super::metrics;
-use super::settings::TelemetrySettings;
+use super::settings::{TelemetryServerAddr, TelemetrySettings};
 use crate::telemetry::log;
 use crate::BootstrapResult;
 use anyhow::Context as _;
@@ -14,18 +14,33 @@ use std::net::SocketAddr;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::task::{Context, Poll};
+use tokio::io::{AsyncRead, AsyncWrite};
 use tokio::net::TcpListener;
+#[cfg(unix)]
+use tokio::net::{TcpStream, UnixListener, UnixStream};
 use tokio::sync::watch;
 
 mod router;
 
 use router::Router;
+
+trait AsyncReadWrite: AsyncRead + AsyncWrite + Send + Unpin + 'static {}
+
+impl AsyncReadWrite for TcpStream {}
+#[cfg(unix)]
+impl AsyncReadWrite for UnixStream {}
+
+enum TelemetryListener {
+    Tcp(TcpListener),
+    #[cfg(unix)]
+    Unix(UnixListener),
+}
 pub use router::{
     BoxError, TelemetryRouteHandler, TelemetryRouteHandlerFuture, TelemetryServerRoute,
 };
 
 pub(super) struct TelemetryServerFuture {
-    listener: TcpListener,
+    listener: TelemetryListener,
     router: Router,
 }
 
@@ -47,27 +62,52 @@ impl TelemetryServerFuture {
                 .map_err(|err| anyhow::anyhow!(err))?;
         }
 
-        let addr = settings.server.addr;
-
-        #[cfg(feature = "settings")]
-        let addr = SocketAddr::from(addr);
-
-        let router = Router::new(custom_routes, settings);
-
-        let listener = {
-            let std_listener = std::net::TcpListener::from(
-                bind_socket(addr).with_context(|| format!("binding to socket {addr:?}"))?,
-            );
+        let router = Router::new(custom_routes, settings.clone());
+
+        let listener = match &settings.server.addr {
+            TelemetryServerAddr::Tcp(addr) => {
+                #[cfg(feature = "settings")]
+                let addr = SocketAddr::from(*addr);
+
+                let std_listener = std::net::TcpListener::from(
+                    bind_socket(addr).with_context(|| format!("binding to TCP socket {addr:?}"))?,
+                );
+                std_listener.set_nonblocking(true)?;
+                let tokio_listener = tokio::net::TcpListener::from_std(std_listener)?;
+                TelemetryListener::Tcp(tokio_listener)
+            }
+            #[cfg(unix)]
+            TelemetryServerAddr::Unix(path) => {
+                if let Some(parent) = path.parent() {
+                    if !parent.exists() {
+                        std::fs::create_dir_all(parent).with_context(|| {
+                            format!("creating parent directories for Unix socket {path:?}")
+                        })?;
+                    }
+                }
 
-            std_listener.set_nonblocking(true)?;
+                if path.exists() {
+                    std::fs::remove_file(path)
+                        .with_context(|| format!("removing existing Unix socket {path:?}"))?;
+                }
 
-            tokio::net::TcpListener::from_std(std_listener)?
+                let unix_listener = UnixListener::bind(path)
+                    .with_context(|| format!("binding to Unix socket {path:?}"))?;
+                TelemetryListener::Unix(unix_listener)
+            }
         };
 
         Ok(Some(TelemetryServerFuture { listener, router }))
     }
-    pub(super) fn local_addr(&self) -> SocketAddr {
-        self.listener.local_addr().unwrap()
+    pub(super) fn local_addr(&self) -> BootstrapResult<String> {
+        match &self.listener {
+            TelemetryListener::Tcp(listener) => Ok(listener.local_addr()?.to_string()),
+            #[cfg(unix)]
+            TelemetryListener::Unix(listener) => match listener.local_addr()?.as_pathname() {
+                Some(path) => Ok(path.display().to_string()),
+                None => Ok("<unnamed>".to_string()),
+            },
+        }
     }
 
     // Adapted from Hyper 0.14 Server stuff and axum::serve::serve.
@@ -87,15 +127,21 @@ impl TelemetryServerFuture {
         let (close_tx, close_rx) = watch::channel(());
         let listener = self.listener;
 
-        pin_mut!(listener);
-
         loop {
             let socket = tokio::select! {
-                conn = listener.accept() => match conn {
+                conn = async {
+                    match &listener {
+                        TelemetryListener::Tcp(listener) => listener.accept().await.map(|(conn, addr)| (Box::new(conn) as Box<dyn AsyncReadWrite>, addr.to_string())),
+                        #[cfg(unix)]
+                        TelemetryListener::Unix(listener) => listener.accept().await.map(|(conn, addr)| {
+                            let addr_str = addr.as_pathname().map(|p| p.display().to_string()).unwrap_or_else(|| "<unnamed>".to_string());
+                            (Box::new(conn) as Box<dyn AsyncReadWrite>, addr_str)
+                        }),
+                    }
+                } => match conn {
                     Ok((conn, _)) => TokioIo::new(conn),
                     Err(e) => {
                         log::warn!("failed to accept connection"; "error" => e);
-
                         continue;
                     }
                 },
@@ -140,13 +186,22 @@ impl Future for TelemetryServerFuture {
         let this = &mut *self;
 
         loop {
-            let socket = match ready!(Pin::new(&mut this.listener).poll_accept(cx)) {
-                Ok((conn, _)) => TokioIo::new(conn),
-                Err(e) => {
-                    log::warn!("failed to accept connection"; "error" => e);
-
-                    continue;
-                }
+            let socket = match &mut this.listener {
+                TelemetryListener::Tcp(listener) => match ready!(listener.poll_accept(cx)) {
+                    Ok((conn, _)) => TokioIo::new(Box::new(conn) as Box<dyn AsyncReadWrite>),
+                    Err(e) => {
+                        log::warn!("failed to accept connection"; "error" => e);
+                        continue;
+                    }
+                },
+                #[cfg(unix)]
+                TelemetryListener::Unix(listener) => match ready!(listener.poll_accept(cx)) {
+                    Ok((conn, _)) => TokioIo::new(Box::new(conn) as Box<dyn AsyncReadWrite>),
+                    Err(e) => {
+                        log::warn!("failed to accept connection"; "error" => e);
+                        continue;
+                    }
+                },
             };
 
             let router = this.router.clone();

foundations/src/telemetry/settings/server.rs

@@ -5,6 +5,53 @@ use crate::settings::settings;
 use std::net::Ipv4Addr;
 #[cfg(not(feature = "settings"))]
 use std::net::SocketAddr;
+use std::path::PathBuf;
+
+/// Telemetry server address - can be either TCP or Unix socket
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
+#[serde(untagged)]
+pub enum TelemetryServerAddr {
+    /// TCP socket address
+    Tcp(SocketAddr),
+    /// Unix domain socket path
+    #[cfg(unix)]
+    Unix(PathBuf),
+}
+
+impl Default for TelemetryServerAddr {
+    fn default() -> Self {
+        let addr: std::net::SocketAddr = (Ipv4Addr::LOCALHOST, 0).into();
+
+        #[cfg(feature = "settings")]
+        let addr = addr.into();
+
+        TelemetryServerAddr::Tcp(addr)
+    }
+}
+
+impl From<SocketAddr> for TelemetryServerAddr {
+    fn from(addr: SocketAddr) -> Self {
+        TelemetryServerAddr::Tcp(addr)
+    }
+}
+
+impl From<std::net::SocketAddr> for TelemetryServerAddr {
+    fn from(addr: std::net::SocketAddr) -> Self {
+        #[cfg(feature = "settings")]
+        let addr = addr.into();
+        TelemetryServerAddr::Tcp(addr)
+    }
+}
+
+#[cfg(unix)]
+impl From<PathBuf> for TelemetryServerAddr {
+    fn from(path: PathBuf) -> Self {
+        TelemetryServerAddr::Unix(path)
+    }
+}
+
+#[cfg(feature = "settings")]
+impl crate::settings::Settings for TelemetryServerAddr {}
 
 /// Telemetry server settings.
 #[cfg_attr(feature = "settings", settings(crate_path = "crate"))]
@@ -16,7 +63,7 @@ pub struct TelemetryServerSettings {
 
     /// Telemetry server address.
     #[serde(default = "TelemetryServerSettings::default_addr")]
-    pub addr: SocketAddr,
+    pub addr: TelemetryServerAddr,
 }
 
 #[cfg(not(feature = "settings"))]
@@ -34,12 +81,12 @@ impl TelemetryServerSettings {
         true
     }
 
-    fn default_addr() -> SocketAddr {
+    fn default_addr() -> TelemetryServerAddr {
         let addr: std::net::SocketAddr = (Ipv4Addr::LOCALHOST, 0).into();
 
         #[cfg(feature = "settings")]
         let addr = addr.into();
 
-        addr
+        TelemetryServerAddr::Tcp(addr)
     }
 }