From 1954990f3280d140d0b83b92c57deb9436bce29a Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Fri, 9 Oct 2026 17:27:20 -0400 Subject: [PATCH 01/11] 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 1829fccf405afdddb224771ceb999bcb0b96642d Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Fri, 9 Oct 2026 17:38:16 -0400 Subject: [PATCH 02/11] 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) => { From e0d97bfb467e2ee93d3dd7bc16f6485ffcb57141 Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Thu, 22 Oct 2026 13:45:54 -0400 Subject: [PATCH 03/11] feat(dogstatsd): turn UDP off for port 0 when a pipe is set With a named pipe configured, DD_DOGSTATSD_PORT=0 now skips the UDP bind instead of listening on a random port that no client can find. This matches dogstatsd_port: 0 in the Datadog Agent and gives customers a pipe-only option on shared plans. Without a pipe, port 0 behaves as before. --- crates/datadog-serverless-compat/src/main.rs | 2 +- crates/dogstatsd/src/dogstatsd.rs | 32 ++++++++++++-------- 2 files changed, 20 insertions(+), 14 deletions(-) diff --git a/crates/datadog-serverless-compat/src/main.rs b/crates/datadog-serverless-compat/src/main.rs index 5a1ff0a6..97183456 100644 --- a/crates/datadog-serverless-compat/src/main.rs +++ b/crates/datadog-serverless-compat/src/main.rs @@ -105,7 +105,7 @@ pub async fn main() { } }; // UDP listener also runs when a named pipe is set, so clients without - // named-pipe support can still send metrics. + // named-pipe support can still send metrics. With a pipe, port 0 turns UDP off. let dd_dogstatsd_port: u16 = env::var("DD_DOGSTATSD_PORT") .ok() .and_then(|port| port.parse::().ok()) diff --git a/crates/dogstatsd/src/dogstatsd.rs b/crates/dogstatsd/src/dogstatsd.rs index 963d31ac..173d13e8 100644 --- a/crates/dogstatsd/src/dogstatsd.rs +++ b/crates/dogstatsd/src/dogstatsd.rs @@ -37,7 +37,8 @@ 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). The UDP socket is also bound when a named pipe - /// is configured, so clients without named-pipe support can still send metrics. + /// is configured, so clients without named-pipe support can still send metrics. With a + /// named pipe, port 0 turns UDP off, as `dogstatsd_port: 0` does in the Datadog Agent. pub port: u16, /// Optional namespace to prepend to all metric names (e.g., "myapp") pub metric_namespace: Option, @@ -240,18 +241,23 @@ impl DogStatsD { // 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) => { - info!("dogstatsd-udp: starting to listen on {}", 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 udp_socket = if config.port == 0 { + info!("dogstatsd-udp: port is 0, so only the named pipe is listening"); + None + } else { + match create_udp_socket(&addr, config.so_rcvbuf).await { + Ok(socket) => { + info!("dogstatsd-udp: starting to listen on {}", 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 + } } }; From 7c477627c08ddbebe5c3bcf5ac3975efc245ca05 Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Thu, 22 Oct 2026 16:15:46 -0400 Subject: [PATCH 04/11] feat(dogstatsd): run UDP and the named pipe in separate tasks With a named pipe set, the pipe server and the UDP reader now each feed the processing queue from their own task, as the trace agent's TCP and named-pipe listeners do. Before, one reader waited on both and then drained UDP without yielding, so sustained UDP traffic could starve the pipe while its messages piled up in an unbounded channel. Beside the pipe, the UDP reader: - skips the non-yielding drain, so tokio's cooperative budget makes it yield regularly even on a one-core host such as Azure Functions EP1; - skips connection-level errors (reset, aborted, refused), as the trace agent's accept loops do; - stops on any other error. If either transport stops on its own, DogStatsD logs an error and stops completely instead of running with one transport. Before, a UDP ConnectionReset looked like "pipe closed" and stopped both transports with only a debug log. datadog-serverless-compat now logs an error when DogStatsD stops without being cancelled. Pipe messages go straight into the bounded queue and are dropped when it is full, as UDP packets are. Their buffers don't return to the UDP buffer pool, which would otherwise cut short later UDP packets. Without a named pipe, UDP behaves as before. --- crates/datadog-serverless-compat/src/main.rs | 6 + crates/dogstatsd/src/dogstatsd.rs | 387 +++++++++++++------ crates/dogstatsd/tests/integration_test.rs | 22 ++ 3 files changed, 302 insertions(+), 113 deletions(-) diff --git a/crates/datadog-serverless-compat/src/main.rs b/crates/datadog-serverless-compat/src/main.rs index 97183456..e2f4f6f6 100644 --- a/crates/datadog-serverless-compat/src/main.rs +++ b/crates/datadog-serverless-compat/src/main.rs @@ -458,8 +458,14 @@ async fn start_dogstatsd_listener( ) .await; + let spin_cancel_token = dogstatsd_cancel_token.clone(); tokio::spawn(async move { dogstatsd_client.spin().await; + if !spin_cancel_token.is_cancelled() { + error!( + "dogstatsd: listener stopped unexpectedly; custom metrics are no longer received" + ); + } }); dogstatsd_cancel_token diff --git a/crates/dogstatsd/src/dogstatsd.rs b/crates/dogstatsd/src/dogstatsd.rs index 173d13e8..26728c63 100644 --- a/crates/dogstatsd/src/dogstatsd.rs +++ b/crates/dogstatsd/src/dogstatsd.rs @@ -90,6 +90,16 @@ impl Packet { fn data(&self) -> &[u8] { &self.buf[..self.len] } + + /// Whether the buffer came from the reader's pool and should go back to it. Named-pipe + /// messages bring their own right-sized buffers, which must not be reused for UDP reads. + fn is_pooled(&self) -> bool { + match self.source { + MessageSource::Network(_) => true, + #[cfg(all(windows, feature = "windows-pipes"))] + MessageSource::NamedPipe(_) => false, + } + } } // BufferReader abstracts transport methods for metric data. @@ -100,18 +110,48 @@ enum BufferReader { /// Mirror reader for testing - replays a fixed buffer #[allow(dead_code)] MirrorTest(Vec, SocketAddr), +} + +/// How DogStatsD receives packets. +enum Transport { + /// One reader task for a single transport: UDP, or the test mirror. + Reader(BufferReader), - /// 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. + /// Windows named pipe, plus UDP when it bound. `udp_socket` is `None` when the port is 0 + /// or another agent on a shared host already owns it. Each transport runs in its own task, + /// like the trace agent's TCP and named-pipe listeners. #[cfg(all(windows, feature = "windows-pipes"))] NamedPipe { pipe_name: Arc, - receiver: Arc>>>, udp_socket: Option, }, } +/// How the UDP reader treats read errors. +#[cfg(all(windows, feature = "windows-pipes"))] +#[derive(Clone, Copy)] +enum UdpErrorPolicy { + /// UDP is the only transport: keep the single-transport behavior. + SoleTransport, + /// UDP runs beside the named pipe. Like the trace agent's accept loops, skip + /// connection-level errors and end the reader on any other error, so the supervisor + /// stops DogStatsD with an error log instead of leaving it half running. Also skip the + /// non-yielding drain so the pipe task always gets to run. + BesideNamedPipe, +} + +/// Connection-level errors that the trace agent's accept loops skip. On a Windows UDP socket, +/// `ConnectionReset` reports an ICMP Port Unreachable reply to an earlier send. +#[cfg(all(windows, feature = "windows-pipes"))] +fn is_connection_error(e: &std::io::Error) -> bool { + matches!( + e.kind(), + std::io::ErrorKind::ConnectionAborted + | std::io::ErrorKind::ConnectionReset + | std::io::ErrorKind::ConnectionRefused + ) +} + impl BufferReader { /// Reads one packet into the provided buffer, waiting if none is available. /// Returns `(bytes_read, source)`. The caller owns the buffer and can @@ -127,38 +167,6 @@ impl BufferReader { buf[..len].copy_from_slice(&data[..len]); Ok((len, MessageSource::Network(*addr))) } - #[cfg(all(windows, feature = "windows-pipes"))] - BufferReader::NamedPipe { - pipe_name, - receiver, - 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", - )), - } - } } } @@ -168,35 +176,22 @@ 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) => 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), + 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), + }, // 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, aggregator_handle: AggregatorHandle, - buffer_reader: BufferReader, + transport: Transport, metric_namespace: Option, buf_size: usize, queue_size: usize, @@ -234,7 +229,7 @@ impl DogStatsD { let addr = format!("{}:{}", config.host, config.port); - let buffer_reader = if let Some(pipe_name_ref) = pipe_name_opt { + let transport = if let Some(pipe_name_ref) = pipe_name_opt { // Windows named pipe transport, plus UDP for clients without named-pipe support #[cfg(all(windows, feature = "windows-pipes"))] { @@ -261,24 +256,9 @@ impl DogStatsD { } }; - let pipe_name = Arc::new(pipe_name_ref.clone()); - - // Create channel for receiving data from client handlers - let (sender, receiver) = tokio::sync::mpsc::unbounded_channel(); - let receiver = Arc::new(tokio::sync::Mutex::new(receiver)); - - // Spawn server accept loop - let server_pipe_name = pipe_name.clone(); - let server_cancel = cancel_token.clone(); - let pipe_buf_size = buf_size; - tokio::spawn(async move { - run_named_pipe_server(server_pipe_name, sender, server_cancel, pipe_buf_size) - .await; - }); - - BufferReader::NamedPipe { - pipe_name, - receiver, + // The pipe server starts in `spin`, where the processing queue exists. + Transport::NamedPipe { + pipe_name: Arc::new(pipe_name_ref.clone()), udp_socket, } } @@ -299,7 +279,7 @@ impl DogStatsD { let socket = create_udp_socket(&addr, config.so_rcvbuf) .await .expect("couldn't create UDP socket"); - BufferReader::UdpSocket(socket) + Transport::Reader(BufferReader::UdpSocket(socket)) }; let queue_size = match config.queue_size { @@ -317,7 +297,7 @@ impl DogStatsD { DogStatsD { cancel_token, aggregator_handle, - buffer_reader, + transport, metric_namespace: config.metric_namespace.clone(), buf_size, queue_size, @@ -336,11 +316,14 @@ impl DogStatsD { /// /// If the queue fills up, the reader drops packets rather /// than blocking. + /// + /// With a Windows named pipe, the pipe server and the UDP reader each feed the queue from + /// their own task. See `spin_named_pipe`. pub async fn spin(self) { let DogStatsD { cancel_token, aggregator_handle, - buffer_reader, + transport, metric_namespace, buf_size, queue_size, @@ -350,39 +333,73 @@ impl DogStatsD { // Buffer pool: processor returns used buffers, reader reuses them. // Avoids a heap allocation + zero-fill per packet in steady state. let (pool_tx, pool_rx) = tokio::sync::mpsc::unbounded_channel(); - let reader_pool_tx = pool_tx.clone(); - let reader_token = cancel_token.clone(); - tokio::spawn(async move { - read_loop( - buffer_reader, - tx, - reader_token, - buf_size, - queue_size, - pool_rx, - reader_pool_tx, - ) - .await; - }); + match transport { + Transport::Reader(buffer_reader) => { + let reader_pool_tx = pool_tx.clone(); + let reader_token = cancel_token.clone(); + tokio::spawn(async move { + let _ = read_loop( + buffer_reader, + tx, + reader_token, + buf_size, + queue_size, + pool_rx, + reader_pool_tx, + #[cfg(all(windows, feature = "windows-pipes"))] + UdpErrorPolicy::SoleTransport, + ) + .await; + }); - process_loop( - &mut rx, - &cancel_token, - &aggregator_handle, - metric_namespace.as_deref(), - pool_tx, - ) - .await; + process_loop( + &mut rx, + &cancel_token, + &aggregator_handle, + metric_namespace.as_deref(), + pool_tx, + ) + .await; + } + #[cfg(all(windows, feature = "windows-pipes"))] + Transport::NamedPipe { + pipe_name, + udp_socket, + } => { + spin_named_pipe( + pipe_name, + udp_socket, + QueueEnds { + tx, + rx, + pool_tx, + pool_rx, + }, + &cancel_token, + &aggregator_handle, + metric_namespace.as_deref(), + buf_size, + queue_size, + ) + .await; + } + } } /// Process a single buffer from the transport. Used by tests via `MirrorTest`. #[cfg(test)] async fn consume_statsd(&mut self) { let mut buf = vec![0u8; self.buf_size]; + // Only one variant exists without the Windows named pipe feature. + #[allow(clippy::infallible_destructuring_match)] + let reader = match &mut self.transport { + Transport::Reader(reader) => reader, + #[cfg(all(windows, feature = "windows-pipes"))] + Transport::NamedPipe { .. } => panic!("consume_statsd needs a single reader"), + }; #[allow(clippy::expect_used)] - let (len, src) = self - .buffer_reader + let (len, src) = reader .read_into(&mut buf) .await .expect("didn't receive data"); @@ -408,6 +425,10 @@ impl DogStatsD { /// Uses `is_cancelled()` as a fast-path check at the top of each iteration, /// and `tokio::select!` on the async `read_into()` to ensure the reader exits /// promptly even on a quiet socket (where `read_into` would block forever). +/// +/// Returns an error only when `policy` is `BesideNamedPipe` and a read fails with an error +/// that isn't connection-level; the caller then stops DogStatsD. +#[allow(clippy::too_many_arguments)] async fn read_loop( mut reader: BufferReader, tx: tokio::sync::mpsc::Sender, @@ -416,8 +437,11 @@ async fn read_loop( queue_capacity: usize, mut pool_rx: tokio::sync::mpsc::UnboundedReceiver>, pool_tx: tokio::sync::mpsc::UnboundedSender>, -) { + #[cfg(all(windows, feature = "windows-pipes"))] policy: UdpErrorPolicy, +) -> std::io::Result<()> { let mut dropped: u64 = 0; + #[cfg_attr(not(all(windows, feature = "windows-pipes")), allow(unused_mut))] + let mut exit = Ok(()); while !cancel.is_cancelled() { // Get a buffer from the pool, or allocate if the pool is empty (cold start). let mut buf = pool_rx.try_recv().unwrap_or_else(|_| vec![0u8; buf_size]); @@ -433,6 +457,20 @@ async fn read_loop( }; let (len, source) = match result { Ok(r) => r, + #[cfg(all(windows, feature = "windows-pipes"))] + Err(e) + if matches!(policy, UdpErrorPolicy::BesideNamedPipe) && is_connection_error(&e) => + { + debug!("DogStatsD UDP read skipped a connection error: {}", e); + let _ = pool_tx.send(buf); + continue; + } + #[cfg(all(windows, feature = "windows-pipes"))] + Err(e) if matches!(policy, UdpErrorPolicy::BesideNamedPipe) => { + let _ = pool_tx.send(buf); + exit = Err(e); + break; + } Err(e) if e.kind() == std::io::ErrorKind::ConnectionReset => { debug!("DogStatsD transport closed: {}", e); break; @@ -451,6 +489,14 @@ async fn read_loop( break; } + // Beside the named pipe, skip the drain below. It never yields, so on a one-core host + // (Azure Functions EP1) it would hold the only worker thread and starve the pipe. Each + // async read instead counts against tokio's cooperative budget, which yields regularly. + #[cfg(all(windows, feature = "windows-pipes"))] + if matches!(policy, UdpErrorPolicy::BesideNamedPipe) { + continue; + } + // Drain any packets already in the kernel buffer without going // through tokio's event loop — just raw non-blocking syscalls. loop { @@ -486,6 +532,7 @@ async fn read_loop( dropped, queue_capacity ); } + exit } /// Sends a packet to the channel. Returns `false` if the channel is closed @@ -515,8 +562,8 @@ fn try_send_packet( } /// Receives packets from the channel, parses them, and forwards metrics -/// to the aggregator. After processing each packet, returns the buffer to -/// the pool for reuse by the reader. On cancellation, drains any remaining +/// to the aggregator. After processing each packet, returns a pooled buffer +/// to the pool for reuse by the reader. On cancellation, drains any remaining /// buffered packets before exiting so no already-read data is lost. async fn process_loop( rx: &mut tokio::sync::mpsc::Receiver, @@ -531,7 +578,9 @@ async fn process_loop( match packet { Some(packet) => { process_packet(packet.data(), &packet.source, aggregator, namespace); - let _ = pool.send(packet.buf); + if packet.is_pooled() { + let _ = pool.send(packet.buf); + } } None => break, } @@ -539,7 +588,9 @@ async fn process_loop( _ = cancel.cancelled() => { while let Ok(packet) = rx.try_recv() { process_packet(packet.data(), &packet.source, aggregator, namespace); - let _ = pool.send(packet.buf); + if packet.is_pooled() { + let _ = pool.send(packet.buf); + } } break; } @@ -684,15 +735,104 @@ async fn create_udp_socket( tokio::net::UdpSocket::from_std(std_socket) } +/// Both ends of the processing queue and the buffer pool, handed to `spin_named_pipe`. +#[cfg(all(windows, feature = "windows-pipes"))] +struct QueueEnds { + tx: tokio::sync::mpsc::Sender, + rx: tokio::sync::mpsc::Receiver, + pool_tx: tokio::sync::mpsc::UnboundedSender>, + pool_rx: tokio::sync::mpsc::UnboundedReceiver>, +} + +/// Runs the named pipe server and, when it bound, the UDP reader, each in its own task feeding +/// the processing queue. This follows the trace agent's TCP and named-pipe listeners: heavy UDP +/// traffic can't starve the pipe, and if either transport stops on its own, DogStatsD logs an +/// error and stops completely instead of running with one transport. +#[cfg(all(windows, feature = "windows-pipes"))] +#[allow(clippy::too_many_arguments)] +async fn spin_named_pipe( + pipe_name: Arc, + udp_socket: Option, + queue: QueueEnds, + cancel_token: &tokio_util::sync::CancellationToken, + aggregator: &AggregatorHandle, + namespace: Option<&str>, + buf_size: usize, + queue_size: usize, +) { + let QueueEnds { + tx, + mut rx, + pool_tx, + pool_rx, + } = queue; + // Cancelled by the caller's token, or here when a transport stops on its own. + let stop = cancel_token.child_token(); + let mut transports = tokio::task::JoinSet::new(); + + if let Some(socket) = udp_socket { + let (tx, stop, pool_tx) = (tx.clone(), stop.clone(), pool_tx.clone()); + transports.spawn(async move { + let result = read_loop( + BufferReader::UdpSocket(socket), + tx, + stop, + buf_size, + queue_size, + pool_rx, + pool_tx, + UdpErrorPolicy::BesideNamedPipe, + ) + .await; + ("UDP", result) + }); + } + { + let stop = stop.clone(); + transports.spawn(async move { + run_named_pipe_server(pipe_name, tx, stop, buf_size).await; + ("named pipe", Ok(())) + }); + } + + let supervise = async { + while let Some(joined) = transports.join_next().await { + if stop.is_cancelled() { + continue; + } + match joined { + Ok((transport, Err(e))) => { + error!("DogStatsD {transport} listener failed: {e}. Stopping DogStatsD.") + } + Ok((transport, Ok(()))) => { + error!( + "DogStatsD {transport} listener stopped unexpectedly. Stopping DogStatsD." + ) + } + Err(e) => error!("DogStatsD listener task failed: {e}. Stopping DogStatsD."), + } + stop.cancel(); + } + }; + + tokio::join!( + supervise, + process_loop(&mut rx, &stop, aggregator, namespace, pool_tx) + ); +} + /// Named Pipe server - accepts client connections and forwards metrics. /// /// Uses a multi-instance approach (like winio in the main agent): /// - Creates new server instance for each client /// - Spawns task to handle each client +/// +/// Clients send complete messages straight into the processing queue. When the queue is full, +/// messages are dropped, as the UDP reader does. #[cfg(all(windows, feature = "windows-pipes"))] async fn run_named_pipe_server( pipe_name: Arc, - sender: tokio::sync::mpsc::UnboundedSender>, + sender: tokio::sync::mpsc::Sender, cancel_token: tokio_util::sync::CancellationToken, buf_size: usize, ) { @@ -739,6 +879,7 @@ async fn run_named_pipe_server( // Byte mode requires manual buffering to handle messages split across reads // so let's track where we're writing each read in the buffer let mut start_write_index = 0; + let mut dropped: u64 = 0; loop { let read_result = tokio::select! { @@ -769,12 +910,26 @@ async fn run_named_pipe_server( if complete_message_size > 0 { // Send complete messages - match sender_clone.send(buf[..complete_message_size].to_vec()) { + let packet = Packet { + buf: buf[..complete_message_size].to_vec(), + len: complete_message_size, + source: MessageSource::NamedPipe(pipe_name_clone.clone()), + }; + match sender_clone.try_send(packet) { + Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => { + dropped += 1; + if dropped.is_power_of_two() { + debug!( + "DogStatsD queue full, {} messages from '{}' dropped so far", + dropped, pipe_name_clone + ); + } + } Err(e) => { error!("Failed to send data from '{}': {}", pipe_name_clone, e); break; } - Ok(_) => { + Ok(()) => { if std::str::from_utf8(&buf[..complete_message_size]).is_ok() { debug!( "Sent {} bytes from '{}'", @@ -796,6 +951,12 @@ async fn run_named_pipe_server( buf.copy_within(complete_message_size..end_index, 0); } } + if dropped > 0 { + error!( + "DogStatsD named pipe client on '{}' disconnected, {} messages dropped", + pipe_name_clone, dropped + ); + } // Server instance is dropped here, automatically cleaned up }); } @@ -958,10 +1119,10 @@ single_machine_performance.rouster.metrics_max_timestamp_latency:1376.90870216|d let mut dogstatsd = DogStatsD { cancel_token, aggregator_handle: handle.clone(), - buffer_reader: BufferReader::MirrorTest( + transport: Transport::Reader(BufferReader::MirrorTest( payload.as_bytes().to_vec(), SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 0), - ), + )), metric_namespace: None, buf_size: custom_buf_size, queue_size: DEFAULT_QUEUE_SIZE, @@ -1018,10 +1179,10 @@ single_machine_performance.rouster.metrics_max_timestamp_latency:1376.90870216|d let mut dogstatsd = DogStatsD { cancel_token, aggregator_handle: handle.clone(), - buffer_reader: BufferReader::MirrorTest( + transport: Transport::Reader(BufferReader::MirrorTest( statsd_string.as_bytes().to_vec(), SocketAddr::new(IpAddr::V4(Ipv4Addr::new(111, 112, 113, 114)), 0), - ), + )), metric_namespace, buf_size: DEFAULT_BUFFER_SIZE, queue_size: DEFAULT_QUEUE_SIZE, diff --git a/crates/dogstatsd/tests/integration_test.rs b/crates/dogstatsd/tests/integration_test.rs index 3d8f78b2..a805daf0 100644 --- a/crates/dogstatsd/tests/integration_test.rs +++ b/crates/dogstatsd/tests/integration_test.rs @@ -602,8 +602,30 @@ async fn test_named_pipe_and_udp_dual_listen() { sleep(Duration::from_millis(100)).await; + // Pipe messages carry right-sized buffers. UDP packets longer than the pipe message must + // still arrive whole, so those buffers must not be reused for UDP reads. Send several, + // because the reader takes a pooled buffer before each wait. + for i in 0..4 { + let metric = format!("dual.udp.after.pipe.{i}.metric.with.a.much.longer.name:1|c"); + socket + .send_to(metric.as_bytes(), ("127.0.0.1", udp_port)) + .await + .expect("UDP send failed"); + sleep(Duration::from_millis(20)).await; + } + + 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"); + for i in 0..4 { + assert!( + series_json.contains(&format!( + "dual.udp.after.pipe.{i}.metric.with.a.much.longer.name" + )), + "UDP packet {i} after a pipe message was not received whole: {series_json}" + ); + } assert!( series_json.contains("dual.udp.metric"), "UDP metric was not received when a pipe name is also configured: {series_json}" From ffb439ee09e05b63b8cb3d8aa4d4fe9ec6d65cc8 Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Thu, 22 Oct 2026 17:19:50 -0400 Subject: [PATCH 05/11] fix(dogstatsd): skip bad UDP datagrams beside the named pipe On Windows, recvfrom reports WSAEMSGSIZE for a datagram larger than the read buffer and WSAENETRESET when a datagram's time to live expired. Both concern one datagram, not the socket, but the reader beside the pipe treated them as fatal, so one UDP datagram over 8192 bytes stopped DogStatsD, pipe included. Beside the pipe, these errors are now logged and skipped, as the single-transport reader already does. Other socket errors still stop DogStatsD with an error log. --- crates/dogstatsd/src/dogstatsd.rs | 24 ++++++- crates/dogstatsd/tests/integration_test.rs | 78 ++++++++++++++++++++++ 2 files changed, 100 insertions(+), 2 deletions(-) diff --git a/crates/dogstatsd/src/dogstatsd.rs b/crates/dogstatsd/src/dogstatsd.rs index 26728c63..9478ae1a 100644 --- a/crates/dogstatsd/src/dogstatsd.rs +++ b/crates/dogstatsd/src/dogstatsd.rs @@ -134,7 +134,8 @@ enum UdpErrorPolicy { /// UDP is the only transport: keep the single-transport behavior. SoleTransport, /// UDP runs beside the named pipe. Like the trace agent's accept loops, skip - /// connection-level errors and end the reader on any other error, so the supervisor + /// connection-level errors. Log and skip errors about one bad datagram, as the + /// single-transport reader does. End the reader on any other error, so the supervisor /// stops DogStatsD with an error log instead of leaving it half running. Also skip the /// non-yielding drain so the pipe task always gets to run. BesideNamedPipe, @@ -152,6 +153,16 @@ fn is_connection_error(e: &std::io::Error) -> bool { ) } +/// Windows `recvfrom` errors that concern one datagram, not the socket: `WSAEMSGSIZE` (10040), +/// a datagram larger than the buffer, and `WSAENETRESET` (10052), which on a datagram socket +/// means its time to live expired. Rust maps neither to an `ErrorKind`. +#[cfg(all(windows, feature = "windows-pipes"))] +fn is_datagram_error(e: &std::io::Error) -> bool { + const WSAEMSGSIZE: i32 = 10040; + const WSAENETRESET: i32 = 10052; + matches!(e.raw_os_error(), Some(WSAEMSGSIZE | WSAENETRESET)) +} + impl BufferReader { /// Reads one packet into the provided buffer, waiting if none is available. /// Returns `(bytes_read, source)`. The caller owns the buffer and can @@ -427,7 +438,8 @@ impl DogStatsD { /// promptly even on a quiet socket (where `read_into` would block forever). /// /// Returns an error only when `policy` is `BesideNamedPipe` and a read fails with an error -/// that isn't connection-level; the caller then stops DogStatsD. +/// that concerns the socket rather than one connection or datagram; the caller then stops +/// DogStatsD. #[allow(clippy::too_many_arguments)] async fn read_loop( mut reader: BufferReader, @@ -466,6 +478,14 @@ async fn read_loop( continue; } #[cfg(all(windows, feature = "windows-pipes"))] + Err(e) + if matches!(policy, UdpErrorPolicy::BesideNamedPipe) && is_datagram_error(&e) => + { + error!("DogStatsD read error: {}", e); + let _ = pool_tx.send(buf); + continue; + } + #[cfg(all(windows, feature = "windows-pipes"))] Err(e) if matches!(policy, UdpErrorPolicy::BesideNamedPipe) => { let _ = pool_tx.send(buf); exit = Err(e); diff --git a/crates/dogstatsd/tests/integration_test.rs b/crates/dogstatsd/tests/integration_test.rs index a805daf0..2d71152e 100644 --- a/crates/dogstatsd/tests/integration_test.rs +++ b/crates/dogstatsd/tests/integration_test.rs @@ -642,6 +642,84 @@ async fn test_named_pipe_and_udp_dual_listen() { handle.shutdown().expect("shutdown failed"); } +#[cfg(test)] +#[cfg(all(windows, feature = "windows-pipes"))] +#[tokio::test] +async fn test_oversized_udp_datagram_keeps_both_transports_running() { + let pipe_name = r"\\.\pipe\test_dogstatsd_oversized_udp"; + let udp_port: u16 = 18131; + 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(); + 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; + + // Larger than the default 8192-byte read buffer. Windows reports this datagram as + // WSAEMSGSIZE, which must drop only this datagram. + let socket = UdpSocket::bind("0.0.0.0:0").await.expect("bind failed"); + let oversized = format!("oversized.metric:1|c|#tag:{}", "x".repeat(9_000)); + socket + .send_to(oversized.as_bytes(), ("127.0.0.1", udp_port)) + .await + .expect("UDP send failed"); + sleep(Duration::from_millis(100)).await; + + socket + .send_to(b"after.oversized.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"after.oversized.pipe.metric:1|c\n") + .await + .expect("pipe write failed"); + client.flush().await.expect("pipe flush failed"); + sleep(Duration::from_millis(100)).await; + + assert!( + !dogstatsd_task.is_finished(), + "DogStatsD stopped after an oversized UDP datagram" + ); + let response = handle.flush().await.expect("flush failed"); + let series_json = serde_json::to_string(&response.series).expect("serialize series"); + assert!( + series_json.contains("after.oversized.udp.metric"), + "UDP metric after an oversized datagram was not received: {series_json}" + ); + assert!( + series_json.contains("after.oversized.pipe.metric"), + "pipe metric after an oversized UDP datagram was not received: {series_json}" + ); + + 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 739137d64b12c02adf8159da9774a4e072301431 Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Thu, 22 Oct 2026 17:53:23 -0400 Subject: [PATCH 06/11] feat(serverless-compat): listen on UDP beside the pipe only when asked With a DogStatsD named pipe set, the binary now listens on UDP only when DD_DOGSTATSD_PORT is set. Without the setting the port stays 0, as before this PR, so a binary with a pipe doesn't take UDP 8125 by default. Several Windows function apps can share one instance, and only one binary can own 8125. A binary without a pipe, such as one bundled with released compat-java, panics when the port is taken. The .NET and Node compatibility layers give their in-process clients the pipe, so their binaries don't need UDP and keep today's behavior. The Java compatibility layer can't, because a JVM can't change its own environment after it starts, so it will set DD_DOGSTATSD_PORT for its binary to keep older Java clients working. Without a pipe, the port rule is unchanged. --- crates/datadog-serverless-compat/src/main.rs | 59 ++++++++++++++++++-- crates/dogstatsd/src/dogstatsd.rs | 11 ++-- 2 files changed, 59 insertions(+), 11 deletions(-) diff --git a/crates/datadog-serverless-compat/src/main.rs b/crates/datadog-serverless-compat/src/main.rs index e2f4f6f6..a764fe59 100644 --- a/crates/datadog-serverless-compat/src/main.rs +++ b/crates/datadog-serverless-compat/src/main.rs @@ -104,12 +104,10 @@ pub async fn main() { None } }; - // UDP listener also runs when a named pipe is set, so clients without - // named-pipe support can still send metrics. With a pipe, port 0 turns UDP off. - let dd_dogstatsd_port: u16 = env::var("DD_DOGSTATSD_PORT") - .ok() - .and_then(|port| port.parse::().ok()) - .unwrap_or(DEFAULT_DOGSTATSD_PORT); + let dd_dogstatsd_port = decide_dogstatsd_port( + dd_dogstatsd_windows_pipe_name.is_some(), + env::var("DD_DOGSTATSD_PORT").ok().as_deref(), + ); 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") @@ -568,6 +566,26 @@ struct EnabledMetricsComponents { start_cpu_metrics_collector: bool, } +/// Chooses the DogStatsD UDP port. 0 means UDP is off. +/// +/// Without a named pipe, UDP is the only transport: use `DD_DOGSTATSD_PORT`, else 8125. +/// +/// With a named pipe, UDP is off unless `DD_DOGSTATSD_PORT` is set. Several Windows function +/// apps can share one instance, and only one binary can own UDP 8125. A binary without a pipe +/// panics when the port is taken, so a binary with a pipe must not take it by default. The .NET +/// and Node compatibility layers point their in-process clients at the pipe, so their binaries +/// don't need UDP. The Java compatibility layer can't: a JVM can't change its own environment +/// after it starts, so older Java DogStatsD clients, and clients that can't open the pipe, +/// still send to UDP 8125. compat-java sets `DD_DOGSTATSD_PORT` for its binary to keep them +/// working. +fn decide_dogstatsd_port(has_pipe: bool, configured_port: Option<&str>) -> u16 { + match configured_port.and_then(|port| port.parse::().ok()) { + Some(port) => port, + None if has_pipe => 0, + None => DEFAULT_DOGSTATSD_PORT, + } +} + /// Determines which components should be started based on configuration. /// /// The metrics aggregator and flusher are started together and shared between dogstatsd and enhanced metrics, @@ -782,3 +800,32 @@ mod metrics_components_tests { } } } + +#[cfg(test)] +mod dogstatsd_port_tests { + use super::{DEFAULT_DOGSTATSD_PORT, decide_dogstatsd_port}; + + #[test] + fn test_decide_dogstatsd_port() { + let cases: &[(bool, Option<&str>, u16)] = &[ + // Without a pipe, UDP is the only transport. + (false, None, DEFAULT_DOGSTATSD_PORT), + (false, Some("18125"), 18125), + (false, Some("not-a-port"), DEFAULT_DOGSTATSD_PORT), + // With a pipe, UDP stays off unless a port is set, so this binary doesn't take + // 8125 from a neighbor without a pipe. + (true, None, 0), + (true, Some("not-a-port"), 0), + (true, Some("8125"), 8125), + (true, Some("0"), 0), + ]; + + for (has_pipe, configured_port, expected) in cases { + assert_eq!( + decide_dogstatsd_port(*has_pipe, *configured_port), + *expected, + "case (has_pipe={has_pipe}, DD_DOGSTATSD_PORT={configured_port:?})" + ); + } + } +} diff --git a/crates/dogstatsd/src/dogstatsd.rs b/crates/dogstatsd/src/dogstatsd.rs index 9478ae1a..bf9304bf 100644 --- a/crates/dogstatsd/src/dogstatsd.rs +++ b/crates/dogstatsd/src/dogstatsd.rs @@ -36,9 +36,9 @@ 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). The UDP socket is also bound when a named pipe - /// is configured, so clients without named-pipe support can still send metrics. With a - /// named pipe, port 0 turns UDP off, as `dogstatsd_port: 0` does in the Datadog Agent. + /// Port to bind UDP socket to (e.g., 8125). With a named pipe, a non-zero port also binds + /// UDP, best-effort, for clients that can't use the pipe; port 0 turns UDP off, as + /// `dogstatsd_port: 0` does in the Datadog Agent. pub port: u16, /// Optional namespace to prepend to all metric names (e.g., "myapp") pub metric_namespace: Option, @@ -211,8 +211,9 @@ pub struct DogStatsD { impl DogStatsD { /// Creates a new DogStatsD server instance. /// - /// The server always binds a UDP socket and also listens on a Windows named pipe when one - /// is configured. + /// Binds UDP on `config.port`. With a Windows named pipe configured, also listens on the + /// pipe; then port 0 turns UDP off and a failed bind is logged, not fatal. Without a pipe, + /// panics if the UDP bind fails. /// Metrics received will be forwarded to the provided aggregator_handle. #[must_use] pub async fn new( From f92a0413794cfe3b5fe69dc80fc6de5858b2e1cb Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Thu, 22 Oct 2026 18:07:20 -0400 Subject: [PATCH 07/11] fix(serverless-compat): report the real DogStatsD port and flag bad ports Review follow-ups for UDP beside the named pipe: - The trace agent's /info statsd_port now follows the same rule as the DogStatsD listener: with a pipe, 0 unless DD_DOGSTATSD_PORT is set. It used to report 0 whenever a pipe was set, even with UDP listening. - The binary warns when DD_DOGSTATSD_PORT is set but isn't a valid port, and the pipe-only log now reads "UDP is off (port 0)" instead of implying someone set port 0. - The oversized-datagram test checks that DogStatsD is still running before it opens the pipe client, so a regression fails with that message, and it sets a receive buffer large enough for the datagram. --- crates/datadog-serverless-compat/src/main.rs | 8 +++- crates/datadog-trace-agent/src/config.rs | 42 ++++++++++++++++---- crates/dogstatsd/src/dogstatsd.rs | 2 +- crates/dogstatsd/tests/integration_test.rs | 13 +++--- 4 files changed, 50 insertions(+), 15 deletions(-) diff --git a/crates/datadog-serverless-compat/src/main.rs b/crates/datadog-serverless-compat/src/main.rs index a764fe59..25d39666 100644 --- a/crates/datadog-serverless-compat/src/main.rs +++ b/crates/datadog-serverless-compat/src/main.rs @@ -104,9 +104,15 @@ pub async fn main() { None } }; + let dd_dogstatsd_port_setting = env::var("DD_DOGSTATSD_PORT").ok(); + if let Some(ref port) = dd_dogstatsd_port_setting + && port.parse::().is_err() + { + warn!("DD_DOGSTATSD_PORT={port:?} is not a valid port, so it is ignored"); + } let dd_dogstatsd_port = decide_dogstatsd_port( dd_dogstatsd_windows_pipe_name.is_some(), - env::var("DD_DOGSTATSD_PORT").ok().as_deref(), + dd_dogstatsd_port_setting.as_deref(), ); let dd_site = env::var("DD_SITE").unwrap_or_else(|_| "datadoghq.com".to_string()); let dd_use_dogstatsd = env::var("DD_USE_DOGSTATSD") diff --git a/crates/datadog-trace-agent/src/config.rs b/crates/datadog-trace-agent/src/config.rs index 186ce0ae..7715aa61 100644 --- a/crates/datadog-trace-agent/src/config.rs +++ b/crates/datadog-trace-agent/src/config.rs @@ -178,13 +178,16 @@ impl Config { 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) + // Reported as `statsd_port` in /info, so it follows the DogStatsD listener's rule in + // datadog-serverless-compat: with a named pipe, UDP is off (0) unless DD_DOGSTATSD_PORT + // is set. + let dd_dogstatsd_port: u16 = match env::var("DD_DOGSTATSD_PORT") + .ok() + .and_then(|port| port.parse::().ok()) + { + Some(port) => port, + None if dd_dogstatsd_windows_pipe_name.is_some() => 0, + None => DEFAULT_DOGSTATSD_PORT, }; let dd_site = env::var("DD_SITE").unwrap_or_else(|_| "datadoghq.com".to_string()); @@ -529,6 +532,7 @@ mod tests { Some("test-spring-app"), ), ("DD_DOGSTATSD_WINDOWS_PIPE_NAME", Some(r"test_pipe")), + ("DD_DOGSTATSD_PORT", None), ], || { let config_res = config::Config::new(); @@ -539,12 +543,34 @@ mod tests { Some(r"\\.\pipe\test_pipe".to_string()) ); - // Port should be overridden to 0 when pipe is set + // With a pipe and no DD_DOGSTATSD_PORT, UDP is off assert_eq!(config.dd_dogstatsd_port, 0); }, ); } + #[test] + #[serial] + #[cfg(any(all(windows, feature = "windows-pipes"), test))] + fn test_dogstatsd_windows_pipe_name_with_port() { + temp_env::with_vars( + [ + ("DD_API_KEY", Some("_not_a_real_key_")), + ( + "ASCSVCRT_SPRING__APPLICATION__NAME", + Some("test-spring-app"), + ), + ("DD_DOGSTATSD_WINDOWS_PIPE_NAME", Some(r"test_pipe")), + ("DD_DOGSTATSD_PORT", Some("8125")), + ], + || { + let config = config::Config::new().unwrap(); + // DogStatsD also listens on UDP, so /info reports the port + assert_eq!(config.dd_dogstatsd_port, 8125); + }, + ); + } + #[test] #[serial] fn test_default_dogstatsd_port() { diff --git a/crates/dogstatsd/src/dogstatsd.rs b/crates/dogstatsd/src/dogstatsd.rs index bf9304bf..b43565eb 100644 --- a/crates/dogstatsd/src/dogstatsd.rs +++ b/crates/dogstatsd/src/dogstatsd.rs @@ -249,7 +249,7 @@ impl DogStatsD { // 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 = if config.port == 0 { - info!("dogstatsd-udp: port is 0, so only the named pipe is listening"); + info!("dogstatsd-udp: UDP is off (port 0), only the named pipe is listening"); None } else { match create_udp_socket(&addr, config.so_rcvbuf).await { diff --git a/crates/dogstatsd/tests/integration_test.rs b/crates/dogstatsd/tests/integration_test.rs index 2d71152e..c221120b 100644 --- a/crates/dogstatsd/tests/integration_test.rs +++ b/crates/dogstatsd/tests/integration_test.rs @@ -663,7 +663,9 @@ async fn test_oversized_udp_datagram_keeps_both_transports_running() { port: udp_port, metric_namespace: None, windows_pipe_name: Some(pipe_name.to_string()), - so_rcvbuf: None, + // Room for the oversized datagram, so Windows delivers it to recvfrom + // instead of dropping it in the kernel. + so_rcvbuf: Some(262_144), buffer_size: None, queue_size: None, }, @@ -686,6 +688,11 @@ async fn test_oversized_udp_datagram_keeps_both_transports_running() { .await .expect("UDP send failed"); sleep(Duration::from_millis(100)).await; + // Checked before opening the pipe: if DogStatsD stopped, the pipe server is gone too. + assert!( + !dogstatsd_task.is_finished(), + "DogStatsD stopped after an oversized UDP datagram" + ); socket .send_to(b"after.oversized.udp.metric:1|c", ("127.0.0.1", udp_port)) @@ -699,10 +706,6 @@ async fn test_oversized_udp_datagram_keeps_both_transports_running() { client.flush().await.expect("pipe flush failed"); sleep(Duration::from_millis(100)).await; - assert!( - !dogstatsd_task.is_finished(), - "DogStatsD stopped after an oversized UDP datagram" - ); let response = handle.flush().await.expect("flush failed"); let series_json = serde_json::to_string(&response.series).expect("serialize series"); assert!( From 010683b6b3b3fa09efea8352a2878d862ff33f80 Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Thu, 22 Oct 2026 20:05:32 -0400 Subject: [PATCH 08/11] refactor(dogstatsd): give UDP beside the pipe its own read loop The UDP reader beside the named pipe now has its own short loop instead of sharing read_loop through a Windows-only policy parameter and Windows-only match arms. read_loop is back to main's code exactly, so the UDP-only path that Linux and the Lambda extension use is visibly unchanged. The new loop reads only with async recv_from, so it yields under tokio's cooperative budget, and it keeps the same error rules: skip connection-level errors, log and skip bad datagrams, and end on anything else. It mirrors the trace agent's separate TCP and named-pipe accept loops. No behavior change. --- crates/dogstatsd/src/dogstatsd.rs | 143 ++++++++++++++++-------------- 1 file changed, 75 insertions(+), 68 deletions(-) diff --git a/crates/dogstatsd/src/dogstatsd.rs b/crates/dogstatsd/src/dogstatsd.rs index b43565eb..4aa08555 100644 --- a/crates/dogstatsd/src/dogstatsd.rs +++ b/crates/dogstatsd/src/dogstatsd.rs @@ -127,20 +127,6 @@ enum Transport { }, } -/// How the UDP reader treats read errors. -#[cfg(all(windows, feature = "windows-pipes"))] -#[derive(Clone, Copy)] -enum UdpErrorPolicy { - /// UDP is the only transport: keep the single-transport behavior. - SoleTransport, - /// UDP runs beside the named pipe. Like the trace agent's accept loops, skip - /// connection-level errors. Log and skip errors about one bad datagram, as the - /// single-transport reader does. End the reader on any other error, so the supervisor - /// stops DogStatsD with an error log instead of leaving it half running. Also skip the - /// non-yielding drain so the pipe task always gets to run. - BesideNamedPipe, -} - /// Connection-level errors that the trace agent's accept loops skip. On a Windows UDP socket, /// `ConnectionReset` reports an ICMP Port Unreachable reply to an earlier send. #[cfg(all(windows, feature = "windows-pipes"))] @@ -351,7 +337,7 @@ impl DogStatsD { let reader_pool_tx = pool_tx.clone(); let reader_token = cancel_token.clone(); tokio::spawn(async move { - let _ = read_loop( + read_loop( buffer_reader, tx, reader_token, @@ -359,8 +345,6 @@ impl DogStatsD { queue_size, pool_rx, reader_pool_tx, - #[cfg(all(windows, feature = "windows-pipes"))] - UdpErrorPolicy::SoleTransport, ) .await; }); @@ -437,11 +421,6 @@ impl DogStatsD { /// Uses `is_cancelled()` as a fast-path check at the top of each iteration, /// and `tokio::select!` on the async `read_into()` to ensure the reader exits /// promptly even on a quiet socket (where `read_into` would block forever). -/// -/// Returns an error only when `policy` is `BesideNamedPipe` and a read fails with an error -/// that concerns the socket rather than one connection or datagram; the caller then stops -/// DogStatsD. -#[allow(clippy::too_many_arguments)] async fn read_loop( mut reader: BufferReader, tx: tokio::sync::mpsc::Sender, @@ -450,11 +429,8 @@ async fn read_loop( queue_capacity: usize, mut pool_rx: tokio::sync::mpsc::UnboundedReceiver>, pool_tx: tokio::sync::mpsc::UnboundedSender>, - #[cfg(all(windows, feature = "windows-pipes"))] policy: UdpErrorPolicy, -) -> std::io::Result<()> { +) { let mut dropped: u64 = 0; - #[cfg_attr(not(all(windows, feature = "windows-pipes")), allow(unused_mut))] - let mut exit = Ok(()); while !cancel.is_cancelled() { // Get a buffer from the pool, or allocate if the pool is empty (cold start). let mut buf = pool_rx.try_recv().unwrap_or_else(|_| vec![0u8; buf_size]); @@ -470,28 +446,6 @@ async fn read_loop( }; let (len, source) = match result { Ok(r) => r, - #[cfg(all(windows, feature = "windows-pipes"))] - Err(e) - if matches!(policy, UdpErrorPolicy::BesideNamedPipe) && is_connection_error(&e) => - { - debug!("DogStatsD UDP read skipped a connection error: {}", e); - let _ = pool_tx.send(buf); - continue; - } - #[cfg(all(windows, feature = "windows-pipes"))] - Err(e) - if matches!(policy, UdpErrorPolicy::BesideNamedPipe) && is_datagram_error(&e) => - { - error!("DogStatsD read error: {}", e); - let _ = pool_tx.send(buf); - continue; - } - #[cfg(all(windows, feature = "windows-pipes"))] - Err(e) if matches!(policy, UdpErrorPolicy::BesideNamedPipe) => { - let _ = pool_tx.send(buf); - exit = Err(e); - break; - } Err(e) if e.kind() == std::io::ErrorKind::ConnectionReset => { debug!("DogStatsD transport closed: {}", e); break; @@ -510,14 +464,6 @@ async fn read_loop( break; } - // Beside the named pipe, skip the drain below. It never yields, so on a one-core host - // (Azure Functions EP1) it would hold the only worker thread and starve the pipe. Each - // async read instead counts against tokio's cooperative budget, which yields regularly. - #[cfg(all(windows, feature = "windows-pipes"))] - if matches!(policy, UdpErrorPolicy::BesideNamedPipe) { - continue; - } - // Drain any packets already in the kernel buffer without going // through tokio's event loop — just raw non-blocking syscalls. loop { @@ -553,7 +499,6 @@ async fn read_loop( dropped, queue_capacity ); } - exit } /// Sends a packet to the channel. Returns `false` if the channel is closed @@ -756,6 +701,76 @@ async fn create_udp_socket( tokio::net::UdpSocket::from_std(std_socket) } +/// Reads UDP beside the named pipe into the processing queue. It does the same job as +/// `read_loop`, so keep the two in step, but differs in two ways: +/// - It reads only with async `recv_from` and skips `read_loop`'s `try_recv_from` drain. That +/// drain never yields, so on a one-core host (Azure Functions EP1) it would hold the only +/// worker thread and starve the pipe. Each async read counts against tokio's cooperative +/// budget, which yields regularly. +/// - Like the trace agent's accept loops, it skips connection-level errors, and it logs and +/// skips errors about one bad datagram, as `read_loop` does. Any other error ends the loop +/// with that error, and `spin_named_pipe` then stops DogStatsD. +#[cfg(all(windows, feature = "windows-pipes"))] +async fn udp_read_loop_beside_pipe( + socket: tokio::net::UdpSocket, + tx: tokio::sync::mpsc::Sender, + stop: tokio_util::sync::CancellationToken, + buf_size: usize, + queue_capacity: usize, + mut pool_rx: tokio::sync::mpsc::UnboundedReceiver>, + pool_tx: tokio::sync::mpsc::UnboundedSender>, +) -> std::io::Result<()> { + let mut dropped: u64 = 0; + let mut exit = Ok(()); + while !stop.is_cancelled() { + // Get a buffer from the pool, or allocate if the pool is empty (cold start). + let mut buf = pool_rx.try_recv().unwrap_or_else(|_| vec![0u8; buf_size]); + + let result = tokio::select! { + r = socket.recv_from(&mut buf) => r, + _ = stop.cancelled() => { + let _ = pool_tx.send(buf); + break; + } + }; + match result { + Ok((0, _)) => { + let _ = pool_tx.send(buf); + } + Ok((len, src)) => { + let packet = Packet { + buf, + len, + source: MessageSource::Network(src), + }; + if !try_send_packet(&tx, packet, &mut dropped, &pool_tx) { + break; + } + } + Err(e) if is_connection_error(&e) => { + debug!("DogStatsD UDP read skipped a connection error: {}", e); + let _ = pool_tx.send(buf); + } + Err(e) if is_datagram_error(&e) => { + error!("DogStatsD read error: {}", e); + let _ = pool_tx.send(buf); + } + Err(e) => { + let _ = pool_tx.send(buf); + exit = Err(e); + break; + } + } + } + if dropped > 0 { + error!( + "DogStatsD UDP reader exiting, {} total packets dropped (queue capacity: {})", + dropped, queue_capacity + ); + } + exit +} + /// Both ends of the processing queue and the buffer pool, handed to `spin_named_pipe`. #[cfg(all(windows, feature = "windows-pipes"))] struct QueueEnds { @@ -794,17 +809,9 @@ async fn spin_named_pipe( if let Some(socket) = udp_socket { let (tx, stop, pool_tx) = (tx.clone(), stop.clone(), pool_tx.clone()); transports.spawn(async move { - let result = read_loop( - BufferReader::UdpSocket(socket), - tx, - stop, - buf_size, - queue_size, - pool_rx, - pool_tx, - UdpErrorPolicy::BesideNamedPipe, - ) - .await; + let result = + udp_read_loop_beside_pipe(socket, tx, stop, buf_size, queue_size, pool_rx, pool_tx) + .await; ("UDP", result) }); } From 2a78aedaefc2294fbd832fb3eaf536e575546fca Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Thu, 22 Oct 2026 20:06:33 -0400 Subject: [PATCH 09/11] docs(dogstatsd): point read_loop at the UDP loop beside the pipe --- crates/dogstatsd/src/dogstatsd.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/crates/dogstatsd/src/dogstatsd.rs b/crates/dogstatsd/src/dogstatsd.rs index 4aa08555..0baafbd6 100644 --- a/crates/dogstatsd/src/dogstatsd.rs +++ b/crates/dogstatsd/src/dogstatsd.rs @@ -421,6 +421,9 @@ impl DogStatsD { /// Uses `is_cancelled()` as a fast-path check at the top of each iteration, /// and `tokio::select!` on the async `read_into()` to ensure the reader exits /// promptly even on a quiet socket (where `read_into` would block forever). +/// +/// With a Windows named pipe, UDP is read by `udp_read_loop_beside_pipe` instead, which does +/// the same job without the drain; keep the two in step. async fn read_loop( mut reader: BufferReader, tx: tokio::sync::mpsc::Sender, From ce70c72ac2dba5ed19d3e760496c8a365e6af2b8 Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Thu, 22 Oct 2026 20:12:06 -0400 Subject: [PATCH 10/11] test(dogstatsd): wait for metrics instead of sleeping in pipe tests The Windows tests for UDP beside the named pipe now share three helpers: start DogStatsD with a pipe (UDP is bound when it returns), open the pipe client with retries, and flush until the expected metrics appear, failing with "DogStatsD stopped" if the server exits first. This replaces fixed 100 ms sleeps that could make a slow Windows runner flake. The check that pipe buffers aren't reused for UDP reads moves out of the dual-listen test into its own test, which waits for the pipe message to be processed before sending the UDP packets. --- crates/dogstatsd/tests/integration_test.rs | 252 +++++++++++---------- 1 file changed, 133 insertions(+), 119 deletions(-) diff --git a/crates/dogstatsd/tests/integration_test.rs b/crates/dogstatsd/tests/integration_test.rs index c221120b..ce655082 100644 --- a/crates/dogstatsd/tests/integration_test.rs +++ b/crates/dogstatsd/tests/integration_test.rs @@ -22,7 +22,11 @@ use tokio_util::sync::CancellationToken; use zstd::zstd_safe::CompressionLevel; #[cfg(all(windows, feature = "windows-pipes"))] -use tokio::{io::AsyncWriteExt, net::windows::named_pipe::ClientOptions}; +use tokio::{ + io::AsyncWriteExt, + net::windows::named_pipe::{ClientOptions, NamedPipeClient}, + time::Instant, +}; #[cfg(test)] #[tokio::test] @@ -548,94 +552,145 @@ async fn test_named_pipe_basic_communication() { handle.shutdown().expect("shutdown failed"); } +/// Starts DogStatsD with a named pipe and, for a non-zero `udp_port`, UDP beside it. UDP is +/// bound when this returns; the pipe server starts in `spin`, so open it with +/// `open_pipe_client`. +#[cfg(all(windows, feature = "windows-pipes"))] +async fn start_dogstatsd_with_pipe( + handle: AggregatorHandle, + pipe_name: &str, + udp_port: u16, + so_rcvbuf: Option, +) -> (tokio::task::JoinHandle<()>, CancellationToken) { + let cancel_token = CancellationToken::new(); + 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, + buffer_size: None, + queue_size: None, + }, + handle, + cancel_token.clone(), + ) + .await; + (tokio::spawn(dogstatsd.spin()), cancel_token) +} + +/// Opens a client on `pipe_name`, retrying until the pipe server has created it. +#[cfg(all(windows, feature = "windows-pipes"))] +async fn open_pipe_client(pipe_name: &str) -> NamedPipeClient { + let deadline = Instant::now() + Duration::from_secs(2); + loop { + match ClientOptions::new().open(pipe_name) { + Ok(client) => return client, + Err(_) if Instant::now() < deadline => sleep(Duration::from_millis(10)).await, + Err(e) => panic!("could not open pipe {pipe_name}: {e}"), + } + } +} + +/// Flushes until every name in `names` has appeared in the flushed series. Panics if DogStatsD +/// stops first, or after a deadline. +#[cfg(all(windows, feature = "windows-pipes"))] +async fn flush_until_seen( + handle: &AggregatorHandle, + dogstatsd_task: &tokio::task::JoinHandle<()>, + names: &[&str], +) { + let deadline = Instant::now() + Duration::from_secs(2); + let mut flushed = String::new(); + while !names.iter().all(|name| flushed.contains(name)) { + assert!( + !dogstatsd_task.is_finished(), + "DogStatsD stopped while waiting for {names:?}" + ); + assert!( + Instant::now() < deadline, + "metrics {names:?} were not received; flushed: {flushed}" + ); + let response = handle.flush().await.expect("flush failed"); + flushed.push_str(&serde_json::to_string(&response.series).expect("serialize series")); + sleep(Duration::from_millis(20)).await; + } +} + #[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; + let (dogstatsd_task, cancel_token) = + start_dogstatsd_with_pipe(handle.clone(), pipe_name, 18130, None).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)) + .send_to(b"dual.udp.metric:1|c", ("127.0.0.1", 18130)) .await .expect("UDP send failed"); - - let mut client = ClientOptions::new().open(pipe_name).expect("client open"); + let mut client = open_pipe_client(pipe_name).await; 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; + flush_until_seen( + &handle, + &dogstatsd_task, + &["dual.udp.metric", "dual.pipe.metric"], + ) + .await; - // Pipe messages carry right-sized buffers. UDP packets longer than the pipe message must - // still arrive whole, so those buffers must not be reused for UDP reads. Send several, - // because the reader takes a pooled buffer before each wait. - for i in 0..4 { - let metric = format!("dual.udp.after.pipe.{i}.metric.with.a.much.longer.name:1|c"); + 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] +async fn test_pipe_buffers_are_not_reused_for_udp_reads() { + let pipe_name = r"\\.\pipe\test_dogstatsd_pipe_buffers"; + let (service, handle) = AggregatorService::new(SortedTags::parse("test:value").unwrap(), 1_024) + .expect("aggregator service creation failed"); + tokio::spawn(service.run()); + let (dogstatsd_task, cancel_token) = + start_dogstatsd_with_pipe(handle.clone(), pipe_name, 18132, None).await; + + // A pipe message's buffer is sized to the message. Once it is processed, it must not go + // back to the pool that UDP reads draw from. + let mut client = open_pipe_client(pipe_name).await; + client + .write_all(b"short.pipe:1|c\n") + .await + .expect("pipe write failed"); + client.flush().await.expect("pipe flush failed"); + flush_until_seen(&handle, &dogstatsd_task, &["short.pipe"]).await; + + // Several UDP packets longer than the pipe message, because the reader takes a pooled + // buffer before each read. A reused pipe buffer would lose one of them. + let socket = UdpSocket::bind("0.0.0.0:0").await.expect("bind failed"); + let names: Vec = (0..4) + .map(|i| format!("udp.after.pipe.{i}.metric.with.a.much.longer.name")) + .collect(); + for name in &names { socket - .send_to(metric.as_bytes(), ("127.0.0.1", udp_port)) + .send_to(format!("{name}:1|c").as_bytes(), ("127.0.0.1", 18132)) .await .expect("UDP send failed"); - sleep(Duration::from_millis(20)).await; - } - - 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"); - for i in 0..4 { - assert!( - series_json.contains(&format!( - "dual.udp.after.pipe.{i}.metric.with.a.much.longer.name" - )), - "UDP packet {i} after a pipe message was not received whole: {series_json}" - ); } - 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}" - ); + let names: Vec<&str> = names.iter().map(String::as_str).collect(); + flush_until_seen(&handle, &dogstatsd_task, &names).await; - // Cleanup cancel_token.cancel(); let result = timeout(Duration::from_millis(500), dogstatsd_task).await; assert!(result.is_ok(), "task should complete after cancellation"); @@ -647,75 +702,34 @@ async fn test_named_pipe_and_udp_dual_listen() { #[tokio::test] async fn test_oversized_udp_datagram_keeps_both_transports_running() { let pipe_name = r"\\.\pipe\test_dogstatsd_oversized_udp"; - let udp_port: u16 = 18131; 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(); - 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()), - // Room for the oversized datagram, so Windows delivers it to recvfrom - // instead of dropping it in the kernel. - so_rcvbuf: Some(262_144), - buffer_size: None, - queue_size: None, - }, - handle, - cancel_token, - ) - .await; - dogstatsd.spin().await; - }) - }; - - sleep(Duration::from_millis(100)).await; + // Room for the oversized datagram, so Windows delivers it to recvfrom instead of dropping + // it in the kernel. + let (dogstatsd_task, cancel_token) = + start_dogstatsd_with_pipe(handle.clone(), pipe_name, 18131, Some(262_144)).await; // Larger than the default 8192-byte read buffer. Windows reports this datagram as - // WSAEMSGSIZE, which must drop only this datagram. + // WSAEMSGSIZE, which must drop only this datagram. If it stopped DogStatsD instead, the + // next wait fails with "DogStatsD stopped". let socket = UdpSocket::bind("0.0.0.0:0").await.expect("bind failed"); let oversized = format!("oversized.metric:1|c|#tag:{}", "x".repeat(9_000)); - socket - .send_to(oversized.as_bytes(), ("127.0.0.1", udp_port)) - .await - .expect("UDP send failed"); - sleep(Duration::from_millis(100)).await; - // Checked before opening the pipe: if DogStatsD stopped, the pipe server is gone too. - assert!( - !dogstatsd_task.is_finished(), - "DogStatsD stopped after an oversized UDP datagram" - ); + for datagram in [oversized.as_bytes(), b"after.oversized.udp.metric:1|c"] { + socket + .send_to(datagram, ("127.0.0.1", 18131)) + .await + .expect("UDP send failed"); + } + flush_until_seen(&handle, &dogstatsd_task, &["after.oversized.udp.metric"]).await; - socket - .send_to(b"after.oversized.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"); + let mut client = open_pipe_client(pipe_name).await; client .write_all(b"after.oversized.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("after.oversized.udp.metric"), - "UDP metric after an oversized datagram was not received: {series_json}" - ); - assert!( - series_json.contains("after.oversized.pipe.metric"), - "pipe metric after an oversized UDP datagram was not received: {series_json}" - ); + flush_until_seen(&handle, &dogstatsd_task, &["after.oversized.pipe.metric"]).await; cancel_token.cancel(); let result = timeout(Duration::from_millis(500), dogstatsd_task).await; From d7af94faca3d057f00f04f79b7f77b8eb59a4719 Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Thu, 22 Oct 2026 20:15:22 -0400 Subject: [PATCH 11/11] docs(dogstatsd): tighten comments on UDP beside the named pipe --- crates/datadog-serverless-compat/src/main.rs | 16 ++++++-------- crates/dogstatsd/src/dogstatsd.rs | 22 ++++++++------------ 2 files changed, 15 insertions(+), 23 deletions(-) diff --git a/crates/datadog-serverless-compat/src/main.rs b/crates/datadog-serverless-compat/src/main.rs index 25d39666..e3932675 100644 --- a/crates/datadog-serverless-compat/src/main.rs +++ b/crates/datadog-serverless-compat/src/main.rs @@ -572,18 +572,14 @@ struct EnabledMetricsComponents { start_cpu_metrics_collector: bool, } -/// Chooses the DogStatsD UDP port. 0 means UDP is off. +/// Chooses the DogStatsD UDP port; 0 means UDP is off. /// -/// Without a named pipe, UDP is the only transport: use `DD_DOGSTATSD_PORT`, else 8125. +/// Without a named pipe, UDP is the only transport: `DD_DOGSTATSD_PORT`, else 8125. /// -/// With a named pipe, UDP is off unless `DD_DOGSTATSD_PORT` is set. Several Windows function -/// apps can share one instance, and only one binary can own UDP 8125. A binary without a pipe -/// panics when the port is taken, so a binary with a pipe must not take it by default. The .NET -/// and Node compatibility layers point their in-process clients at the pipe, so their binaries -/// don't need UDP. The Java compatibility layer can't: a JVM can't change its own environment -/// after it starts, so older Java DogStatsD clients, and clients that can't open the pipe, -/// still send to UDP 8125. compat-java sets `DD_DOGSTATSD_PORT` for its binary to keep them -/// working. +/// With a named pipe, UDP is off unless `DD_DOGSTATSD_PORT` is set. Function apps on one +/// Windows instance share ports, and a binary without a pipe panics if another one owns 8125, +/// so a binary with a pipe must not take it by default. Callers whose clients still send UDP, +/// such as datadog-serverless-compat-java, set the port. fn decide_dogstatsd_port(has_pipe: bool, configured_port: Option<&str>) -> u16 { match configured_port.and_then(|port| port.parse::().ok()) { Some(port) => port, diff --git a/crates/dogstatsd/src/dogstatsd.rs b/crates/dogstatsd/src/dogstatsd.rs index 0baafbd6..78520f0b 100644 --- a/crates/dogstatsd/src/dogstatsd.rs +++ b/crates/dogstatsd/src/dogstatsd.rs @@ -118,8 +118,7 @@ enum Transport { Reader(BufferReader), /// Windows named pipe, plus UDP when it bound. `udp_socket` is `None` when the port is 0 - /// or another agent on a shared host already owns it. Each transport runs in its own task, - /// like the trace agent's TCP and named-pipe listeners. + /// or another agent on a shared host already owns it. See `spin_named_pipe`. #[cfg(all(windows, feature = "windows-pipes"))] NamedPipe { pipe_name: Arc, @@ -127,8 +126,8 @@ enum Transport { }, } -/// Connection-level errors that the trace agent's accept loops skip. On a Windows UDP socket, -/// `ConnectionReset` reports an ICMP Port Unreachable reply to an earlier send. +/// Errors about a peer, not the socket. On a Windows UDP socket, `ConnectionReset` reports an +/// ICMP Port Unreachable reply to an earlier send. #[cfg(all(windows, feature = "windows-pipes"))] fn is_connection_error(e: &std::io::Error) -> bool { matches!( @@ -704,15 +703,12 @@ async fn create_udp_socket( tokio::net::UdpSocket::from_std(std_socket) } -/// Reads UDP beside the named pipe into the processing queue. It does the same job as -/// `read_loop`, so keep the two in step, but differs in two ways: -/// - It reads only with async `recv_from` and skips `read_loop`'s `try_recv_from` drain. That -/// drain never yields, so on a one-core host (Azure Functions EP1) it would hold the only -/// worker thread and starve the pipe. Each async read counts against tokio's cooperative -/// budget, which yields regularly. -/// - Like the trace agent's accept loops, it skips connection-level errors, and it logs and -/// skips errors about one bad datagram, as `read_loop` does. Any other error ends the loop -/// with that error, and `spin_named_pipe` then stops DogStatsD. +/// Reads UDP beside the named pipe. Same job as `read_loop`; keep the two in step. It differs +/// in two ways: +/// - No `try_recv_from` drain. The drain never yields, so on a one-core host it would starve +/// the pipe task; async `recv_from` yields under tokio's cooperative budget. +/// - It skips peer and single-datagram errors and returns any other error, which makes +/// `spin_named_pipe` stop DogStatsD. #[cfg(all(windows, feature = "windows-pipes"))] async fn udp_read_loop_beside_pipe( socket: tokio::net::UdpSocket,