Observability Pipelines Worker for Google Chronicle を認証するには、Google Security Operations の担当者に連絡して Google Developer Service Account Credential を取得してください。このクレデンシャルは JSON ファイルであり、DD_OP_DATA_DIR/config に配置する必要があります。詳細については、API 認証情報の取得を参照してください。

Worker の Google Chronicle 宛先を設定するには、以下の手順を行います:

1. Google Chronicle インスタンスのカスタマー ID を入力します。
2. 先ほどダウンロードした資格情報 JSON ファイルへのパスを入力します。
3. ドロップダウンメニューで JSON または Raw エンコーディングを選択します。
4. ログタイプを入力します。ログの特定のフィールドに基づいて異なるログタイプに振り分けたい場合は、テンプレート構文を参照してください。

注: Google Chronicle 宛先に送信するログにはインジェスションラベルが必須です。たとえば、A10 ロードバランサーのログであれば、インジェスションラベルとして A10_LOAD_BALANCER を付与する必要があります。利用可能なログタイプと対応するインジェスションラベルの一覧については、Google Cloud のデフォルトパーサーでログタイプをサポートするを参照してください。

To use the CrowdStrike NG-SIEM destination, you need to set up a CrowdStrike data connector using the HEC/HTTP Event Connector. See Step 1: Set up the HEC/HTTP event data connector for instructions. When you set up the data connector, you are given a HEC API key and URL, which you use when you configure the Observability Pipelines Worker later on.

1. Select JSON or Raw encoding in the dropdown menu.
2. Optionally, enable compressions and select an algorithm (gzip or zlib) in the dropdown menu.
3. Optionally, 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.

Datadog の宛先に関して必要な構成手順はありません。

If you do not have a Datadog Log Archive configured for Observability Pipelines, configure a Log Archive for your cloud provider (Amazon S3, Google Cloud Storage, or Azure Storage).

Note: You need to have the Datadog integration for your cloud provider installed to set up Datadog Log Archives. See the AWS integration, Google Cloud Platform, and Azure integration documentation for more information.

To set up the destination, follow the instructions for the cloud provider you are using to archive your logs.

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.

Prerequisites

To set up the Microsoft Sentinel destination, you need the following information:

Do the following to get that information:

1. Create or identify a Data Collection Rule (DCR).
   1. In the Azure Portal, navigate to Azure Monitor → Data Collection Rules.
   2. Create a DCR or use an existing one for custom logs. See Data collection rules (DCRs) in Azure Monitor to learn more about creating or viewing DCRs.
   3. Take note of the DCR Immutable ID and, if you are using private links, the DCR’s Data Collection Endpoint (DCE). You need this information when you set up the Microsoft Sentinel destination.
   4. Define a custom table (for example, Custom-MyLogs_CL) in the DCR, which is where Observability Pipelines sends logs to.
2. Get the ingestion URL.
   1. In the DCR, locate the Logs Ingestion API endpoint. The endpoint has the format: https://<DCE-ID>.ingest.monitor.azure.com/dataCollectionRules/<DCR-Immutable-ID>/streams/<Stream-Name>?api-version=2023-01-01, where the <Stream-Name> typically matches your custom table (for example, Custom-MyLogs_CL).
   2. The ingestion URL is needed when you set up you Microsoft Sentinel destination’s environment variable.
3. To authenticate the Observability Pipelines Worker with Microsoft Sentinel:
   1. In the Azure Portal, navigate to Azure AD > App Registrations and register an Azure Active Directory (AD) application. See Register an application in Microsoft Entra ID for information on creating a new application.
   2. Generate a Client Secret.
   3. Assign it the Monitoring Metrics Publisher role on the Log Analytics workspace
   4. Take note of the Tenant ID, Client ID, and Client Secret. You need this information when you set up the Microsoft Sentinel destination.

Set up destination in Observability Pipelines

To set up the Microsoft Sentinel destination in Observability Pipelines:

