Skip to content

Add spike protection processor to Observability Pipelines #4498

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
dd-octo-sts wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Add spike protection processor to Observability Pipelines #4498

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
164 changes: 164 additions & 0 deletions .generator/schemas/v2/openapi.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84707,6 +84707,7 @@ components:
- $ref: "#/components/schemas/ObservabilityPipelineRenameFieldsProcessor"
- $ref: "#/components/schemas/ObservabilityPipelineSampleProcessor"
- $ref: "#/components/schemas/ObservabilityPipelineSensitiveDataScannerProcessor"
- $ref: "#/components/schemas/ObservabilityPipelineSpikeProtectionProcessor"
- $ref: "#/components/schemas/ObservabilityPipelineSplitArrayProcessor"
- $ref: "#/components/schemas/ObservabilityPipelineThrottleProcessor"
- $ref: "#/components/schemas/ObservabilityPipelineAddMetricTagsProcessor"
Expand Down Expand Up @@ -89159,6 +89160,169 @@ components:
- type
- attributes
type: object
ObservabilityPipelineSpikeProtectionProcessor:
description: |-
The `spike_protection` processor learns the normal ingestion volume for the logs that match a specified filter and takes action when the volume spikes unexpectedly.

**Supported pipeline types:** logs
properties:
action_when_missing_partitions:
$ref: "#/components/schemas/ObservabilityPipelineSpikeProtectionProcessorActionWhenMissingPartitions"
display_name:
$ref: "#/components/schemas/ObservabilityPipelineComponentDisplayName"
enabled:
description: Indicates whether the processor is enabled.
example: true
type: boolean
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: "spike-protection-processor"
type: string
include:
description: A Datadog search query used to determine which logs this processor targets.
example: "service:my-service"
type: string
minimal_threshold:
description: The minimum volume, in bytes, that must be observed within 24 hours before spike protection can activate. Defaults to 100 GiB.
example: 107374182400
format: int64
type: integer
minimal_threshold_overrides:
description: A list of per-partition overrides for the minimal threshold, identified by matching field values. Requires `partition_fields` to be set.
items:
$ref: "#/components/schemas/ObservabilityPipelineSpikeProtectionProcessorThresholdOverride"
type: array
name:
description: Name of the spike protection rule.
example: MySpikeProtection
type: string
partition_fields:
description: A list of fields used to segment log traffic. Volume is learned and evaluated independently for each unique combination of these field values. A maximum of five fields is supported.
items:
description: The name of a log field used to partition spike detection.
maxLength: 128
type: string
maxItems: 5
type: array
uniqueItems: true
spike_action:
$ref: "#/components/schemas/ObservabilityPipelineSpikeProtectionProcessorSpikeAction"
spike_threshold:
description: The number of standard deviations above the learned mean volume that must be exceeded before a spike is detected. Defaults to `5`.
example: 5
format: double
type: number
type:
$ref: "#/components/schemas/ObservabilityPipelineSpikeProtectionProcessorType"
required:
- id
- type
- include
- name
- spike_action
- enabled
type: object
x-pipeline-types: [logs]
ObservabilityPipelineSpikeProtectionProcessorActionWhenMissingPartitions:
default: ignore
description: |-
How to handle logs that are missing one of the `partition_fields`. Options:
- `ignore`: Forward the event without applying spike protection.
- `default_bucket`: Assign the event to a default partition.
enum:
- ignore
- default_bucket
example: ignore
type: string
x-enum-varnames:
- IGNORE
- DEFAULT_BUCKET
ObservabilityPipelineSpikeProtectionProcessorMinimalThreshold:
description: The minimal threshold to apply to the matching partition.
oneOf:
- $ref: "#/components/schemas/ObservabilityPipelineSpikeProtectionProcessorMinimalThresholdBytes"
- $ref: "#/components/schemas/ObservabilityPipelineSpikeProtectionProcessorMinimalThresholdAlwaysKeep"
ObservabilityPipelineSpikeProtectionProcessorMinimalThresholdAlwaysKeep:
description: Never activates spike protection for the matching partition, regardless of volume.
properties:
type:
$ref: "#/components/schemas/ObservabilityPipelineSpikeProtectionProcessorMinimalThresholdAlwaysKeepType"
required:
- type
type: object
ObservabilityPipelineSpikeProtectionProcessorMinimalThresholdAlwaysKeepType:
default: always_keep
description: The minimal threshold type. The value should always be `always_keep`.
enum:
- always_keep
example: always_keep
type: string
x-enum-varnames:
- ALWAYS_KEEP
ObservabilityPipelineSpikeProtectionProcessorMinimalThresholdBytes:
description: Sets a specific byte threshold for the matching partition.
properties:
bytes:
description: The minimal threshold, in bytes, for the matching partition.
example: 107374182400
format: int64
type: integer
type:
$ref: "#/components/schemas/ObservabilityPipelineSpikeProtectionProcessorMinimalThresholdBytesType"
required:
- type
- bytes
type: object
ObservabilityPipelineSpikeProtectionProcessorMinimalThresholdBytesType:
default: threshold
description: The minimal threshold type. The value should always be `threshold`.
enum:
- threshold
example: threshold
type: string
x-enum-varnames:
- THRESHOLD
ObservabilityPipelineSpikeProtectionProcessorSpikeAction:
description: |-
The action to take with matching logs while spike protection is active. Options:
- `drop`: Drop the event.
- `no_action`: Let the event pass through.
- `overflow_routing`: Route to an overflow destination.

