From 5aa2a6b1ebe21e7ca314d5f08dcf371b9bca60c8 Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Fri, 9 Oct 2026 17:27:20 -0400 Subject: [PATCH 1/2] feat(dogstatsd): listen on UDP and the named pipe together When DD_DOGSTATSD_WINDOWS_PIPE_NAME is set, dogstatsd now keeps its UDP listener alongside the pipe, as #94 did for the trace agent. Clients that can't use a pipe, such as older java-dogstatsd-client versions, still reach the binary over 8125. With a pipe set, the UDP bind is best-effort, as #129 did for TCP 8126. Several function apps share one VM on Elastic Premium and Dedicated plans, so a second app's binary finds 8125 taken and must keep running on its pipe. Without a pipe, a failed bind still panics. datadog-serverless-compat no longer forces the dogstatsd port to 0 when a pipe is set, and logs both listeners. SVLS-9467 --- crates/datadog-serverless-compat/src/main.rs | 17 ++- crates/dogstatsd/src/dogstatsd.rs | 103 +++++++++++++++---- crates/dogstatsd/tests/integration_test.rs | 72 +++++++++++++ 3 files changed, 160 insertions(+), 32 deletions(-) diff --git a/crates/datadog-serverless-compat/src/main.rs b/crates/datadog-serverless-compat/src/main.rs index cdcdcabb..cdb4005d 100644 --- a/crates/datadog-serverless-compat/src/main.rs +++ b/crates/datadog-serverless-compat/src/main.rs @@ -104,14 +104,12 @@ pub async fn main() { None } }; - let dd_dogstatsd_port: u16 = if dd_dogstatsd_windows_pipe_name.is_some() { - 0 // Override to 0 when using Windows named pipe - } else { - env::var("DD_DOGSTATSD_PORT") - .ok() - .and_then(|port| port.parse::().ok()) - .unwrap_or(DEFAULT_DOGSTATSD_PORT) - }; + // UDP listener also runs when a named pipe is set, so clients without + // named-pipe support can still send metrics. + let dd_dogstatsd_port: u16 = env::var("DD_DOGSTATSD_PORT") + .ok() + .and_then(|port| port.parse::().ok()) + .unwrap_or(DEFAULT_DOGSTATSD_PORT); let dd_site = env::var("DD_SITE").unwrap_or_else(|_| "datadoghq.com".to_string()); let dd_use_dogstatsd = env::var("DD_USE_DOGSTATSD") .map(|val| val.to_lowercase() != "false") @@ -251,10 +249,9 @@ pub async fn main() { dd_dogstatsd_windows_pipe_name.clone(), ) .await; + info!("dogstatsd-udp: starting to listen on port {dd_dogstatsd_port}"); if let Some(ref windows_pipe_name) = dd_dogstatsd_windows_pipe_name { info!("dogstatsd-pipe: starting to listen on pipe {windows_pipe_name}"); - } else { - info!("dogstatsd-udp: starting to listen on port {dd_dogstatsd_port}"); } } else { info!("dogstatsd disabled"); diff --git a/crates/dogstatsd/src/dogstatsd.rs b/crates/dogstatsd/src/dogstatsd.rs index ada4a578..c314343b 100644 --- a/crates/dogstatsd/src/dogstatsd.rs +++ b/crates/dogstatsd/src/dogstatsd.rs @@ -17,7 +17,10 @@ use tracing::{debug, error, trace}; // Windows-specific imports #[cfg(all(windows, feature = "windows-pipes"))] -use {std::sync::Arc, tokio::io::AsyncReadExt, tokio::net::windows::named_pipe::ServerOptions}; +use { + std::sync::Arc, tokio::io::AsyncReadExt, tokio::net::windows::named_pipe::ServerOptions, + tracing::warn, +}; // Default buffer size for receiving DogStatsD packets (one read call). // Used for both UDP recv_from and Windows named pipe reads. @@ -31,7 +34,8 @@ const DEFAULT_QUEUE_SIZE: usize = 1024; pub struct DogStatsDConfig { /// Host to bind UDP socket to (e.g., "127.0.0.1") pub host: String, - /// Port to bind UDP socket to (e.g., 8125), will be 0 if we're using a Named Pipe + /// Port to bind UDP socket to (e.g., 8125). The UDP socket is also bound when a named pipe + /// is configured, so clients without named-pipe support can still send metrics. pub port: u16, /// Optional namespace to prepend to all metric names (e.g., "myapp") pub metric_namespace: Option, @@ -94,11 +98,14 @@ enum BufferReader { #[allow(dead_code)] MirrorTest(Vec, SocketAddr), - /// Windows named pipe reader (Windows-only transport) + /// Windows named pipe reader (Windows-only transport), read alongside the UDP socket. + /// `udp_socket` is `None` when the UDP bind failed, e.g. because another agent on a + /// shared host already owns the port. #[cfg(all(windows, feature = "windows-pipes"))] NamedPipe { pipe_name: Arc, receiver: Arc>>>, + udp_socket: Option, }, } @@ -121,17 +128,34 @@ impl BufferReader { BufferReader::NamedPipe { pipe_name, receiver, - } => match receiver.lock().await.recv().await { - Some(data) => { - let len = data.len().min(buf.len()); - buf[..len].copy_from_slice(&data[..len]); - Ok((len, MessageSource::NamedPipe(pipe_name.clone()))) + udp_socket, + } => { + // Wait on whichever transport delivers first. The UDP arm stays pending + // when the UDP bind failed. + let pipe_data = tokio::select! { + udp_result = async { + match udp_socket.as_ref() { + Some(socket) => socket.recv_from(buf).await, + None => std::future::pending().await, + } + } => { + let (amt, src) = udp_result?; + return Ok((amt, MessageSource::Network(src))); + } + pipe_data = async { receiver.lock().await.recv().await } => pipe_data, + }; + match pipe_data { + Some(data) => { + let len = data.len().min(buf.len()); + buf[..len].copy_from_slice(&data[..len]); + Ok((len, MessageSource::NamedPipe(pipe_name.clone()))) + } + None => Err(std::io::Error::new( + std::io::ErrorKind::ConnectionReset, + "named pipe channel closed", + )), } - None => Err(std::io::Error::new( - std::io::ErrorKind::ConnectionReset, - "named pipe channel closed", - )), - }, + } } } @@ -141,17 +165,30 @@ impl BufferReader { /// kernel buffer without re-entering tokio's event loop. fn try_read_into(&mut self, buf: &mut [u8]) -> std::io::Result> { match self { - BufferReader::UdpSocket(socket) => match socket.try_recv_from(buf) { - Ok((amt, src)) => Ok(Some((amt, MessageSource::Network(src)))), - Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => Ok(None), - Err(e) => Err(e), - }, + BufferReader::UdpSocket(socket) => try_recv_udp(socket, buf), + // Drain only the UDP side; the named pipe channel is read by `read_into`. + #[cfg(all(windows, feature = "windows-pipes"))] + BufferReader::NamedPipe { + udp_socket: Some(socket), + .. + } => try_recv_udp(socket, buf), // Non-UDP transports don't support non-blocking reads _ => Ok(None), } } } +fn try_recv_udp( + socket: &tokio::net::UdpSocket, + buf: &mut [u8], +) -> std::io::Result> { + match socket.try_recv_from(buf) { + Ok((amt, src)) => Ok(Some((amt, MessageSource::Network(src)))), + Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => Ok(None), + Err(e) => Err(e), + } +} + /// DogStatsD server to receive, parse, and forward metrics. pub struct DogStatsD { cancel_token: tokio_util::sync::CancellationToken, @@ -165,7 +202,8 @@ pub struct DogStatsD { impl DogStatsD { /// Creates a new DogStatsD server instance. /// - /// The server will bind to either a UDP socket or Windows named pipe based on the config. + /// The server always binds a UDP socket and also listens on a Windows named pipe when one + /// is configured. /// Metrics received will be forwarded to the provided aggregator_handle. #[must_use] pub async fn new( @@ -191,10 +229,30 @@ impl DogStatsD { None => DEFAULT_BUFFER_SIZE, }; + let addr = format!("{}:{}", config.host, config.port); + let buffer_reader = if let Some(pipe_name_ref) = pipe_name_opt { - // Windows named pipe transport + // Windows named pipe transport, plus UDP for clients without named-pipe support #[cfg(all(windows, feature = "windows-pipes"))] { + // UDP bind is best-effort when a named pipe is configured. Multiple Azure apps + // share the same VM on certain Windows Azure plans, and only one agent can own + // the port. The named pipe is unique per app, so a UDP bind failure is not fatal. + let udp_socket = match create_udp_socket(&addr, config.so_rcvbuf).await { + Ok(socket) => { + debug!("DogStatsD listening on UDP {}", addr); + Some(socket) + } + Err(e) => { + warn!( + "DogStatsD could not bind UDP {}: {}. Named-pipe transport is active; \ + metrics sent over UDP will not be received by this agent.", + addr, e + ); + None + } + }; + let pipe_name = Arc::new(pipe_name_ref.clone()); // Create channel for receiving data from client handlers @@ -213,6 +271,7 @@ impl DogStatsD { BufferReader::NamedPipe { pipe_name, receiver, + udp_socket, } } #[cfg(not(all(windows, feature = "windows-pipes")))] @@ -225,8 +284,8 @@ impl DogStatsD { ); } } else { - // UDP socket for all platforms - let addr = format!("{}:{}", config.host, config.port); + // UDP socket for all platforms. Without a named pipe, UDP is the only transport, so a + // bind failure is fatal. // TODO (UDS socket) #[allow(clippy::expect_used)] let socket = create_udp_socket(&addr, config.so_rcvbuf) diff --git a/crates/dogstatsd/tests/integration_test.rs b/crates/dogstatsd/tests/integration_test.rs index f390fa48..3d8f78b2 100644 --- a/crates/dogstatsd/tests/integration_test.rs +++ b/crates/dogstatsd/tests/integration_test.rs @@ -548,6 +548,78 @@ async fn test_named_pipe_basic_communication() { handle.shutdown().expect("shutdown failed"); } +#[cfg(test)] +#[cfg(all(windows, feature = "windows-pipes"))] +#[tokio::test] +async fn test_named_pipe_and_udp_dual_listen() { + let pipe_name = r"\\.\pipe\test_dogstatsd_dual_listen"; + let udp_port: u16 = 18130; + let (service, handle) = AggregatorService::new(SortedTags::parse("test:value").unwrap(), 1_024) + .expect("aggregator service creation failed"); + tokio::spawn(service.run()); + + let cancel_token = CancellationToken::new(); + + // Both transports are deliberately set on the same server: a non-zero UDP + // port AND a pipe name. They must come up concurrently. + let dogstatsd_task = { + let handle = handle.clone(); + let cancel_token = cancel_token.clone(); + tokio::spawn(async move { + let dogstatsd = DogStatsD::new( + &DogStatsDConfig { + host: "127.0.0.1".to_string(), + port: udp_port, + metric_namespace: None, + windows_pipe_name: Some(pipe_name.to_string()), + so_rcvbuf: None, + buffer_size: None, + queue_size: None, + }, + handle, + cancel_token, + ) + .await; + dogstatsd.spin().await; + }) + }; + + sleep(Duration::from_millis(100)).await; + + // One metric per transport, distinguishable by metric name. + let socket = UdpSocket::bind("0.0.0.0:0").await.expect("bind failed"); + socket + .send_to(b"dual.udp.metric:1|c", ("127.0.0.1", udp_port)) + .await + .expect("UDP send failed"); + + let mut client = ClientOptions::new().open(pipe_name).expect("client open"); + client + .write_all(b"dual.pipe.metric:1|c\n") + .await + .expect("pipe write failed"); + client.flush().await.expect("pipe flush failed"); + + sleep(Duration::from_millis(100)).await; + + let response = handle.flush().await.expect("flush failed"); + let series_json = serde_json::to_string(&response.series).expect("serialize series"); + assert!( + series_json.contains("dual.udp.metric"), + "UDP metric was not received when a pipe name is also configured: {series_json}" + ); + assert!( + series_json.contains("dual.pipe.metric"), + "pipe metric was not received when UDP is also listening: {series_json}" + ); + + // Cleanup + cancel_token.cancel(); + let result = timeout(Duration::from_millis(500), dogstatsd_task).await; + assert!(result.is_ok(), "task should complete after cancellation"); + handle.shutdown().expect("shutdown failed"); +} + #[cfg(test)] #[cfg(all(windows, feature = "windows-pipes"))] #[tokio::test] From da8801ab8d319dc504aa2ff258cfc4523ce74f2a Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Fri, 9 Oct 2026 17:38:16 -0400 Subject: [PATCH 2/2] fix(dogstatsd): log the UDP listener only when it binds With a pipe set, main.rs logged "starting to listen on port 8125" even when the best-effort UDP bind had failed. The dogstatsd crate now logs the UDP listener at info when it binds, next to its existing warning when it doesn't, and main.rs logs only the pipe in that case. SVLS-9467 --- crates/datadog-serverless-compat/src/main.rs | 4 +++- crates/dogstatsd/src/dogstatsd.rs | 8 +++++--- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/crates/datadog-serverless-compat/src/main.rs b/crates/datadog-serverless-compat/src/main.rs index cdb4005d..5a1ff0a6 100644 --- a/crates/datadog-serverless-compat/src/main.rs +++ b/crates/datadog-serverless-compat/src/main.rs @@ -249,9 +249,11 @@ pub async fn main() { dd_dogstatsd_windows_pipe_name.clone(), ) .await; - info!("dogstatsd-udp: starting to listen on port {dd_dogstatsd_port}"); if let Some(ref windows_pipe_name) = dd_dogstatsd_windows_pipe_name { + // The dogstatsd crate logs whether the best-effort UDP bind succeeded. info!("dogstatsd-pipe: starting to listen on pipe {windows_pipe_name}"); + } else { + info!("dogstatsd-udp: starting to listen on port {dd_dogstatsd_port}"); } } else { info!("dogstatsd disabled"); diff --git a/crates/dogstatsd/src/dogstatsd.rs b/crates/dogstatsd/src/dogstatsd.rs index c314343b..963d31ac 100644 --- a/crates/dogstatsd/src/dogstatsd.rs +++ b/crates/dogstatsd/src/dogstatsd.rs @@ -18,8 +18,10 @@ use tracing::{debug, error, trace}; // Windows-specific imports #[cfg(all(windows, feature = "windows-pipes"))] use { - std::sync::Arc, tokio::io::AsyncReadExt, tokio::net::windows::named_pipe::ServerOptions, - tracing::warn, + std::sync::Arc, + tokio::io::AsyncReadExt, + tokio::net::windows::named_pipe::ServerOptions, + tracing::{info, warn}, }; // Default buffer size for receiving DogStatsD packets (one read call). @@ -240,7 +242,7 @@ impl DogStatsD { // the port. The named pipe is unique per app, so a UDP bind failure is not fatal. let udp_socket = match create_udp_socket(&addr, config.so_rcvbuf).await { Ok(socket) => { - debug!("DogStatsD listening on UDP {}", addr); + info!("dogstatsd-udp: starting to listen on {}", addr); Some(socket) } Err(e) => {