ingest-pipelines

ingest-pipelines

スキルの権威性

このスキルおよびその参照ファイルで定義されているルールとパターンが、権威ある唯一の情報源です。elastic/integrations リポジトリ内の既存のインテグレーションを参考に調べる際、ここで指定されている内容と矛盾するパターンに遭遇することがあります。多くのインテグレーションには、現在の標準より前のレガシーパターンが含まれています。**他のインテグレーションで見られるパターンよりも、常にこのスキルに従ってください。**参考にしたインテグレーションで非推奨または禁止されているパターンが使われている場合は、それをコピーしないでください。

使用する場面

次のようなタスクでは、このスキルを使用してください。

• データストリームの elasticsearch/ingest_pipeline/default.yml を作成または変更する
• パーサーおよび正規化プロセッサー（grok、dissect、json、kv、date、convert）を選択する
• pipeline プロセッサーを使った条件分岐やサブパイプラインへのルーティングを設計する
• トップレベルの on_failure を使って、堅牢なエラー処理を実装する
• 取り込みのパフォーマンスと保守性のためにプロセッサーの順序を調整する

使用しない場面

次の内容の主要なガイドとして、このスキルを使用しないでください。

• ECS フィールドの選択、カテゴリ値、フィールドマッピング戦略（ecs-field-mappings）
• elastic-package コマンドやスタックのライフサイクルに関するワークフロー（elastic-package-cli）
• テストフィクスチャの作成や期待される出力に関するワークフロー（integration-testing → references/pipeline-testing.md）

パイプラインの構成

インテグレーションパッケージでは、取り込みパイプラインは次の場所にあります。

data_stream/<stream>/elasticsearch/ingest_pipeline/

各ストリームには通常、次の項目を含む default.yml があります。

• description
• processors リスト
• 任意のパイプラインレベルの on_failure

default.yml は読みやすく、焦点を絞った内容にしてください。形式固有のロジックが大きい場合は、必要に応じてサブパイプラインに移してください。

ECS バージョン

processors の先頭（すでに使用している導入プロセッサーがあれば、その後）で、パイプラインの ECS 参照バージョンを明示的に設定してください。**9.3.0 を使用してください。**古い ECS バージョンに固定しないでください。**例外：**オーケストレーターがエンティティデータストリーム（event.kind: asset）であると指定した場合は、代わりに 9.5.0 を使用してください。エンティティのリーフフィールド（entity.attributes.*、entity.lifecycle.*、entity.relationships.*）は 9.3.0 には存在しないため、そのバージョンに固定すると field is undefined のビルドエラーが発生します。

  - set:
      field: ecs.version
      tag: set_ecs_version
      value: '9.3.0'

rename と set（ECS へのマッピング）

カスタムフィールドまたはベンダーフィールドからECS フィールドに値を移す場合は、rename プロセッサーを優先してください。これにより、ソースフィールドが削除され、データの重複を避けられます。ソースフィールドを保持する必要がある場合、または rename が適用できない場合に限り、copy_from を使った set を使用してください。

プロセッサータグ

パイプライン内のすべてのプロセッサーに tag を付けてください（失敗する可能性のあるプロセッサーだけではありません）。タグを付けると、失敗やテレメトリの原因となったステップを特定できます。

CEL 専用の先行プロセッサー（Agentless メタデータとエラーのみのドキュメント）

CEL ベースのインテグレーションに限り、該当する場合は標準の message → event.original 処理より前に、次のプロセッサーを含めてください。

• remove：すべての値が文字列の場合、Agentless メタデータフィールド（organization、division、team）を削除して、ECS との衝突を防ぎます。ignore_missing: true と条件付きの if を使用してください。
• terminate：ドキュメントがコレクターからのエラープレースホルダーである場合、処理を停止します（ctx.error?.message != null && ctx.message == null && ctx.event?.original == null）。

非 CEL インテグレーション（ログ、syslog、filebeat 形式の入力）では、このブロックをそのままコピーしてはいけません。これらのフィールドとエラー形式は CEL／Agentless の経路に固有のものです。create-integration スキルを参照してください。オーケストレーターがこのブロックを想定するのは、データストリームが CEL 入力を使用する場合に限られます。

