feat: eat: Add hierarchical configurable compression levels

bottlecap/Cargo.toml

@@ -62,7 +62,7 @@ datadog-trace-protobuf = { git = "https://github.com/DataDog/libdatadog", rev =
 datadog-trace-utils = { git = "https://github.com/DataDog/libdatadog", rev = "9405db9cb4ef733f3954c3ee77ce71a502e98e50" , features = ["mini_agent"] }
 datadog-trace-normalization = { git = "https://github.com/DataDog/libdatadog",  rev = "9405db9cb4ef733f3954c3ee77ce71a502e98e50" }
 datadog-trace-obfuscation = { git = "https://github.com/DataDog/libdatadog", rev = "9405db9cb4ef733f3954c3ee77ce71a502e98e50"  }
-dogstatsd = { git = "https://github.com/DataDog/serverless-components", rev = "abfec752b0638a9e4096e1465acd4bb2651edfa7", default-features = false }
+dogstatsd = { git = "https://github.com/DataDog/serverless-components", rev = "b0b0cb9310d8d8f2038c00a46e3267e21dc3e287", default-features = false }
 datadog-fips = { git = "https://github.com/DataDog/serverless-components", rev = "fa1d2f4ea2c4c2596144a1f362935e56cf0cb3c7", default-features = false }
 libddwaf = { version = "1.26.0", git = "https://github.com/DataDog/libddwaf-rust", rev = "1d57bf0ca49782723e556ba327ee7f378978aaa7", default-features = false, features = ["serde", "dynamic"] }
 

bottlecap/src/bin/bottlecap/main.rs

@@ -1048,6 +1048,7 @@ fn start_metrics_flushers(
         https_proxy: config.proxy_https.clone(),
         timeout: Duration::from_secs(config.flush_timeout),
         retry_strategy: DsdRetryStrategy::Immediate(3),
+        compression_level: config.metrics_config_compression_level,
     };
     flushers.push(MetricsFlusher::new(flusher_config));
 
@@ -1076,6 +1077,7 @@ fn start_metrics_flushers(
                 https_proxy: config.proxy_https.clone(),
                 timeout: Duration::from_secs(config.flush_timeout),
                 retry_strategy: DsdRetryStrategy::Immediate(3),
+                compression_level: config.metrics_config_compression_level,
             };
             flushers.push(MetricsFlusher::new(additional_flusher_config));
         }

bottlecap/src/config/env.rs

