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:

examples/http_server/main.rs

@@ -56,7 +56,7 @@ async fn main() -> BootstrapResult<()> {
         custom_server_routes: vec![],
     })?;
 
-    if let Some(tele_serv_addr) = tele_driver.server_addr() {
+    if let Some(tele_serv_addr) = tele_driver.addr() {
         log::info!("Telemetry server is listening on http://{}", tele_serv_addr);
     }
 

foundations/src/lib.rs

@@ -69,7 +69,7 @@ mod utils;
 #[cfg(feature = "cli")]
 pub mod cli;
 
-#[cfg(feature = "settings")]
+#[cfg(any(feature = "settings", feature = "telemetry-server"))]
 pub mod settings;
 
 #[cfg(any(

foundations/src/settings/addr.rs

@@ -0,0 +1,72 @@
+//! Network address types that support both TCP and Unix domain sockets.
+//!
+//! This module provides the [`ListenAddr`] enum, a flexible address type that can represent
+//! either TCP socket addresses or Unix domain socket paths.
+
+use super::net::SocketAddr;
+use super::Settings;
+use serde::{Deserialize, Serialize};
+use std::fmt;
+
+/// Address that can be either TCP socket or Unix domain socket endpoint
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(untagged)]
+pub enum ListenAddr {
+    /// TCP network socket address
+    Tcp(SocketAddr),
+    /// Unix domain socket path
+    #[cfg(unix)]
+    Unix(std::path::PathBuf),
+}
+
+impl Default for ListenAddr {
+    fn default() -> Self {
+        let addr: std::net::SocketAddr = (std::net::Ipv4Addr::LOCALHOST, 0).into();
+        ListenAddr::Tcp(addr.into())
+    }
+}
+
+impl From<SocketAddr> for ListenAddr {
+    fn from(addr: SocketAddr) -> Self {
+        ListenAddr::Tcp(addr)
+    }
+}
+
+impl From<std::net::SocketAddr> for ListenAddr {
+    fn from(addr: std::net::SocketAddr) -> Self {
+        ListenAddr::Tcp(addr.into())
+    }
+}
+
+#[cfg(unix)]
+impl From<std::path::PathBuf> for ListenAddr {
+    fn from(path: std::path::PathBuf) -> Self {
+        ListenAddr::Unix(path)
+    }
+}
+
+#[cfg(unix)]
+impl From<&str> for ListenAddr {
+    fn from(path: &str) -> Self {
+        ListenAddr::Unix(std::path::PathBuf::from(path))
+    }
+}
+
+#[cfg(unix)]
+impl From<String> for ListenAddr {
+    fn from(path: String) -> Self {
+        ListenAddr::Unix(std::path::PathBuf::from(path))
+    }
+}
+
+impl fmt::Display for ListenAddr {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            ListenAddr::Tcp(addr) => write!(f, "{addr}"),
+            #[cfg(unix)]
+            ListenAddr::Unix(path) => write!(f, "{}", path.display()),
+        }
+    }
+}
+
+impl Settings for ListenAddr {}

foundations/src/settings/mod.rs

@@ -142,6 +142,7 @@
 
 mod basic_impls;
 
+pub mod addr;
 pub mod collections;
 pub mod net;
 

foundations/src/telemetry/driver.rs

@@ -9,7 +9,8 @@ use std::task::{Context, Poll};
 
 feature_use!(cfg(feature = "telemetry-server"), {
     use super::server::TelemetryServerFuture;
-    use std::net::SocketAddr;
+    use crate::settings::addr::ListenAddr;
+    use crate::settings::net::SocketAddr;
 });
 
 /// A future that drives async telemetry functionality and that is returned
