Adding custom mapper support to Observability Pipelines OCSF Mapper

.generator/schemas/v2/openapi.yaml

@@ -41365,6 +41365,7 @@ components:
       example: CloudTrail Account Change
       oneOf:
       - $ref: '#/components/schemas/ObservabilityPipelineOcsfMappingLibrary'
+      - $ref: '#/components/schemas/ObservabilityPipelineOcsfMappingCustom'
     ObservabilityPipelineOcsfMapperProcessorType:
       default: ocsf_mapper
       description: The processor type. The value should always be `ocsf_mapper`.
@@ -41374,6 +41375,116 @@ components:
       type: string
       x-enum-varnames:
       - OCSF_MAPPER
+    ObservabilityPipelineOcsfMappingCustom:
+      description: Custom OCSF mapping configuration for transforming logs.
+      properties:
+        mapping:
+          description: A list of field mapping rules for transforming log fields to
+            OCSF schema fields.
+          items:
+            $ref: '#/components/schemas/ObservabilityPipelineOcsfMappingCustomFieldMapping'
+          type: array
+        metadata:
+          $ref: '#/components/schemas/ObservabilityPipelineOcsfMappingCustomMetadata'
+        version:
+          description: The version of the custom mapping configuration.
+          example: 1
+          format: int64
+          type: integer
+      required:
+      - mapping
+      - metadata
+      - version
+      type: object
+    ObservabilityPipelineOcsfMappingCustomFieldMapping:
+      description: Defines a single field mapping rule for transforming a source field
+        to an OCSF destination field.
+      properties:
+        default:
+          description: The default value to use if the source field is missing or
+            empty.
+          example: ''
+        dest:
+          description: The destination OCSF field path.
+          example: device.type
+          type: string
+        lookup:
+          $ref: '#/components/schemas/ObservabilityPipelineOcsfMappingCustomLookup'
+        source:
+          description: The source field path from the log event.
+          example: host.type
+        sources:
+          description: Multiple source field paths for combined mapping.
+          example:
+          - field1
+          - field2
+        value:
+          description: A static value to use for the destination field.
+          example: static_value
+      required:
+      - dest
+      type: object
+    ObservabilityPipelineOcsfMappingCustomLookup:
+      description: Lookup table configuration for mapping source values to destination
+        values.
+      properties:
+        default:
+          description: The default value to use if no lookup match is found.
+          example: unknown
+        table:
+          description: A list of lookup table entries for value transformation.
+          items:
+            $ref: '#/components/schemas/ObservabilityPipelineOcsfMappingCustomLookupTableEntry'
+          type: array
+      type: object
+    ObservabilityPipelineOcsfMappingCustomLookupTableEntry:
+      description: A single entry in a lookup table for value transformation.
+      properties:
+        contains:
+          description: The substring to match in the source value.
+          example: Desktop
+          type: string
+        equals:
+          description: The exact value to match in the source.
+          example: desktop
+        equals_source:
+          description: The source field to match against.
+          example: device_type
+          type: string
+        matches:
+          description: A regex pattern to match in the source value.
+          example: ^Desktop.*
+          type: string
+        not_matches:
+          description: A regex pattern that must not match the source value.
+          example: ^Mobile.*
+          type: string
+        value:
+          description: The value to use when a match is found.
+          example: desktop
+      type: object
+    ObservabilityPipelineOcsfMappingCustomMetadata:
+      description: Metadata for the custom OCSF mapping.
+      properties:
+        class:
+          description: The OCSF event class name.
+          example: Device Inventory Info
+          type: string
+        profiles:
+          description: A list of OCSF profiles to apply.
+          example:
+          - container
+          items:
+            type: string
+          type: array
+        version:
+          description: The OCSF schema version.
+          example: 1.3.0
+          type: string
+      required:
+      - class
+      - version
+      type: object
     ObservabilityPipelineOcsfMappingLibrary:
       description: Predefined library mappings for common log formats.
       enum:

cassettes/features/v2/observability_pipelines/Validate-an-observability-pipeline-with-OCSF-mapper-custom-mapping-returns-OK-response.frozen

@@ -0,0 +1 @@
+2026-02-10T14:12:05.668Z

cassettes/features/v2/observability_pipelines/Validate-an-observability-pipeline-with-OCSF-mapper-invalid-custom-mapping-returns-Bad-Request-response.frozen

@@ -0,0 +1 @@
+2026-02-10T14:12:06.064Z

