add options to change dogstatsd configuration

Author: duncanista

bottlecap/Cargo.lock

@@ -71,8 +71,8 @@ libdd-trace-utils = { git = "https://github.com/DataDog/libdatadog", rev = "158b
 libdd-trace-normalization = { git = "https://github.com/DataDog/libdatadog",  rev = "158b59471f1132e3cb36023fa3c46ccb2dd0eda1" }
 libdd-trace-obfuscation = { git = "https://github.com/DataDog/libdatadog", rev = "158b59471f1132e3cb36023fa3c46ccb2dd0eda1"  }
 libdd-trace-stats = { git = "https://github.com/DataDog/libdatadog", rev = "158b59471f1132e3cb36023fa3c46ccb2dd0eda1"  }
-dogstatsd = { git = "https://github.com/DataDog/serverless-components", rev = "086c9375e0bec1f50e268827d9347a5e3b73b79d", default-features = false }
-datadog-fips = { git = "https://github.com/DataDog/serverless-components", rev = "086c9375e0bec1f50e268827d9347a5e3b73b79d", default-features = false }
+dogstatsd = { git = "https://github.com/DataDog/serverless-components", rev = "659a19f036f7f42776e1746ae60cbbc8396a7593", default-features = false }
+datadog-fips = { git = "https://github.com/DataDog/serverless-components", rev = "659a19f036f7f42776e1746ae60cbbc8396a7593", default-features = false }
 libddwaf = { version = "1.28.1", git = "https://github.com/DataDog/libddwaf-rust", rev = "d1534a158d976bd4f747bf9fcc58e0712d2d17fc", default-features = false, features = ["serde"] }
 
 [dev-dependencies]

bottlecap/Cargo.toml

@@ -1194,6 +1194,9 @@ async fn start_dogstatsd(
         host: EXTENSION_HOST.to_string(),
         port: DOGSTATSD_PORT,
         metric_namespace: config.statsd_metric_namespace.clone(),
+        so_rcvbuf: config.dogstatsd_so_rcvbuf,
+        buffer_size: config.dogstatsd_buffer_size,
+        queue_size: config.dogstatsd_queue_size,
     };
     let cancel_token = tokio_util::sync::CancellationToken::new();
     let dogstatsd_agent = DogStatsD::new(

bottlecap/src/bin/bottlecap/main.rs

@@ -277,6 +277,24 @@ pub struct EnvConfig {
     #[serde(deserialize_with = "deserialize_optional_string")]
     pub statsd_metric_namespace: Option<String>,
 
+    /// @env `DD_DOGSTATSD_SO_RCVBUF`
+    /// Size of the receive buffer for `DogStatsD` UDP packets, in bytes (`SO_RCVBUF`).
+    /// Increase to reduce packet loss under high-throughput metric bursts.
+    #[serde(deserialize_with = "deserialize_option_lossless")]
+    pub dogstatsd_so_rcvbuf: Option<usize>,
+
+    /// @env `DD_DOGSTATSD_BUFFER_SIZE`
+    /// Maximum size of a single read from any transport (UDP or named pipe), in bytes.
+    /// Defaults to 8192.
+    #[serde(deserialize_with = "deserialize_option_lossless")]
+    pub dogstatsd_buffer_size: Option<usize>,
+
+    /// @env `DD_DOGSTATSD_QUEUE_SIZE`
+    /// Internal queue capacity between the socket reader and metric processor.
+    /// Defaults to 1024. Increase if the processor can't keep up with burst traffic.
+    #[serde(deserialize_with = "deserialize_option_lossless")]
+    pub dogstatsd_queue_size: Option<usize>,
+
     // OTLP
     //
     // - APM / Traces
@@ -554,6 +572,11 @@ fn merge_config(config: &mut Config, env_config: &EnvConfig) {
         config.statsd_metric_namespace = parse_metric_namespace(namespace);
     }
 
+    // DogStatsD
+    merge_option!(config, env_config, dogstatsd_so_rcvbuf);
+    merge_option!(config, env_config, dogstatsd_buffer_size);
+    merge_option!(config, env_config, dogstatsd_queue_size);
+
     // OTLP
     merge_option_to_value!(config, env_config, otlp_config_traces_enabled);
     merge_option_to_value!(
@@ -830,6 +853,11 @@ mod tests {
             );
             jail.set_env("DD_OTLP_CONFIG_LOGS_ENABLED", "true");
 
+            // DogStatsD
+            jail.set_env("DD_DOGSTATSD_SO_RCVBUF", "1048576");
+            jail.set_env("DD_DOGSTATSD_BUFFER_SIZE", "65507");
+            jail.set_env("DD_DOGSTATSD_QUEUE_SIZE", "2048");
+
             // AWS Lambda
             jail.set_env(
                 "DD_API_KEY_SECRET_ARN",
@@ -984,6 +1012,9 @@ mod tests {
                 otlp_config_traces_probabilistic_sampler_sampling_percentage: Some(50),
                 otlp_config_logs_enabled: true,
                 statsd_metric_namespace: None,
+                dogstatsd_so_rcvbuf: Some(1048576),
+                dogstatsd_buffer_size: Some(65507),
+                dogstatsd_queue_size: Some(2048),
                 api_key_secret_arn: "arn:aws:secretsmanager:region:account:secret:datadog-api-key"
                     .to_string(),
                 kms_api_key: "test-kms-key".to_string(),
@@ -1170,4 +1201,43 @@ mod tests {
             Ok(())
         });
     }
+
+    #[test]
+    fn test_dogstatsd_config_from_env() {
+        figment::Jail::expect_with(|jail| {
+            jail.clear_env();
+            jail.set_env("DD_DOGSTATSD_SO_RCVBUF", "1048576");
+            jail.set_env("DD_DOGSTATSD_BUFFER_SIZE", "65507");
+            jail.set_env("DD_DOGSTATSD_QUEUE_SIZE", "2048");
+
+            let mut config = Config::default();
+            let env_config_source = EnvConfigSource;
+            env_config_source
+                .load(&mut config)
+                .expect("Failed to load config");
+
+            assert_eq!(config.dogstatsd_so_rcvbuf, Some(1048576));
+            assert_eq!(config.dogstatsd_buffer_size, Some(65507));
+            assert_eq!(config.dogstatsd_queue_size, Some(2048));
+            Ok(())
+        });
+    }
+
+    #[test]
+    fn test_dogstatsd_config_defaults_to_none() {
+        figment::Jail::expect_with(|jail| {
+            jail.clear_env();
+
+            let mut config = Config::default();
+            let env_config_source = EnvConfigSource;
+            env_config_source
+                .load(&mut config)
+                .expect("Failed to load config");
+
+            assert_eq!(config.dogstatsd_so_rcvbuf, None);
+            assert_eq!(config.dogstatsd_buffer_size, None);
+            assert_eq!(config.dogstatsd_queue_size, None);
+            Ok(())
+        });
+    }
 }

bottlecap/src/config/env.rs

@@ -300,6 +300,17 @@ pub struct Config {
     // Metrics
     pub metrics_config_compression_level: i32,
     pub statsd_metric_namespace: Option<String>,
+    /// Size of the receive buffer for `DogStatsD` UDP packets, in bytes (`SO_RCVBUF`).
+    /// Increase to reduce packet loss under high-throughput metric bursts.
+    /// If None, uses the OS default.
+    pub dogstatsd_so_rcvbuf: Option<usize>,
+    /// Maximum size of a single read from any transport (UDP or named pipe), in bytes.
+    /// Defaults to 8192. For UDP, the client must batch metrics into packets of
+    /// this size for the increase to take effect.
+    pub dogstatsd_buffer_size: Option<usize>,
+    /// Internal queue capacity between the socket reader and metric processor.
+    /// Defaults to 1024. Increase if the processor can't keep up with burst traffic.
+    pub dogstatsd_queue_size: Option<usize>,
 
     // OTLP
     //
@@ -418,6 +429,14 @@ impl Default for Config {
             metrics_config_compression_level: 3,
             statsd_metric_namespace: None,
 
+            // DogStatsD
+            // Defaults to None, which uses the OS default.
+            dogstatsd_so_rcvbuf: None,
+            // Defaults to 8192 internally.
+            dogstatsd_buffer_size: None,
+            // Defaults to 1024 internally.
+            dogstatsd_queue_size: None,
+
             // OTLP
             otlp_config_traces_enabled: true,
             otlp_config_traces_span_name_as_resource_name: false,

bottlecap/src/config/mod.rs

@@ -93,6 +93,17 @@ pub struct YamlConfig {
     // Metrics
     pub metrics_config: MetricsConfig,
 
+    // DogStatsD
+    /// Size of the receive buffer for `DogStatsD` UDP packets, in bytes (`SO_RCVBUF`).
+    #[serde(deserialize_with = "deserialize_option_lossless")]
+    pub dogstatsd_so_rcvbuf: Option<usize>,
+    /// Maximum size of a single read from any transport (UDP or named pipe), in bytes.
+    #[serde(deserialize_with = "deserialize_option_lossless")]
+    pub dogstatsd_buffer_size: Option<usize>,
+    /// Internal queue capacity between the socket reader and metric processor.
+    #[serde(deserialize_with = "deserialize_option_lossless")]
+    pub dogstatsd_queue_size: Option<usize>,
+
     // OTLP
     pub otlp_config: Option<OtlpConfig>,
 
@@ -477,6 +488,11 @@ fn merge_config(config: &mut Config, yaml_config: &YamlConfig) {
         compression_level
     );
 
+    // DogStatsD
+    merge_option!(config, yaml_config, dogstatsd_so_rcvbuf);
+    merge_option!(config, yaml_config, dogstatsd_buffer_size);
+    merge_option!(config, yaml_config, dogstatsd_queue_size);
+
     // APM
     merge_hashmap!(config, yaml_config, service_mapping);
     merge_string!(config, apm_dd_url, yaml_config.apm_config, apm_dd_url);
@@ -814,6 +830,10 @@ trace_aws_service_representation_enabled: true
 metrics_config:
   compression_level: 3
 
+dogstatsd_so_rcvbuf: 1048576
+dogstatsd_buffer_size: 65507
+dogstatsd_queue_size: 2048
+
 # OTLP
 otlp_config:
   receiver:
@@ -1009,6 +1029,9 @@ api_security_sample_delay: 60 # Seconds
                 apm_filter_tags_regex_require: None,
                 apm_filter_tags_regex_reject: None,
                 statsd_metric_namespace: None,
+                dogstatsd_so_rcvbuf: Some(1048576),
+                dogstatsd_buffer_size: Some(65507),
+                dogstatsd_queue_size: Some(2048),
             };
 
             // Assert that
@@ -1017,4 +1040,51 @@ api_security_sample_delay: 60 # Seconds
             Ok(())
         });
     }
+
+    #[test]
+    fn test_yaml_dogstatsd_config() {
+        figment::Jail::expect_with(|jail| {
+            jail.clear_env();
+            jail.create_file(
+                "datadog.yaml",
+                r#"
+dogstatsd_so_rcvbuf: 524288
+dogstatsd_buffer_size: 16384
+dogstatsd_queue_size: 512
+"#,
+            )?;
+            let mut config = Config::default();
+            let yaml_config_source = YamlConfigSource {
+                path: Path::new("datadog.yaml").to_path_buf(),
+            };
+            yaml_config_source
+                .load(&mut config)
+                .expect("Failed to load config");
+
+            assert_eq!(config.dogstatsd_so_rcvbuf, Some(524288));
+            assert_eq!(config.dogstatsd_buffer_size, Some(16384));
+            assert_eq!(config.dogstatsd_queue_size, Some(512));
+            Ok(())
+        });
+    }
+
+    #[test]
+    fn test_yaml_dogstatsd_config_defaults_to_none() {
+        figment::Jail::expect_with(|jail| {
+            jail.clear_env();
+            jail.create_file("datadog.yaml", "")?;
+            let mut config = Config::default();
+            let yaml_config_source = YamlConfigSource {
+                path: Path::new("datadog.yaml").to_path_buf(),
+            };
+            yaml_config_source
+                .load(&mut config)
+                .expect("Failed to load config");
+
+            assert_eq!(config.dogstatsd_so_rcvbuf, None);
+            assert_eq!(config.dogstatsd_buffer_size, None);
+            assert_eq!(config.dogstatsd_queue_size, None);
+            Ok(())
+        });
+    }
 }