1. Enter the client ID for your application, such as 550e8400-e29b-41d4-a716-446655440000.
2. Enter the directory ID for your tenant, such as 72f988bf-86f1-41af-91ab-2d7cd011db47. This is the Azure AD tenant ID.
3. Enter the name of the table, such as Custom-MyLogs, to which you are sending logs.
4. Enter the Data Collection Rule (DCR) immutable ID, such as dcr-000a00a000a00000a000000aa000a0aa.

Select the data center region (US or EU) of your New Relic account.

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.

Select your SentinelOne logs environment in the dropdown menu.

• Splunk HEC アドレス:
  • Observability Pipelines Worker がリッスンして、本来 Splunk インデクサー向けであるログを受信するバインドアドレスです。例えば、0.0.0.0:8088。
    注: /services/collector/event は自動的にエンドポイントに付加されます。
  • 環境変数 DD_OP_SOURCE_SPLUNK_HEC_ADDRESS に格納されます。

以下のフィールドはオプションです。

1. Encoding ドロップダウンメニューで、パイプラインの出力を JSON、Logfmt、または Raw テキストでエンコードするかを選択します。デコードが選択されていない場合、デフォルトで JSON にデコードされます。
2. Sumo Logic コレクターのソースに設定されたデフォルトの name 値を上書きするには、source name を入力します。
3. Sumo Logic コレクターのソースに設定されたデフォルトの host 値を上書きするには、host name を入力します。
4. Sumo Logic コレクターのソースに設定されたデフォルトの category 値を上書きするには、category name を入力します。
5. カスタムヘッダーフィールドと値を追加するには、Add Header をクリックします。

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.

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.

重複排除 (deduplicate) プロセッサーは、ボリュームとノイズを削減するためにデータの重複を削除します。一度に 5,000 件のメッセージをキャッシュし、そこに対して受信ログを比較します。たとえば、複数の同一の警告ログが連続で送信される場合に、ユニークな警告ログのみを保持したいときに使用できます。

Deduplicate プロセッサーの設定方法

1. フィルタークエリを定義します。指定したフィルタークエリに一致するログだけが処理されます。重複排除されたログと、フィルタークエリに一致しないログはパイプラインの次のステップに送られます。
2. Type of deduplication ドロップダウンメニューで、下記のフィールドについて Match するか、または Ignore するかを選択します。
   • Match を選択した場合、ログが通過した後、下記で指定するすべてのフィールドの値が同一である将来のログが削除されます。
   • Ignore を選択した場合、ログが通過した後、下記で指定するフィールドを除いたすべてのフィールドの値が同一である将来のログが削除されます。
3. Match する、または Ignore するフィールドを入力します。最低 1 つのフィールドが必要で、最大 3 つのフィールドを指定できます。
   • サブフィールドを指定する場合は、<OUTER_FIELD>.<INNER_FIELD> のようなパス表記を使用します。詳細は、パス表記の例を参照してください。
4. Add field をクリックして、フィルターの対象にしたいフィールドを追加します。

パス表記の例

以下のようなメッセージ構造があった場合、double_inner_value という値を持つキーを指すには、outer_key.inner_key.double_inner_key のように記述します:

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

リマッププロセッサーは、個々のログデータ内のフィールドを追加 (add)、削除 (drop)、またはリネーム (rename) できます。このプロセッサーを使用して、ログに追加のコンテキストを付与したり、ログ容量を削減するために価値の低いフィールドを削除したり、重要な属性全体で命名を標準化したりします。はじめるには、ドロップダウンメニューから add field、drop field、または rename field を選択してください。

フィールドを追加

add field を使用して、ログに新しいキーと値のペアを追加します。

add field プロセッサーの設定方法

1. filter query を定義します。指定されたフィルタークエリに一致するログだけが処理されます。クエリに一致する・しないに関係なく、すべてのログはパイプラインの次のステップに送られます。
2. 追加したいフィールド名と値を入力します。ネストされたフィールドを指定する場合は、パス表記の <OUTER_FIELD>.<INNER_FIELD> 形式を使用してください。すべての値は文字列として格納されます。 注: 追加しようとしているフィールドがすでに存在する場合、Worker はエラーをスローし、既存のフィールドは変更されません。

