enhancement(datadog_logs sink): default to zstd compression

Cargo.toml

@@ -307,7 +307,7 @@ opendal = { version = "0.54", default-features = false, features = ["services-we
 
 # Tower
 tower = { version = "0.5.2", default-features = false, features = ["buffer", "limit", "retry", "timeout", "util", "balance", "discover"] }
-tower-http = { version = "0.4.4", default-features = false, features = ["compression-full", "decompression-gzip", "trace"] }
+tower-http = { version = "0.4.4", default-features = false, features = ["compression-full", "decompression-full", "trace"] }
 # Serde
 serde.workspace = true
 serde-toml-merge = { version = "0.3.11", default-features = false }

changelog.d/datadog-logs-zstd.enhancement.md

@@ -0,0 +1,6 @@
+The `datadog_logs` sink now defaults to `zstd` compression instead of no compression. This results in
+better network efficiency and higher throughput. You can explicitly set `compression = "none"` to
+restore the previous behavior of no compression, or set `compression = "gzip"` if you were previously
+using gzip compression explicitly.
+
+authors: jszwedko pront

regression/cases/datadog_agent_remap_datadog_logs/vector/vector.yaml

@@ -42,7 +42,6 @@ sinks:
     inputs: [ "parse_message" ]
     endpoint:        "http://localhost:8080"
     default_api_key: "DEADBEEF"
-    compression: "gzip"
     healthcheck:
       enabled: false
     buffer:

regression/cases/datadog_agent_remap_datadog_logs_acks/vector/vector.yaml

@@ -42,7 +42,6 @@ sinks:
     inputs: [ "parse_message" ]
     endpoint:        "http://localhost:8080"
     default_api_key: "DEADBEEF"
-    compression: "gzip"
     healthcheck:
       enabled: false
     buffer:

src/components/validation/resources/http.rs

@@ -329,10 +329,29 @@ impl HttpResourceOutputContext<'_> {
                 let mut decoder = decoder.clone();
 
                 async move {
+                    // Extract the Content-Encoding header before consuming the request
+                    let content_encoding = request
+                        .headers()
+                        .get("content-encoding")
+                        .and_then(|v| v.to_str().ok())
+                        .map(|s| s.to_string());
+
                     match request.into_body().collect().await.map(Collected::to_bytes) {
                         Err(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
                         Ok(body) => {
                             let byte_size = body.len();
+
+                            // Validation tests should not use compression - error if we receive compressed data
+                            if let Some(encoding) = &content_encoding
+                                && encoding != "identity"
+                            {
+                                error!(
+                                    "Received compressed data (Content-Encoding: {encoding}). \
+                                        Validation tests assert on bytes sizes and compressed size might not be deterministic."
+                                );
+                                return StatusCode::BAD_REQUEST.into_response();
+                            }
+
                             let mut body = BytesMut::from(&body[..]);
                             loop {
                                 match decoder.decode_eof(&mut body) {
@@ -381,7 +400,7 @@ impl HttpResourceOutputContext<'_> {
                                     }
                                     Err(_) => {
                                         error!(
-                                            "HTTP server failed to decode {:?}",
+                                            "HTTP server failed to decode body: {:?}",
                                             String::from_utf8_lossy(&body)
                                         );
                                         return StatusCode::INTERNAL_SERVER_ERROR.into_response();

src/sinks/datadog/logs/config.rs

@@ -43,14 +43,16 @@ impl SinkBatchSettings for DatadogLogsDefaultBatchSettings {
 
 /// Configuration for the `datadog_logs` sink.
 #[configurable_component(sink("datadog_logs", "Publish log events to Datadog."))]
-#[derive(Clone, Debug, Default)]
+#[derive(Clone, Debug, Derivative)]
+#[derivative(Default)]
 #[serde(deny_unknown_fields)]
 pub struct DatadogLogsConfig {
     #[serde(flatten)]
     pub local_dd_common: LocalDatadogCommonConfig,
 
     #[configurable(derived)]
-    #[serde(default)]
+    #[derivative(Default(value = "default_compression()"))]
+    #[serde(default = "default_compression")]
     pub compression: Option<Compression>,
 
     #[configurable(derived)]
@@ -73,6 +75,10 @@ pub struct DatadogLogsConfig {
     pub conforms_as_agent: bool,
 }
 
+const fn default_compression() -> Option<Compression> {
+    Some(Compression::zstd_default())
+}
+
 impl GenerateConfig for DatadogLogsConfig {
     fn generate_config() -> toml::Value {
         toml::from_str(indoc! {r#"
@@ -155,7 +161,7 @@ impl DatadogLogsConfig {
             protocol,
             conforms_as_agent,
         )
-        .compression(self.compression.unwrap_or_default())
+        .compression(self.compression.or_else(default_compression).unwrap())
         .build();
 
         Ok(VectorSink::from_event_streamsink(sink))
@@ -226,6 +232,56 @@ mod test {
         crate::test_util::test_generate_config::<DatadogLogsConfig>();
     }
 
+    #[test]
+    fn compression_config_field() {
+        // Verify the default compression function returns zstd
+        assert_eq!(default_compression(), Some(Compression::zstd_default()));
+
+        // Test 1: Config deserialized without compression field gets zstd default
+        // (due to #[serde(default = "default_compression")])
+        let config_yaml = indoc! {r#"
+            default_api_key: "test_key"
+        "#};
+
+        let config: DatadogLogsConfig = serde_yaml::from_str(config_yaml).unwrap();
+        // The serde default applies immediately during deserialization
+        assert!(matches!(config.compression, Some(Compression::Zstd(_))));
+
+        // Test 2: When explicitly set to "none", it should be Some(Compression::None)
+        let config_yaml_with_none = indoc! {r#"
+            default_api_key: "test_key"
+            compression: "none"
+        "#};
+
+        let config_no_compression: DatadogLogsConfig =
+            serde_yaml::from_str(config_yaml_with_none).unwrap();
+        assert_eq!(config_no_compression.compression, Some(Compression::None));
+
+        // Test 3: When explicitly set to "zstd", it should be Some(Compression::Zstd)
+        let config_yaml_with_zstd = indoc! {r#"
+            default_api_key: "test_key"
+            compression: "zstd"
+        "#};
+
+        let config_zstd: DatadogLogsConfig = serde_yaml::from_str(config_yaml_with_zstd).unwrap();
+        assert!(matches!(
+            config_zstd.compression,
+            Some(Compression::Zstd(_))
+        ));
+
+        // Test 4: When explicitly set to "gzip", it should be Some(Compression::Gzip)
+        let config_yaml_with_gzip = indoc! {r#"
+            default_api_key: "test_key"
+            compression: "gzip"
+        "#};
+
+        let config_gzip: DatadogLogsConfig = serde_yaml::from_str(config_yaml_with_gzip).unwrap();
+        assert!(matches!(
+            config_gzip.compression,
+            Some(Compression::Gzip(_))
+        ));
+    }
+
     impl ValidatableComponent for DatadogLogsConfig {
         fn validation_configuration() -> ValidationConfiguration {
             let endpoint = "http://127.0.0.1:9005".to_string();
@@ -235,6 +291,8 @@ mod test {
                     default_api_key: Some("unused".to_string().into()),
                     ..Default::default()
                 },
+                // Disable compression for validation tests to ensure byte counting is accurate
+                compression: Some(Compression::None),
                 ..Default::default()
             };
 

tests/e2e/datadog-logs/data/vector.toml

@@ -14,4 +14,3 @@ type = "datadog_logs"
 default_api_key = "unused"
 endpoint = "http://fakeintake-vector:80"
 batch.timeout_secs = 1
-compression = "gzip"

website/cue/reference/components/sinks/generated/datadog_logs.cue

@@ -69,28 +69,31 @@ generated: components: sinks: datadog_logs: configuration: {
 			All compression algorithms use the default compression level unless otherwise specified.
 			"""
 		required: false
-		type: string: enum: {
-			gzip: """
-				[Gzip][gzip] compression.
+		type: string: {
+			default: "zstd"
+			enum: {
+				gzip: """
+					[Gzip][gzip] compression.
 
-				[gzip]: https://www.gzip.org/
-				"""
-			none: "No compression."
-			snappy: """
-				[Snappy][snappy] compression.
+					[gzip]: https://www.gzip.org/
+					"""
+				none: "No compression."
+				snappy: """
+					[Snappy][snappy] compression.
 
-				[snappy]: https://github.com/google/snappy/blob/main/docs/README.md
-				"""
-			zlib: """
-				[Zlib][zlib] compression.
+					[snappy]: https://github.com/google/snappy/blob/main/docs/README.md
+					"""
+				zlib: """
+					[Zlib][zlib] compression.
 
-				[zlib]: https://zlib.net/
-				"""
-			zstd: """
-				[Zstandard][zstd] compression.
+					[zlib]: https://zlib.net/
+					"""
+				zstd: """
+					[Zstandard][zstd] compression.
 
-				[zstd]: https://facebook.github.io/zstd/
-				"""
+					[zstd]: https://facebook.github.io/zstd/
+					"""
+			}
 		}
 	}
 	conforms_as_agent: {