標準の開始処理：ECS、任意の CEL ブロック、JSE00001、その後 event.original を解析

任意の CEL 専用プロセッサーの後、パイプラインは次の構成にします。すべての解析（json、csv、grok など）は event.original に対して実行します。後続のプロセッサーで event.original を上書きまたは変更しないでください。構造化フィールドは、ほかのパス（たとえば json、_temp.*、ECS フィールド）に生成します。

description: Parse <dataset> events.
processors:
  - set:
      field: ecs.version
      tag: set_ecs_version
      value: '9.3.0'

  # --- CEL input only (omit for log/syslog-only streams) ---
  - remove:
      field:
        - organization
        - division
        - team
      ignore_missing: true
      if: ctx.organization instanceof String && ctx.division instanceof String && ctx.team instanceof String
      tag: remove_agentless_tags
      description: >-
        Removes the fields added by Agentless as metadata,
        as they can collide with ECS fields.
  - terminate:
      tag: data_collection_error
      if: ctx.error?.message != null && ctx.message == null && ctx.event?.original == null
      description: error message set and no data to process.
  # --- end CEL-only ---

  - rename:
      field: message
      tag: rename_message_to_event_original
      target_field: event.original
      ignore_missing: true
      description: Renames the original `message` field to `event.original` to store a copy of the original message. The `event.original` field is not touched if the document already has one; it may happen when Logstash sends the document.
      if: ctx.event?.original == null
  - remove:
      field: message
      tag: remove_message
      ignore_missing: true
      description: The `message` field is no longer required if the document has an `event.original` field.
      if: ctx.event?.original != null

  # Parse (always read from event.original; do not modify event.original)
  - json:
      field: event.original
      target_field: json
      tag: parse_json
      if: ctx.event?.original != null

  # ... normalize, enrich, ECS categorization, cleanup ...

  - append:
      field: tags
      value: preserve_original_event
      allow_duplicates: false
      if: ctx.error?.message != null