When `overflow_routing` is used, there must be a destination whose `inputs` reference this processor with the `<processor-id>.overflow_events` suffix to route the overflowing events.
enum:
- drop
- no_action
- overflow_routing
example: drop
type: string
x-enum-varnames:
- DROP
- NO_ACTION
- OVERFLOW_ROUTING
ObservabilityPipelineSpikeProtectionProcessorThresholdOverride:
description: Defines a minimal threshold that applies to a specific partition instead of the processor-wide `minimal_threshold`.
properties:
fields:
description: The set of field matchers that identifies the partition this override applies to. Every field name listed must also appear in `partition_fields`.
items:
$ref: "#/components/schemas/ObservabilityPipelineFieldValue"
type: array
minimal_threshold:
$ref: "#/components/schemas/ObservabilityPipelineSpikeProtectionProcessorMinimalThreshold"
required:
- fields
- minimal_threshold
type: object
ObservabilityPipelineSpikeProtectionProcessorType:
default: spike_protection
description: The processor type. The value should always be `spike_protection`.
enum:
- spike_protection
example: spike_protection
type: string
x-enum-varnames:
- SPIKE_PROTECTION
ObservabilityPipelineSplitArrayProcessor:
description: |-
The `split_array` processor splits array fields into separate events based on configured rules.
Expand Down
145 changes: 145 additions & 0 deletions examples/v2/observability-pipelines/ValidatePipeline_3654691150.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,145 @@
// Validate an observability pipeline with spike protection processor threshold overrides returns
// "OK" response

import com.datadog.api.client.ApiClient;
import com.datadog.api.client.ApiException;
import com.datadog.api.client.v2.api.ObservabilityPipelinesApi;
import com.datadog.api.client.v2.model.ObservabilityPipelineConfig;
import com.datadog.api.client.v2.model.ObservabilityPipelineConfigDestinationItem;
import com.datadog.api.client.v2.model.ObservabilityPipelineConfigProcessorGroup;
import com.datadog.api.client.v2.model.ObservabilityPipelineConfigProcessorItem;
import com.datadog.api.client.v2.model.ObservabilityPipelineConfigSourceItem;
import com.datadog.api.client.v2.model.ObservabilityPipelineDataAttributes;
import com.datadog.api.client.v2.model.ObservabilityPipelineDatadogAgentSource;
import com.datadog.api.client.v2.model.ObservabilityPipelineDatadogAgentSourceType;
import com.datadog.api.client.v2.model.ObservabilityPipelineDatadogLogsDestination;
import com.datadog.api.client.v2.model.ObservabilityPipelineDatadogLogsDestinationType;
import com.datadog.api.client.v2.model.ObservabilityPipelineFieldValue;
import com.datadog.api.client.v2.model.ObservabilityPipelineSpec;
import com.datadog.api.client.v2.model.ObservabilityPipelineSpecData;
import com.datadog.api.client.v2.model.ObservabilityPipelineSpikeProtectionProcessor;
import com.datadog.api.client.v2.model.ObservabilityPipelineSpikeProtectionProcessorActionWhenMissingPartitions;
import com.datadog.api.client.v2.model.ObservabilityPipelineSpikeProtectionProcessorMinimalThreshold;
import com.datadog.api.client.v2.model.ObservabilityPipelineSpikeProtectionProcessorMinimalThresholdAlwaysKeep;
import com.datadog.api.client.v2.model.ObservabilityPipelineSpikeProtectionProcessorMinimalThresholdAlwaysKeepType;
import com.datadog.api.client.v2.model.ObservabilityPipelineSpikeProtectionProcessorMinimalThresholdBytes;
import com.datadog.api.client.v2.model.ObservabilityPipelineSpikeProtectionProcessorMinimalThresholdBytesType;
import com.datadog.api.client.v2.model.ObservabilityPipelineSpikeProtectionProcessorSpikeAction;
import com.datadog.api.client.v2.model.ObservabilityPipelineSpikeProtectionProcessorThresholdOverride;
import com.datadog.api.client.v2.model.ObservabilityPipelineSpikeProtectionProcessorType;
import com.datadog.api.client.v2.model.ValidationResponse;
import java.util.Arrays;
import java.util.Collections;