フィールドを削除

drop field を使用して、指定したフィルタに一致したログデータからフィールドを削除します。オブジェクトを削除することも可能なため、ネストされたキーを削除する場合にもこのプロセッサーを使用できます。

drop field プロセッサーの設定方法

1. filter query を定義します。指定されたフィルタークエリに一致するログだけが処理されます。クエリに一致する・しないに関係なく、すべてのログはパイプラインの次のステップに送られます。
2. 削除したいフィールドのキーを入力します。ネストされたフィールドを指定する場合は、パス表記の <OUTER_FIELD>.<INNER_FIELD> 形式を使用してください。 注: 指定したキーが存在しない場合、そのログには変更が加えられません。

フィールドをリネーム

rename field を使用して、ログ内のフィールド名を変更します。

rename field プロセッサーの設定方法

1. filter query を定義します。指定されたフィルタークエリに一致するログだけが処理されます。クエリに一致する・しないに関係なく、すべてのログはパイプラインの次のステップに送られます。
2. Source field にリネームしたいフィールドの名前を入力します。ネストされたフィールドを指定する場合は、パス表記の <OUTER_FIELD>.<INNER_FIELD> 形式を使用してください。リネーム後は元のフィールドが削除されますが、後述の Preserve source tag チェックボックスを有効にすると元のフィールドを保持できます。
   注: 指定したソースキーが存在しない場合、デフォルトで null 値がターゲットに適用されます。
3. Target field にソースフィールドをリネーム後の名前として入力します。ネストされたフィールドを指定する場合は、パス表記の <OUTER_FIELD>.<INNER_FIELD> 形式を使用してください。
   注: 指定したターゲットフィールドがすでに存在する場合、Worker はエラーをスローし、既存のターゲットフィールドを上書きしません。
4. 必要に応じて、Preserve source tag チェックボックスをオンにすると、元のソースフィールドを保持して、ソースキーの情報を指定したターゲットキーに重複させることができます。このボックスをオフにした場合は、リネーム後にソースキーが削除されます。

パス表記例

以下のようなメッセージ構造があった場合、double_inner_value という値を持つキーを指すには、outer_key.inner_key.double_inner_key のように記述します:

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

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.
     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"
}

このプロセッサーは、指定されたフィルタークエリに一致するログをフィルタリングし、一致しないすべてのログを削除します。このプロセッサーでログが削除された場合、このプロセッサーより下位のプロセッサーはいずれも、そのログを受け取りません。このプロセッサーは、デバッグや警告などの不要なログを除外することができます。

フィルタープロセッサーをセットアップするには

• フィルタークエリを定義します。クエリを指定すると、それに一致するログのみを通過させ、それ以外のログはすべて削除します。

Many types of logs are meant to be used for telemetry to track trends, such as KPIs, over long periods of time. Generating metrics from your logs is a cost-effective way to summarize log data from high-volume logs, such as CDN logs, VPC flow logs, firewall logs, and networks logs. Use the generate metrics processor to generate either a count metric of logs that match a query or a distribution metric of a numeric value contained in the logs, such as a request duration.

Note: The metrics generated are custom metrics and billed accordingly. See Custom Metrics Billing for more information.

To set up the processor:

Click Manage Metrics to create new metrics or edit existing metrics. This opens a side panel.

• If you have not created any metrics yet, enter the metric parameters as described in the Add a metric section to create a metric.
• If you have already created metrics, click on the metric’s row in the overview table to edit or delete it. Use the search bar to find a specific metric by its name, and then select the metric to edit or delete it. Click Add Metric to add another metric.

Add a metric

