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

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

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 のデフォルトパーサーでログタイプをサポートするを参照してください。

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.

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

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

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

リマッププロセッサーは、個々のログデータ内のフィールドを追加 (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"
}

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.

クォータプロセッサは、指定したフィルターに一致するログのロギングトラフィックを測定します。構成された日次クォータが 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.

重複排除 (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"
}

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. オプションとして、正規表現に一致するすべてのイベントにタグを追加し、イベントのフィルタリング、分析、アラートを行うことができます。

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