Add Windows Named Pipes to Dogstatsd

LICENSE-3rdparty.csv

@@ -120,12 +120,11 @@ nix,https://github.com/nix-rust/nix,MIT,The nix-rust Project Developers
 nu-ansi-term,https://github.com/nushell/nu-ansi-term,MIT,"ogham@bsago.me, Ryan Scheel (Havvy) <ryan.havvy@gmail.com>, Josh Triplett <josh@joshtriplett.org>, The Nushell Project Developers"
 num-traits,https://github.com/rust-num/num-traits,MIT OR Apache-2.0,The Rust Project Developers
 once_cell,https://github.com/matklad/once_cell,MIT OR Apache-2.0,Aleksey Kladov <aleksey.kladov@gmail.com>
-openssl-probe,https://github.com/alexcrichton/openssl-probe,MIT OR Apache-2.0,Alex Crichton <alex@alexcrichton.com>
+openssl-probe,https://github.com/rustls/openssl-probe,MIT OR Apache-2.0,Alex Crichton <alex@alexcrichton.com>
 ordered-float,https://github.com/reem/rust-ordered-float,MIT,"Jonathan Reem <jonathan.reem@gmail.com>, Matt Brubeck <mbrubeck@limpet.net>"
 parking,https://github.com/smol-rs/parking,Apache-2.0 OR MIT,"Stjepan Glavina <stjepang@gmail.com>, The Rust Project Developers"
 parking_lot,https://github.com/Amanieu/parking_lot,MIT OR Apache-2.0,Amanieu d'Antras <amanieu@gmail.com>
 parking_lot_core,https://github.com/Amanieu/parking_lot,MIT OR Apache-2.0,Amanieu d'Antras <amanieu@gmail.com>
-paste,https://github.com/dtolnay/paste,MIT OR Apache-2.0,David Tolnay <dtolnay@gmail.com>
 path-tree,https://github.com/viz-rs/path-tree,MIT OR Apache-2.0,Fangdun Tsai <cfddream@gmail.com>
 percent-encoding,https://github.com/servo/rust-url,MIT OR Apache-2.0,The rust-url developers
 petgraph,https://github.com/petgraph/petgraph,MIT OR Apache-2.0,"bluss, mitchmindtree"
@@ -162,7 +161,7 @@ regex-automata,https://github.com/rust-lang/regex,MIT OR Apache-2.0,"The Rust Pr
 regex-syntax,https://github.com/rust-lang/regex,MIT OR Apache-2.0,"The Rust Project Developers, Andrew Gallant <jamslam@gmail.com>"
 reqwest,https://github.com/seanmonstar/reqwest,MIT OR Apache-2.0,Sean McArthur <sean@seanmonstar.com>
 ring,https://github.com/briansmith/ring,Apache-2.0 AND ISC,The ring Authors
-rmp,https://github.com/3Hren/msgpack-rust,MIT,Evgeny Safronov <division494@gmail.com>
+rmp,https://github.com/3Hren/msgpack-rust,MIT,"Evgeny Safronov <division494@gmail.com>, Kornel <kornel@geekhood.net>"
 rmp-serde,https://github.com/3Hren/msgpack-rust,MIT,Evgeny Safronov <division494@gmail.com>
 rmpv,https://github.com/3Hren/msgpack-rust,MIT,Evgeny Safronov <division494@gmail.com>
 rustc-hash,https://github.com/rust-lang/rustc-hash,Apache-2.0 OR MIT,The Rust Project Developers
@@ -285,6 +284,7 @@ zeroize,https://github.com/RustCrypto/utils,Apache-2.0 OR MIT,The RustCrypto Pro
 zerotrie,https://github.com/unicode-org/icu4x,Unicode-3.0,The ICU4X Project Developers
 zerovec,https://github.com/unicode-org/icu4x,Unicode-3.0,The ICU4X Project Developers
 zerovec-derive,https://github.com/unicode-org/icu4x,Unicode-3.0,Manish Goregaokar <manishsmail@gmail.com>