1. Enter 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. Note: Since a single processor can generate multiple metrics, you can define a different filter query for each metric.
2. Enter a name for the metric.
3. In the Define parameters section, select the metric type (count, gauge, or distribution). See the Count metric example and Distribution metric example. Also see Metrics Types for more information.
   • For gauge and distribution metric types, select a log field which has a numeric (or parseable numeric string) value that is used for the value of the generated metric.
   • For the distribution metric type, the log field’s value can be an array of (parseable) numerics, which is used for the generated metric’s sample set.
   • The Group by field determines how the metric values are grouped together. For example, if you have hundreds of hosts spread across four regions, grouping by region allows you to graph one line for every region. The fields listed in the Group by setting are set as tags on the configured metric.
4. Click Add Metric.

Metrics Types

You can generate these types of metrics for your logs. See the Metrics Types and Distributions documentation for more details.

Count metric example

For this status:error log example:

{"status": "error", "env": "prod", "host": "ip-172-25-222-111.ec2.internal"}

To create a count metric that counts the number of logs that contain "status":"error" and groups them by env and host, enter the following information:

Distribution metric example

For this example of an API response log:

{
    "timestamp": "2018-10-15T17:01:33Z",
    "method": "GET",
    "status": 200,
    "request_body": "{"information"}",
    "response_time_seconds: 10
}

To create a distribution metric that measures the average time it takes for an API call to be made, enter the following information:

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.

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.
   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.

This processor converts the specified field into JSON objects.

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.

This processor parses Extensible Markup Language (XML) so the data can be processed and sent to different destinations. XML is a log format used to store and transport structured data. It is organized in a tree-like structure to represent nested information and uses tags and attributes to define the data. For example, this is XML data using only tags (<recipe>,<type>, and <name>) and no attributes:

<recipe>
    <type>pasta</type>
    <name>Carbonara</name>
</recipe>

This is an XML example where the tag recipe has the attribute type:

<recipe>
    <recipe type="pasta">
    <name>Carbonara</name>
</recipe>

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 match the filter query, are sent to the next step in the pipeline.
2. Enter the path to the log field on which you want to parse XML. Use the path notation <OUTER_FIELD>.<INNER_FIELD> to match subfields. See the Path notation example below.
3. Optionally, in the Enter text key field, input the key name to use for the text node when XML attributes are appended. See the text key example. If the field is left empty, value is used as the key name.
4. Optionally, select Always use text key if you want to store text inside an object using the text key even when no attributes exist.
5. Optionally, toggle Include XML attributes on if you want to include XML attributes. You can then choose to add the attribute prefix you want to use. See attribute prefix example. If the field is left empty, the original attribute key is used.
6. Optionally, select if you want to convert data types into numbers, Booleans, or nulls.
   • If Numbers is selected, numbers are parsed as integers and floats.
   • If Booleans is selected, true and false are parsed as Booleans.
   • If Nulls is selected, the string null is parsed as null.

Path notation example

For the following message structure, use outer_key.inner_key.double_inner_key to refer to the key with the value double_inner_value.

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

Always use text key example

If Always use text key is selected, the text key is the default (value), and you have the following XML:

<recipe>
    <recipe type="pasta">
    <name>Carbonara</name>
</recipe>

The XML is converted to:

{
    "recipe": {
        "type": "pasta",
        "value": "Carbonara"
        }
}

Text key example

If the key is text and you have the following XML:

<recipe>
    <recipe type="pasta">
    <name>Carbonara</name>
</recipe>

The XML is converted to:

{
    "recipe": {
        "type": "pasta",
        "text": "Carbonara"
        }
}

Attribute prefix example

If you enable Include XML attributes, the attribute is added as a prefix to each XML attribute. For example, if the attribute prefix is @ and you have the following XML:

<recipe type="pasta">Carbonara</recipe>

Then it is converted to the JSON:

{
    "recipe": {
        "@type": "pasta",
        "<text key>": "Carbonara"
        }
}

クォータプロセッサは、指定したフィルターに一致するログのロギングトラフィックを測定します。構成された日次クォータが 24 時間のローリングウィンドウ内で達成されると、プロセッサは追加のログをドロップするか、Datadog モニターを使用してアラートを送信することができます。プロセッサは、総ボリュームまたはイベントの総数を追跡するように構成できます。パイプラインは、Worker の複数の Remote Configuration デプロイメント間でクォータを識別するために、クォータの名前を使用します。