cassettes/features/v2/observability_pipelines/Validate-an-observability-pipeline-with-OCSF-mapper-library-mapping-returns-OK-response.frozen

@@ -0,0 +1 @@
+2026-02-10T14:12:05.285Z

examples/v2/observability-pipelines/ValidatePipeline_3024756866.rb

@@ -0,0 +1,89 @@
+# Validate an observability pipeline with OCSF mapper custom mapping 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::ObservabilityPipelineDatadogLogsDestination.new({
+            id: "datadog-logs-destination",
+            inputs: [
+              "my-processor-group",
+            ],
+            type: DatadogAPIClient::V2::ObservabilityPipelineDatadogLogsDestinationType::DATADOG_LOGS,
+          }),
+        ],
+        processor_groups: [
+          DatadogAPIClient::V2::ObservabilityPipelineConfigProcessorGroup.new({
+            enabled: true,
+            id: "my-processor-group",
+            include: "service:my-service",
+            inputs: [
+              "datadog-agent-source",
+            ],
+            processors: [
+              DatadogAPIClient::V2::ObservabilityPipelineOcsfMapperProcessor.new({
+                enabled: true,
+                id: "ocsf-mapper-processor",
+                include: "service:my-service",
+                mappings: [
+                  DatadogAPIClient::V2::ObservabilityPipelineOcsfMapperProcessorMapping.new({
+                    include: "source:custom",
+                    mapping: DatadogAPIClient::V2::ObservabilityPipelineOcsfMappingCustom.new({
+                      mapping: [
+                        DatadogAPIClient::V2::ObservabilityPipelineOcsfMappingCustomFieldMapping.new({
+                          default: "",
+                          dest: "time",
+                          source: "timestamp",
+                        }),
+                        DatadogAPIClient::V2::ObservabilityPipelineOcsfMappingCustomFieldMapping.new({
+                          default: "",
+                          dest: "severity",
+                          source: "level",
+                        }),
+                        DatadogAPIClient::V2::ObservabilityPipelineOcsfMappingCustomFieldMapping.new({
+                          default: "",
+                          dest: "device.type",
+                          lookup: DatadogAPIClient::V2::ObservabilityPipelineOcsfMappingCustomLookup.new({
+                            table: [
+                              DatadogAPIClient::V2::ObservabilityPipelineOcsfMappingCustomLookupTableEntry.new({
+                                contains: "Desktop",
+                                value: "desktop",
+                              }),
+                            ],
+                          }),
+                          source: "host.type",
+                        }),
+                      ],
+                      metadata: DatadogAPIClient::V2::ObservabilityPipelineOcsfMappingCustomMetadata.new({
+                        _class: "Device Inventory Info",
+                        profiles: [
+                          "container",
+                        ],
+                        version: "1.3.0",
+                      }),
+                      version: 1,
+                    }),
+                  }),
+                ],
+                type: DatadogAPIClient::V2::ObservabilityPipelineOcsfMapperProcessorType::OCSF_MAPPER,
+              }),
+            ],
+          }),
+        ],
+        sources: [
+          DatadogAPIClient::V2::ObservabilityPipelineDatadogAgentSource.new({
+            id: "datadog-agent-source",
+            type: DatadogAPIClient::V2::ObservabilityPipelineDatadogAgentSourceType::DATADOG_AGENT,
+          }),
+        ],
+      }),
+      name: "OCSF Custom Mapper Pipeline",
+    }),
+    type: "pipelines",
+  }),
+})
+p api_instance.validate_pipeline(body)

examples/v2/observability-pipelines/ValidatePipeline_3565101276.rb