@@ -21,7 +22,7 @@ feature_use!(cfg(feature = "telemetry-server"), {
 /// [security syscall-related]: `crate::security`
 pub struct TelemetryDriver {
     #[cfg(feature = "telemetry-server")]
-    server_addr: Option<SocketAddr>,
+    addr: Option<ListenAddr>,
 
     #[cfg(feature = "telemetry-server")]
     server_fut: Option<TelemetryServerFuture>,
@@ -36,7 +37,7 @@ impl TelemetryDriver {
     ) -> Self {
         Self {
             #[cfg(feature = "telemetry-server")]
-            server_addr: server_fut.as_ref().map(|fut| fut.local_addr()),
+            addr: server_fut.as_ref().and_then(|fut| fut.local_addr().ok()),
 
             #[cfg(feature = "telemetry-server")]
             server_fut,
@@ -47,10 +48,23 @@ impl TelemetryDriver {
 
     /// Address of the telemetry server.
     ///
-    /// Returns `None` if the server wasn't spawned.
+    /// Returns `None` if the server wasn't spawned or if it was spawned with a Unix socket.
     #[cfg(feature = "telemetry-server")]
+    #[deprecated(since = "4.6.0", note = "addr should be used instead")]
     pub fn server_addr(&self) -> Option<SocketAddr> {
-        self.server_addr
+        match self.addr.as_ref()? {
+            ListenAddr::Tcp(addr) => Some(*addr),
+            #[cfg(unix)]
+            ListenAddr::Unix(_) => None,
+        }
+    }
+
+    /// Address of the temeletry server.
+    ///
+    /// Returns `None` if the server weasn't spawned.
+    #[cfg(feature = "telemetry-server")]
+    pub fn addr(&self) -> Option<&ListenAddr> {
+        self.addr.as_ref()
     }
 
     /// Instructs the telemetry driver and server to perform an orderly shutdown when the given

foundations/src/telemetry/server/mod.rs

@@ -1,6 +1,7 @@
 #[cfg(feature = "metrics")]
 use super::metrics;
 use super::settings::TelemetrySettings;
+use crate::settings::addr::ListenAddr;
 use crate::telemetry::log;
 use crate::BootstrapResult;
 use anyhow::Context as _;
@@ -14,18 +15,127 @@ 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;
+
+enum TelemetryStream {
+    Tcp(TcpStream),
+    #[cfg(unix)]
+    Unix(UnixStream),
+}
+
+impl AsyncRead for TelemetryStream {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut tokio::io::ReadBuf<'_>,
+    ) -> Poll<std::io::Result<()>> {
+        match self.get_mut() {
+            TelemetryStream::Tcp(stream) => Pin::new(stream).poll_read(cx, buf),
+            #[cfg(unix)]
+            TelemetryStream::Unix(stream) => Pin::new(stream).poll_read(cx, buf),
+        }
+    }
+}
+
+impl AsyncWrite for TelemetryStream {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<Result<usize, std::io::Error>> {
+        match self.get_mut() {
+            TelemetryStream::Tcp(stream) => Pin::new(stream).poll_write(cx, buf),
+            #[cfg(unix)]
+            TelemetryStream::Unix(stream) => Pin::new(stream).poll_write(cx, buf),
+        }
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
+        match self.get_mut() {
+            TelemetryStream::Tcp(stream) => Pin::new(stream).poll_flush(cx),
+            #[cfg(unix)]
+            TelemetryStream::Unix(stream) => Pin::new(stream).poll_flush(cx),
+        }
+    }
+
+    fn poll_shutdown(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<(), std::io::Error>> {
+        match self.get_mut() {
+            TelemetryStream::Tcp(stream) => Pin::new(stream).poll_shutdown(cx),
+            #[cfg(unix)]
+            TelemetryStream::Unix(stream) => Pin::new(stream).poll_shutdown(cx),
+        }
+    }
+}
+
+enum TelemetryListener {
+    Tcp(TcpListener),
+    #[cfg(unix)]
+    Unix(UnixListener),
+}
+
+impl TelemetryListener {
+    pub(crate) fn local_addr(&self) -> BootstrapResult<ListenAddr> {
+        match self {
+            TelemetryListener::Tcp(listener) => Ok(listener.local_addr()?.into()),
+            #[cfg(unix)]
+            TelemetryListener::Unix(listener) => match listener.local_addr()?.as_pathname() {
+                Some(path) => Ok(path.to_path_buf().into()),
+                None => Err(anyhow::anyhow!("unix socket listener has no pathname")),
+            },
+        }
+    }
+
+    pub(crate) async fn accept(&self) -> std::io::Result<TelemetryStream> {
+        match self {
+            TelemetryListener::Tcp(listener) => listener
+                .accept()
+                .await
+                .map(|(conn, _)| TelemetryStream::Tcp(conn)),
+            #[cfg(unix)]
+            TelemetryListener::Unix(listener) => listener
+                .accept()
+                .await
+                .map(|(conn, _)| TelemetryStream::Unix(conn)),
+        }
+    }
+
+    pub(crate) fn poll_accept(
+        &mut self,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<std::io::Result<TelemetryStream>> {
+        match self {
+            TelemetryListener::Tcp(listener) => match std::task::ready!(listener.poll_accept(cx)) {
+                Ok((conn, _)) => std::task::Poll::Ready(Ok(TelemetryStream::Tcp(conn))),
+                Err(e) => std::task::Poll::Ready(Err(e)),
+            },
+            #[cfg(unix)]
+            TelemetryListener::Unix(listener) => {
+                match std::task::ready!(listener.poll_accept(cx)) {
+                    Ok((conn, _)) => std::task::Poll::Ready(Ok(TelemetryStream::Unix(conn))),
+                    Err(e) => std::task::Poll::Ready(Err(e)),
+                }
+            }
+        }
+    }
+}
+
 pub use router::{
     BoxError, TelemetryRouteHandler, TelemetryRouteHandlerFuture, TelemetryServerRoute,
 };
 
 pub(super) struct TelemetryServerFuture {
-    listener: TcpListener,
+    listener: TelemetryListener,
     router: Router,
 }
 
@@ -47,27 +157,40 @@ 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:?}"))?,
-            );
-
-            std_listener.set_nonblocking(true)?;
+        let router = Router::new(custom_routes, Arc::clone(&settings));
+
+        let listener = match &settings.server.addr {
+            ListenAddr::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)]
+            ListenAddr::Unix(path) => {
+                // Remove existing socket file if it exists to avoid bind errors
+                if path.exists() {
+                    if let Err(e) = std::fs::remove_file(path) {
+                        log::warn!("failed to remove existing Unix socket file"; "path" => %path.display(), "error" => e);
+                    }
+                }
 
-            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<ListenAddr> {
+        self.listener.local_addr()
     }
 
     // Adapted from Hyper 0.14 Server stuff and axum::serve::serve.
@@ -87,15 +210,12 @@ 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 {
-                    Ok((conn, _)) => TokioIo::new(conn),
+                    Ok(conn) => TokioIo::new(conn),
                     Err(e) => {
                         log::warn!("failed to accept connection"; "error" => e);
-
                         continue;
                     }
                 },
@@ -140,11 +260,10 @@ 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),
+            let socket = match ready!(this.listener.poll_accept(cx)) {
+                Ok(conn) => TokioIo::new(conn),
                 Err(e) => {
                     log::warn!("failed to accept connection"; "error" => e);
-
                     continue;
                 }
             };