+zmij,https://github.com/dtolnay/zmij,MIT,David Tolnay <dtolnay@gmail.com>
 zstd,https://github.com/gyscos/zstd-rs,MIT,Alexandre Bury <alexandre.bury@gmail.com>
 zstd-safe,https://github.com/gyscos/zstd-rs,MIT OR Apache-2.0,Alexandre Bury <alexandre.bury@gmail.com>
 zstd-sys,https://github.com/gyscos/zstd-rs,MIT OR Apache-2.0,Alexandre Bury <alexandre.bury@gmail.com>

crates/datadog-serverless-compat/src/main.rs

@@ -65,10 +65,36 @@ pub async fn main() {
     };
 
     let dd_api_key: Option<String> = env::var("DD_API_KEY").ok();
-    let dd_dogstatsd_port: u16 = env::var("DD_DOGSTATSD_PORT")
-        .ok()
-        .and_then(|port| port.parse::<u16>().ok())
-        .unwrap_or(DEFAULT_DOGSTATSD_PORT);
+
+    // Windows named pipe name for DogStatsD.
+    // Normalize by adding \\.\pipe\ prefix if not present
+    let dd_dogstatsd_windows_pipe_name: Option<String> = {
+        #[cfg(windows)]
+        {
+            env::var("DD_DOGSTATSD_WINDOWS_PIPE_NAME")
+                .ok()
+                .map(|pipe_name| {
+                    if pipe_name.starts_with("\\\\.\\pipe\\") || pipe_name.starts_with(r"\\.\pipe\")
+                    {
+                        pipe_name
+                    } else {
+                        format!(r"\\.\pipe\{}", pipe_name)
+                    }
+                })
+        }
+        #[cfg(not(windows))]
+        {
+            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::<u16>().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")
@@ -152,9 +178,14 @@ pub async fn main() {
             https_proxy,
             dogstatsd_tags,
             dd_statsd_metric_namespace,
+            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}");
+        }
         (metrics_flusher, Some(aggregator_handle))
     } else {
         info!("dogstatsd disabled");
@@ -181,6 +212,7 @@ async fn start_dogstatsd(
     https_proxy: Option<String>,
     dogstatsd_tags: &str,
     metric_namespace: Option<String>,
+    windows_pipe_name: Option<String>,
 ) -> (CancellationToken, Option<Flusher>, AggregatorHandle) {
     // 1. Create the aggregator service
     #[allow(clippy::expect_used)]
@@ -197,6 +229,7 @@ async fn start_dogstatsd(
         host: AGENT_HOST.to_string(),
         port,
         metric_namespace,
+        windows_pipe_name,
     };
     let dogstatsd_cancel_token = tokio_util::sync::CancellationToken::new();
 

crates/datadog-trace-agent/src/config.rs

@@ -83,6 +83,7 @@ pub struct Config {
     pub dd_apm_receiver_port: u16,
     pub dd_apm_windows_pipe_name: Option<String>,
     pub dd_dogstatsd_port: u16,
+    pub dd_dogstatsd_windows_pipe_name: Option<String>,
     pub env_type: trace_utils::EnvironmentType,
     pub app_name: Option<String>,
     pub max_request_content_length: usize,
@@ -118,11 +119,25 @@ impl Config {
             anyhow::anyhow!("Unable to identify environment. Shutting down Mini Agent.")
         })?;
 
-        let dd_apm_windows_pipe_name: Option<String> =
-            env::var("DD_APM_WINDOWS_PIPE_NAME").ok().map(|pipe_name| {
-                // Prepend \\.\pipe\ prefix to match datadog-agent behavior
-                format!(r"\\.\pipe\{}", pipe_name)
-            });
+        // Windows named pipe name for APM receiver.
+        // Normalize by adding \\.\pipe\ prefix if not present
+        let dd_apm_windows_pipe_name: Option<String> = {
+            #[cfg(any(windows, test))]
+            {
+                env::var("DD_APM_WINDOWS_PIPE_NAME").ok().map(|pipe_name| {
+                    if pipe_name.starts_with("\\\\.\\pipe\\") || pipe_name.starts_with(r"\\.\pipe\")
+                    {
+                        pipe_name
+                    } else {
+                        format!(r"\\.\pipe\{}", pipe_name)
+                    }
+                })
+            }
+            #[cfg(not(any(windows, test)))]
+            {
+                None
+            }
+        };
         let dd_apm_receiver_port: u16 = if dd_apm_windows_pipe_name.is_some() {
             0 // Override to 0 when using Windows named pipe
         } else {
@@ -132,10 +147,35 @@ impl Config {
                 .unwrap_or(DEFAULT_APM_RECEIVER_PORT)
         };
 
-        let dd_dogstatsd_port: u16 = env::var("DD_DOGSTATSD_PORT")
-            .ok()
-            .and_then(|port| port.parse::<u16>().ok())
-            .unwrap_or(DEFAULT_DOGSTATSD_PORT);
+        let dd_dogstatsd_windows_pipe_name: Option<String> = {
+            #[cfg(any(windows, test))]
+            {
+                env::var("DD_DOGSTATSD_WINDOWS_PIPE_NAME")
+                    .ok()
+                    .map(|pipe_name| {
+                        if pipe_name.starts_with("\\\\.\\pipe\\")
+                            || pipe_name.starts_with(r"\\.\pipe\")
+                        {
+                            pipe_name
+                        } else {
+                            format!(r"\\.\pipe\{}", pipe_name)
+                        }
+                    })
+            }
+            #[cfg(not(any(windows, test)))]
+            {
+                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::<u16>().ok())
+                .unwrap_or(DEFAULT_DOGSTATSD_PORT)
+        };
+
         let dd_site = env::var("DD_SITE").unwrap_or_else(|_| "datadoghq.com".to_string());
 
         // construct the trace & trace stats intake urls based on DD_SITE env var (to flush traces &
@@ -185,6 +225,7 @@ impl Config {
             dd_apm_receiver_port,
             dd_apm_windows_pipe_name,
             dd_dogstatsd_port,
+            dd_dogstatsd_windows_pipe_name,
             dd_site,
             trace_intake: Endpoint {
                 url: hyper::Uri::from_str(&trace_intake_url).unwrap(),
@@ -337,51 +378,73 @@ mod tests {
 
     #[test]
     #[serial]
-    fn test_default_dogstatsd_port() {
+    fn test_apm_windows_pipe_name() {
         env::set_var("DD_API_KEY", "_not_a_real_key_");
         env::set_var("ASCSVCRT_SPRING__APPLICATION__NAME", "test-spring-app");
+        env::set_var("DD_APM_WINDOWS_PIPE_NAME", r"test_pipe");
         let config_res = config::Config::new();
         assert!(config_res.is_ok());
         let config = config_res.unwrap();
-        assert_eq!(config.dd_dogstatsd_port, 8125);
+        assert_eq!(
+            config.dd_apm_windows_pipe_name,
+            Some(r"\\.\pipe\test_pipe".to_string())
+        );
+
+        // Port should be overridden to 0 when pipe is set
+        assert_eq!(config.dd_apm_receiver_port, 0);
         env::remove_var("DD_API_KEY");
         env::remove_var("ASCSVCRT_SPRING__APPLICATION__NAME");
+        env::remove_var("DD_APM_WINDOWS_PIPE_NAME");
     }
 
     #[test]
     #[serial]
-    fn test_custom_dogstatsd_port() {
+    fn test_dogstatsd_windows_pipe_name() {
         env::set_var("DD_API_KEY", "_not_a_real_key_");
         env::set_var("ASCSVCRT_SPRING__APPLICATION__NAME", "test-spring-app");
-        env::set_var("DD_DOGSTATSD_PORT", "18125");
+        env::set_var("DD_DOGSTATSD_WINDOWS_PIPE_NAME", r"test_pipe");
         let config_res = config::Config::new();
-        println!("{:?}", config_res);
         assert!(config_res.is_ok());
         let config = config_res.unwrap();
-        assert_eq!(config.dd_dogstatsd_port, 18125);
+        assert_eq!(
+            config.dd_dogstatsd_windows_pipe_name,
+            Some(r"\\.\pipe\test_pipe".to_string())
+        );
+
+        // Port should be overridden to 0 when pipe is set
+        assert_eq!(config.dd_dogstatsd_port, 0);
         env::remove_var("DD_API_KEY");
         env::remove_var("ASCSVCRT_SPRING__APPLICATION__NAME");
-        env::remove_var("DD_DOGSTATSD_PORT");
+        env::remove_var("DD_DOGSTATSD_WINDOWS_PIPE_NAME");
     }
 
     #[test]
     #[serial]
-    fn test_apm_windows_pipe_name() {
+    fn test_default_dogstatsd_port() {
         env::set_var("DD_API_KEY", "_not_a_real_key_");
         env::set_var("ASCSVCRT_SPRING__APPLICATION__NAME", "test-spring-app");
-        env::set_var("DD_APM_WINDOWS_PIPE_NAME", r"test_pipe");
         let config_res = config::Config::new();
         assert!(config_res.is_ok());
         let config = config_res.unwrap();
-        assert_eq!(
-            config.dd_apm_windows_pipe_name,
-            Some(r"\\.\pipe\test_pipe".to_string())
-        );
-        // Port should be overridden to 0 when pipe is set
-        assert_eq!(config.dd_apm_receiver_port, 0);
+        assert_eq!(config.dd_dogstatsd_port, 8125);
         env::remove_var("DD_API_KEY");
         env::remove_var("ASCSVCRT_SPRING__APPLICATION__NAME");
-        env::remove_var("DD_APM_WINDOWS_PIPE_NAME");
+    }
+
+    #[test]
+    #[serial]
+    fn test_custom_dogstatsd_port() {
+        env::set_var("DD_API_KEY", "_not_a_real_key_");
+        env::set_var("ASCSVCRT_SPRING__APPLICATION__NAME", "test-spring-app");
+        env::set_var("DD_DOGSTATSD_PORT", "18125");
+        let config_res = config::Config::new();
+        println!("{:?}", config_res);
+        assert!(config_res.is_ok());
+        let config = config_res.unwrap();
+        assert_eq!(config.dd_dogstatsd_port, 18125);
+        env::remove_var("DD_API_KEY");
+        env::remove_var("ASCSVCRT_SPRING__APPLICATION__NAME");
+        env::remove_var("DD_DOGSTATSD_PORT");
     }
 
     #[test]

crates/datadog-trace-agent/src/trace_processor.rs

@@ -206,6 +206,7 @@ mod tests {
             dd_apm_receiver_port: 8126,
             dd_apm_windows_pipe_name: None,
             dd_dogstatsd_port: 8125,
+            dd_dogstatsd_windows_pipe_name: None,
             env_type: trace_utils::EnvironmentType::CloudFunction,
             os: "linux".to_string(),
             obfuscation_config: ObfuscationConfig::new().unwrap(),

crates/datadog-trace-agent/tests/integration_test.rs

@@ -26,6 +26,7 @@ pub fn create_tcp_test_config() -> Config {
         dd_apm_receiver_port: 8126,
         dd_apm_windows_pipe_name: None,
         dd_dogstatsd_port: 8125,
+        dd_dogstatsd_windows_pipe_name: None,
         env_type: trace_utils::EnvironmentType::AzureFunction,
         app_name: Some("test-app".to_string()),
         max_request_content_length: 10_000_000,

crates/dogstatsd/Cargo.toml

@@ -19,7 +19,7 @@ reqwest = { version = "0.12.4", features = ["json", "http2"], default-features =
 serde = { version = "1.0.197", default-features = false, features = ["derive"] }
 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"] }
+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 }
 regex = { version = "1.10.6", default-features = false }