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:

examples/v2/observability-pipelines/ValidatePipeline_253109967.py

@@ -0,0 +1,148 @@
+"""
+Validate an observability pipeline with ClickHouse destination with all fields set returns "OK" response
+"""
+
+from datadog_api_client import ApiClient, Configuration
+from datadog_api_client.v2.api.observability_pipelines_api import ObservabilityPipelinesApi
+from datadog_api_client.v2.model.observability_pipeline_buffer_options_memory_type import (
+    ObservabilityPipelineBufferOptionsMemoryType,
+)
+from datadog_api_client.v2.model.observability_pipeline_buffer_options_when_full import (
+    ObservabilityPipelineBufferOptionsWhenFull,
+)
+from datadog_api_client.v2.model.observability_pipeline_clickhouse_destination import (
+    ObservabilityPipelineClickhouseDestination,
+)
+from datadog_api_client.v2.model.observability_pipeline_clickhouse_destination_auth import (
+    ObservabilityPipelineClickhouseDestinationAuth,
+)
+from datadog_api_client.v2.model.observability_pipeline_clickhouse_destination_auth_strategy import (
+    ObservabilityPipelineClickhouseDestinationAuthStrategy,
+)
+from datadog_api_client.v2.model.observability_pipeline_clickhouse_destination_batch import (
+    ObservabilityPipelineClickhouseDestinationBatch,
+)
+from datadog_api_client.v2.model.observability_pipeline_clickhouse_destination_batch_encoding import (
+    ObservabilityPipelineClickhouseDestinationBatchEncoding,
+)
+from datadog_api_client.v2.model.observability_pipeline_clickhouse_destination_batch_encoding_codec import (
+    ObservabilityPipelineClickhouseDestinationBatchEncodingCodec,
+)
+from datadog_api_client.v2.model.observability_pipeline_clickhouse_destination_compression_algorithm import (
+    ObservabilityPipelineClickhouseDestinationCompressionAlgorithm,
+)
+from datadog_api_client.v2.model.observability_pipeline_clickhouse_destination_compression_object import (
+    ObservabilityPipelineClickhouseDestinationCompressionObject,
+)
+from datadog_api_client.v2.model.observability_pipeline_clickhouse_destination_format import (
+    ObservabilityPipelineClickhouseDestinationFormat,
+)
+from datadog_api_client.v2.model.observability_pipeline_clickhouse_destination_type import (
+    ObservabilityPipelineClickhouseDestinationType,
+)
+from datadog_api_client.v2.model.observability_pipeline_config import ObservabilityPipelineConfig
+from datadog_api_client.v2.model.observability_pipeline_config_processor_group import (
+    ObservabilityPipelineConfigProcessorGroup,
+)
+from datadog_api_client.v2.model.observability_pipeline_data_attributes import ObservabilityPipelineDataAttributes
+from datadog_api_client.v2.model.observability_pipeline_datadog_agent_source import (
+    ObservabilityPipelineDatadogAgentSource,
+)
+from datadog_api_client.v2.model.observability_pipeline_datadog_agent_source_type import (
+    ObservabilityPipelineDatadogAgentSourceType,
+)
+from datadog_api_client.v2.model.observability_pipeline_filter_processor import ObservabilityPipelineFilterProcessor
+from datadog_api_client.v2.model.observability_pipeline_filter_processor_type import (
+    ObservabilityPipelineFilterProcessorType,
+)
+from datadog_api_client.v2.model.observability_pipeline_memory_buffer_size_options import (
+    ObservabilityPipelineMemoryBufferSizeOptions,
+)
+from datadog_api_client.v2.model.observability_pipeline_spec import ObservabilityPipelineSpec
+from datadog_api_client.v2.model.observability_pipeline_spec_data import ObservabilityPipelineSpecData
+from datadog_api_client.v2.model.observability_pipeline_tls import ObservabilityPipelineTls
+
+body = ObservabilityPipelineSpec(
+    data=ObservabilityPipelineSpecData(
+        attributes=ObservabilityPipelineDataAttributes(
+            config=ObservabilityPipelineConfig(
+                destinations=[
+                    ObservabilityPipelineClickhouseDestination(
+                        id="clickhouse-destination",
+                        inputs=[
+                            "my-processor-group",
+                        ],
+                        type=ObservabilityPipelineClickhouseDestinationType.CLICKHOUSE,
+                        endpoint_url_key="CLICKHOUSE_ENDPOINT_URL",
+                        database="my_database",
+                        table="application_logs",
+                        format=ObservabilityPipelineClickhouseDestinationFormat.ARROW_STREAM,
+                        skip_unknown_fields=True,
+                        date_time_best_effort=True,
+                        compression=ObservabilityPipelineClickhouseDestinationCompressionObject(
+                            algorithm=ObservabilityPipelineClickhouseDestinationCompressionAlgorithm.GZIP,
+                            level=6,
+                        ),
+                        auth=ObservabilityPipelineClickhouseDestinationAuth(
+                            strategy=ObservabilityPipelineClickhouseDestinationAuthStrategy.BASIC,
+                            username_key="CLICKHOUSE_USERNAME",
+                            password_key="CLICKHOUSE_PASSWORD",
+                        ),
+                        batch=ObservabilityPipelineClickhouseDestinationBatch(
+                            max_events=1000,
+                            timeout_secs=1,
+                        ),
+                        batch_encoding=ObservabilityPipelineClickhouseDestinationBatchEncoding(
+                            codec=ObservabilityPipelineClickhouseDestinationBatchEncodingCodec.ARROW_STREAM,
+                            allow_nullable_fields=True,
+                        ),
+                        tls=ObservabilityPipelineTls(
+                            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=ObservabilityPipelineMemoryBufferSizeOptions(
+                            type=ObservabilityPipelineBufferOptionsMemoryType.MEMORY,
+                            max_events=500,
+                            when_full=ObservabilityPipelineBufferOptionsWhenFull.BLOCK,
+                        ),
+                    ),
+                ],
+                processor_groups=[
+                    ObservabilityPipelineConfigProcessorGroup(
+                        enabled=True,
+                        id="my-processor-group",
+                        include="service:my-service",
+                        inputs=[
+                            "datadog-agent-source",
+                        ],
+                        processors=[
+                            ObservabilityPipelineFilterProcessor(
+                                enabled=True,
+                                id="filter-processor",
+                                include="status:error",
+                                type=ObservabilityPipelineFilterProcessorType.FILTER,
+                            ),
+                        ],
+                    ),
+                ],
+                sources=[
+                    ObservabilityPipelineDatadogAgentSource(
+                        id="datadog-agent-source",
+                        type=ObservabilityPipelineDatadogAgentSourceType.DATADOG_AGENT,
+                    ),
+                ],
+            ),
+            name="Pipeline with ClickHouse Destination All Fields",
+        ),
+        type="pipelines",
+    ),
+)
+
+configuration = Configuration()
+with ApiClient(configuration) as api_client:
+    api_instance = ObservabilityPipelinesApi(api_client)
+    response = api_instance.validate_pipeline(body=body)
+
+    print(response)

src/datadog_api_client/v2/model/observability_pipeline_clickhouse_destination.py

@@ -24,6 +24,7 @@
     from datadog_api_client.v2.model.observability_pipeline_clickhouse_destination_batch_encoding import (
         ObservabilityPipelineClickhouseDestinationBatchEncoding,
     )
+    from datadog_api_client.v2.model.observability_pipeline_buffer_options import ObservabilityPipelineBufferOptions
     from datadog_api_client.v2.model.observability_pipeline_clickhouse_destination_compression import (
         ObservabilityPipelineClickhouseDestinationCompression,
     )
@@ -34,6 +35,15 @@
     from datadog_api_client.v2.model.observability_pipeline_clickhouse_destination_type import (
         ObservabilityPipelineClickhouseDestinationType,
     )
+    from datadog_api_client.v2.model.observability_pipeline_disk_buffer_options import (
+        ObservabilityPipelineDiskBufferOptions,
+    )
+    from datadog_api_client.v2.model.observability_pipeline_memory_buffer_options import (
+        ObservabilityPipelineMemoryBufferOptions,
+    )
+    from datadog_api_client.v2.model.observability_pipeline_memory_buffer_size_options import (
+        ObservabilityPipelineMemoryBufferSizeOptions,
+    )
     from datadog_api_client.v2.model.observability_pipeline_clickhouse_destination_compression_object import (
         ObservabilityPipelineClickhouseDestinationCompressionObject,
     )
@@ -51,6 +61,7 @@ def openapi_types(_):
         from datadog_api_client.v2.model.observability_pipeline_clickhouse_destination_batch_encoding import (
             ObservabilityPipelineClickhouseDestinationBatchEncoding,
         )
+        from datadog_api_client.v2.model.observability_pipeline_buffer_options import ObservabilityPipelineBufferOptions
         from datadog_api_client.v2.model.observability_pipeline_clickhouse_destination_compression import (
             ObservabilityPipelineClickhouseDestinationCompression,
         )
@@ -66,6 +77,7 @@ def openapi_types(_):
             "auth": (ObservabilityPipelineClickhouseDestinationAuth,),
             "batch": (ObservabilityPipelineClickhouseDestinationBatch,),
             "batch_encoding": (ObservabilityPipelineClickhouseDestinationBatchEncoding,),
+            "buffer": (ObservabilityPipelineBufferOptions,),
             "compression": (ObservabilityPipelineClickhouseDestinationCompression,),
             "database": (str,),
             "date_time_best_effort": (bool,),
@@ -83,6 +95,7 @@ def openapi_types(_):
         "auth": "auth",
         "batch": "batch",
         "batch_encoding": "batch_encoding",
+        "buffer": "buffer",
         "compression": "compression",
         "database": "database",
         "date_time_best_effort": "date_time_best_effort",
@@ -105,6 +118,13 @@ def __init__(
         auth: Union[ObservabilityPipelineClickhouseDestinationAuth, UnsetType] = unset,
         batch: Union[ObservabilityPipelineClickhouseDestinationBatch, UnsetType] = unset,
         batch_encoding: Union[ObservabilityPipelineClickhouseDestinationBatchEncoding, UnsetType] = unset,
+        buffer: Union[
+            ObservabilityPipelineBufferOptions,
+            ObservabilityPipelineDiskBufferOptions,
+            ObservabilityPipelineMemoryBufferOptions,
+            ObservabilityPipelineMemoryBufferSizeOptions,
+            UnsetType,
+        ] = unset,
         compression: Union[
             ObservabilityPipelineClickhouseDestinationCompression,
             str,
@@ -135,6 +155,9 @@ def __init__(
             Required when ``format`` is ``arrow_stream``. The ``codec`` field must be set to ``arrow_stream``.
         :type batch_encoding: ObservabilityPipelineClickhouseDestinationBatchEncoding, optional
 
+        :param buffer: Configuration for buffer settings on destination components.
+        :type buffer: ObservabilityPipelineBufferOptions, optional
+
         :param compression: 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).
@@ -183,6 +206,8 @@ def __init__(
             kwargs["batch"] = batch
         if batch_encoding is not unset:
             kwargs["batch_encoding"] = batch_encoding
+        if buffer is not unset:
+            kwargs["buffer"] = buffer
         if compression is not unset:
             kwargs["compression"] = compression
         if database is not unset:

tests/v2/cassettes/test_scenarios/test_validate_an_observability_pipeline_with_clickhouse_destination_with_all_fields_set_returns_ok_response.frozen

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

tests/v2/cassettes/test_scenarios/test_validate_an_observability_pipeline_with_clickhouse_destination_with_all_fields_set_returns_ok_response.yaml

@@ -0,0 +1,23 @@
+interactions:
+- request:
+    body: '{"data":{"attributes":{"config":{"destinations":[{"auth":{"password_key":"CLICKHOUSE_PASSWORD","strategy":"basic","username_key":"CLICKHOUSE_USERNAME"},"batch":{"max_events":1000,"timeout_secs":1},"batch_encoding":{"allow_nullable_fields":true,"codec":"arrow_stream"},"buffer":{"max_events":500,"type":"memory","when_full":"block"},"compression":{"algorithm":"gzip","level":6},"database":"my_database","date_time_best_effort":true,"endpoint_url_key":"CLICKHOUSE_ENDPOINT_URL","format":"arrow_stream","id":"clickhouse-destination","inputs":["my-processor-group"],"skip_unknown_fields":true,"table":"application_logs","tls":{"ca_file":"/path/to/ca.crt","crt_file":"/path/to/cert.crt","key_file":"/path/to/key.key","key_pass_key":"TLS_KEY_PASSPHRASE"},"type":"clickhouse"}],"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"}}'
+    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":[]}
+
+        '
+    headers:
+      content-type:
+      - application/vnd.api+json
+    status:
+      code: 200
+      message: OK
+version: 1

tests/v2/features/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