on_failure:
  - append:
      field: error.message
      value: >-
        Processor '{{{ _ingest.on_failure_processor_type }}}'
        {{{#_ingest.on_failure_processor_tag}}}with tag '{{{ _ingest.on_failure_processor_tag }}}'
        {{{/_ingest.on_failure_processor_tag}}}failed with message '{{{ _ingest.on_failure_message }}}'
  - set:
      field: event.kind
      tag: set_pipeline_error_to_event_kind
      value: pipeline_error
  - append:
      field: tags
      value: preserve_original_event
      allow_duplicates: false

単一パスのパターン（線形パイプライン）

1 つのパーサーフローですべてのイベントを処理する場合は、このパターンを使用します。標準の開始処理（ECS バージョン、任意の CEL 専用ブロック、JSE00001 による名前変更と削除、event.original を変更せずに解析）、各ステップにタグを付けた中間プロセッサー、そして前述のパイプラインレベルの on_failure と preserve_original_event の条件付き append を組み合わせます。

中間セクションの例（参考）：

  - grok:
      field: event.original
      patterns:
        - '^...$'
      tag: parse_main

  - date:
      field: some.time
      target_field: '@timestamp'
      formats: [ISO8601]
      tag: parse_timestamp
  - convert:
      field: http.response.status_code
      type: long
      ignore_missing: true
      tag: convert_status

  - user_agent:
      field: user_agent.original
      ignore_missing: true
      tag: enrich_user_agent
  - geoip:
      field: source.ip
      target_field: source.geo
      if: ctx.source?.ip != null
      ignore_missing: true
      tag: enrich_source_geo
  - geoip:
      database_file: GeoLite2-ASN.mmdb
      field: source.ip
      target_field: source.as
      if: ctx.source?.ip != null
      properties:
        - asn
        - organization_name
      ignore_missing: true
      tag: enrich_source_asn
  - rename:
      field: source.as.asn
      target_field: source.as.number
      ignore_missing: true
      tag: rename_source_asn
  - rename:
      field: source.as.organization_name
      target_field: source.as.organization.name
      ignore_missing: true
      tag: rename_source_as_org

  - set:
      field: event.kind
      tag: set_event_kind
      value: event
  - append:
      field: event.category
      tag: append_event_category_web
      value: web
  - remove:
      field: temp
      ignore_missing: true
      tag: remove_temp

分岐パターン（ルーター＋サブパイプライン）

イベント形式やオブジェクトモデルが異なる場合は、分岐を使用します。

• 形式に基づく分岐（たとえば JSON とテキスト）
• クラス／カテゴリに基づく分岐（たとえば OCSF クラス／カテゴリのルーティング）
• オブジェクトの存在に基づく分岐（ctx.ocsf.user != null）

パターン：

processors:
  - pipeline:
      name: '{{ IngestPipeline "pipeline_branch_json" }}'
      if: ctx.event?.original != null && ctx.event.original.startsWith('{')
      ignore_missing_pipeline: true
      tag: route_json
  - pipeline:
      name: '{{ IngestPipeline "pipeline_branch_text" }}'
      if: ctx.event?.original != null && !ctx.event.original.startsWith('{')
      ignore_missing_pipeline: true
      tag: route_text

大規模なインテグレーションでは、default.yml をルーターとして維持し、分岐ロジックは次のようなファイルに記述します。

• pipeline_object_<name>.yml
• pipeline_category_<name>.yml

amazon_security_lake の完全なパターンについては、references/branching-patterns.md を参照してください。

複数のログタイプを扱うインテグレーションのサブパイプラインルーティング

データストリームが複数の異なるログタイプを受け取る場合（たとえば、同じストリームにトラフィック、認証、DNS のログを出力するファイアウォール）、すべての解析を単一の巨大な default.yml に実装しないでください。default.yml はログタイプを判別して、タイプごとの専用サブパイプラインに処理を委譲する、簡潔なルーターとして使用します。

ファイル構成

elasticsearch/ingest_pipeline/
  default.yml              # router only — detects log type, calls sub-pipelines
  pipeline-<type>.yml      # one file per log type (e.g. pipeline-traffic.yml)

default.yml のルーターパターン

標準の開始処理と同じ ecs.version、メッセージ用の JSE00001 rename／remove の組み合わせ、および完全なパイプラインレベルの on_failure を使用します。ルーターはサブパイプラインを分岐させるだけで、ペイロードは解析しません。

processors:
  - set:
      field: ecs.version
      tag: set_ecs_version
      value: '9.3.0'
  - rename:
      field: message
      tag: rename_message_to_event_original
      target_field: event.original
      ignore_missing: true
      if: ctx.event?.original == null
  - remove:
      field: message
      tag: remove_message
      ignore_missing: true
      if: ctx.event?.original != null
  - pipeline:
      name: '{{ IngestPipeline "pipeline-traffic" }}'
      if: 'ctx.event?.original != null && ctx.event.original.contains("TRAFFIC")'
      tag: route_traffic
  - pipeline:
      name: '{{ IngestPipeline "pipeline-auth" }}'
      if: 'ctx.event?.original != null && ctx.event.original.contains("AUTH")'
      tag: route_auth
  - pipeline:
      name: '{{ IngestPipeline "pipeline-dns" }}'
      if: 'ctx.event?.original != null && ctx.event.original.contains("DNS")'
      tag: route_dns
on_failure:
  - append:
      field: error.message
      value: >-
        Processor '{{{ _ingest.on_failure_processor_type }}}'
        {{{#_ingest.on_failure_processor_tag}}}with tag '{{{ _ingest.on_failure_processor_tag }}}'
        {{{/_ingest.on_failure_processor_tag}}}failed with message '{{{ _ingest.on_failure_message }}}'
  - set:
      field: event.kind
      tag: set_pipeline_error_to_event_kind
      value: pipeline_error
  - append:
      field: tags
      value: preserve_original_event
      allow_duplicates: false

ルール

• default.yml にはルーティングロジックと on_failure の処理のみを記述します。フィールド解析は行いません。
• 各サブパイプラインは、それぞれのログタイプについて、解析、ECS マッピング、カテゴリ分類を処理します。
• 各サブパイプラインには、独自の on_failure ブロックが必要です。
• サブパイプラインのファイル名は pipeline-<type>.yml とし、<type> はルーティング条件で使用するログタイプ識別子に一致させます。
• ログタイプごとに、命名規則 test-<package>-<datastream>-<type>-sample.log に従ったパイプラインテスト用フィクスチャファイルを用意します。

プロセッサーの順序とパフォーマンス

• コストの低い存在チェックを、コストの高い処理より先に実行します
• 対象範囲外のレコードは早い段階で破棄します
• 区切り形式が安定している場合は、grok より dissect を優先します
• 組み込みプロセッサーで実現できる場合は、script プロセッサーを使用しないでください — set、rename、remove、append、convert、dissect、grok、gsub、lowercase、uppercase、trim は、いずれも Painless より高速で、レビューも容易です。references/processor-cookbook.md → プロセッサーのパフォーマンスガイドにあるコスト階層を参照してください。
• エンリッチメントプロセッサー（geoip、user_agent）は必要な場合にのみ使用します
• grok パターンには必ず ^ と $ を付けてアンカーします。アンカーがないと、正規表現エンジンは部分一致を探して入力文字列全体を走査するため、処理が遅くなり、ノイズの多いログ行で誤った結果が発生する可能性があります

processor の値での Mustache テンプレート構文

Ingest pipeline の processor では、value、message などの文字列パラメーターでフィールド値を参照するために Mustache テンプレートを使用します。三重波括弧 {{{field}}} と一重引用符を使用してください。二重波括弧や二重引用符は使用しないでください。

# CORRECT — triple braces, single quotes
- append:
    field: related.user
    value: '{{{user.target.email}}}'
    allow_duplicates: false
    if: ctx.user?.target?.email != null

# WRONG — double braces HTML-escape the value; double quotes
- append:
    field: related.user
    value: "{{user.target.email}}"
    allow_duplicates: false
    if: ctx.user?.target?.email != null

理由：Mustache の二重波括弧 {{...}} は値を HTML エンコードするため（たとえば、& は & になります）、Ingest pipeline のデータが破損します。三重波括弧 {{{...}}} は値をそのまま出力します。一重引用符を使用すると、YAML が波括弧を解釈するのを防げます。

例外： pipeline.name の {{ IngestPipeline "..." }} は、ビルド時に処理される Go テンプレートディレクティブであり、Mustache テンプレートではありません。正しく二重波括弧を使用します。

エラー処理の要点

エラー報告の主な仕組みとして、パイプラインレベルの on_failure を使用してください。

推奨される基本パターン（順序が重要です）：

• _ingest.on_failure_* 変数を使って、コンテキストを含む error.message を最初に append します（完全なテンプレートは標準の冒頭の例を参照してください）
• event.kind: pipeline_error を set します（set processor に tag を指定します）
• トリアージ用に失敗したドキュメントを保持する必要がある場合は、preserve_original_event を tags に append します
• 失敗する可能性のある processor だけでなく、すべての processor に tag を指定します

processor レベルの on_failure は、ローカルなクリーンアップやフォールバック解析に使用し、主なグローバルエラーメッセージの経路としては使用しないでください。

完全な例とトレードオフ（ignore_failure、fail、processor レベルの on_failure）については、references/error-handling-patterns.md を参照してください。

event.original の処理（JSE00001）

elastic-package build バリデーターは、パイプラインが message から event.original への名前変更を正しく処理しているかを検証します。このチェックは JSE00001 として知られています。新しいパッケージは準拠する必要があります。一部のレガシーパッケージでは、validation.yml でこのチェックを除外しています。

必須の 2 processor パターン

message フィールドを使用するすべてのパイプラインに、両方の processor を含める必要があります（通常は ecs.version の後、該当する場合は CEL のみの remove／terminate ステップの後に配置します）。

- rename:
    field: message
    tag: rename_message_to_event_original
    target_field: event.original
    ignore_missing: true
    description: Renames the original `message` field to `event.original` to store a copy of the original message. The `event.original` field is not touched if the document already has one; it may happen when Logstash sends the document.
    if: ctx.event?.original == null
- remove:
    field: message
    tag: remove_message
    ignore_missing: true
    description: The `message` field is no longer required if the document has an `event.original` field.
    if: ctx.event?.original != null

ステップ 1（rename）：event.original に値がまだ設定されていない場合に限り、message を event.original に移します（先行するパイプラインまたは Logstash がすでに設定している場合でも、冪等に処理されます）。

ステップ 2（remove）：event.original が存在する場合（名前変更後、または上流のプロデューサーから渡された場合）、冗長な message フィールドを削除します。

パイプラインの最後に event.original を削除する processor を追加しないでください

一部の既存インテグレーションには、preserve_original_event が tags にない場合に、パイプラインの最後で event.original を削除する remove processor が含まれています。このパターンは非推奨であり、新しいパイプラインでは使用しないでください。 ストレージ最適化のための event.original の削除は、現在、インテグレーションの外部にある別の最終パイプラインで処理されます。このパターンがまだ含まれている参照インテグレーションから、コピーしないでください。これはレガシーです。

スタック側のこの最終パイプラインは preserve_original_event タグを処理するため、この切り替えを公開するパッケージでは、切り替えを機能させるためのパイプライン内 remove は不要です。つまり、これがなくても切り替えは何もしないわけではありません。この processor の追加は決して推奨しないでください。レビュー時に指摘すべきなのは、この processor が存在することであり、存在しないことではありません。

参照

上に示した 2 processor の JSE00001 パターン（message の rename と remove）は必須であり、これで完結しています。この 2 つ以外に event.original を処理する processor を追加しないでください。

タイムゾーンの処理（tz_offset）

tz_offset manifest var を含むデータストリーム（メッセージにタイムゾーンが含まれない syslog ストリーム）では、日付の解析前に、パイプラインの早い段階で _conf.tz_offset から event.timezone を設定してください。

- set:
    field: event.timezone
    tag: set_event_timezone
    value: '{{{_conf.tz_offset}}}'
    if: ctx._conf?.tz_offset != null && ctx._conf.tz_offset != ''

これにより、タイムゾーン要素を含まないタイムスタンプを解析する際に、date processor が正しいタイムゾーンを適用できます。

Syslog の構造化データ（RFC 5424 SD-ELEMENT）の解析

ベンダーの key=value ペイロードと RFC 5424 の SD-ELEMENT ブロックには、3 つの方法があります。trim_value を使った KV（最も簡単な方法、方法 1）、正規表現による分割を使った SYSLOG5424SD grok ＋ KV（方法 2）、埋め込みの等号や混在した引用符があるエッジケース向けの Painless（方法 3）です。

方法 1 または 2 を推奨します。KV のエッジケースで必要な場合にのみ、Painless を使用してください。

詳しいコード例、主要な設定、リファレンス実装については、references/grok-recipes.md → Syslog structured data strategies を参照してください。

数値として渡される keyword フィールド

識別子、プロトコルコード、その他の不透明な値を持つフィールドは、ソースデータで数値として渡される場合でも、fields.yml で keyword として宣言する必要があります。一般的な例は次のとおりです。

• ネットワークプロトコル番号（network.iana_number）
• 識別子として使用されるポート番号
• エラーコード、結果コード、ステータスコード
• SNMP OID、イベント ID、オブジェクトクラスコード

これらの値を文字列化する convert プロセッサは追加しないでください。Elasticsearch はインデックス作成時に数値を keyword 文字列へ暗黙的に変換するため、パイプラインでは生の数値をそのまま渡せます。

fields.yml でのフィールド宣言：

- name: network.iana_number
  type: keyword
  description: IANA protocol number.

テストランナーは生の値の型を宣言されたフィールド型と比較するため、6（long）は keyword との不一致として検出されます。フィクスチャで値を不自然に文字列化する必要がないよう、パイプラインのテスト設定でフィールドを numeric_keyword_fields に宣言してください。設定の構文については、integration-testing/references/pipeline-testing.md を参照してください。

ベンダーフィールドの命名

ベンダーフィールド名は、ソースに記載されているとおりに維持してください。ベンダー固有のフィールド名の変更、形式変更、正規化は行わないでください。許可される変更は、ベンダーフィールドを ECS フィールドにマッピングする場合（例：src_ip を source.ip に変更する場合）のみです。ベンダーフィールドに対応する ECS フィールドがない場合は、ソースの元の名前を使用してベンダー名前空間のプレフィックス（例：vendor.product.field_name）の下に保持してください。

related.ip の設定

ドキュメント内に存在するすべての IP アドレスを related.ip に追加する必要があります。 これには、イベントの種類に応じて、ソース、宛先、クライアント、サーバー、ホスト、その他すべての IP フィールドが含まれます。

IP フィールドごとに 1 つの append プロセッサを使用し、フィールドが存在しない場合に何もしないよう ignore_missing: true を設定してください。すべての IP フィールドを設定した後（たとえば、geoip、convert、ECS の名前変更手順の後）、クリーンアップ用の remove プロセッサの前に、これらのプロセッサを配置してください。

  - append:
      field: related.ip
      tag: append_source_ip_to_related
      value: '{{{source.ip}}}'
      allow_duplicates: false
      if: ctx.source?.ip != null
  - append:
      field: related.ip
      tag: append_destination_ip_to_related
      value: '{{{destination.ip}}}'
      allow_duplicates: false
      if: ctx.destination?.ip != null
  # repeat the same pattern for client.ip, server.ip, host.ip, and any other IP fields the pipeline sets

ルール：

• 値の重複を避けるため、すべての append で allow_duplicates: false を使用してください。
• イベントにフィールドがない場合にスキップするよう、すべてのプロセッサに if ガードを追加してください。
• パイプラインが実際に書き込む IP フィールドごとに append を 1 つ追加してください。パイプラインが設定しないフィールド用のプロセッサは追加しないでください。

Painless スクリプトのベストプラクティス

script プロセッサーを書く前に、同じ処理を行える組み込みプロセッサーがないか、必ず確認してください。 script は汎用プロセッサーの中で最も遅いものです（Painless のコンパイルとドキュメントごとの実行が必要です）。次の操作には専用のプロセッサーがあり、より低コストでレビューもしやすくなります。

必要な処理	script ではなく、こちらのプロセッサーを使用
フィールドのコピー、移動、名前変更	rename、または copy_from を指定した set
定数値または導出値の設定	set
リストへの値の追加	append
フィールドの型の変更	convert
区切り文字を含む文字列から部分文字列を抽出	dissect
正規表現を使って部分文字列を抽出	grok
文字列内の文字の置換	gsub
大文字・小文字の正規化	lowercase / uppercase

script を使うのは、組み込みプロセッサーの組み合わせではロジックを表現できない場合に限ってください。たとえば、5 件以上のエントリーを含む ECS カテゴリ分類のルックアップテーブル（パターン A）、複雑な条件付き算術処理、あるいは dissect と grok では実際に対応できない特殊な文字列解析などです。

大文字・小文字を区別しない比較 — 大文字・小文字が予測できない場合は equalsIgnoreCase() を使用してください

Syslog とベンダー製デバイスでは、大文字・小文字の使い方に一貫性がないことがよくあります。そのため、ベンダー固有の自由記述フィールドを比較する Painless スクリプトでは、== ではなく equalsIgnoreCase() を使用してください。ただし、一律に適用せず、状況に応じて判断してください。

• デバイス、ファームウェアのバージョン、ログソースによってベンダーフィールドの値の大文字・小文字が変わる可能性がある場合は、equalsIgnoreCase() を使用してください（例：allow/Allow/ALLOW のようなアクションフィールド、重要度を表す文字列、自由記述のステータスフィールド）。
• API または仕様で固定の小文字の列挙値が定義され、値が常に仕様どおりに渡される場合は、== を使用してください（例：ECS のカテゴリ分類フィールド、小文字のみの列挙値として文書化された API レスポンスフィールド）。固定列挙値のフィールドに equalsIgnoreCase() を追加しても、メリットがないまま冗長になるだけです。

// Correct for unpredictable vendor casing
if (ctx.vendor?.action?.equalsIgnoreCase('allow')) { ... }

// Correct for a fixed lowercase API enum — == is appropriate here
if (ctx.json?.event_type == 'login') { ... }

// Incorrect for unpredictable casing — breaks on "Allow", "ALLOW"
if (ctx.vendor?.action == 'allow') { ... }

スクリプト本体では ctx に直接アクセスする — null 安全演算子は使用しないでください

script プロセッサーの source ブロックでは、ctx フィールドに直接アクセスしてください。null 安全演算子 ?. ではなく、明示的な null チェックを使用してください。

// Correct — direct access with explicit null check
if (ctx.source != null && ctx.source.ip != null) { ... }

// Incorrect — null-safe operator in a script body
if (ctx.source?.ip != null) { ... }

注：プロセッサーの if 条件（YAML）では、null 安全演算子 ?. を使用できます。こちらは Painless の実行コンテキストが異なります。

- append:
    field: related.ip
    value: '{{{source.ip}}}'
    if: ctx.source?.ip != null

その他のルール

• すべての script プロセッサーに tag と description を設定してください。
• スクリプトは短く、対象を絞ってください。複雑なロジックは複数の script プロセッサーに分散させず、スクリプト内のヘルパー変数にまとめてください。
• 組み込みプロセッサーで対応できる場合は script を使用しないでください — このセクション冒頭の必須チェックリスト表を参照してください。

ECS カテゴリ分類マッピング

ソースイベントの種類やアクションを event.category、event.type、event.outcome、event.action にマッピングする場合は、references/processor-cookbook.md → ECS カテゴリ分類マッピングのパターンにあるパターンを使用してください。

• パターン A（params ルックアップテーブルを使うスクリプト）：5 件以上のマッピングに推奨されます。マッピングデータを params に置くと、Painless のコンパイルキャッシュを活用でき、スクリプト本体も汎用的に保てます。
• パターン B（条件付きの set プロセッサー）：スクリプトを使うほどではない、5 件未満のマッピングに適しています。
• パターン C（サブパイプライン）：100 件以上のマッピングでは、カテゴリ分類を専用のサブパイプラインファイルに切り出してください。

一括した append プロセッサー（イベントの種類ごとに 2 個、25 種類で 50 個以上のプロセッサーが必要）や、params を使わないインラインの Painless if / else チェーンは使用しないでください（コンパイルキャッシュの利点が失われます）。これらは明確なアンチパターンです。詳細はクックブックを参照してください。

Grok のベストプラクティス

• 構造が固定されている場合は dissect を優先してください
• 可能な限り単純な grok パターンを使用してください
• grok パターンの先頭と末尾には、必ず ^ と $ を付けてください。

  # Correct — anchored, fails fast on non-matching lines
  patterns:
    - '^%{IPORHOST:source.ip} %{USER:user.name} %{DATA:message}$'
  
  # Incorrect — unanchored, scans the whole string for a partial match
  patterns:
    - '%{IPORHOST:source.ip} %{USER:user.name} %{DATA:message}'
  

• 不要なバックトラッキングを引き起こすカスタム正規表現は避けてください
• すべての grok プロセッサー（およびその他すべてのプロセッサー）に tag を追加してください

grok の構文（3 つの式形式、インライン正規表現、型変換、pattern_definitions）、Syslog ヘッダーの分割方法、よくある間違いについては、references/grok-recipes.md を参照してください。

使用禁止のパターン

以下のパターンは多くのレガシーな統合に存在しますが、新規または更新するパイプラインでは使用してはいけません。参照用の統合からコピーしないでください。

event.ingested を設定しない

event.ingested フィールドは、統合パイプラインの外部で Elasticsearch によって管理されます。統合パイプラインに event.ingested 用の set プロセッサーを追加しないでください。以下のようなパターンも含まれます。

# PROHIBITED — do not use
- set:
    field: event.ingested
    value: '{{{_ingest.timestamp}}}'

パイプラインでは、元のイベントのタイムスタンプから @timestamp を設定してください。ソースデータに複数のタイムスタンプが含まれる場合は、次のようにマッピングします。

• @timestamp：ソースデータから解析した主要なイベントタイムスタンプです。必須です。
• event.created：ソースシステムでイベントが最初に作成または記録された日時です（@timestamp と異なる場合）。
• event.start：アクティビティまたは期間が開始した日時です（例：セッションの開始、接続の開始）。
• event.end：アクティビティまたは期間が終了した日時です（例：セッションの終了、接続の切断）。

ソースのタイムスタンプが event.created、event.start、または event.end の意味に当てはまらない場合は、fields.yml で type: date を指定したベンダー名前空間内のカスタムフィールドにマッピングし、適切な target_field を指定した date プロセッサーを使用します。

preserve_duplicate_custom_fields を使用しない

preserve_duplicate_custom_fields タグのパターン、つまり set と copy_from を使ってソースフィールドを ECS フィールドにコピーし、条件に応じて元のフィールドを保持するパターンは、レガシーなアンチパターンです。新規または更新するパイプラインでは使用しないでください。preserve_duplicate_custom_fields のマニフェスト変数、タグ、条件ロジックを追加しないでください。

代わりに、次のフィールドマッピングルールに従います。

• ソースフィールドを ECS フィールドにマッピングする場合は、rename を使って直接移動します。ソースフィールドは削除され、重複は発生しません。
• 型変換が必要な場合（例：文字列から日付、文字列から long）、適切なプロセッサー（date、convert、copy_from を指定した set）を使って ECS のターゲットフィールドに値を設定し、パイプライン末尾のクリーンアップセクションでソースフィールドを remove します。
• 元のベンダーフィールドと ECS のコピーの両方を保持する必要があるパイプラインは決して設計しないでください。正規の格納場所は ECS フィールドです。

参照用の統合でこのパターンを見つけた場合は無視してください。これはレガシーなパターンです。

最後に event.original の削除プロセッサーを追加しない

上記の JSE00001 セクションに記載されているとおり、パイプラインの最後に event.original 用の remove プロセッサーを追加しないでください。この処理は別個の最終パイプラインで行います。

参考資料

• references/processor-cookbook.md — プロセッサーの選択、解析／正規化／エンリッチメントの例、ECS カテゴリ分類のマッピングパターン（パターン A／B／C とアンチパターン）
• references/branching-patterns.md
• references/error-handling-patterns.md
• references/grok-recipes.md — grok の構文、型変換、syslog ヘッダーのレシピ、よくある間違い、パターンライブラリーへのリンク
• references/builder-subagent-guidance.md — サブエージェントの運用マニュアル：スコープの境界、スキルの読み込み順序、入力データのパス（CEL 優先と Direct）、パイプライン構築の 9 ステップのワークフロー、「生成された出力を確認し、期待される JSON を手作業で編集しない」、報告の取り決め。オーケストレーターはタスクプロンプトにこのファイルのパスを渡してサブエージェントを起動します。サブエージェントは独立した新しいコンテキスト内で自らこのファイルを読み込みます。内容をタスクプロンプトに埋め込んだり、貼り付けたりしないでください。
• references/cdr-pipeline-requirements.md — クラウドセキュリティ／CDR 統合のみ（cloudsecurity_cdr カテゴリー）：イベントのカテゴリ分類、必須／推奨フィールド、プロセッサーのパターン、レビュー用チェックリスト
• entity-mappings/references/entity-pipeline-patterns.md — エンティティ／インベントリのデータストリームのみ（event.kind: asset）：カテゴリ分類のプロセッサー、entity.id のミラーリング、ブール値の変換、リレーションシップオブジェクト、アンチパターン、パイプラインのレビューチェックリスト