Add Windows Named Pipes to Dogstatsd

LICENSE-3rdparty.csv

@@ -198,6 +198,7 @@ security-framework-sys,https://github.com/kornelski/rust-security-framework,MIT
 semver,https://github.com/dtolnay/semver,MIT OR Apache-2.0,David Tolnay <[email protected]>
 serde,https://github.com/serde-rs/serde,MIT OR Apache-2.0,"Erick Tryzelaar <[email protected]>, David Tolnay <[email protected]>"
 serde_bytes,https://github.com/serde-rs/bytes,MIT OR Apache-2.0,David Tolnay <[email protected]>
+serde_core,https://github.com/serde-rs/serde,MIT OR Apache-2.0,"Erick Tryzelaar <[email protected]>, David Tolnay <[email protected]>"
 serde_derive,https://github.com/serde-rs/serde,MIT OR Apache-2.0,"Erick Tryzelaar <[email protected]>, David Tolnay <[email protected]>"
 serde_json,https://github.com/serde-rs/json,MIT OR Apache-2.0,"Erick Tryzelaar <[email protected]>, David Tolnay <[email protected]>"
 serde_regex,https://github.com/tailhook/serde-regex,MIT OR Apache-2.0,[email protected]

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")
@@ -149,9 +175,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");
@@ -178,6 +209,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)]
@@ -194,6 +226,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

@@ -74,6 +74,7 @@ impl Tags {
 pub struct Config {
     pub dd_site: 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,
@@ -105,6 +106,8 @@ impl Config {
             .ok()
             .and_then(|port| port.parse::<u16>().ok())
             .unwrap_or(DEFAULT_DOGSTATSD_PORT);
+        let dd_dogstatsd_windows_pipe_name: Option<String> =
+            env::var("DD_DOGSTATSD_WINDOWS_PIPE_NAME").ok();
         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 &
@@ -141,6 +144,7 @@ impl Config {
             stats_flush_interval: 3,
             verify_env_timeout: 100,
             dd_dogstatsd_port,
+            dd_dogstatsd_windows_pipe_name,
             dd_site,
             trace_intake: Endpoint {
                 url: hyper::Uri::from_str(&trace_intake_url).unwrap(),
@@ -315,6 +319,24 @@ mod tests {
         env::remove_var("DD_DOGSTATSD_PORT");
     }
 
+    #[test]
+    #[serial]
+    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_WINDOWS_PIPE_NAME", r"\\.\pipe\dogstatsd");
+        let config_res = config::Config::new();
+        assert!(config_res.is_ok());
+        let config = config_res.unwrap();
+        assert_eq!(
+            config.dd_dogstatsd_windows_pipe_name,
+            Some(r"\\.\pipe\dogstatsd".to_string())
+        );
+        env::remove_var("DD_API_KEY");
+        env::remove_var("ASCSVCRT_SPRING__APPLICATION__NAME");
+        env::remove_var("DD_DOGSTATSD_WINDOWS_PIPE_NAME");
+    }
+
     fn test_config_with_dd_tags(dd_tags: &str) -> config::Config {
         env::set_var("DD_API_KEY", "_not_a_real_key_");
         env::set_var("ASCSVCRT_SPRING__APPLICATION__NAME", "test-spring-app");

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

@@ -193,7 +193,10 @@ impl MiniAgent {
                     ),
                 }
             }
-            (_, INFO_ENDPOINT_PATH) => match Self::info_handler(config.dd_dogstatsd_port) {
+            (_, INFO_ENDPOINT_PATH) => match Self::info_handler(
+                config.dd_dogstatsd_port,
+                config.dd_dogstatsd_windows_pipe_name.as_deref(),
+            ) {
                 Ok(res) => Ok(res),
                 Err(err) => log_and_create_http_response(
                     &format!("Info endpoint error: {err}"),
@@ -208,7 +211,18 @@ impl MiniAgent {
         }
     }
 
-    fn info_handler(dd_dogstatsd_port: u16) -> http::Result<hyper_migration::HttpResponse> {
+    fn info_handler(
+        dd_dogstatsd_port: u16,
+        dd_dogstatsd_windows_pipe_name: Option<&str>,
+    ) -> http::Result<hyper_migration::HttpResponse> {
+        let mut config_json = serde_json::json!({
+            "statsd_port": dd_dogstatsd_port
+        });
+
+        if let Some(pipe_name) = dd_dogstatsd_windows_pipe_name {
+            config_json["statsd_windows_pipe_name"] = serde_json::json!(pipe_name);
+        }
+
         let response_json = json!(
             {
                 "endpoints": [
@@ -217,9 +231,7 @@ impl MiniAgent {
                     INFO_ENDPOINT_PATH
                 ],
                 "client_drop_p0s": true,
-                "config": {
-                    "statsd_port": dd_dogstatsd_port
-                }
+                "config": config_json
             }
         );
         Response::builder()

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

@@ -198,6 +198,7 @@ mod tests {
             },
             dd_site: "datadoghq.com".to_string(),
             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/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 }