Debug test_named_pipe_max_errors

Lewis-E · Lewis-E · commit 3001c286e319 · 2025-12-30T10:19:04.000-05:00
diff --git a/crates/dogstatsd/Cargo.toml b/crates/dogstatsd/Cargo.toml
@@ -21,7 +21,7 @@ serde_json = { version = "1.0.116", default-features = false, features = ["alloc
 thiserror = { version = "1.0.58", default-features = false }
 tokio = { version = "1.37.0", default-features = false, features = ["macros", "rt-multi-thread", "net", "io-util"] }
 tokio-util = { version = "0.7.11", default-features = false }
-tracing = { version = "0.1.40", default-features = false }
+tracing = { version = "0.1.40", default-features = false, features = ["attributes"] }
 regex = { version = "1.10.6", default-features = false }
 zstd = { version = "0.13.3", default-features = false }
 datadog-fips = { path = "../datadog-fips", default-features = false }
@@ -30,6 +30,7 @@ datadog-fips = { path = "../datadog-fips", default-features = false }
 mockito = { version = "1.5.0", default-features = false }
 proptest = "1.4.0"
 tracing-test = { version = "0.2.5", default-features = false }
+tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }
 
 [features]
 default = [ "reqwest/rustls-tls" ]
diff --git a/crates/dogstatsd/src/dogstatsd.rs b/crates/dogstatsd/src/dogstatsd.rs
@@ -89,6 +89,15 @@ impl BufferReader {
     /// This is the main entry point for receiving metric data.
     /// Note: Different transports have different blocking behaviors.
     async fn read(&self) -> std::io::Result<(Vec<u8>, MessageSource)> {
+        debug!(
+            "BufferReader::read() called, variant: {}",
+            match self {
+                BufferReader::UdpSocket(_) => "UdpSocket",
+                BufferReader::MirrorTest(_, _) => "MirrorTest",
+                #[cfg(windows)]
+                BufferReader::NamedPipe { .. } => "NamedPipe",
+            }
+        );
         match self {
             BufferReader::UdpSocket(socket) => {
                 // UDP socket: blocks until a packet arrives
@@ -137,18 +146,30 @@ async fn handle_pipe_error_with_backoff(
     );
 
     if *consecutive_errors >= MAX_NAMED_PIPE_ERRORS {
-        return Err(std::io::Error::other(format!(
-            "Too many consecutive {} errors: {}",
-            error_type, error
-        )));
+        let err_msg = format!("Too many consecutive {} errors: {}", error_type, error);
+        debug!(
+            "MAX_NAMED_PIPE_ERRORS ({}) reached: {}",
+            MAX_NAMED_PIPE_ERRORS, err_msg
+        );
+        return Err(std::io::Error::other(err_msg));
     }
 
     let backoff_ms = 10u64 * (1 << *consecutive_errors);
+    debug!(
+        "Backing off {}ms after {} error (attempt {})",
+        backoff_ms, error_type, consecutive_errors
+    );
 
     // Sleep with cancellation support for clean, fast shutdown
     tokio::select! {
-        _ = sleep(Duration::from_millis(backoff_ms)) => Ok(true),
-        _ = cancel_token.cancelled() => Ok(false),
+        _ = sleep(Duration::from_millis(backoff_ms)) => {
+            debug!("Backoff sleep completed");
+            Ok(true)
+        },
+        _ = cancel_token.cancelled() => {
+            debug!("Backoff cancelled via token");
+            Ok(false)
+        },
     }
 }
 
@@ -162,21 +183,34 @@ async fn read_from_named_pipe(
     pipe_name: &str,
     cancel_token: &tokio_util::sync::CancellationToken,
 ) -> std::io::Result<Vec<u8>> {
+    debug!("read_from_named_pipe() called for: {}", pipe_name);
     let mut consecutive_errors = 0;
     let mut current_pipe: Option<tokio::net::windows::named_pipe::NamedPipeServer> = None;
     let mut needs_connection = true; // Track whether we need to wait for a client
 
+    debug!(
+        "Entering retry loop, cancel_token.is_cancelled() = {}",
+        cancel_token.is_cancelled()
+    );
     // Let named pipes cancel cleanly when the server is shut down
     while !cancel_token.is_cancelled() {
         // Create pipe if needed (initial startup or after error)
         if current_pipe.is_none() {
+            debug!("Attempting to create named pipe: {}", pipe_name);
             match ServerOptions::new().create(pipe_name) {
                 Ok(new_pipe) => {
+                    debug!("Successfully created named pipe: {}", pipe_name);
                     consecutive_errors = 0; // Reset on successful pipe creation
                     current_pipe = Some(new_pipe);
                     needs_connection = true;
                 }
                 Err(e) => {
+                    debug!(
+                        "Failed to create named pipe: {} - ErrorKind: {:?}, Error: {}",
+                        pipe_name,
+                        e.kind(),
+                        e
+                    );
                     match handle_pipe_error_with_backoff(
                         &mut consecutive_errors,
                         "pipe creation",
@@ -185,8 +219,17 @@ async fn read_from_named_pipe(
                     )
                     .await?
                     {
-                        true => continue,
-                        false => break,
+                        true => {
+                            debug!(
+                                "Retrying after pipe creation error backoff (attempt {})",
+                                consecutive_errors
+                            );
+                            continue;
+                        }
+                        false => {
+                            debug!("Breaking loop - cancelled during pipe creation error backoff");
+                            break;
+                        }
                     }
                 }
             }
@@ -197,19 +240,23 @@ async fn read_from_named_pipe(
             #[allow(clippy::expect_used)]
             let pipe = current_pipe.as_ref().expect("pipe must exist");
 
+            debug!("Waiting for client to connect to named pipe: {}", pipe_name);
             let connect_result = tokio::select! {
                 result = pipe.connect() => result,
                 _ = cancel_token.cancelled() => {
+                    debug!("Named pipe connect cancelled via token");
                     break;
                 }
             };
 
             match connect_result {
                 Ok(()) => {
+                    debug!("Client connected successfully to named pipe: {}", pipe_name);
                     consecutive_errors = 0;
                     needs_connection = false;
                 }
                 Err(e) => {
+                    debug!("Connection failed for named pipe: {} - {:?}", pipe_name, e);
                     // Connection failed - disconnect and recreate pipe on next iteration
                     if let Some(pipe) = current_pipe.as_ref() {
                         let _ = pipe.disconnect(); // Ignore disconnect errors, we're recreating anyway
@@ -223,8 +270,14 @@ async fn read_from_named_pipe(
                     )
                     .await?
                     {
-                        true => continue,
-                        false => break,
+                        true => {
+                            debug!("Retrying after connection error backoff");
+                            continue;
+                        }
+                        false => {
+                            debug!("Breaking loop - cancelled during connection error backoff");
+                            break;
+                        }
                     }
                 }
             }
@@ -306,8 +359,13 @@ impl DogStatsD {
         aggregator_handle: AggregatorHandle,
         cancel_token: tokio_util::sync::CancellationToken,
     ) -> DogStatsD {
+        debug!(
+            "DogStatsD::new() called with windows_pipe_name: {:?}, host: {}, port: {}",
+            config.windows_pipe_name, config.host, config.port
+        );
         #[allow(unused_variables)] // pipe_name unused on non-Windows
         let buffer_reader = if let Some(ref pipe_name) = config.windows_pipe_name {
+            debug!("Creating NamedPipe BufferReader for: {}", pipe_name);
             #[cfg(windows)]
             {
                 BufferReader::NamedPipe {
@@ -321,6 +379,10 @@ impl DogStatsD {
                 panic!("Named pipes are only supported on Windows.")
             }
         } else {
+            debug!(
+                "Creating UdpSocket BufferReader for: {}:{}",
+                config.host, config.port
+            );
             // UDP socket for all platforms
             let addr = format!("{}:{}", config.host, config.port);
             // TODO (UDS socket)
@@ -330,6 +392,7 @@ impl DogStatsD {
                 .expect("couldn't bind to address");
             BufferReader::UdpSocket(socket)
         };
+        debug!("BufferReader created successfully");
 
         DogStatsD {
             cancel_token,
@@ -341,21 +404,31 @@ impl DogStatsD {
 
     /// Main event loop that continuously receives and processes metrics.
     pub async fn spin(self) {
+        debug!(
+            "spin() starting, cancel_token.is_cancelled() = {}",
+            self.cancel_token.is_cancelled()
+        );
         let mut spin_cancelled = false;
         while !spin_cancelled {
+            debug!("spin() loop iteration, calling consume_statsd()...");
             self.consume_statsd().await;
+            debug!("consume_statsd() returned");
             spin_cancelled = self.cancel_token.is_cancelled();
+            debug!("spin_cancelled = {}", spin_cancelled);
         }
+        debug!("spin() exiting");
     }
 
     /// Receive one batch of metrics from the transport layer and process them.
     async fn consume_statsd(&self) {
+        debug!("consume_statsd() called, about to read from buffer_reader");
         #[allow(clippy::expect_used)]
         let (buf, src) = self
             .buffer_reader
             .read()
             .await
             .expect("didn't receive data");
+        debug!("buffer_reader.read() completed with {} bytes", buf.len());
 
         #[allow(clippy::expect_used)]
         let msgs = std::str::from_utf8(&buf).expect("couldn't parse as string");
diff --git a/crates/dogstatsd/tests/integration_test.rs b/crates/dogstatsd/tests/integration_test.rs
@@ -1,15 +1,17 @@
 // Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/
 // SPDX-License-Identifier: Apache-2.0
 
-use dogstatsd::metric::SortedTags;
 use dogstatsd::{
     aggregator::Aggregator as MetricsAggregator,
     aggregator_service::{AggregatorHandle, AggregatorService},
     api_key::ApiKeyFactory,
     constants::CONTEXTS,
-    datadog::{DdDdUrl, MetricsIntakeUrlPrefix, MetricsIntakeUrlPrefixOverride},
+    datadog::{
+        DdApi, DdDdUrl, MetricsIntakeUrlPrefix, MetricsIntakeUrlPrefixOverride, RetryStrategy,
+    },
     dogstatsd::{DogStatsD, DogStatsDConfig},
-    flusher::{Flusher, FlusherConfig},
+    flusher::{Flusher, FlusherConfig, ShippingError},
+    metric::{parse, SortedTags},
 };
 use mockito::Server;
 use std::sync::Arc;
@@ -21,13 +23,14 @@ use tokio_util::sync::CancellationToken;
 use zstd::zstd_safe::CompressionLevel;
 
 #[cfg(windows)]
-use tokio::{io::AsyncWriteExt, net::windows::named_pipe::ClientOptions};
+use tokio::{
+    io::AsyncWriteExt, net::windows::named_pipe::ClientOptions,
+    net::windows::named_pipe::ServerOptions,
+};
 
 #[cfg(test)]
 #[tokio::test]
 async fn dogstatsd_server_ships_series() {
-    use dogstatsd::datadog::RetryStrategy;
-
     let mut mock_server = Server::new_async().await;
 
     let mock = mock_server
@@ -122,9 +125,6 @@ async fn start_dogstatsd(aggregator_handle: AggregatorHandle) -> CancellationTok
 #[cfg(test)]
 #[tokio::test]
 async fn test_send_with_retry_immediate_failure() {
-    use dogstatsd::datadog::{DdApi, DdDdUrl, RetryStrategy};
-    use dogstatsd::metric::{parse, SortedTags};
-
     let mut server = Server::new_async().await;
     let mock = server
         .mock("POST", "/api/v2/series")
@@ -170,9 +170,6 @@ async fn test_send_with_retry_immediate_failure() {
 #[cfg(test)]
 #[tokio::test]
 async fn test_send_with_retry_linear_backoff_success() {
-    use dogstatsd::datadog::{DdApi, DdDdUrl, RetryStrategy};
-    use dogstatsd::metric::{parse, SortedTags};
-
     let mut server = Server::new_async().await;
     let mock = server
         .mock("POST", "/api/v2/series")
@@ -232,10 +229,6 @@ async fn test_send_with_retry_linear_backoff_success() {
 #[cfg(test)]
 #[tokio::test]
 async fn test_send_with_retry_immediate_failure_after_one_attempt() {
-    use dogstatsd::datadog::{DdApi, DdDdUrl, RetryStrategy};
-    use dogstatsd::flusher::ShippingError;
-    use dogstatsd::metric::{parse, SortedTags};
-
     let mut server = Server::new_async().await;
     let mock = server
         .mock("POST", "/api/v2/series")
@@ -454,26 +447,47 @@ async fn test_named_pipe_cancellation() {
 #[cfg(windows)]
 #[tokio::test]
 async fn test_named_pipe_max_errors() {
-    use tokio::net::windows::named_pipe::ServerOptions;
+    use tracing::info;
+    use tracing_subscriber::{fmt, EnvFilter};
+
+    // Initialize subscriber that will capture ALL logs including from dogstatsd crate
+    let _ = fmt()
+        .with_env_filter(
+            EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("debug")),
+        )
+        .with_test_writer()
+        .try_init();
 
     let pipe_name = r"\\.\pipe\test_dogstatsd_max_errors";
+    info!(
+        "TEST START: test_named_pipe_max_errors with pipe: {}",
+        pipe_name
+    );
+
     let (service, handle) = AggregatorService::new(SortedTags::parse("test:value").unwrap(), 1_024)
         .expect("aggregator service creation failed");
+    info!("Aggregator service created");
     tokio::spawn(service.run());
+    info!("Aggregator service spawned");
 
     let cancel_token = CancellationToken::new();
+    info!("Cancel token created");
 
     // Create a pipe with first_pipe_instance(true) to block other instances
+    info!("Creating blocking pipe with first_pipe_instance(true)...");
     let _blocking_pipe = ServerOptions::new()
         .first_pipe_instance(true)
         .create(pipe_name)
         .expect("failed to create blocking pipe");
+    info!("Blocking pipe created successfully - should prevent other instances");
 
     // Start DogStatsD server - should fail to create pipe due to existing instance
+    info!("Spawning DogStatsD task (should fail to create pipe and panic after 5 retries)...");
     let dogstatsd_task = {
         let handle = handle.clone();
         let cancel_token_clone = cancel_token.clone();
         tokio::spawn(async move {
+            info!("DogStatsD task starting...");
             let dogstatsd = DogStatsD::new(
                 &DogStatsDConfig {
                     host: String::new(),
@@ -485,14 +499,22 @@ async fn test_named_pipe_max_errors() {
                 cancel_token_clone,
             )
             .await;
+            info!("DogStatsD::new() completed, calling spin()...");
             dogstatsd.spin().await;
+            info!("DogStatsD spin() completed (should not reach here if panics)");
         })
     };
+    info!("DogStatsD task spawned, now waiting with 1500ms timeout...");
 
     // Task should complete after exceeding MAX_NAMED_PIPE_ERRORS (5) attempts
     // With backoff: 20ms + 40ms + 80ms + 160ms + 320ms = ~620ms + overhead
     let result = timeout(Duration::from_millis(1500), dogstatsd_task).await;
 
+    info!(
+        "Timeout completed, result: {:?}",
+        result.as_ref().map(|_| "Ok").map_err(|_| "Timeout")
+    );
+
     // The task should panic because read_from_named_pipe returns an error after max retries
     assert!(
         result.is_ok(),
@@ -501,13 +523,19 @@ async fn test_named_pipe_max_errors() {
 
     // Verify the task actually panicked (completed with error)
     if let Ok(join_result) = result {
+        info!("Task completed, checking if it panicked...");
         assert!(
             join_result.is_err(),
             "task should have panicked after exceeding max consecutive errors"
         );
     }
 
     // Cleanup
+    info!("Test cleanup: cancelling token and shutting down...");
     cancel_token.cancel();
+    info!("Blocking pipe being dropped...");
+    drop(_blocking_pipe);
+    info!("Blocking pipe dropped");
     handle.shutdown().expect("shutdown failed");
+    info!("TEST END: test_named_pipe_max_errors");
 }