public class Example {
public static void main(String[] args) {
ApiClient defaultClient = ApiClient.getDefaultApiClient();
defaultClient.setAccessToken(System.getenv("DD_BEARER_TOKEN"));
ObservabilityPipelinesApi apiInstance = new ObservabilityPipelinesApi(defaultClient);

ObservabilityPipelineSpec body =
new ObservabilityPipelineSpec()
.data(
new ObservabilityPipelineSpecData()
.attributes(
new ObservabilityPipelineDataAttributes()
.config(
new ObservabilityPipelineConfig()
.destinations(
Collections.singletonList(
new ObservabilityPipelineConfigDestinationItem(
new ObservabilityPipelineDatadogLogsDestination()
.id("datadog-logs-destination")
.inputs(
Collections.singletonList(
"my-processor-group"))
.type(
ObservabilityPipelineDatadogLogsDestinationType
.DATADOG_LOGS))))
.processorGroups(
Collections.singletonList(
new ObservabilityPipelineConfigProcessorGroup()
.enabled(true)
.id("my-processor-group")
.include("service:my-service")
.inputs(
Collections.singletonList(
"datadog-agent-source"))
.processors(
Collections.singletonList(
new ObservabilityPipelineConfigProcessorItem(
new ObservabilityPipelineSpikeProtectionProcessor()
.enabled(true)
.id("spike-protection-processor")
.include("*")
.type(
ObservabilityPipelineSpikeProtectionProcessorType
.SPIKE_PROTECTION)
.name(
"MyPartitionedSpikeProtection")
.spikeAction(
ObservabilityPipelineSpikeProtectionProcessorSpikeAction
.NO_ACTION)
.partitionFields(
Collections.singletonList(
"service"))
.actionWhenMissingPartitions(
ObservabilityPipelineSpikeProtectionProcessorActionWhenMissingPartitions
.DEFAULT_BUCKET)
.minimalThresholdOverrides(
Arrays.asList(
new ObservabilityPipelineSpikeProtectionProcessorThresholdOverride()
.fields(
Collections
.singletonList(
new ObservabilityPipelineFieldValue()
.name(
"service")
.value(
"checkout")))
.minimalThreshold(
new ObservabilityPipelineSpikeProtectionProcessorMinimalThreshold(
new ObservabilityPipelineSpikeProtectionProcessorMinimalThresholdBytes()
.type(
ObservabilityPipelineSpikeProtectionProcessorMinimalThresholdBytesType
.THRESHOLD)
.bytes(
53687091200L))),
new ObservabilityPipelineSpikeProtectionProcessorThresholdOverride()
.fields(
Collections
.singletonList(
new ObservabilityPipelineFieldValue()
.name(
"service")
.value(
"payments")))
.minimalThreshold(
new ObservabilityPipelineSpikeProtectionProcessorMinimalThreshold(
new ObservabilityPipelineSpikeProtectionProcessorMinimalThresholdAlwaysKeep()
.type(
ObservabilityPipelineSpikeProtectionProcessorMinimalThresholdAlwaysKeepType
.ALWAYS_KEEP))))))))))
.sources(
Collections.singletonList(
new ObservabilityPipelineConfigSourceItem(
new ObservabilityPipelineDatadogAgentSource()
.id("datadog-agent-source")
.type(
ObservabilityPipelineDatadogAgentSourceType
.DATADOG_AGENT)))))
.name("Pipeline with Spike Protection Threshold Overrides"))
.type("pipelines"));

try {
ValidationResponse result = apiInstance.validatePipeline(body);
System.out.println(result);
} catch (ApiException e) {
System.err.println("Exception when calling ObservabilityPipelinesApi#validatePipeline");
System.err.println("Status code: " + e.getCode());
System.err.println("Reason: " + e.getResponseBody());
System.err.println("Response headers: " + e.getResponseHeaders());
e.printStackTrace();
}
}
}
Loading
Loading