例えば、このプロセッサを構成して、過去 24 時間以内に特定のサービスから 1000 万件のイベントを受信した後に、新しいログを削除するか、削除せずにアラートをトリガーするように構成できます。

クォータプロセッサを設定するには、次の手順に従ってください、

1. クォータプロセッサの名前を入力します。
2. フィルタークエリを定義します。指定したフィルタークエリに一致するログのみが日次制限にカウントされます。
   • クォータフィルターに一致し、かつ日次クォータ内のログは、パイプラインの次のステップに送信されます。
   • クォータフィルターに一致しないログも、パイプラインの次のステップに送信されます。
3. Unit for quota (クォータの単位) ドロップダウンメニューで、クォータを Events の数か、バイト単位の Volume で測定するかを選択します。
4. 日次クォータの上限を設定し、希望するクォータの大きさの単位を選択します。
5. クォータが上限に達した場合にすべてのイベントを削除したい場合は、Drop events (イベントを削除) のチェックボックスをオンにします。クォータが上限に達したときにアラートを送信するモニターをセットアップする場合は、チェックを外しておきます。
   • 日次クォータの上限に達した後にクォータフィルターに一致するログが受信され、Drop events (イベントを削除) オプションが選択されている場合、それらのログは削除されます。この場合、フィルタークエリに一致しなかったログのみがパイプラインの次のステップに送信されます。
   • 日次クォータの上限に達した後でも Drop events (イベントを削除) オプションが選択されていない場合、クォータフィルターに一致するログと一致しなかったログが共にパイプラインの次のステップに送信されます。
6. オプション: 特定のサービスまたはリージョンフィールドにクォータを設定したい場合は、Add Field をクリックします。 a. パーティション分割したいフィールド名を入力します。詳細はパーティションの例を参照してください。 i. クォータをパーティションに一致するイベントのみに適用したい場合は、Ignore when missing を選択します。詳細は「欠落時に無視」オプションの例を参照してください。 ii. オプション: パーティション化されたフィールドに異なるクォータを設定したい場合は、Overrides をクリックします。

• CSV の構造例については、Download as CSV をクリックします。
• オーバーライド CSV をドラッグアンドドロップしてアップロードします。または、Browse をクリックしてファイルを選択してアップロードすることもできます。詳細はオーバーライドの例を参照してください。 b. もう 1 つのパーティションを追加したい場合は、Add Field をクリックします。

例

パーティションの例

特定のサービスまたはリージョンにクォータを設定したい場合は、Partition by を使用します。例えば、1 日に 10 件のイベントのクォータを設定し、service フィールドでイベントをグループ化したい場合、service を Partition by フィールドに入力します。

「欠落時に無視」オプションの例

クォータをパーティションに一致するイベントのみに適用したい場合は、Ignore when missing を選択します。例えば、Worker が次のイベントセットを受信した場合:

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

そして Ignore when missing が選択されている場合、Worker は:

• service:a と source:foo を持つログのセットを作成します
• service:b と source:bar を持つログのセットを作成します
• 最後の 3 つのイベントを無視します

クォータは 2 つのログセットに適用され、最後の 3 つのイベントには適用されません。

Ignore when missing が選択されていない場合、クォータは 5 つのすべてのイベントに適用されます。

オーバーライドの例

service でパーティション分割し、2 つのサービス a と b がある場合、オーバーライドを使用してそれぞれに異なるクォータを適用できます。例えば、service:a に 5,000 バイトのクォータ制限、service:b に 50 イベントの制限を設定したい場合、オーバーライドルールは次のようになります。

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.

Use this processor to remap logs to Open Cybersecurity Schema Framework (OCSF) events. OCSF schema event classes are set for a specific log source and type. You can add multiple mappings to one processor. Note: Datadog recommends that the OCSF processor be the last processor in your pipeline, so that remapping is done after the logs have been processed by all the other processors.