@@ -107,6 +107,12 @@ pub struct EnvConfig {
     /// @env `DD_TAGS`
     #[serde(deserialize_with = "deserialize_key_value_pairs")]
     pub tags: HashMap<String, String>,
+    /// @env `DD_COMPRESSION_LEVEL`
+    ///
+    /// Global level `compression_level` parameter accepts values from 0 (no compression)
+    /// to 9 (maximum compression but higher resource usage). This value is effective only if
+    /// the individual component doesn't specify its own.
+    pub compression_level: Option<i32>,
 
     // Logs
     /// @env `DD_LOGS_CONFIG_LOGS_DD_URL`
@@ -229,6 +235,12 @@ pub struct EnvConfig {
     #[serde(deserialize_with = "deserialize_optional_bool_from_anything")]
     pub trace_propagation_http_baggage_enabled: Option<bool>,
 
+    /// @env `DD_METRICS_CONFIG_COMPRESSION_LEVEL`
+    /// The metrics compresses traces before sending them. The `compression_level` parameter
+    /// accepts values from 0 (no compression) to 9 (maximum compression but
+    /// higher resource usage).
+    pub metrics_config_compression_level: Option<i32>,
+
     // OTLP
     //
     // - APM / Traces
@@ -393,10 +405,18 @@ fn merge_config(config: &mut Config, env_config: &EnvConfig) {
     merge_string!(config, env_config, url);
     merge_hashmap!(config, env_config, additional_endpoints);
 
+    merge_option_to_value!(config, env_config, compression_level);
+
     // Logs
     merge_string!(config, env_config, logs_config_logs_dd_url);
     merge_option!(config, env_config, logs_config_processing_rules);
     merge_option_to_value!(config, env_config, logs_config_use_compression);
+    merge_option_to_value!(
+        config,
+        logs_config_compression_level,
+        env_config,
+        compression_level
+    );
     merge_option_to_value!(config, env_config, logs_config_compression_level);
     merge_vec!(config, env_config, logs_config_additional_endpoints);
 
@@ -414,6 +434,12 @@ fn merge_config(config: &mut Config, env_config: &EnvConfig) {
         env_config,
         apm_config_obfuscation_http_remove_paths_with_digits
     );
+    merge_option_to_value!(
+        config,
+        apm_config_compression_level,
+        env_config,
+        compression_level
+    );
     merge_option_to_value!(config, env_config, apm_config_compression_level);
     merge_vec!(config, env_config, apm_features);
     merge_hashmap!(config, env_config, apm_additional_endpoints);
@@ -429,6 +455,15 @@ fn merge_config(config: &mut Config, env_config: &EnvConfig) {
     merge_option_to_value!(config, env_config, trace_propagation_extract_first);
     merge_option_to_value!(config, env_config, trace_propagation_http_baggage_enabled);
 
+    // Metrics
+    merge_option_to_value!(
+        config,
+        metrics_config_compression_level,
+        env_config,
+        compression_level
+    );
+    merge_option_to_value!(config, env_config, metrics_config_compression_level);
+
     // OTLP
     merge_option_to_value!(config, env_config, otlp_config_traces_enabled);
     merge_option_to_value!(
@@ -588,6 +623,7 @@ mod tests {
             jail.set_env("DD_SERVICE", "test-service");
             jail.set_env("DD_VERSION", "1.0.0");
             jail.set_env("DD_TAGS", "team:test-team,project:test-project");
+            jail.set_env("DD_COMPRESSION_LEVEL", "4");
 
             // Logs
             jail.set_env("DD_LOGS_CONFIG_LOGS_DD_URL", "https://logs.datadoghq.com");
@@ -596,7 +632,7 @@ mod tests {
                 r#"[{"type":"exclude_at_match","name":"exclude","pattern":"exclude"}]"#,
             );
             jail.set_env("DD_LOGS_CONFIG_USE_COMPRESSION", "false");
-            jail.set_env("DD_LOGS_CONFIG_COMPRESSION_LEVEL", "3");
+            jail.set_env("DD_LOGS_CONFIG_COMPRESSION_LEVEL", "1");
             jail.set_env(
                 "DD_LOGS_CONFIG_ADDITIONAL_ENDPOINTS",
                 "[{\"api_key\": \"apikey2\", \"Host\": \"agent-http-intake.logs.datadoghq.com\", \"Port\": 443, \"is_reliable\": true}]",
@@ -615,7 +651,7 @@ mod tests {
                 "DD_APM_CONFIG_OBFUSCATION_HTTP_REMOVE_PATHS_WITH_DIGITS",
                 "true",
             );
-            jail.set_env("DD_APM_CONFIG_COMPRESSION_LEVEL", "3");
+            jail.set_env("DD_APM_CONFIG_COMPRESSION_LEVEL", "2");
             jail.set_env(
                 "DD_APM_FEATURES",
                 "enable_otlp_compute_top_level_by_span_kind,enable_stats_by_span_kind",
@@ -632,6 +668,7 @@ mod tests {
                 "env:^test.*$ debug:^true$",
             );
 
+            jail.set_env("DD_METRICS_CONFIG_COMPRESSION_LEVEL", "3");
             // Trace Propagation
             jail.set_env("DD_TRACE_PROPAGATION_STYLE", "datadog");
             jail.set_env("DD_TRACE_PROPAGATION_STYLE_EXTRACT", "b3");
@@ -721,6 +758,7 @@ mod tests {
                 site: "test-site".to_string(),
                 api_key: "test-api-key".to_string(),
                 log_level: LogLevel::Debug,
+                compression_level: 4,
                 flush_timeout: 42,
                 proxy_https: Some("https://proxy.example.com".to_string()),
                 proxy_no_proxy: vec!["localhost".to_string(), "127.0.0.1".to_string()],
@@ -752,7 +790,7 @@ mod tests {
                     replace_placeholder: None,
                 }]),
                 logs_config_use_compression: false,
-                logs_config_compression_level: 3,
+                logs_config_compression_level: 1,
                 logs_config_additional_endpoints: vec![LogsAdditionalEndpoint {
                     api_key: "apikey2".to_string(),
                     host: "agent-http-intake.logs.datadoghq.com".to_string(),
@@ -772,7 +810,7 @@ mod tests {
                 ),
                 apm_config_obfuscation_http_remove_query_string: true,
                 apm_config_obfuscation_http_remove_paths_with_digits: true,
-                apm_config_compression_level: 3,
+                apm_config_compression_level: 2,
                 apm_features: vec![
                     "enable_otlp_compute_top_level_by_span_kind".to_string(),
                     "enable_stats_by_span_kind".to_string(),
@@ -808,6 +846,7 @@ mod tests {
                 trace_propagation_extract_first: true,
                 trace_propagation_http_baggage_enabled: true,
                 trace_aws_service_representation_enabled: true,
+                metrics_config_compression_level: 3,
                 otlp_config_traces_enabled: false,
                 otlp_config_traces_span_name_as_resource_name: true,
                 otlp_config_traces_span_name_remappings: HashMap::from([(

bottlecap/src/config/mod.rs

@@ -245,6 +245,10 @@ pub struct Config {
     // Timeout for the request to flush data to Datadog endpoint
     pub flush_timeout: u64,
 
+    // Global config of compression levels.
+    // It would be overridden by the setup for the individual component
+    pub compression_level: i32,
+
     // Proxy
     pub proxy_https: Option<String>,
     pub proxy_no_proxy: Vec<String>,
@@ -291,6 +295,9 @@ pub struct Config {
     pub trace_propagation_http_baggage_enabled: bool,
     pub trace_aws_service_representation_enabled: bool,
 
+    // Metrics
+    pub metrics_config_compression_level: i32,
+
     // OTLP
     //
     // - APM / Traces
@@ -368,6 +375,8 @@ impl Default for Config {
             version: None,
             tags: HashMap::new(),
 
+            compression_level: 6,
+
             // Logs
             logs_config_logs_dd_url: String::default(),
             logs_config_processing_rules: None,
@@ -397,6 +406,9 @@ impl Default for Config {
             trace_propagation_extract_first: false,
             trace_propagation_http_baggage_enabled: false,
 
+            // Metrics
+            metrics_config_compression_level: 6,
+
             // OTLP
             otlp_config_traces_enabled: true,
             otlp_config_traces_span_name_as_resource_name: false,

bottlecap/src/config/yaml.rs

@@ -38,6 +38,8 @@ pub struct YamlConfig {
 
     pub flush_timeout: Option<u64>,
 
+    pub compression_level: Option<i32>,
+
     // Proxy
     pub proxy: ProxyConfig,
     // nit: this should probably be in the endpoints section
@@ -78,6 +80,9 @@ pub struct YamlConfig {
     #[serde(deserialize_with = "deserialize_optional_bool_from_anything")]
     pub trace_propagation_http_baggage_enabled: Option<bool>,
 
+    // Metrics
+    pub metrics_config: MetricsConfig,
+
     // OTLP
     pub otlp_config: Option<OtlpConfig>,
 
@@ -131,6 +136,15 @@ pub struct LogsConfig {
     pub additional_endpoints: Vec<LogsAdditionalEndpoint>,
 }
 
+/// Metrics specific config
+///
+#[derive(Debug, PartialEq, Deserialize, Clone, Copy, Default)]
+#[serde(default)]
+#[allow(clippy::module_name_repetitions)]
+pub struct MetricsConfig {
+    pub compression_level: Option<i32>,
+}
+
 /// APM Config
 ///
 
@@ -373,6 +387,7 @@ fn merge_config(config: &mut Config, yaml_config: &YamlConfig) {
     merge_option!(config, yaml_config, version);
     merge_hashmap!(config, yaml_config, tags);
 
+    merge_option_to_value!(config, yaml_config, compression_level);
     // Proxy
     merge_option!(config, proxy_https, yaml_config.proxy, https);
     merge_option_to_value!(config, proxy_no_proxy, yaml_config.proxy, no_proxy);
@@ -401,6 +416,12 @@ fn merge_config(config: &mut Config, yaml_config: &YamlConfig) {
         yaml_config.logs_config,
         use_compression
     );
+    merge_option_to_value!(
+        config,
+        logs_config_compression_level,
+        yaml_config,
+        compression_level
+    );
     merge_option_to_value!(
         config,
         logs_config_compression_level,
@@ -414,6 +435,20 @@ fn merge_config(config: &mut Config, yaml_config: &YamlConfig) {
         additional_endpoints
     );
 
+    merge_option_to_value!(
+        config,
+        metrics_config_compression_level,
+        yaml_config,
+        compression_level
+    );
+
+    merge_option_to_value!(
+        config,
+        metrics_config_compression_level,
+        yaml_config.metrics_config,
+        compression_level
+    );
+
     // APM
     merge_hashmap!(config, yaml_config, service_mapping);
     merge_string!(config, apm_dd_url, yaml_config.apm_config, apm_dd_url);
@@ -423,6 +458,12 @@ fn merge_config(config: &mut Config, yaml_config: &YamlConfig) {
         yaml_config.apm_config,
         replace_tags
     );
+    merge_option_to_value!(
+        config,
+        apm_config_compression_level,
+        yaml_config,
+        compression_level
+    );
     merge_option_to_value!(
         config,
         apm_config_compression_level,
@@ -667,7 +708,7 @@ site: "test-site"
 api_key: "test-api-key"
 log_level: "debug"
 flush_timeout: 42
-
+compression_level: 4
 # Proxy
 proxy:
   https: "https://proxy.example.com"
@@ -699,7 +740,7 @@ logs_config:
       type: "exclude_at_match"
       pattern: "test-pattern"
   use_compression: false
-  compression_level: 3
+  compression_level: 1
   additional_endpoints:
     - api_key: "apikey2"
       Host: "agent-http-intake.logs.datadoghq.com"
@@ -714,7 +755,7 @@ apm_config:
     http:
       remove_query_string: true
       remove_paths_with_digits: true
-  compression_level: 3
+  compression_level: 2
   features:
     - "enable_otlp_compute_top_level_by_span_kind"
     - "enable_stats_by_span_kind"
@@ -734,6 +775,9 @@ trace_propagation_extract_first: true
 trace_propagation_http_baggage_enabled: true
 trace_aws_service_representation_enabled: true
 
+metrics_config:
+  compression_level: 3
+
 # OTLP
 otlp_config:
   receiver:
@@ -801,6 +845,7 @@ extension_version: "compatibility"
                 api_key: "test-api-key".to_string(),
                 log_level: LogLevel::Debug,
                 flush_timeout: 42,
+                compression_level: 4,
                 proxy_https: Some("https://proxy.example.com".to_string()),
                 proxy_no_proxy: vec!["localhost".to_string(), "127.0.0.1".to_string()],
                 http_protocol: Some("http1".to_string()),
@@ -831,7 +876,7 @@ extension_version: "compatibility"
                     replace_placeholder: None,
                 }]),
                 logs_config_use_compression: false,
-                logs_config_compression_level: 3,
+                logs_config_compression_level: 1,
                 logs_config_additional_endpoints: vec![LogsAdditionalEndpoint {
                     api_key: "apikey2".to_string(),
                     host: "agent-http-intake.logs.datadoghq.com".to_string(),
@@ -846,7 +891,7 @@ extension_version: "compatibility"
                 apm_replace_tags: Some(vec![]),
                 apm_config_obfuscation_http_remove_query_string: true,
                 apm_config_obfuscation_http_remove_paths_with_digits: true,
-                apm_config_compression_level: 3,
+                apm_config_compression_level: 2,
                 apm_features: vec![
                     "enable_otlp_compute_top_level_by_span_kind".to_string(),
                     "enable_stats_by_span_kind".to_string(),
@@ -866,6 +911,7 @@ extension_version: "compatibility"
                 trace_propagation_extract_first: true,
                 trace_propagation_http_baggage_enabled: true,
                 trace_aws_service_representation_enabled: true,
+                metrics_config_compression_level: 3,
                 otlp_config_traces_enabled: false,
                 otlp_config_traces_span_name_as_resource_name: true,
                 otlp_config_traces_span_name_remappings: HashMap::from([(

bottlecap/src/traces/trace_processor.rs

@@ -361,7 +361,7 @@ impl TraceProcessor for ServerlessTraceProcessor {
         };
 
         let builder = SendDataBuilder::new(body_size, payload, header_tags, &endpoint)
-            .with_compression(Compression::Zstd(6))
+            .with_compression(Compression::Zstd(config.apm_config_compression_level))
             .with_retry_strategy(RetryStrategy::new(
                 1,
                 100,