Add aggregate event-time configuration to Observability Pipelines

.generator/schemas/v2/openapi.yaml

@@ -82253,6 +82253,8 @@ components:
 
         **Supported pipeline types:** metrics
       properties:
+        aggregation_timing:
+          $ref: "#/components/schemas/ObservabilityPipelineAggregateProcessorAggregationTiming"
         display_name:
           $ref: "#/components/schemas/ObservabilityPipelineComponentDisplayName"
         enabled:
@@ -82287,6 +82289,31 @@ components:
         - enabled
       type: object
       x-pipeline-types: [metrics]
+    ObservabilityPipelineAggregateProcessorAggregationTiming:
+      description: Configures how metrics are assigned to aggregation windows. When omitted, metrics are grouped using system time.
+      properties:
+        allowed_lateness_secs:
+          description: Grace period, in seconds, for late-arriving metrics when using event time. Defaults to 10 seconds when omitted.
+          example: 10
+          format: int64
+          maximum: 3600
+          minimum: 0
+          type: integer
+        type:
+          $ref: "#/components/schemas/ObservabilityPipelineAggregateProcessorAggregationTimingType"
+      required:
+        - type
+      type: object
+    ObservabilityPipelineAggregateProcessorAggregationTimingType:
+      description: Determines whether metrics are assigned to aggregation windows based on when they are processed or their timestamps.
+      enum:
+        - system_time
+        - event_time
+      example: event_time
+      type: string
+      x-enum-varnames:
+        - SYSTEM_TIME
+        - EVENT_TIME
     ObservabilityPipelineAggregateProcessorMode:
       description: The aggregation mode applied to metrics that share the same name and tags within the interval.
       enum:

api/datadogV2/model_observability_pipeline_aggregate_processor.go