To set up this processor:

Click Manage mappings. This opens a modal:

• If you have already added mappings, click on a mapping in the list to edit or delete it. You can use the search bar to find a mapping by its name. Click Add Mapping if you want to add another mapping. Select Library Mapping or Custom Mapping and click Continue.
• If you have not added any mappings yet, select Library Mapping or Custom Mapping. Click Continue.

Filter query syntax

Each processor has a corresponding filter query in their fields. Processors only process logs that match their filter query. And for all processors except the filter processor, logs that do not match the query are sent to the next step of the pipeline. For the filter processor, logs that do not match the query are dropped.

For any attribute, tag, or key:value pair that is not a reserved attribute, your query must start with @. Conversely, to filter reserved attributes, you do not need to append @ in front of your filter query.

For example, to filter out and drop status:info logs, your filter can be set as NOT (status:info). To filter out and drop system-status:info, your filter must be set as NOT (@system-status:info).

Filter query examples:

• NOT (status:debug): This filters for only logs that do not have the status DEBUG.
• status:ok service:flask-web-app: This filters for all logs with the status OK from your flask-web-app service.
  • This query can also be written as: status:ok AND service:flask-web-app.
• host:COMP-A9JNGYK OR host:COMP-J58KAS: This filter query only matches logs from the labeled hosts.
• @user.status:inactive: This filters for logs with the status inactive nested under the user attribute.

Queries run in the Observability Pipelines Worker are case sensitive. Learn more about writing filter queries in Datadog’s Log Search Syntax.

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.

Sensitive Data Scanner プロセッサはログをスキャンして、PII、PCI、カスタムの機密データなどの機密情報を検出し、マスキングまたはハッシュ化します。機密データをスキャンするには、当社のライブラリから定義済みのルールを選択するか、カスタムの正規表現ルールを入力できます。

Sensitive Data Scanner プロセッサをセットアップするには

1. フィルタークエリを定義します。指定したフィルタークエリに一致するログのみがスキャンおよび処理されます。フィルタークエリに一致するかどうかにかかわらず、すべてのログはパイプラインの次のステップに送信されます。
2. Add Scanning Rule をクリックします。
3. スキャンルールに名前を付けます。
4. Select scanning rule type フィールドで、ライブラリからルールを作成するか、カスタムルールを作成するかを選択します。
   • ライブラリからルールを作成する場合は、使用するライブラリパターンを選択します。
   • カスタムルールを作成する場合は、データに対して確認する正規表現パターンを入力します。
5. Scan entire or part of event セクションで、ドロップダウンメニューの Entire Event (イベント全体)、Specific Attributes (特定の属性)、Exclude Attributes (属性の除外) からスキャン対象を選択します。
   • Specific Attributes (特定の属性) を選択した場合は、Add Field をクリックし、スキャンする特定の属性を入力します。最大 3 つのフィールドを追加できます。ネストされたキーにアクセスするには、パス記法 (outer_key.inner_key) を使用します。ネストされたデータを持つ特定の属性では、すべてのネストされたデータがスキャンされます。
   • Exclude Attributes (属性の除外) を選択した場合は、Add Field をクリックし、スキャンから除外する特定の属性を入力します。最大 3 つのフィールドを追加できます。ネストされたキーにアクセスするには、パス記法 (outer_key.inner_key) を使用します。ネストされたデータを持つ指定された属性については、すべてのネストされたデータが除外されます。
6. Define action on match セクションで、一致した情報に対して実行するアクションを選択します。注: マスキング、部分的なマスキング、およびハッシュ化はすべて元に戻せないアクションです。
   • 情報をマスキングする場合は、一致したデータを置き換えるテキストを指定します。
   • 情報を部分的にマスキングする場合は、マスキングする文字数を指定し、部分的なマスキングを一致したデータの先頭または末尾に適用するかどうかを指定します。
   • 注: ハッシュ化を選択した場合、一致した UTF-8 バイトは FarmHash の 64 ビットフィンガープリントでハッシュ化されます。
