diff --git a/crates/datadog-serverless-compat/src/main.rs b/crates/datadog-serverless-compat/src/main.rs index cdcdcabb..e3932675 100644 --- a/crates/datadog-serverless-compat/src/main.rs +++ b/crates/datadog-serverless-compat/src/main.rs @@ -104,14 +104,16 @@ 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) - }; + 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(), + 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") .map(|val| val.to_lowercase() != "false") @@ -252,6 +254,7 @@ pub async fn main() { ) .await; 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}"); @@ -459,8 +462,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 @@ -563,6 +572,22 @@ 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: `DD_DOGSTATSD_PORT`, else 8125. +/// +/// 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, + 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, @@ -777,3 +802,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/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 ada4a578..78520f0b 100644 --- a/crates/dogstatsd/src/dogstatsd.rs +++ b/crates/dogstatsd/src/dogstatsd.rs @@ -17,7 +17,12 @@ 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::{info, warn}, +}; // Default buffer size for receiving DogStatsD packets (one read call). // Used for both UDP recv_from and Windows named pipe reads. @@ -31,7 +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), will be 0 if we're using a Named Pipe + /// 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, @@ -83,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. @@ -93,15 +110,44 @@ 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) + /// 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. See `spin_named_pipe`. #[cfg(all(windows, feature = "windows-pipes"))] NamedPipe { pipe_name: Arc, - receiver: Arc>>>, + udp_socket: Option, }, } +/// 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!( + e.kind(), + std::io::ErrorKind::ConnectionAborted + | std::io::ErrorKind::ConnectionReset + | std::io::ErrorKind::ConnectionRefused + ) +} + +/// 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 @@ -117,21 +163,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, - } => 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()))) - } - None => Err(std::io::Error::new( - std::io::ErrorKind::ConnectionReset, - "named pipe channel closed", - )), - }, } } @@ -156,7 +187,7 @@ impl BufferReader { 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, @@ -165,7 +196,9 @@ 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. + /// 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( @@ -191,28 +224,39 @@ impl DogStatsD { None => DEFAULT_BUFFER_SIZE, }; - let buffer_reader = if let Some(pipe_name_ref) = pipe_name_opt { - // Windows named pipe transport + let addr = format!("{}:{}", config.host, config.port); + + 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"))] { - 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; - }); + // 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 = if config.port == 0 { + 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 { + 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 + } + } + }; - 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, } } #[cfg(not(all(windows, feature = "windows-pipes")))] @@ -225,14 +269,14 @@ 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) .await .expect("couldn't create UDP socket"); - BufferReader::UdpSocket(socket) + Transport::Reader(BufferReader::UdpSocket(socket)) }; let queue_size = match config.queue_size { @@ -250,7 +294,7 @@ impl DogStatsD { DogStatsD { cancel_token, aggregator_handle, - buffer_reader, + transport, metric_namespace: config.metric_namespace.clone(), buf_size, queue_size, @@ -269,11 +313,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, @@ -283,39 +330,71 @@ 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 { + read_loop( + buffer_reader, + tx, + reader_token, + buf_size, + queue_size, + pool_rx, + reader_pool_tx, + ) + .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"); @@ -341,6 +420,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, @@ -448,8 +530,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, @@ -464,7 +546,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, } @@ -472,7 +556,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; } @@ -617,15 +703,163 @@ async fn create_udp_socket( tokio::net::UdpSocket::from_std(std_socket) } +/// 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, + 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 { + 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 = + udp_read_loop_beside_pipe(socket, tx, stop, buf_size, queue_size, pool_rx, pool_tx) + .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, ) { @@ -672,6 +906,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! { @@ -702,12 +937,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 '{}'", @@ -729,6 +978,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 }); } @@ -891,10 +1146,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, @@ -951,10 +1206,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 f390fa48..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,6 +552,191 @@ 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 (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, 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", 18130)) + .await + .expect("UDP send failed"); + 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"); + + flush_until_seen( + &handle, + &dogstatsd_task, + &["dual.udp.metric", "dual.pipe.metric"], + ) + .await; + + 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(format!("{name}:1|c").as_bytes(), ("127.0.0.1", 18132)) + .await + .expect("UDP send failed"); + } + let names: Vec<&str> = names.iter().map(String::as_str).collect(); + flush_until_seen(&handle, &dogstatsd_task, &names).await; + + 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_oversized_udp_datagram_keeps_both_transports_running() { + let pipe_name = r"\\.\pipe\test_dogstatsd_oversized_udp"; + let (service, handle) = AggregatorService::new(SortedTags::parse("test:value").unwrap(), 1_024) + .expect("aggregator service creation failed"); + tokio::spawn(service.run()); + // 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. 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)); + 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; + + 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"); + flush_until_seen(&handle, &dogstatsd_task, &["after.oversized.pipe.metric"]).await; + + 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]