@@ -0,0 +1,55 @@
+# Validate an observability pipeline with OCSF mapper library mapping 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::ObservabilityPipelineDatadogLogsDestination.new({
+            id: "datadog-logs-destination",
+            inputs: [
+              "my-processor-group",
+            ],
+            type: DatadogAPIClient::V2::ObservabilityPipelineDatadogLogsDestinationType::DATADOG_LOGS,
+          }),
+        ],
+        processor_groups: [
+          DatadogAPIClient::V2::ObservabilityPipelineConfigProcessorGroup.new({
+            enabled: true,
+            id: "my-processor-group",
+            include: "service:my-service",
+            inputs: [
+              "datadog-agent-source",
+            ],
+            processors: [
+              DatadogAPIClient::V2::ObservabilityPipelineOcsfMapperProcessor.new({
+                enabled: true,
+                id: "ocsf-mapper-processor",
+                include: "service:my-service",
+                type: DatadogAPIClient::V2::ObservabilityPipelineOcsfMapperProcessorType::OCSF_MAPPER,
+                mappings: [
+                  DatadogAPIClient::V2::ObservabilityPipelineOcsfMapperProcessorMapping.new({
+                    include: "source:cloudtrail",
+                    mapping: DatadogAPIClient::V2::ObservabilityPipelineOcsfMappingLibrary::CLOUDTRAIL_ACCOUNT_CHANGE,
+                  }),
+                ],
+              }),
+            ],
+          }),
+        ],
+        sources: [
+          DatadogAPIClient::V2::ObservabilityPipelineDatadogAgentSource.new({
+            id: "datadog-agent-source",
+            type: DatadogAPIClient::V2::ObservabilityPipelineDatadogAgentSourceType::DATADOG_AGENT,
+          }),
+        ],
+      }),
+      name: "OCSF Mapper Pipeline",
+    }),
+    type: "pipelines",
+  }),
+})
+p api_instance.validate_pipeline(body)

features/v2/observability_pipelines.feature

@@ -167,3 +167,26 @@ Feature: Observability 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 OCSF mapper custom mapping returns "OK" response
+    Given new "ValidatePipeline" request
+    And body with value {"data": {"attributes": {"config": {"destinations": [{"id": "datadog-logs-destination", "inputs": ["my-processor-group"], "type": "datadog_logs"}], "processor_groups": [{"enabled": true, "id": "my-processor-group", "include": "service:my-service", "inputs": ["datadog-agent-source"], "processors": [{"enabled": true, "id": "ocsf-mapper-processor", "include": "service:my-service", "mappings": [{"include": "source:custom", "mapping": {"mapping": [{"default": "", "dest": "time", "source": "timestamp"}, {"default": "", "dest": "severity", "source": "level"}, {"default": "", "dest": "device.type", "lookup": {"table": [{"contains": "Desktop", "value": "desktop"}]}, "source": "host.type"}], "metadata": {"class": "Device Inventory Info", "profiles": ["container"], "version": "1.3.0"}, "version": 1}}], "type": "ocsf_mapper"}]}], "sources": [{"id": "datadog-agent-source", "type": "datadog_agent"}]}, "name": "OCSF Custom Mapper Pipeline"}, "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 OCSF mapper invalid custom mapping returns "Bad Request" response
+    Given new "ValidatePipeline" request
+    And body with value {"data": {"attributes": {"config": {"destinations": [{"id": "datadog-logs-destination", "inputs": ["my-processor-group"], "type": "datadog_logs"}], "processor_groups": [{"enabled": true, "id": "my-processor-group", "include": "service:my-service", "inputs": ["datadog-agent-source"], "processors": [{"enabled": true, "id": "ocsf-mapper-processor", "include": "service:my-service", "mappings": [{"include": "source:custom", "mapping": {"mapping": [{"dest": "time", "source": "timestamp"}], "metadata": {"class": "Invalid Class", "profiles": ["container"], "version": "1.3.0"}, "version": 0}}], "type": "ocsf_mapper"}]}], "sources": [{"id": "datadog-agent-source", "type": "datadog_agent"}]}, "name": "OCSF Invalid Mapper Pipeline"}, "type": "pipelines"}}
+    When the request is sent
+    Then the response status is 400 Bad Request
+
+  @team:DataDog/observability-pipelines
+  Scenario: Validate an observability pipeline with OCSF mapper library mapping returns "OK" response
+    Given new "ValidatePipeline" request
+    And body with value {"data": {"attributes": {"config": {"destinations": [{"id": "datadog-logs-destination", "inputs": ["my-processor-group"], "type": "datadog_logs"}], "processor_groups": [{"enabled": true, "id": "my-processor-group", "include": "service:my-service", "inputs": ["datadog-agent-source"], "processors": [{"enabled": true, "id": "ocsf-mapper-processor", "include": "service:my-service", "type": "ocsf_mapper", "mappings": [{"include": "source:cloudtrail", "mapping": "CloudTrail Account Change"}]}]}], "sources": [{"id": "datadog-agent-source", "type": "datadog_agent"}]}, "name": "OCSF Mapper Pipeline"}, "type": "pipelines"}}
+    When the request is sent
+    Then the response status is 200 OK
+    And the response "errors" has length 0