Add buffer configuration to ClickHouse destination

.generator/schemas/v2/openapi.yaml

@@ -64398,6 +64398,8 @@ components:
           $ref: "#/components/schemas/ObservabilityPipelineClickhouseDestinationBatch"
         batch_encoding:
           $ref: "#/components/schemas/ObservabilityPipelineClickhouseDestinationBatchEncoding"
+        buffer:
+          $ref: "#/components/schemas/ObservabilityPipelineBufferOptions"
         compression:
           $ref: "#/components/schemas/ObservabilityPipelineClickhouseDestinationCompression"
         database:

cassettes/features/v2/observability_pipelines/Validate-an-observability-pipeline-with-ClickHouse-destination-with-all-fields-set-returns-OK-response.frozen

@@ -0,0 +1 @@
+2026-06-24T16:45:05.037Z

examples/v2/observability-pipelines/ValidatePipeline_253109967.rb

@@ -0,0 +1,83 @@
+# Validate an observability pipeline with ClickHouse destination with all fields set returns "OK" response
+
+require "datadog_api_client"
+api_instance = DatadogAPIClient::V2::ObservabilityPipelinesAPI.new
+
+body = DatadogAPIClient::V2::ObservabilityPipelineSpec.new({
+  data: DatadogAPIClient::V2::ObservabilityPipelineSpecData.new({
+    attributes: DatadogAPIClient::V2::ObservabilityPipelineDataAttributes.new({
+      config: DatadogAPIClient::V2::ObservabilityPipelineConfig.new({
+        destinations: [
+          DatadogAPIClient::V2::ObservabilityPipelineClickhouseDestination.new({
+            id: "clickhouse-destination",
+            inputs: [
+              "my-processor-group",
+            ],
+            type: DatadogAPIClient::V2::ObservabilityPipelineClickhouseDestinationType::CLICKHOUSE,
+            endpoint_url_key: "CLICKHOUSE_ENDPOINT_URL",
+            database: "my_database",
+            table: "application_logs",
+            format: DatadogAPIClient::V2::ObservabilityPipelineClickhouseDestinationFormat::ARROW_STREAM,
+            skip_unknown_fields: true,
+            date_time_best_effort: true,
+            compression: DatadogAPIClient::V2::ObservabilityPipelineClickhouseDestinationCompressionObject.new({
+              algorithm: DatadogAPIClient::V2::ObservabilityPipelineClickhouseDestinationCompressionAlgorithm::GZIP,
+              level: 6,
+            }),
+            auth: DatadogAPIClient::V2::ObservabilityPipelineClickhouseDestinationAuth.new({
+              strategy: DatadogAPIClient::V2::ObservabilityPipelineClickhouseDestinationAuthStrategy::BASIC,
+              username_key: "CLICKHOUSE_USERNAME",
+              password_key: "CLICKHOUSE_PASSWORD",
+            }),
+            batch: DatadogAPIClient::V2::ObservabilityPipelineClickhouseDestinationBatch.new({
+              max_events: 1000,
+              timeout_secs: 1,
+            }),
+            batch_encoding: DatadogAPIClient::V2::ObservabilityPipelineClickhouseDestinationBatchEncoding.new({
+              codec: DatadogAPIClient::V2::ObservabilityPipelineClickhouseDestinationBatchEncodingCodec::ARROW_STREAM,
+              allow_nullable_fields: true,
+            }),
+            tls: DatadogAPIClient::V2::ObservabilityPipelineTls.new({
+              crt_file: "/path/to/cert.crt",
+              ca_file: "/path/to/ca.crt",
+              key_file: "/path/to/key.key",
+              key_pass_key: "TLS_KEY_PASSPHRASE",
+            }),
+            buffer: DatadogAPIClient::V2::ObservabilityPipelineMemoryBufferSizeOptions.new({
+              type: DatadogAPIClient::V2::ObservabilityPipelineBufferOptionsMemoryType::MEMORY,
+              max_events: 500,
+              when_full: DatadogAPIClient::V2::ObservabilityPipelineBufferOptionsWhenFull::BLOCK,
+            }),
+          }),
+        ],
+        processor_groups: [
+          DatadogAPIClient::V2::ObservabilityPipelineConfigProcessorGroup.new({
+            enabled: true,
+            id: "my-processor-group",
+            include: "service:my-service",
+            inputs: [
+              "datadog-agent-source",
+            ],
+            processors: [
+              DatadogAPIClient::V2::ObservabilityPipelineFilterProcessor.new({
+                enabled: true,
+                id: "filter-processor",
+                include: "status:error",
+                type: DatadogAPIClient::V2::ObservabilityPipelineFilterProcessorType::FILTER,
+              }),
+            ],
+          }),
+        ],
+        sources: [
+          DatadogAPIClient::V2::ObservabilityPipelineDatadogAgentSource.new({
+            id: "datadog-agent-source",
+            type: DatadogAPIClient::V2::ObservabilityPipelineDatadogAgentSourceType::DATADOG_AGENT,
+          }),
+        ],
+      }),
+      name: "Pipeline with ClickHouse Destination All Fields",
+    }),
+    type: "pipelines",
+  }),
+})
+p api_instance.validate_pipeline(body)

