1. Optionally, toggle the switch to enable Buffering Options.
   Note: Buffering options is in Preview. Contact your account manager to request access.
   • If left disabled, the maximum size for buffering is 500 events.
   • If enabled:
     1. Select the buffer type you want to set (Memory or Disk).
     2. Enter the buffer size and select the unit.

The following fields are optional:

1. Enter the name of the Splunk index you want your data in. This has to be an allowed index for your HEC. See template syntax if you want to route logs to different indexes based on specific fields in your logs.
2. Select whether the timestamp should be auto-extracted. If set to true, Splunk extracts the timestamp from the message with the expected format of yyyy-mm-dd hh:mm:ss.
3. Optionally, set the sourcetype to override Splunk’s default value, which is httpevent for HEC data. See template syntax if you want to route logs to different source types based on specific fields in your logs.
4. Optionally, toggle the switch to enable Buffering Options.
   Note: Buffering options is in Preview. Contact your account manager to request access.
   • If left disabled, the maximum size for buffering is 500 events.
   • If enabled:
     1. Select the buffer type you want to set (Memory or Disk).
     2. Enter the buffer size and select the unit.

The following fields are optional:

1. In the Encoding dropdown menu, select whether you want to encode your pipeline’s output in JSON, Logfmt, or Raw text. If no decoding is selected, the decoding defaults to JSON.
2. Enter a source name to override the default name value configured for your Sumo Logic collector’s source.
3. Enter a host name to override the default host value configured for your Sumo Logic collector’s source.
4. Enter a category name to override the default category value configured for your Sumo Logic collector’s source.
5. Click Add Header to add any custom header fields and values.
6. Optionally, toggle the switch to enable Buffering Options.
   Note: Buffering options is in Preview. Contact your account manager to request access.
   • If left disabled, the maximum size for buffering is 500 events.
   • If enabled:
     1. Select the buffer type you want to set (Memory or Disk).
     2. Enter the buffer size and select the unit.

The rsyslog and syslog-ng destinations match these log fields to the following Syslog fields:

The following destination settings are optional:

