Adding compression optional field to Amazon S3 source (#1467)

Co-authored-by: ci.datadog-api-spec <packages@datadoghq.com>

Author: api-clients-generation-pipeline[bot]

.generator/schemas/v2/openapi.yaml

@@ -43045,12 +43045,14 @@ components:
     ObservabilityPipelineAmazonS3Source:
       description: |-
         The `amazon_s3` source ingests logs from an Amazon S3 bucket.
-        It supports AWS authentication and TLS encryption.
+        It supports AWS authentication, TLS encryption, and configurable compression.
 
         **Supported pipeline types:** logs
       properties:
         auth:
           $ref: "#/components/schemas/ObservabilityPipelineAwsAuth"
+        compression:
+          $ref: "#/components/schemas/ObservabilityPipelineAmazonS3SourceCompression"
         id:
           description: The unique identifier for this component. Used in other parts of the pipeline to reference this component (for example, as the `input` to downstream components).
           example: aws-s3-source
@@ -43073,6 +43075,20 @@ components:
         - region
       type: object
       x-pipeline-types: [logs]
+    ObservabilityPipelineAmazonS3SourceCompression:
+      description: Compression format for objects retrieved from the S3 bucket. Use `auto` to detect compression from the object's Content-Encoding header or file extension.
+      enum:
+        - auto
+        - none
+        - gzip
+        - zstd
+      example: gzip
+      type: string
+      x-enum-varnames:
+        - AUTO
+        - NONE
+        - GZIP
+        - ZSTD
     ObservabilityPipelineAmazonS3SourceType:
       default: amazon_s3
       description: The source type. Always `amazon_s3`.

examples/v2_observability-pipelines_ValidatePipeline_99164570.rs

@@ -0,0 +1,85 @@
+// Validate an observability pipeline with amazon S3 source compression returns
+// "OK" response
+use datadog_api_client::datadog;
+use datadog_api_client::datadogV2::api_observability_pipelines::ObservabilityPipelinesAPI;
+use datadog_api_client::datadogV2::model::ObservabilityPipelineAmazonS3Source;
+use datadog_api_client::datadogV2::model::ObservabilityPipelineAmazonS3SourceCompression;
+use datadog_api_client::datadogV2::model::ObservabilityPipelineAmazonS3SourceType;
+use datadog_api_client::datadogV2::model::ObservabilityPipelineConfig;
+use datadog_api_client::datadogV2::model::ObservabilityPipelineConfigDestinationItem;
+use datadog_api_client::datadogV2::model::ObservabilityPipelineConfigProcessorGroup;
+use datadog_api_client::datadogV2::model::ObservabilityPipelineConfigProcessorItem;
+use datadog_api_client::datadogV2::model::ObservabilityPipelineConfigSourceItem;
+use datadog_api_client::datadogV2::model::ObservabilityPipelineDataAttributes;
+use datadog_api_client::datadogV2::model::ObservabilityPipelineDatadogLogsDestination;
+use datadog_api_client::datadogV2::model::ObservabilityPipelineDatadogLogsDestinationType;
+use datadog_api_client::datadogV2::model::ObservabilityPipelineFilterProcessor;
+use datadog_api_client::datadogV2::model::ObservabilityPipelineFilterProcessorType;
+use datadog_api_client::datadogV2::model::ObservabilityPipelineSpec;
+use datadog_api_client::datadogV2::model::ObservabilityPipelineSpecData;
+
+#[tokio::main]
+async fn main() {
+    let body =
+        ObservabilityPipelineSpec::new(
+            ObservabilityPipelineSpecData::new(
+                ObservabilityPipelineDataAttributes::new(
+                    ObservabilityPipelineConfig::new(
+                        vec![
+                            ObservabilityPipelineConfigDestinationItem::ObservabilityPipelineDatadogLogsDestination(
+                                Box::new(
+                                    ObservabilityPipelineDatadogLogsDestination::new(
+                                        "datadog-logs-destination".to_string(),
+                                        vec!["my-processor-group".to_string()],
+                                        ObservabilityPipelineDatadogLogsDestinationType::DATADOG_LOGS,
+                                    ),
+                                ),
+                            )
+                        ],
+                        vec![
+                            ObservabilityPipelineConfigSourceItem::ObservabilityPipelineAmazonS3Source(
+                                Box::new(
+                                    ObservabilityPipelineAmazonS3Source::new(
+                                        "amazon-s3-source".to_string(),
+                                        "us-east-1".to_string(),
+                                        ObservabilityPipelineAmazonS3SourceType::AMAZON_S3,
+                                    ).compression(ObservabilityPipelineAmazonS3SourceCompression::GZIP),
+                                ),
+                            )
+                        ],
+                    ).processor_groups(
+                        vec![
+                            ObservabilityPipelineConfigProcessorGroup::new(
+                                true,
+                                "my-processor-group".to_string(),
+                                "service:my-service".to_string(),
+                                vec!["amazon-s3-source".to_string()],
+                                vec![
+                                    ObservabilityPipelineConfigProcessorItem::ObservabilityPipelineFilterProcessor(
+                                        Box::new(
+                                            ObservabilityPipelineFilterProcessor::new(
+                                                true,
+                                                "filter-processor".to_string(),
+                                                "service:my-service".to_string(),
+                                                ObservabilityPipelineFilterProcessorType::FILTER,
+                                            ),
+                                        ),
+                                    )
+                                ],
+                            )
+                        ],
+                    ),
+                    "Pipeline with S3 Source Compression".to_string(),
+                ),
+                "pipelines".to_string(),
+            ),
+        );
+    let configuration = datadog::Configuration::new();
+    let api = ObservabilityPipelinesAPI::with_config(configuration);
+    let resp = api.validate_pipeline(body).await;
+    if let Ok(value) = resp {
+        println!("{:#?}", value);
+    } else {
+        println!("{:#?}", resp.unwrap_err());
+    }
+}

src/datadogV2/model/mod.rs

@@ -5452,6 +5452,8 @@ pub mod model_observability_pipeline_amazon_data_firehose_source_type;
 pub use self::model_observability_pipeline_amazon_data_firehose_source_type::ObservabilityPipelineAmazonDataFirehoseSourceType;
 pub mod model_observability_pipeline_amazon_s3_source;
 pub use self::model_observability_pipeline_amazon_s3_source::ObservabilityPipelineAmazonS3Source;
+pub mod model_observability_pipeline_amazon_s3_source_compression;
+pub use self::model_observability_pipeline_amazon_s3_source_compression::ObservabilityPipelineAmazonS3SourceCompression;
 pub mod model_observability_pipeline_amazon_s3_source_type;
 pub use self::model_observability_pipeline_amazon_s3_source_type::ObservabilityPipelineAmazonS3SourceType;
 pub mod model_observability_pipeline_fluent_bit_source;

src/datadogV2/model/model_observability_pipeline_amazon_s3_source.rs

@@ -7,7 +7,7 @@ use serde_with::skip_serializing_none;
 use std::fmt::{self, Formatter};
 
 /// The `amazon_s3` source ingests logs from an Amazon S3 bucket.
-/// It supports AWS authentication and TLS encryption.
+/// It supports AWS authentication, TLS encryption, and configurable compression.
 ///
 /// **Supported pipeline types:** logs
 #[non_exhaustive]
@@ -18,6 +18,10 @@ pub struct ObservabilityPipelineAmazonS3Source {
     /// If omitted, the system’s default credentials are used (for example, the IAM role and environment variables).
     #[serde(rename = "auth")]
     pub auth: Option<crate::datadogV2::model::ObservabilityPipelineAwsAuth>,
+    /// Compression format for objects retrieved from the S3 bucket. Use `auto` to detect compression from the object's Content-Encoding header or file extension.
+    #[serde(rename = "compression")]
+    pub compression:
+        Option<crate::datadogV2::model::ObservabilityPipelineAmazonS3SourceCompression>,
     /// The unique identifier for this component. Used in other parts of the pipeline to reference this component (for example, as the `input` to downstream components).
     #[serde(rename = "id")]
     pub id: String,
@@ -48,6 +52,7 @@ impl ObservabilityPipelineAmazonS3Source {
     ) -> ObservabilityPipelineAmazonS3Source {
         ObservabilityPipelineAmazonS3Source {
             auth: None,
+            compression: None,
             id,
             region,
             tls: None,
@@ -63,6 +68,14 @@ impl ObservabilityPipelineAmazonS3Source {
         self
     }
 
+    pub fn compression(
+        mut self,
+        value: crate::datadogV2::model::ObservabilityPipelineAmazonS3SourceCompression,
+    ) -> Self {
+        self.compression = Some(value);
+        self
+    }
+
     pub fn tls(mut self, value: crate::datadogV2::model::ObservabilityPipelineTls) -> Self {
         self.tls = Some(value);
         self
@@ -100,6 +113,9 @@ impl<'de> Deserialize<'de> for ObservabilityPipelineAmazonS3Source {
                 M: MapAccess<'a>,
             {
                 let mut auth: Option<crate::datadogV2::model::ObservabilityPipelineAwsAuth> = None;
+                let mut compression: Option<
+                    crate::datadogV2::model::ObservabilityPipelineAmazonS3SourceCompression,
+                > = None;
                 let mut id: Option<String> = None;
                 let mut region: Option<String> = None;
                 let mut tls: Option<crate::datadogV2::model::ObservabilityPipelineTls> = None;
@@ -121,6 +137,21 @@ impl<'de> Deserialize<'de> for ObservabilityPipelineAmazonS3Source {
                             }
                             auth = Some(serde_json::from_value(v).map_err(M::Error::custom)?);
                         }
+                        "compression" => {
+                            if v.is_null() {
+                                continue;
+                            }
+                            compression =
+                                Some(serde_json::from_value(v).map_err(M::Error::custom)?);
+                            if let Some(ref _compression) = compression {
+                                match _compression {
+                                    crate::datadogV2::model::ObservabilityPipelineAmazonS3SourceCompression::UnparsedObject(_compression) => {
+                                        _unparsed = true;
+                                    },
+                                    _ => {}
+                                }
+                            }
+                        }
                         "id" => {
                             id = Some(serde_json::from_value(v).map_err(M::Error::custom)?);
                         }
@@ -163,6 +194,7 @@ impl<'de> Deserialize<'de> for ObservabilityPipelineAmazonS3Source {
 
                 let content = ObservabilityPipelineAmazonS3Source {
                     auth,
+                    compression,
                     id,
                     region,
                     tls,

src/datadogV2/model/model_observability_pipeline_amazon_s3_source_compression.rs

@@ -0,0 +1,57 @@
+// Unless explicitly stated otherwise all files in this repository are licensed under the Apache-2.0 License.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2019-Present Datadog, Inc.
+
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
+
+#[non_exhaustive]
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum ObservabilityPipelineAmazonS3SourceCompression {
+    AUTO,
+    NONE,
+    GZIP,
+    ZSTD,
+    UnparsedObject(crate::datadog::UnparsedObject),
+}
+
+impl ToString for ObservabilityPipelineAmazonS3SourceCompression {
+    fn to_string(&self) -> String {
+        match self {
+            Self::AUTO => String::from("auto"),
+            Self::NONE => String::from("none"),
+            Self::GZIP => String::from("gzip"),
+            Self::ZSTD => String::from("zstd"),
+            Self::UnparsedObject(v) => v.value.to_string(),
+        }
+    }
+}
+
+impl Serialize for ObservabilityPipelineAmazonS3SourceCompression {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        match self {
+            Self::UnparsedObject(v) => v.serialize(serializer),
+            _ => serializer.serialize_str(self.to_string().as_str()),
+        }
+    }
+}
+
+impl<'de> Deserialize<'de> for ObservabilityPipelineAmazonS3SourceCompression {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let s: String = String::deserialize(deserializer)?;
+        Ok(match s.as_str() {
+            "auto" => Self::AUTO,
+            "none" => Self::NONE,
+            "gzip" => Self::GZIP,
+            "zstd" => Self::ZSTD,
+            _ => Self::UnparsedObject(crate::datadog::UnparsedObject {
+                value: serde_json::Value::String(s.into()),
+            }),
+        })
+    }
+}

tests/scenarios/cassettes/v2/observability_pipelines/Validate-an-observability-pipeline-with-amazon-S3-source-compression-returns-OK-response.frozen

@@ -0,0 +1 @@
+2026-04-08T12:44:25.060Z

tests/scenarios/cassettes/v2/observability_pipelines/Validate-an-observability-pipeline-with-amazon-S3-source-compression-returns-OK-response.json

@@ -0,0 +1,39 @@
+{
+  "http_interactions": [
+    {
+      "request": {
+        "body": {
+          "string": "{\"data\":{\"attributes\":{\"config\":{\"destinations\":[{\"id\":\"datadog-logs-destination\",\"inputs\":[\"my-processor-group\"],\"type\":\"datadog_logs\"}],\"processor_groups\":[{\"enabled\":true,\"id\":\"my-processor-group\",\"include\":\"service:my-service\",\"inputs\":[\"amazon-s3-source\"],\"processors\":[{\"enabled\":true,\"id\":\"filter-processor\",\"include\":\"service:my-service\",\"type\":\"filter\"}]}],\"sources\":[{\"compression\":\"gzip\",\"id\":\"amazon-s3-source\",\"region\":\"us-east-1\",\"type\":\"amazon_s3\"}]},\"name\":\"Pipeline with S3 Source Compression\"},\"type\":\"pipelines\"}}",
+          "encoding": null
+        },
+        "headers": {
+          "Accept": [
+            "application/json"
+          ],
+          "Content-Type": [
+            "application/json"
+          ]
+        },
+        "method": "post",
+        "uri": "https://api.datadoghq.com/api/v2/obs-pipelines/pipelines/validate"
+      },
+      "response": {
+        "body": {
+          "string": "{\"errors\":[]}\n",
+          "encoding": null
+        },
+        "headers": {
+          "Content-Type": [
+            "application/vnd.api+json"
+          ]
+        },
+        "status": {
+          "code": 200,
+          "message": "OK"
+        }
+      },
+      "recorded_at": "Wed, 08 Apr 2026 12:44:25 GMT"
+    }
+  ],
+  "recorded_with": "VCR 6.0.0"
+}

tests/scenarios/features/v2/observability_pipelines.feature

@@ -207,6 +207,14 @@ Feature: Observability Pipelines
     Then the response status is 200 OK
     And the response "errors" has length 0
 
+  @team:DataDog/observability-pipelines
+  Scenario: Validate an observability pipeline with amazon S3 source compression returns "OK" response
+    Given new "ValidatePipeline" request
+    And body with value {"data": {"attributes": {"config": {"destinations": [{"id": "datadog-logs-destination", "inputs": ["my-processor-group"], "type": "datadog_logs"}], "processor_groups": [{"enabled": true, "id": "my-processor-group", "include": "service:my-service", "inputs": ["amazon-s3-source"], "processors": [{"enabled": true, "id": "filter-processor", "include": "service:my-service", "type": "filter"}]}], "sources": [{"id": "amazon-s3-source", "type": "amazon_s3", "region": "us-east-1", "compression": "gzip"}]}, "name": "Pipeline with S3 Source Compression"}, "type": "pipelines"}}
+    When the request is sent
+    Then the response status is 200 OK
+    And the response "errors" has length 0
+
   @team:DataDog/observability-pipelines
   Scenario: Validate an observability pipeline with destination secret key returns "OK" response
     Given new "ValidatePipeline" request