7. オプションとして、正規表現に一致するすべてのイベントにタグを追加し、イベントのフィルタリング、分析、アラートを行うことができます。

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

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

This processor splits nested arrays into distinct events so that you can query, filter, alert, and visualize data within an array. The arrays need to already be parsed. For example, the processor can process [item_1, item_2], but cannot process "[item_1, item2]". The items in the array can be JSON objects, strings, integers, floats, or Booleans. All unmodified fields are added to the child events. For example, if you are sending the following items to the Observability Pipelines Worker:

{
    "host": "my-host",
    "env": "prod",
    "batched_items": [item_1, item_2]
}

Use the Split Array processor to send each item in batched_items as a separate event:

{
    "host": "my-host",
    "env": "prod",
    "batched_items": item_1
}

{
    "host": "my-host",
    "env": "prod",
    "batched_items": item_2
}

See the split array example for a more detailed example.

To set up this processor:

Click Manage arrays to split to add an array to split or edit an existing array to split. This opens a side panel.

• If you have not created any arrays yet, enter the array parameters as described in the Add a new array section below.
• If you have already created arrays, click on the array’s row in the table to edit or delete it. Use the search bar to find a specific array, and then select the array to edit or delete it. Click Add Array to Split to add a new array.

Add a new array

1. 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.
2. Enter the path to the array field. Use the path notation <OUTER_FIELD>.<INNER_FIELD> to match subfields. See the Path notation example below.
3. Click Save.

Split array example

This is an example event:

{
    "ddtags": ["tag1", "tag2"],
    "host": "my-host",
    "env": "prod",
    "message": {
        "isMessage": true,
        "myfield" : {
            "timestamp":14500000,
            "firstarray":["one", 2]
        },
    },
    "secondarray": [
    {
        "some":"json",
        "Object":"works"
    }, 44]
}

If the processor is splitting the arrays "message.myfield.firstarray" and "secondarray", it outputs child events that are identical to the parent event, except for the values of "message.myfield.firstarray" and "secondarray", which becomes a single item from their respective original array. Each child event is a unique combination of items from the two arrays, so four child events (2 items * 2 items = 4 combinations) are created in this example.

{
    "ddtags": ["tag1", "tag2"],
    "host": "my-host",
    "env": "prod",
    "message": {
        "isMessage": true,
        "myfield" : {"timestamp":14500000, "firstarray":"one"},
    },
    "secondarray": {
        "some":"json",
        "Object":"works"
    }
}

{
    "ddtags": ["tag1", "tag2"],
    "host": "my-host",
    "env": "prod",
    "message": {
        "isMessage": true,
        "myfield" : {"timestamp":14500000, "firstarray":"one"},
        },
    "secondarray": 44
}

{
    "ddtags": ["tag1", "tag2"],
    "host": "my-host",
    "env": "prod",
    "message": {
        "isMessage": true,
        "myfield" : {"timestamp":14500000, "firstarray":2},
        },
    "secondarray": {
            "some":"json",
            "object":"works"
        }
}

{
    "ddtags": ["tag1", "tag2"],
    "host": "my-host",
    "env": "prod",
    "message": {
        "isMessage": true,
        "myfield" : {"timestamp":14500000, "firstarray":2},
        },
    "secondarray": 44
}

Path notation example

For the following message structure, use outer_key.inner_key.double_inner_key to refer to the key with the value double_inner_value.

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

Use this processor to set a limit on the number of logs sent within a specific time window. For example, you can set a limit so that only 100 logs are sent per second. Setting a rate limit can help you catch any spikes in log ingestion and prevent unexpected billing costs.

To set up the processor:

1. Define a filter query. Only logs that match the specified filter query are processed. All matched logs get throttled. Logs that are sent within the throttle limit and logs that do not match the filter are sent to the next step. Logs sent after the throttle limit has been reached, are dropped.
2. Set the window threshold. This is the number of events allowed for a given bucket during the set time window.
3. Set the time window.
4. Optionally, click Add Field if you want to group by a field.