1. Toggle the switch to enable TLS. If you enable TLS, the following certificate and key files are required:
   • Server Certificate Path: The path to the certificate file that has been signed by your Certificate Authority (CA) Root File in DER or PEM (X.509).
   • CA Certificate Path: The path to the certificate file that is your Certificate Authority (CA) Root File in DER or PEM (X.509).
   • Private Key Path: The path to the .key private key file that belongs to your Server Certificate Path in DER or PEM (PKCS#8) format.
2. Enter the number of seconds to wait before sending TCP keepalive probes on an idle connection.
3. Optionally, toggle the switch to enable Buffering Options.
   Note: Buffering options is in Preview. Contact your account manager to request access.
   • If left disabled, the maximum size for buffering is 500 events.
   • If enabled:
     1. Select the buffer type you want to set (Memory or Disk).
     2. Enter the buffer size and select the unit.

To set up the Worker’s Google Chronicle destination:

1. Enter the customer ID for your Google Chronicle instance.
2. If you have a credentials JSON file, enter the path to your credentials JSON file. The credentials file must be placed under DD_OP_DATA_DIR/config. Alternatively, you can use the GOOGLE_APPLICATION_CREDENTIALS environment variable to provide the credential path.
   • If you’re using workload identity on Google Kubernetes Engine (GKE), the GOOGLE_APPLICATION_CREDENTIALS is provided for you.
   • The Worker uses standard Google authentication methods.
3. Select JSON or Raw encoding in the dropdown menu.
4. Enter the log type. See template syntax if you want to route logs to different log types based on specific fields in your logs.
5. Optionally, toggle the switch to enable Buffering Options.
   Note: Buffering options is in Preview. Contact your account manager to request access.
   • If left disabled, the maximum size for buffering is 500 events.
   • If enabled:
     1. Select the buffer type you want to set (Memory or Disk).
     2. Enter the buffer size and select the unit.

Note: Logs sent to the Google Chronicle destination must have ingestion labels. For example, if the logs are from a A10 load balancer, it must have the ingestion label A10_LOAD_BALANCER. See Google Cloud’s Support log types with a default parser for a list of available log types and their respective ingestion labels.

The following fields are optional:

1. Enter the name for the Elasticsearch index. See template syntax if you want to route logs to different indexes based on specific fields in your logs.
2. Enter the Elasticsearch version.
3. Optionally, toggle the switch to enable Buffering Options.
   Note: Buffering options is in Preview. Contact your account manager to request access.
   • If left disabled, the maximum size for buffering is 500 events.
   • If enabled:
     1. Select the buffer type you want to set (Memory or Disk).
     2. Enter the buffer size and select the unit.

1. Optionally, enter the name of the OpenSearch index. See template syntax if you want to route logs to different indexes based on specific fields in your logs.
2. Optionally, toggle the switch to enable Buffering Options.
   Note: Buffering options is in Preview. Contact your account manager to request access.
   • If left disabled, the maximum size for buffering is 500 events.
   • If enabled:
     1. Select the buffer type you want to set (Memory or Disk).
     2. Enter the buffer size and select the unit.

1. Optionally, enter the name of the Amazon OpenSearch index. See template syntax if you want to route logs to different indexes based on specific fields in your logs.
2. Select an authentication strategy, Basic or AWS. For AWS, enter the AWS region.
3. Optionally, toggle the switch to enable Buffering Options.
   Note: Buffering options is in Preview. Contact your account manager to request access.
   • If left disabled, the maximum size for buffering is 500 events.
   • If enabled:
     1. Select the buffer type you want to set (Memory or Disk).
     2. Enter the buffer size and select the unit.

This processor filters for logs that match the specified filter query and drops all non-matching logs. If a log is dropped at this processor, then none of the processors below this one receives that log. This processor can filter out unnecessary logs, such as debug or warning logs.

To set up the filter processor:

• Define a filter query. The query you specify filters for and passes on only logs that match it, dropping all other logs.

The remap processor can add, drop, or rename fields within your individual log data. Use this processor to enrich your logs with additional context, remove low-value fields to reduce volume, and standardize naming across important attributes. Select add field, drop field, or rename field in the dropdown menu to get started.

See the Remap Reserved Attributes guide on how to use the Edit Fields processor to remap attributes.

Add field

Use add field to append a new key-value field to your log.

To set up the add field processor:

1. Define a filter query. Only logs that match the specified filter query are processed. All logs, regardless of whether they do or do not match the filter query, are sent to the next step in the pipeline.
2. Enter the field and value you want to add. To specify a nested field for your key, use the path notation: <OUTER_FIELD>.<INNER_FIELD>. All values are stored as strings. Note: If the field you want to add already exists, the Worker throws an error and the existing field remains unchanged.

Drop field

Use drop field to drop a field from logging data that matches the filter you specify below. It can delete objects, so you can use the processor to drop nested keys.

To set up the drop field processor:

1. Define a filter query. Only logs that match the specified filter query are processed. All logs, regardless of whether they do or do not match the filter query, are sent to the next step in the pipeline.
2. Enter the key of the field you want to drop. To specify a nested field for your specified key, use the path notation: <OUTER_FIELD>.<INNER_FIELD>. Note: If your specified key does not exist, your log will be unimpacted.

Rename field

Use rename field to rename a field within your log.

To set up the rename field processor:

1. Define a filter query. Only logs that match the specified filter query are processed. All logs, regardless of whether they do or do not match the filter query, are sent to the next step in the pipeline.
2. Enter the name of the field you want to rename in the Source field. To specify a nested field for your key, use the path notation: <OUTER_FIELD>.<INNER_FIELD>. Once renamed, your original field is deleted unless you enable the Preserve source tag checkbox described below.
   Note: If the source key you specify doesn’t exist, a default null value is applied to your target.
3. In the Target field, enter the name you want the source field to be renamed to. To specify a nested field for your specified key, use the path notation: <OUTER_FIELD>.<INNER_FIELD>.
   Note: If the target field you specify already exists, the Worker throws an error and does not overwrite the existing target field.
4. Optionally, check the Preserve source tag box if you want to retain the original source field and duplicate the information from your source key to your specified target key. If this box is not checked, the source key is dropped after it is renamed.

Path notation example

For the following message structure:

{
    "outer_key": {
        "inner_key": "inner_value",
        "a": {
            "double_inner_key": "double_inner_value",
            "b": "b value"
        },
        "c": "c value"
    },
    "d": "d value"
}

• Use outer_key.inner_key to refer to the key with the value inner_value.
• Use outer_key.inner_key.double_inner_key to refer to the key with the value double_inner_value.

This processor samples your logging traffic for a representative subset at the rate that you define, dropping the remaining logs. As an example, you can use this processor to sample 20% of logs from a noisy non-critical service.

The sampling only applies to logs that match your filter query and does not impact other logs. If a log is dropped at this processor, none of the processors below receives that log.

To set up the sample processor:

1. Define a filter query. Only logs that match the specified filter query are sampled at the specified retention rate below. The sampled logs and the logs that do not match the filter query are sent to the next step in the pipeline.
2. Enter your desired sampling rate in the Retain field. For example, entering 2 means 2% of logs are retained out of all the logs that match the filter query.
3. Optionally, enter a Group By field to create separate sampling groups for each unique value for that field. For example, status:error and status:info are two unique field values. Each bucket of events with the same field is sampled independently. Click Add Field if you want to add more fields to partition by. See the group-by example.

Group-by example

If you have the following setup for the sample processor:

• Filter query: env:staging
• Retain: 40% of matching logs
• Group by: status and host

Then, 40% of logs for each unique combination of status and service from env:staging is retained. For example:

• 40% of logs with status:info and service:networks are retained.
• 40% of logs with status:info and service:core-web are retained.
• 40% of logs with status:error and service:networks are retained.
• 40% of logs with status:error and service:core-web are retained.

This processor parses logs using the grok parsing rules that are available for a set of sources. The rules are automatically applied to logs based on the log source. Therefore, logs must have a source field with the source name. If this field is not added when the log is sent to the Observability Pipelines Worker, you can use the Add field processor to add it.

If the source field of a log matches one of the grok parsing rule sets, the log’s message field is checked against those rules. If a rule matches, the resulting parsed data is added in the message field as a JSON object, overwriting the original message.

If there isn’t a source field on the log, or no rule matches the log message, then no changes are made to the log and it is sent to the next step in the pipeline.

Datadog’s Grok patterns differ from the standard Grok pattern, where Datadog’s Grok implementation provides:

• Matchers that include options for how you define parsing rules
• Filters for post-processing of extracted data
• A set of built-in patterns tailored to common log formats

See Parsing for more information on Datadog’s Grok patterns.

To set up the grok parser, define a filter query. Only logs that match the specified filter query are processed. All logs, regardless of whether they match the filter query, are sent to the next step in the pipeline.

To test log samples for out-of-the-box rules:

1. Click the Preview Library Rules button.
2. Search or select a source in the dropdown menu.
3. Enter a log sample to test the parsing rules for that source.

To add a custom parsing rule:

1. Click Add Custom Rule.
2. If you want to clone a library rule, select Clone library rule and then the library source from the dropdown menu.
3. If you want to create a custom rule, select Custom and then enter the source. The parsing rules are applied to logs with that source.
4. Enter log samples to test the parsing rules.
5. Enter the rules for parsing the logs. See Parsing for more information on writing parsing rules with Datadog Grok patterns.
   Note: The url, useragent, and csv filters are not available.
6. Click Advanced Settings if you want to add helper rules. See Using helper rules to factorize multiple parsing rules for more information.
7. Click Add Rule.

The quota processor measures the logging traffic for logs that match the filter you specify. When the configured daily quota is met inside the 24-hour rolling window, the processor can either keep or drop additional logs, or send them to a storage bucket. For example, you can configure this processor to drop new logs or trigger an alert without dropping logs after the processor has received 10 million events from a certain service in the last 24 hours.

You can also use field-based partitioning, such as service, env, status. Each unique fields uses a separate quota bucket with its own daily quota limit. See Partition example for more information.

Notes:

• Each pipeline can have up to 1000 buckets. If you need to increase the bucket limit, contact your account manager.
• The pipeline uses the name of the quota to identify the quota across multiple Remote Configuration deployments of the Worker.

To set up the quota processor:

1. Enter a name for the quota processor.
2. Define a filter query. Only logs that match the specified filter query are counted towards the daily limit.
   • Logs that match the quota filter and are within the daily quota are sent to the next step in the pipeline.
   • Logs that do not match the quota filter are sent to the next step of the pipeline.
3. In the Unit for quota dropdown menu, select if you want to measure the quota by the number of Events or by the Volume in bytes.
4. Set the daily quota limit and select the unit of magnitude for your desired quota.
5. Optional, Click Add Field if you want to set a quota on a specific service or region field.
   a. Enter the field name you want to partition by. See the Partition example for more information.
   i. Select the Ignore when missing if you want the quota applied only to events that match the partition. See the Ignore when missing example for more information.
   ii. Optional: Click Overrides if you want to set different quotas for the partitioned field.
   - Click Download as CSV for an example of how to structure the CSV.
   - Drag and drop your overrides CSV to upload it. You can also click Browse to select the file to upload it. See the Overrides example for more information.
   b. Click Add Field if you want to add another partition.
6. In the When quota is met dropdown menu, select if you want to drop events, keep events, or send events to overflow destination, when the quota has been met.
   1. If you select send events to overflow destination, an overflow destination is added with the following cloud storage options: Amazon S3, Azure Blob, and Google Cloud.
   2. Select the cloud storage you want to send overflow logs to. See the setup instructions for your cloud storage: Amazon S3, Azure Blob Storage, or Google Cloud Storage.

Examples

Partition example

Use Partition by if you want to set a quota on a specific service or region. For example, if you want to set a quota for 10 events per day and group the events by the service field, enter service into the Partition by field.

Example for the “ignore when missing” option

Select Ignore when missing if you want the quota applied only to events that match the partition. For example, if the Worker receives the following set of events:

{"service":"a", "source":"foo", "message": "..."}
{"service":"b", "source":"bar", "message": "..."}
{"service":"b", "message": "..."}
{"source":"redis", "message": "..."}
{"message": "..."}

And the Ignore when missing is selected, then the Worker:

• creates a set for logs with service:a and source:foo
• creates a set for logs with service:b and source:bar
• ignores the last three events

The quota is applied to the two sets of logs and not to the last three events.

If the Ignore when missing is not selected, the quota is applied to all five events.

Overrides example

If you are partitioning by service and have two services: a and b, you can use overrides to apply different quotas for them. For example, if you want service:a to have a quota limit of 5,000 bytes and service:b to have a limit of 50 events, the override rules look like this:

The reduce processor groups multiple log events into a single log, based on the fields specified and the merge strategies selected. Logs are grouped at 10-second intervals. After the interval has elapsed for the group, the reduced log for that group is sent to the next step in the pipeline.

To set up the reduce processor:

1. Define a filter query. Only logs that match the specified filter query are processed. Reduced logs and logs that do not match the filter query are sent to the next step in the pipeline.
2. In the Group By section, enter the field you want to group the logs by.
3. Click Add Group by Field to add additional fields.
4. In the Merge Strategy section:
   • In On Field, enter the name of the field you want to merge the logs on.
   • Select the merge strategy in the Apply dropdown menu. This is the strategy used to combine events. See the following Merge strategies section for descriptions of the available strategies.
   • Click Add Merge Strategy to add additional strategies.

Merge strategies

These are the available merge strategies for combining log events.

The deduplicate processor removes copies of data to reduce volume and noise. It caches 5,000 messages at a time and compares your incoming logs traffic against the cached messages. For example, this processor can be used to keep only unique warning logs in the case where multiple identical warning logs are sent in succession.

To set up the deduplicate processor:

1. Define a filter query. Only logs that match the specified filter query are processed. Deduped logs and logs that do not match the filter query are sent to the next step in the pipeline.
2. In the Type of deduplication dropdown menu, select whether you want to Match on or Ignore the fields specified below.
   • If Match is selected, then after a log passes through, future logs that have the same values for all of the fields you specify below are removed.
   • If Ignore is selected, then after a log passes through, future logs that have the same values for all of their fields, except the ones you specify below, are removed.
3. Enter the fields you want to match on, or ignore. At least one field is required, and you can specify a maximum of three fields.
   • Use the path notation <OUTER_FIELD>.<INNER_FIELD> to match subfields. See the Path notation example below.
4. Click Add field to add additional fields you want to filter on.

Path notation example

For the following message structure:

{
    "outer_key": {
        "inner_key": "inner_value",
        "a": {
            "double_inner_key": "double_inner_value",
            "b": "b value"
        },
        "c": "c value"
    },
    "d": "d value"
}

• Use outer_key.inner_key to refer to the key with the value inner_value.
• Use outer_key.inner_key.double_inner_key to refer to the key with the value double_inner_value.

The Sensitive Data Scanner processor scans logs to detect and redact or hash sensitive information such as PII, PCI, and custom sensitive data. You can pick from Datadog’s library of predefined rules, or input custom Regex rules to scan for sensitive data.

To set up the processor:

1. Define a filter query. Only logs that match the specified filter query are scanned and processed. All logs are sent to the next step in the pipeline, regardless of whether they match the filter query.
2. Click Add Scanning Rule.
3. Select one of the following:

This processor adds a field with the name of the host that sent the log. For example, hostname: 613e197f3526. Note: If the hostname already exists, the Worker throws an error and does not overwrite the existing hostname.

To set up this processor:

• Define a filter query. Only logs that match the specified filter query are processed. All logs, regardless of whether they do or do not match the filter query, are sent to the next step in the pipeline.

This processor parses the specified JSON field into objects. For example, if you have a message field that contains stringified JSON:

{
    "foo": "bar",
    "team": "my-team",
    "message": "{\"level\":\"info\",\"timestamp\":\"2024-01-15T10:30:00Z\",\"service\":\"user-service\",\"user_id\":\"12345\",\"action\":\"login\",\"success\":true,\"ip_address\":\"192.168.1.100\"}"
    "app_id":"streaming-services",
    "ddtags": [
    "kube_service:my-service",
    "k8_deployment :your-host"
    ]
}

Use the Parse JSON processor to parse the message field so the message field has all the attributes within a nested object.

This output contains the message field with the parsed JSON:

{
    "foo": "bar",
    "team": "my-team",
    "message": {
        "action": "login",
        "ip_address": "192.168.1.100",
        "level": "info",
        "service": "user-service",
        "success": true,
        "timestamp": "2024-01-15T10:30:00Z",
        "user_id": "12345"
    }
    "app_id":"streaming-services",
    "ddtags": [
    "kube_service:my-service",
    "k8_deployment :your-host"
    ]
}

To set up this processor:

1. Define a filter query. Only logs that match the specified filter query are processed. All logs, regardless of whether they do or do not match the filter query, are sent to the next step in the pipeline.
2. Enter the name of the field you want to parse JSON on.
   Note: The parsed JSON overwrites what was originally contained in the field.

Use this processor to enrich your logs with information from a reference table, which could be a local file or database.

To set up the enrichment table processor:

1. Define a filter query. Only logs that match the specified filter query are processed. All logs, regardless of whether they do or do not match the filter query, are sent to the next step in the pipeline.
2. Enter the source attribute of the log. The source attribute’s value is what you want to find in the reference table.
3. Enter the target attribute. The target attribute’s value stores, as a JSON object, the information found in the reference table.
4. Select the type of reference table you want to use, File or GeoIP.
   • For the File type:
     1. Enter the file path.
        Note: All file paths are made relative to the configuration data directory, which is /var/lib/observability-pipelines-worker/config/ by default. See Advanced Configurations for more information. The file must be owned by the observability-pipelines-worker group and observability-pipelines-worker user, or at least readable by the group or user.
     2. Enter the column name. The column name in the enrichment table is used for matching the source attribute value. See the Enrichment file example.
   • For the GeoIP type, enter the GeoIP path.

Enrichment file example

For this example, merchant_id is used as the source attribute and merchant_info as the target attribute.

This is the example reference table that the enrichment processor uses:

merch_id is set as the column name the processor uses to find the source attribute’s value. Note: The source attribute’s value does not have to match the column name.

If the enrichment processor receives a log with "merchant_id":"536":

• The processor looks for the value 536 in the reference table’s merch_id column.
• After it finds the value, it adds the entire row of information from the reference table to the merchant_info attribute as a JSON object:

merchant_info {
    "merchant_name":"Cindy's Couches",
    "city":"Boulder",
    "state":"Colorado"
}