chore(config): clamp zstd compressor level to ADP minimum

The core Agent streams a fully-resolved config, so its schema default of 1
for serializer_zstd_compressor_level arrives as a concrete value that would
otherwise overwrite ADP's intended level-3 floor. Add a ZSTD_MINIMUM const in
the config layer and clamp the translated value up to it, preserving ADP's
harder-compression behavior instead of silently regressing to the Agent
default.

Author: webern

lib/agent-data-plane-config-system/src/translators/datadog_translator.rs

@@ -22,7 +22,7 @@ use agent_data_plane_config::control::ListenAddress;
 use agent_data_plane_config::domains::dogstatsd::{
     FilterAction, MapperProfile, MetricMapping, MetricTagFilterEntry, OriginTagCardinality,
 };
-use agent_data_plane_config::shared::ForwarderHttpProtocol;
+use agent_data_plane_config::shared::{ForwarderHttpProtocol, ZSTD_MINIMUM};
 use agent_data_plane_config::SalukiConfiguration;
 use bytesize::ByteSize;
 use datadog_agent_config::{drive, DatadogConfigWitness, DatadogConfiguration, TranslateError, TranslateErrors};
@@ -952,7 +952,13 @@ impl DatadogConfigWitness for DatadogTranslator<'_> {
     }
 
     fn consume_serializer_zstd_compressor_level(&mut self, value: i64) {
-        self.config.shared.endpoints.compression.zstd_compressor_level = value as i32;
+        // TODO: The core Agent streams a fully-resolved config, so its schema default of `1` for
+        // `serializer_zstd_compressor_level` arrives here as a concrete value. Without this clamp it
+        // would silently override ADP's intended `ZSTD_MINIMUM` floor. Clamping up cannot tell an
+        // operator who deliberately set `1` apart from the Agent's default `1`, so both are raised to
+        // the floor. Follow-up work needs to track each value's source (user-set vs Agent default) so
+        // we can honor an explicit low level while still applying our floor to the Agent default.
+        self.config.shared.endpoints.compression.zstd_compressor_level = (value as i32).max(ZSTD_MINIMUM);
     }
 
     fn consume_site(&mut self, value: String) {
@@ -1143,4 +1149,29 @@ mod tests {
         // The error is still surfaced, so startup's strict gate rejects the config.
         assert!(errors.is_some(), "an unknown action must record a translation error");
     }
+
+    #[test]
+    fn zstd_compressor_level_clamps_up_to_the_minimum() {
+        use agent_data_plane_config::shared::ZSTD_MINIMUM;
+
+        // The Agent's schema default of `1` arrives as a concrete value; it must be raised to the
+        // ADP floor rather than lowering our compression level.
+        let datadog: DatadogConfiguration = serde_json::from_value(json!({
+            "serializer_zstd_compressor_level": 1,
+        }))
+        .expect("datadog source deserializes");
+        let (config, _) = DatadogTranslator::new(&datadog, SalukiConfiguration::default()).translate();
+        assert_eq!(
+            config.shared.endpoints.compression.zstd_compressor_level, ZSTD_MINIMUM,
+            "a level below the floor is clamped up to it"
+        );
+
+        // A level at or above the floor is used as provided.
+        let datadog: DatadogConfiguration = serde_json::from_value(json!({
+            "serializer_zstd_compressor_level": 7,
+        }))
+        .expect("datadog source deserializes");
+        let (config, _) = DatadogTranslator::new(&datadog, SalukiConfiguration::default()).translate();
+        assert_eq!(config.shared.endpoints.compression.zstd_compressor_level, 7);
+    }
 }

lib/agent-data-plane-config/src/shared.rs

@@ -118,6 +118,13 @@ pub struct Tls {
     pub sslkeylogfile: String,
 }
 
+/// Lowest zstd compression level ADP will use for payloads. ADP compresses harder than the core
+/// Agent, whose schema default for `serializer_zstd_compressor_level` is `1`. Because the Agent
+/// streams a fully-resolved config, that `1` arrives as a concrete value and would otherwise
+/// overwrite ADP's intended floor, so the translator clamps anything below this minimum back up to
+/// it.
+pub const ZSTD_MINIMUM: i32 = 3;
+
 /// Payload compression settings applied before transmission.
 #[derive(Clone, Debug, Default, PartialEq, Serialize)]
 pub struct Compression {