@@ -14,6 +14,8 @@ import (
 //
 // **Supported pipeline types:** metrics
 type ObservabilityPipelineAggregateProcessor struct {
+	// Configures how metrics are assigned to aggregation windows. When omitted, metrics are grouped using system time.
+	AggregationTiming *ObservabilityPipelineAggregateProcessorAggregationTiming `json:"aggregation_timing,omitempty"`
 	// The display name for a component.
 	DisplayName *string `json:"display_name,omitempty"`
 	// Indicates whether the processor is enabled.
@@ -58,6 +60,34 @@ func NewObservabilityPipelineAggregateProcessorWithDefaults() *ObservabilityPipe
 	return &this
 }
 
+// GetAggregationTiming returns the AggregationTiming field value if set, zero value otherwise.
+func (o *ObservabilityPipelineAggregateProcessor) GetAggregationTiming() ObservabilityPipelineAggregateProcessorAggregationTiming {
+	if o == nil || o.AggregationTiming == nil {
+		var ret ObservabilityPipelineAggregateProcessorAggregationTiming
+		return ret
+	}
+	return *o.AggregationTiming
+}
+
+// GetAggregationTimingOk returns a tuple with the AggregationTiming field value if set, nil otherwise
+// and a boolean to check if the value has been set.
+func (o *ObservabilityPipelineAggregateProcessor) GetAggregationTimingOk() (*ObservabilityPipelineAggregateProcessorAggregationTiming, bool) {
+	if o == nil || o.AggregationTiming == nil {
+		return nil, false
+	}
+	return o.AggregationTiming, true
+}
+
+// HasAggregationTiming returns a boolean if a field has been set.
+func (o *ObservabilityPipelineAggregateProcessor) HasAggregationTiming() bool {
+	return o != nil && o.AggregationTiming != nil
+}
+
+// SetAggregationTiming gets a reference to the given ObservabilityPipelineAggregateProcessorAggregationTiming and assigns it to the AggregationTiming field.
+func (o *ObservabilityPipelineAggregateProcessor) SetAggregationTiming(v ObservabilityPipelineAggregateProcessorAggregationTiming) {
+	o.AggregationTiming = &v
+}
+
 // GetDisplayName returns the DisplayName field value if set, zero value otherwise.
 func (o *ObservabilityPipelineAggregateProcessor) GetDisplayName() string {
 	if o == nil || o.DisplayName == nil {
@@ -230,6 +260,9 @@ func (o ObservabilityPipelineAggregateProcessor) MarshalJSON() ([]byte, error) {
 	if o.UnparsedObject != nil {
 		return datadog.Marshal(o.UnparsedObject)
 	}
+	if o.AggregationTiming != nil {
+		toSerialize["aggregation_timing"] = o.AggregationTiming
+	}
 	if o.DisplayName != nil {
 		toSerialize["display_name"] = o.DisplayName
 	}
@@ -249,13 +282,14 @@ func (o ObservabilityPipelineAggregateProcessor) MarshalJSON() ([]byte, error) {
 // UnmarshalJSON deserializes the given payload.
 func (o *ObservabilityPipelineAggregateProcessor) UnmarshalJSON(bytes []byte) (err error) {
 	all := struct {
-		DisplayName  *string                                      `json:"display_name,omitempty"`
-		Enabled      *bool                                        `json:"enabled"`
-		Id           *string                                      `json:"id"`
-		Include      *string                                      `json:"include"`
-		IntervalSecs *int64                                       `json:"interval_secs"`
-		Mode         *ObservabilityPipelineAggregateProcessorMode `json:"mode"`
-		Type         *ObservabilityPipelineAggregateProcessorType `json:"type"`
+		AggregationTiming *ObservabilityPipelineAggregateProcessorAggregationTiming `json:"aggregation_timing,omitempty"`
+		DisplayName       *string                                                   `json:"display_name,omitempty"`
+		Enabled           *bool                                                     `json:"enabled"`
+		Id                *string                                                   `json:"id"`
+		Include           *string                                                   `json:"include"`
+		IntervalSecs      *int64                                                    `json:"interval_secs"`
+		Mode              *ObservabilityPipelineAggregateProcessorMode              `json:"mode"`
+		Type              *ObservabilityPipelineAggregateProcessorType              `json:"type"`
 	}{}
 	if err = datadog.Unmarshal(bytes, &all); err != nil {
 		return datadog.Unmarshal(bytes, &o.UnparsedObject)
@@ -280,12 +314,16 @@ func (o *ObservabilityPipelineAggregateProcessor) UnmarshalJSON(bytes []byte) (e
 	}
 	additionalProperties := make(map[string]interface{})
 	if err = datadog.UnmarshalUseNumber(bytes, &additionalProperties); err == nil {
-		datadog.DeleteKeys(additionalProperties, &[]string{"display_name", "enabled", "id", "include", "interval_secs", "mode", "type"})
+		datadog.DeleteKeys(additionalProperties, &[]string{"aggregation_timing", "display_name", "enabled", "id", "include", "interval_secs", "mode", "type"})
 	} else {
 		return err
 	}
 
 	hasInvalidField := false
+	if all.AggregationTiming != nil && all.AggregationTiming.UnparsedObject != nil && o.UnparsedObject == nil {
+		hasInvalidField = true
+	}
+	o.AggregationTiming = all.AggregationTiming
 	o.DisplayName = all.DisplayName
 	o.Enabled = *all.Enabled
 	o.Id = *all.Id

api/datadogV2/model_observability_pipeline_aggregate_processor_aggregation_timing.go

@@ -0,0 +1,146 @@
+// 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"
+)
+
+// ObservabilityPipelineAggregateProcessorAggregationTiming Configures how metrics are assigned to aggregation windows. When omitted, metrics are grouped using system time.
+type ObservabilityPipelineAggregateProcessorAggregationTiming struct {
+	// Grace period, in seconds, for late-arriving metrics when using event time. Defaults to 10 seconds when omitted.
+	AllowedLatenessSecs *int64 `json:"allowed_lateness_secs,omitempty"`
+	// Determines whether metrics are assigned to aggregation windows based on when they are processed or their timestamps.
+	Type ObservabilityPipelineAggregateProcessorAggregationTimingType `json:"type"`
+	// 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:"-"`
+}
+
+// NewObservabilityPipelineAggregateProcessorAggregationTiming instantiates a new ObservabilityPipelineAggregateProcessorAggregationTiming 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 NewObservabilityPipelineAggregateProcessorAggregationTiming(typeVar ObservabilityPipelineAggregateProcessorAggregationTimingType) *ObservabilityPipelineAggregateProcessorAggregationTiming {
+	this := ObservabilityPipelineAggregateProcessorAggregationTiming{}
+	this.Type = typeVar
+	return &this
+}
+
+// NewObservabilityPipelineAggregateProcessorAggregationTimingWithDefaults instantiates a new ObservabilityPipelineAggregateProcessorAggregationTiming 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 NewObservabilityPipelineAggregateProcessorAggregationTimingWithDefaults() *ObservabilityPipelineAggregateProcessorAggregationTiming {
+	this := ObservabilityPipelineAggregateProcessorAggregationTiming{}
+	return &this
+}
+
+// GetAllowedLatenessSecs returns the AllowedLatenessSecs field value if set, zero value otherwise.
+func (o *ObservabilityPipelineAggregateProcessorAggregationTiming) GetAllowedLatenessSecs() int64 {
+	if o == nil || o.AllowedLatenessSecs == nil {
+		var ret int64
+		return ret
+	}
+	return *o.AllowedLatenessSecs
+}
+
+// GetAllowedLatenessSecsOk returns a tuple with the AllowedLatenessSecs field value if set, nil otherwise
+// and a boolean to check if the value has been set.
+func (o *ObservabilityPipelineAggregateProcessorAggregationTiming) GetAllowedLatenessSecsOk() (*int64, bool) {
+	if o == nil || o.AllowedLatenessSecs == nil {
+		return nil, false
+	}
+	return o.AllowedLatenessSecs, true
+}
+
+// HasAllowedLatenessSecs returns a boolean if a field has been set.
+func (o *ObservabilityPipelineAggregateProcessorAggregationTiming) HasAllowedLatenessSecs() bool {
+	return o != nil && o.AllowedLatenessSecs != nil
+}
+
+// SetAllowedLatenessSecs gets a reference to the given int64 and assigns it to the AllowedLatenessSecs field.
+func (o *ObservabilityPipelineAggregateProcessorAggregationTiming) SetAllowedLatenessSecs(v int64) {
+	o.AllowedLatenessSecs = &v
+}
+
+// GetType returns the Type field value.
+func (o *ObservabilityPipelineAggregateProcessorAggregationTiming) GetType() ObservabilityPipelineAggregateProcessorAggregationTimingType {
+	if o == nil {
+		var ret ObservabilityPipelineAggregateProcessorAggregationTimingType
+		return ret
+	}
+	return o.Type
+}
+
+// GetTypeOk returns a tuple with the Type field value
+// and a boolean to check if the value has been set.
+func (o *ObservabilityPipelineAggregateProcessorAggregationTiming) GetTypeOk() (*ObservabilityPipelineAggregateProcessorAggregationTimingType, bool) {
+	if o == nil {
+		return nil, false
+	}
+	return &o.Type, true
+}
+
+// SetType sets field value.
+func (o *ObservabilityPipelineAggregateProcessorAggregationTiming) SetType(v ObservabilityPipelineAggregateProcessorAggregationTimingType) {
+	o.Type = v
+}
+
+// MarshalJSON serializes the struct using spec logic.
+func (o ObservabilityPipelineAggregateProcessorAggregationTiming) MarshalJSON() ([]byte, error) {
+	toSerialize := map[string]interface{}{}
+	if o.UnparsedObject != nil {
+		return datadog.Marshal(o.UnparsedObject)
+	}
+	if o.AllowedLatenessSecs != nil {
+		toSerialize["allowed_lateness_secs"] = o.AllowedLatenessSecs
+	}
+	toSerialize["type"] = o.Type
+
+	for key, value := range o.AdditionalProperties {
+		toSerialize[key] = value
+	}
+	return datadog.Marshal(toSerialize)
+}
+
+// UnmarshalJSON deserializes the given payload.
+func (o *ObservabilityPipelineAggregateProcessorAggregationTiming) UnmarshalJSON(bytes []byte) (err error) {
+	all := struct {
+		AllowedLatenessSecs *int64                                                        `json:"allowed_lateness_secs,omitempty"`
+		Type                *ObservabilityPipelineAggregateProcessorAggregationTimingType `json:"type"`
+	}{}
+	if err = datadog.Unmarshal(bytes, &all); err != nil {
+		return datadog.Unmarshal(bytes, &o.UnparsedObject)
+	}
+	if all.Type == nil {
+		return fmt.Errorf("required field type missing")
+	}
+	additionalProperties := make(map[string]interface{})
+	if err = datadog.UnmarshalUseNumber(bytes, &additionalProperties); err == nil {
+		datadog.DeleteKeys(additionalProperties, &[]string{"allowed_lateness_secs", "type"})
+	} else {
+		return err
+	}
+
+	hasInvalidField := false
+	o.AllowedLatenessSecs = all.AllowedLatenessSecs
+	if !all.Type.IsValid() {
+		hasInvalidField = true
+	} else {
+		o.Type = *all.Type
+	}
+
+	if len(additionalProperties) > 0 {
+		o.AdditionalProperties = additionalProperties
+	}
+
+	if hasInvalidField {
+		return datadog.Unmarshal(bytes, &o.UnparsedObject)
+	}
+
+	return nil
+}

api/datadogV2/model_observability_pipeline_aggregate_processor_aggregation_timing_type.go

@@ -0,0 +1,66 @@
+// 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"
+)
+
+// ObservabilityPipelineAggregateProcessorAggregationTimingType Determines whether metrics are assigned to aggregation windows based on when they are processed or their timestamps.
+type ObservabilityPipelineAggregateProcessorAggregationTimingType string
+
+// List of ObservabilityPipelineAggregateProcessorAggregationTimingType.
+const (
+	OBSERVABILITYPIPELINEAGGREGATEPROCESSORAGGREGATIONTIMINGTYPE_SYSTEM_TIME ObservabilityPipelineAggregateProcessorAggregationTimingType = "system_time"
+	OBSERVABILITYPIPELINEAGGREGATEPROCESSORAGGREGATIONTIMINGTYPE_EVENT_TIME  ObservabilityPipelineAggregateProcessorAggregationTimingType = "event_time"
+)
+
+var allowedObservabilityPipelineAggregateProcessorAggregationTimingTypeEnumValues = []ObservabilityPipelineAggregateProcessorAggregationTimingType{
+	OBSERVABILITYPIPELINEAGGREGATEPROCESSORAGGREGATIONTIMINGTYPE_SYSTEM_TIME,
+	OBSERVABILITYPIPELINEAGGREGATEPROCESSORAGGREGATIONTIMINGTYPE_EVENT_TIME,
+}
+
+// GetAllowedValues reeturns the list of possible values.
+func (v *ObservabilityPipelineAggregateProcessorAggregationTimingType) GetAllowedValues() []ObservabilityPipelineAggregateProcessorAggregationTimingType {
+	return allowedObservabilityPipelineAggregateProcessorAggregationTimingTypeEnumValues
+}
+
+// UnmarshalJSON deserializes the given payload.
+func (v *ObservabilityPipelineAggregateProcessorAggregationTimingType) UnmarshalJSON(src []byte) error {
+	var value string
+	err := datadog.Unmarshal(src, &value)
+	if err != nil {
+		return err
+	}
+	*v = ObservabilityPipelineAggregateProcessorAggregationTimingType(value)
+	return nil
+}
+
+// NewObservabilityPipelineAggregateProcessorAggregationTimingTypeFromValue returns a pointer to a valid ObservabilityPipelineAggregateProcessorAggregationTimingType
+// for the value passed as argument, or an error if the value passed is not allowed by the enum.
+func NewObservabilityPipelineAggregateProcessorAggregationTimingTypeFromValue(v string) (*ObservabilityPipelineAggregateProcessorAggregationTimingType, error) {
+	ev := ObservabilityPipelineAggregateProcessorAggregationTimingType(v)
+	if ev.IsValid() {
+		return &ev, nil
+	}
+	return nil, fmt.Errorf("invalid value '%v' for ObservabilityPipelineAggregateProcessorAggregationTimingType: valid values are %v", v, allowedObservabilityPipelineAggregateProcessorAggregationTimingTypeEnumValues)
+}
+
+// IsValid return true if the value is valid for the enum, false otherwise.
+func (v ObservabilityPipelineAggregateProcessorAggregationTimingType) IsValid() bool {
+	for _, existing := range allowedObservabilityPipelineAggregateProcessorAggregationTimingTypeEnumValues {
+		if existing == v {
+			return true
+		}
+	}
+	return false
+}
+
+// Ptr returns reference to ObservabilityPipelineAggregateProcessorAggregationTimingType value.
+func (v ObservabilityPipelineAggregateProcessorAggregationTimingType) Ptr() *ObservabilityPipelineAggregateProcessorAggregationTimingType {
+	return &v
+}