[pull] master from DataDog:master

.generator/schemas/v2/openapi.yaml

@@ -39514,6 +39514,8 @@ components:
 
         **Supported pipeline types:** logs'
       properties:
+        cache:
+          $ref: '#/components/schemas/ObservabilityPipelineDedupeProcessorCache'
         display_name:
           $ref: '#/components/schemas/ObservabilityPipelineComponentDisplayName'
         enabled:
@@ -39551,6 +39553,19 @@ components:
       type: object
       x-pipeline-types:
       - logs
+    ObservabilityPipelineDedupeProcessorCache:
+      description: Configuration for the cache used to detect duplicates.
+      properties:
+        num_events:
+          description: The number of events to cache for duplicate detection.
+          example: 5000
+          format: int64
+          maximum: 1000000000
+          minimum: 1
+          type: integer
+      required:
+      - num_events
+      type: object
     ObservabilityPipelineDedupeProcessorMode:
       description: The deduplication mode to apply to the fields.
       enum:

api/datadogV2/model_observability_pipeline_dedupe_processor.go

@@ -14,6 +14,8 @@ import (
 //
 // **Supported pipeline types:** logs
 type ObservabilityPipelineDedupeProcessor struct {
+	// Configuration for the cache used to detect duplicates.
+	Cache *ObservabilityPipelineDedupeProcessorCache `json:"cache,omitempty"`
 	// The display name for a component.
 	DisplayName *string `json:"display_name,omitempty"`
 	// Indicates whether the processor is enabled.
@@ -58,6 +60,34 @@ func NewObservabilityPipelineDedupeProcessorWithDefaults() *ObservabilityPipelin
 	return &this
 }
 
+// GetCache returns the Cache field value if set, zero value otherwise.
+func (o *ObservabilityPipelineDedupeProcessor) GetCache() ObservabilityPipelineDedupeProcessorCache {
+	if o == nil || o.Cache == nil {
+		var ret ObservabilityPipelineDedupeProcessorCache
+		return ret
+	}
+	return *o.Cache
+}
+
+// GetCacheOk returns a tuple with the Cache field value if set, nil otherwise
+// and a boolean to check if the value has been set.
+func (o *ObservabilityPipelineDedupeProcessor) GetCacheOk() (*ObservabilityPipelineDedupeProcessorCache, bool) {
+	if o == nil || o.Cache == nil {
+		return nil, false
+	}
+	return o.Cache, true
+}
+
+// HasCache returns a boolean if a field has been set.
+func (o *ObservabilityPipelineDedupeProcessor) HasCache() bool {
+	return o != nil && o.Cache != nil
+}
+
+// SetCache gets a reference to the given ObservabilityPipelineDedupeProcessorCache and assigns it to the Cache field.
+func (o *ObservabilityPipelineDedupeProcessor) SetCache(v ObservabilityPipelineDedupeProcessorCache) {
+	o.Cache = &v
+}
+
 // GetDisplayName returns the DisplayName field value if set, zero value otherwise.
 func (o *ObservabilityPipelineDedupeProcessor) GetDisplayName() string {
 	if o == nil || o.DisplayName == nil {
@@ -230,6 +260,9 @@ func (o ObservabilityPipelineDedupeProcessor) MarshalJSON() ([]byte, error) {
 	if o.UnparsedObject != nil {
 		return datadog.Marshal(o.UnparsedObject)
 	}
+	if o.Cache != nil {
+		toSerialize["cache"] = o.Cache
+	}
 	if o.DisplayName != nil {
 		toSerialize["display_name"] = o.DisplayName
 	}
@@ -249,13 +282,14 @@ func (o ObservabilityPipelineDedupeProcessor) MarshalJSON() ([]byte, error) {
 // UnmarshalJSON deserializes the given payload.
 func (o *ObservabilityPipelineDedupeProcessor) UnmarshalJSON(bytes []byte) (err error) {
 	all := struct {
-		DisplayName *string                                   `json:"display_name,omitempty"`
-		Enabled     *bool                                     `json:"enabled"`
-		Fields      *[]string                                 `json:"fields"`
-		Id          *string                                   `json:"id"`
-		Include     *string                                   `json:"include"`
-		Mode        *ObservabilityPipelineDedupeProcessorMode `json:"mode"`
-		Type        *ObservabilityPipelineDedupeProcessorType `json:"type"`
+		Cache       *ObservabilityPipelineDedupeProcessorCache `json:"cache,omitempty"`
+		DisplayName *string                                    `json:"display_name,omitempty"`
+		Enabled     *bool                                      `json:"enabled"`
+		Fields      *[]string                                  `json:"fields"`
+		Id          *string                                    `json:"id"`
+		Include     *string                                    `json:"include"`
+		Mode        *ObservabilityPipelineDedupeProcessorMode  `json:"mode"`
+		Type        *ObservabilityPipelineDedupeProcessorType  `json:"type"`
 	}{}
 	if err = datadog.Unmarshal(bytes, &all); err != nil {
 		return datadog.Unmarshal(bytes, &o.UnparsedObject)
@@ -280,12 +314,16 @@ func (o *ObservabilityPipelineDedupeProcessor) UnmarshalJSON(bytes []byte) (err
 	}
 	additionalProperties := make(map[string]interface{})
 	if err = datadog.Unmarshal(bytes, &additionalProperties); err == nil {
-		datadog.DeleteKeys(additionalProperties, &[]string{"display_name", "enabled", "fields", "id", "include", "mode", "type"})
+		datadog.DeleteKeys(additionalProperties, &[]string{"cache", "display_name", "enabled", "fields", "id", "include", "mode", "type"})
 	} else {
 		return err
 	}
 
 	hasInvalidField := false
+	if all.Cache != nil && all.Cache.UnparsedObject != nil && o.UnparsedObject == nil {
+		hasInvalidField = true
+	}
+	o.Cache = all.Cache
 	o.DisplayName = all.DisplayName
 	o.Enabled = *all.Enabled
 	o.Fields = *all.Fields

api/datadogV2/model_observability_pipeline_dedupe_processor_cache.go

@@ -0,0 +1,101 @@
+// 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.
+
+package datadogV2
+
+import (
+	"fmt"
+
+	"github.com/DataDog/datadog-api-client-go/v2/api/datadog"
+)
+
+// ObservabilityPipelineDedupeProcessorCache Configuration for the cache used to detect duplicates.
+type ObservabilityPipelineDedupeProcessorCache struct {
+	// The number of events to cache for duplicate detection.
+	NumEvents int64 `json:"num_events"`
+	// UnparsedObject contains the raw value of the object if there was an error when deserializing into the struct
+	UnparsedObject       map[string]interface{} `json:"-"`
+	AdditionalProperties map[string]interface{} `json:"-"`
+}
+
+// NewObservabilityPipelineDedupeProcessorCache instantiates a new ObservabilityPipelineDedupeProcessorCache object.
+// This constructor will assign default values to properties that have it defined,
+// and makes sure properties required by API are set, but the set of arguments
+// will change when the set of required properties is changed.
+func NewObservabilityPipelineDedupeProcessorCache(numEvents int64) *ObservabilityPipelineDedupeProcessorCache {
+	this := ObservabilityPipelineDedupeProcessorCache{}
+	this.NumEvents = numEvents
+	return &this
+}
+
+// NewObservabilityPipelineDedupeProcessorCacheWithDefaults instantiates a new ObservabilityPipelineDedupeProcessorCache object.
+// This constructor will only assign default values to properties that have it defined,
+// but it doesn't guarantee that properties required by API are set.
+func NewObservabilityPipelineDedupeProcessorCacheWithDefaults() *ObservabilityPipelineDedupeProcessorCache {
+	this := ObservabilityPipelineDedupeProcessorCache{}
+	return &this
+}
+
+// GetNumEvents returns the NumEvents field value.
+func (o *ObservabilityPipelineDedupeProcessorCache) GetNumEvents() int64 {
+	if o == nil {
+		var ret int64
+		return ret
+	}
+	return o.NumEvents
+}
+
+// GetNumEventsOk returns a tuple with the NumEvents field value
+// and a boolean to check if the value has been set.
+func (o *ObservabilityPipelineDedupeProcessorCache) GetNumEventsOk() (*int64, bool) {
+	if o == nil {
+		return nil, false
+	}
+	return &o.NumEvents, true
+}
+
+// SetNumEvents sets field value.
+func (o *ObservabilityPipelineDedupeProcessorCache) SetNumEvents(v int64) {
+	o.NumEvents = v
+}
+
+// MarshalJSON serializes the struct using spec logic.
+func (o ObservabilityPipelineDedupeProcessorCache) MarshalJSON() ([]byte, error) {
+	toSerialize := map[string]interface{}{}
+	if o.UnparsedObject != nil {
+		return datadog.Marshal(o.UnparsedObject)
+	}
+	toSerialize["num_events"] = o.NumEvents
+
+	for key, value := range o.AdditionalProperties {
+		toSerialize[key] = value
+	}
+	return datadog.Marshal(toSerialize)
+}
+
+// UnmarshalJSON deserializes the given payload.
+func (o *ObservabilityPipelineDedupeProcessorCache) UnmarshalJSON(bytes []byte) (err error) {
+	all := struct {
+		NumEvents *int64 `json:"num_events"`
+	}{}
+	if err = datadog.Unmarshal(bytes, &all); err != nil {
+		return datadog.Unmarshal(bytes, &o.UnparsedObject)
+	}
+	if all.NumEvents == nil {
+		return fmt.Errorf("required field num_events missing")
+	}
+	additionalProperties := make(map[string]interface{})
+	if err = datadog.Unmarshal(bytes, &additionalProperties); err == nil {
+		datadog.DeleteKeys(additionalProperties, &[]string{"num_events"})
+	} else {
+		return err
+	}
+	o.NumEvents = *all.NumEvents
+
+	if len(additionalProperties) > 0 {
+		o.AdditionalProperties = additionalProperties
+	}
+
+	return nil
+}

examples/v2/observability-pipelines/CreatePipeline_3363445359.go

@@ -0,0 +1,82 @@
+// Create a pipeline with dedupe processor with cache returns "OK" response
+
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+
+	"github.com/DataDog/datadog-api-client-go/v2/api/datadog"
+	"github.com/DataDog/datadog-api-client-go/v2/api/datadogV2"
+)
+
+func main() {
+	body := datadogV2.ObservabilityPipelineSpec{
+		Data: datadogV2.ObservabilityPipelineSpecData{
+			Attributes: datadogV2.ObservabilityPipelineDataAttributes{
+				Config: datadogV2.ObservabilityPipelineConfig{
+					Destinations: []datadogV2.ObservabilityPipelineConfigDestinationItem{
+						datadogV2.ObservabilityPipelineConfigDestinationItem{
+							ObservabilityPipelineDatadogLogsDestination: &datadogV2.ObservabilityPipelineDatadogLogsDestination{
+								Id: "datadog-logs-destination",
+								Inputs: []string{
+									"my-processor-group",
+								},
+								Type: datadogV2.OBSERVABILITYPIPELINEDATADOGLOGSDESTINATIONTYPE_DATADOG_LOGS,
+							}},
+					},
+					ProcessorGroups: []datadogV2.ObservabilityPipelineConfigProcessorGroup{
+						{
+							Enabled: true,
+							Id:      "my-processor-group",
+							Include: "service:my-service",
+							Inputs: []string{
+								"datadog-agent-source",
+							},
+							Processors: []datadogV2.ObservabilityPipelineConfigProcessorItem{
+								datadogV2.ObservabilityPipelineConfigProcessorItem{
+									ObservabilityPipelineDedupeProcessor: &datadogV2.ObservabilityPipelineDedupeProcessor{
+										Enabled: true,
+										Id:      "dedupe-processor",
+										Include: "service:my-service",
+										Type:    datadogV2.OBSERVABILITYPIPELINEDEDUPEPROCESSORTYPE_DEDUPE,
+										Fields: []string{
+											"message",
+										},
+										Mode: datadogV2.OBSERVABILITYPIPELINEDEDUPEPROCESSORMODE_MATCH,
+										Cache: &datadogV2.ObservabilityPipelineDedupeProcessorCache{
+											NumEvents: 5000,
+										},
+									}},
+							},
+						},
+					},
+					Sources: []datadogV2.ObservabilityPipelineConfigSourceItem{
+						datadogV2.ObservabilityPipelineConfigSourceItem{
+							ObservabilityPipelineDatadogAgentSource: &datadogV2.ObservabilityPipelineDatadogAgentSource{
+								Id:   "datadog-agent-source",
+								Type: datadogV2.OBSERVABILITYPIPELINEDATADOGAGENTSOURCETYPE_DATADOG_AGENT,
+							}},
+					},
+				},
+				Name: "Pipeline with Dedupe Cache",
+			},
+			Type: "pipelines",
+		},
+	}
+	ctx := datadog.NewDefaultContext(context.Background())
+	configuration := datadog.NewConfiguration()
+	apiClient := datadog.NewAPIClient(configuration)
+	api := datadogV2.NewObservabilityPipelinesApi(apiClient)
+	resp, r, err := api.CreatePipeline(ctx, body)
+
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error when calling `ObservabilityPipelinesApi.CreatePipeline`: %v\n", err)
+		fmt.Fprintf(os.Stderr, "Full HTTP response: %v\n", r)
+	}
+
+	responseContent, _ := json.MarshalIndent(resp, "", "  ")
+	fmt.Fprintf(os.Stdout, "Response from `ObservabilityPipelinesApi.CreatePipeline`:\n%s\n", responseContent)
+}

examples/v2/observability-pipelines/CreatePipeline_581245895.go

@@ -0,0 +1,79 @@
+// Create a pipeline with dedupe processor without cache returns "OK" response
+
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+
+	"github.com/DataDog/datadog-api-client-go/v2/api/datadog"
+	"github.com/DataDog/datadog-api-client-go/v2/api/datadogV2"
+)
+
+func main() {
+	body := datadogV2.ObservabilityPipelineSpec{
+		Data: datadogV2.ObservabilityPipelineSpecData{
+			Attributes: datadogV2.ObservabilityPipelineDataAttributes{
+				Config: datadogV2.ObservabilityPipelineConfig{
+					Destinations: []datadogV2.ObservabilityPipelineConfigDestinationItem{
+						datadogV2.ObservabilityPipelineConfigDestinationItem{
+							ObservabilityPipelineDatadogLogsDestination: &datadogV2.ObservabilityPipelineDatadogLogsDestination{
+								Id: "datadog-logs-destination",
+								Inputs: []string{
+									"my-processor-group",
+								},
+								Type: datadogV2.OBSERVABILITYPIPELINEDATADOGLOGSDESTINATIONTYPE_DATADOG_LOGS,
+							}},
+					},
+					ProcessorGroups: []datadogV2.ObservabilityPipelineConfigProcessorGroup{
+						{
+							Enabled: true,
+							Id:      "my-processor-group",
+							Include: "service:my-service",
+							Inputs: []string{
+								"datadog-agent-source",
+							},
+							Processors: []datadogV2.ObservabilityPipelineConfigProcessorItem{
+								datadogV2.ObservabilityPipelineConfigProcessorItem{
+									ObservabilityPipelineDedupeProcessor: &datadogV2.ObservabilityPipelineDedupeProcessor{
+										Enabled: true,
+										Id:      "dedupe-processor",
+										Include: "service:my-service",
+										Type:    datadogV2.OBSERVABILITYPIPELINEDEDUPEPROCESSORTYPE_DEDUPE,
+										Fields: []string{
+											"message",
+										},
+										Mode: datadogV2.OBSERVABILITYPIPELINEDEDUPEPROCESSORMODE_MATCH,
+									}},
+							},
+						},
+					},
+					Sources: []datadogV2.ObservabilityPipelineConfigSourceItem{
+						datadogV2.ObservabilityPipelineConfigSourceItem{
+							ObservabilityPipelineDatadogAgentSource: &datadogV2.ObservabilityPipelineDatadogAgentSource{
+								Id:   "datadog-agent-source",
+								Type: datadogV2.OBSERVABILITYPIPELINEDATADOGAGENTSOURCETYPE_DATADOG_AGENT,
+							}},
+					},
+				},
+				Name: "Pipeline with Dedupe No Cache",
+			},
+			Type: "pipelines",
+		},
+	}
+	ctx := datadog.NewDefaultContext(context.Background())
+	configuration := datadog.NewConfiguration()
+	apiClient := datadog.NewAPIClient(configuration)
+	api := datadogV2.NewObservabilityPipelinesApi(apiClient)
+	resp, r, err := api.CreatePipeline(ctx, body)
+
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error when calling `ObservabilityPipelinesApi.CreatePipeline`: %v\n", err)
+		fmt.Fprintf(os.Stderr, "Full HTTP response: %v\n", r)
+	}
+
+	responseContent, _ := json.MarshalIndent(resp, "", "  ")
+	fmt.Fprintf(os.Stdout, "Response from `ObservabilityPipelinesApi.CreatePipeline`:\n%s\n", responseContent)
+}

tests/scenarios/cassettes/TestScenarios/v2/Feature_Observability_Pipelines/Scenario_Create_a_pipeline_with_dedupe_processor_with_cache_returns_OK_response.freeze

@@ -0,0 +1 @@
+2026-02-09T09:53:31.840Z