features/v2/observability_pipelines.feature

@@ -192,6 +192,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 ClickHouse destination with all fields set returns "OK" response
+    Given new "ValidatePipeline" request
+    And body with value {"data": {"attributes": {"config": {"destinations": [{"id": "clickhouse-destination", "inputs": ["my-processor-group"], "type": "clickhouse", "endpoint_url_key": "CLICKHOUSE_ENDPOINT_URL", "database": "my_database", "table": "application_logs", "format": "arrow_stream", "skip_unknown_fields": true, "date_time_best_effort": true, "compression": {"algorithm": "gzip", "level": 6}, "auth": {"strategy": "basic", "username_key": "CLICKHOUSE_USERNAME", "password_key": "CLICKHOUSE_PASSWORD"}, "batch": {"max_events": 1000, "timeout_secs": 1}, "batch_encoding": {"codec": "arrow_stream", "allow_nullable_fields": true}, "tls": {"crt_file": "/path/to/cert.crt", "ca_file": "/path/to/ca.crt", "key_file": "/path/to/key.key", "key_pass_key": "TLS_KEY_PASSPHRASE"}, "buffer": {"type": "memory", "max_events": 500, "when_full": "block"}}], "processor_groups": [{"enabled": true, "id": "my-processor-group", "include": "service:my-service", "inputs": ["datadog-agent-source"], "processors": [{"enabled": true, "id": "filter-processor", "include": "status:error", "type": "filter"}]}], "sources": [{"id": "datadog-agent-source", "type": "datadog_agent"}]}, "name": "Pipeline with ClickHouse Destination All Fields"}, "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 HTTP server source valid_tokens returns "OK" response
     Given new "ValidatePipeline" request

lib/datadog_api_client/v2/models/observability_pipeline_clickhouse_destination.rb

@@ -34,6 +34,9 @@ class ObservabilityPipelineClickhouseDestination
     # Required when `format` is `arrow_stream`. The `codec` field must be set to `arrow_stream`.
     attr_accessor :batch_encoding
 
+    # Configuration for buffer settings on destination components.
+    attr_accessor :buffer
+
     # Compression setting for outbound HTTP requests to ClickHouse.
     # Can be specified as a shorthand string (`"gzip"` or `"none"`) or as an object
     # with an `algorithm` field and an optional `level` (gzip only, 1–9).
@@ -84,6 +87,7 @@ def self.attribute_map
         :'auth' => :'auth',
         :'batch' => :'batch',
         :'batch_encoding' => :'batch_encoding',
+        :'buffer' => :'buffer',
         :'compression' => :'compression',
         :'database' => :'database',
         :'date_time_best_effort' => :'date_time_best_effort',
@@ -105,6 +109,7 @@ def self.openapi_types
         :'auth' => :'ObservabilityPipelineClickhouseDestinationAuth',
         :'batch' => :'ObservabilityPipelineClickhouseDestinationBatch',
         :'batch_encoding' => :'ObservabilityPipelineClickhouseDestinationBatchEncoding',
+        :'buffer' => :'ObservabilityPipelineBufferOptions',
         :'compression' => :'ObservabilityPipelineClickhouseDestinationCompression',
         :'database' => :'String',
         :'date_time_best_effort' => :'Boolean',
@@ -157,6 +162,10 @@ def initialize(attributes = {})
         self.batch_encoding = attributes[:'batch_encoding']
       end
 
+      if attributes.key?(:'buffer')
+        self.buffer = attributes[:'buffer']
+      end
+
       if attributes.key?(:'compression')
         self.compression = attributes[:'compression']
       end
@@ -284,6 +293,7 @@ def ==(o)
           auth == o.auth &&
           batch == o.batch &&
           batch_encoding == o.batch_encoding &&
+          buffer == o.buffer &&
           compression == o.compression &&
           database == o.database &&
           date_time_best_effort == o.date_time_best_effort &&
@@ -302,7 +312,7 @@ def ==(o)
     # @return [Integer] Hash code
     # @!visibility private
     def hash
-      [auth, batch, batch_encoding, compression, database, date_time_best_effort, endpoint_url_key, format, id, inputs, skip_unknown_fields, table, tls, type, additional_properties].hash
+      [auth, batch, batch_encoding, buffer, compression, database, date_time_best_effort, endpoint_url_key, format, id, inputs, skip_unknown_fields, table, tls, type, additional_properties].hash
     end
   end
 end