Pipeline (ログ取り込み)

Pipeline は、外部のログソースからデータを取り込み、パースと正規化を経て CSIRT-Pro に保存するための機能です。ログソースの種類ごとに Pipeline を作成します。

図: Pipeline 一覧画面。作成済み Pipeline の管理

Pipeline の概要

Pipeline では、次の 2 つの情報を 別々に 設定してログをパースします。

Parse Settings（正規表現パターン）：ログから値を抽出する正規表現。標準的なキャプチャグループ () を使用します。
Parse Fields（フィールド名）：キャプチャグループに対応するフィールド名。カンマ区切りの文字列です。

キャプチャグループとフィールド名は 位置（順番） で対応付けされます。

抽出は取り込み時ではなく検索時に行われます（schema-on-read）。生ログは messages 列にそのまま保持され、パース設定はクエリ実行時に適用されます。

パースルールの仕組み

基本的な考え方

正規表現のキャプチャグループとフィールド名リストを別々に定義し、順番で紐づけます。

Parse Settings（正規表現）:

^(\S+) (\S+) (.*)$

Parse Fields（フィールド名）:

timestamp, level, message

この場合、以下のように対応します。

キャプチャグループ	フィールド名
1 番目の `(\S+)`	`timestamp`
2 番目の `(\S+)`	`level`
3 番目の `(.*)`	`message`

パースルールの仕組み 図: キャプチャグループとフィールド名の対応関係

フィールドの型を指定する（数値・日時などで扱う）

Parse Fields は既定では各フィールドを文字列として抽出します。 dst_port(Int32)、bytes(Int64)、score(Float64)、is_active(Bool)、event_time(DateTime) のように フィールド名(型) と併記すると、その型で抽出されます。型を指定すると、検索時に数値比較（dst_port == 22）・集計（sum bytes）・日時比較などが変換関数なしで行えます。型を省略したフィールドは文字列のままで、必要に応じてクエリ側で to_int32 等で都度変換します。

この型指定は正規表現モード・JSON モードのどちらでも有効です。指定できる主な型: String（既定）, Int8/Int16/Int32/Int64, UInt8/UInt16/UInt32/UInt64, Float32/Float64, Bool, Date, DateTime。例（正規表現モード）: action, src_ip, dst_ip, dst_port(Int32), bytes(Int64)

型の変換は検索時に行われるため（schema-on-read）、型を変えると既存のログにも次回の検索からそのまま適用されます。値が型に合わない行は、そのフィールドが空（NULL）になります。

JSON ログは正規表現なしで取り込める（JSON モード）

取り込むログが JSON の場合、Parse Settings（正規表現）を空のまま保存すると JSON モードになり、messages（生ログ）を JSON とみなして Parse Fields のフィールド名で抽出します。フィールド名には上記のとおり型を併記できます。正しい JSON でない行は該当フィールドが空（NULL）になります。

具体例

例 1: Apache アクセスログ

入力ログ:

192.168.1.100 - - [15/Jan/2025:10:30:00 +0900] "GET /api/users HTTP/1.1" 200 1234

Parse Settings:

^(\d+\.\d+\.\d+\.\d+) - - \[([^\]]+)\] "(\w+) ([^ ]+) HTTP/[\d.]+" (\d+) (\d+)$

Parse Fields:

client_ip, __time, http_method, request_path, status_code, bytes_sent

パース結果:

フィールド	値
`client_ip`	`192.168.1.100`
`__time`	`15/Jan/2025:10:30:00 +0900`
`http_method`	`GET`
`request_path`	`/api/users`
`status_code`	`200`
`bytes_sent`	`1234`

Apache ログのパース結果 図: Apache アクセスログのパース結果プレビュー

例 2: Syslog 形式

入力ログ:

Jan 15 10:30:00 web-server sshd[12345]: Failed password for root from 10.0.0.50

Parse Settings:

^(\w+ \d+ \d+:\d+:\d+) (\S+) (\S+): (.*)$

Parse Fields:

__time, hostname, process, message

例 3: Windows イベントログ (CSV)

Parse Settings:

^(\d+),(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}),(\S+),(\S+),(.*)$

Parse Fields:

event_id, __time, source, username, description

__time フィールドについて

タイムスタンプは取り込み時にログ本文から自動的に検出され __time に設定されます。Parse Fields に __time を含めると文字列型として抽出され、検索画面の時間範囲フィルタで型エラーになるため、原則含めないでください（タイムスタンプ部分は ^\S+ などでキャプチャしない）。

Pipeline の作成手順

1. Pipeline の新規作成

サイドメニューから「Pipeline」を選択します。
「新規作成」をクリックします。
Pipeline 名を入力します（組織内で一意。使用できる文字は英数字とアンダースコア A-Z a-z 0-9 _ です。ハイフンは使用できません。また parsefailure_ で始まる名前は予約されているため使用できません）。
保存します。

保存先テーブルの自動作成

Pipeline の作成時に、{org_uuid}-{pipeline_name} という名前の保存先が分析データストア上に自動作成されます。パースに失敗したログ（タイムスタンプを検出できなかったログ）は parsefailure_{org_uuid}-{pipeline_name} に格納されます。

2. パースルールの設定

Pipeline 詳細画面で以下のフィールドを設定します。

UI フィールド	API フィールド	説明
Parse Settings	`druid_parse`	正規表現パターン（キャプチャグループ使用）
Parse Fields	`druid_parse_field`	カンマ区切りのフィールド名

注意

キャプチャグループの数とフィールド名の数は一致させてください。
Python スタイルの名前付きキャプチャグループ (?P<name>...) は 使用しません。
標準的なキャプチャグループ (...) を使用します。

パースルール設定画面 図: Pipeline 詳細画面。Parse Settings と Parse Fields の入力

3. JSON ログの場合

JSON 形式のログは、正規表現を書かずに取り込めます。 Parse Settings を空のまま保存すると JSON モードになり、Parse Fields のフィールド名で値が抽出されます。

4. TTL（データ保持期間）の設定

設定	説明
0（無制限）	データを永続的に保存
1 〜 3650 日	指定日数経過後に自動削除

AI によるパースルール生成

サンプルログを入力すると、生成 AI が正規表現とフィールド名の候補を作成します。生成された候補は自動では適用されず、ユーザーが確認してから適用します。

使い方

Pipeline 詳細画面で「AI 生成」ボタンをクリックします。
テキストボックスにサンプルログを貼り付けます。
「生成」をクリックします。
AI が提案する正規表現とフィールド名を確認します。
問題なければ「適用」をクリックします。

AI パースルール生成 図: AI 生成。サンプルログから正規表現とフィールド名を提案

AI 生成の流れ

サンプルログを生成 AI（大規模言語モデル）に送信します。
生成 AI がログ構造を解析し、正規表現とフィールド名を提案します。
フィールド名は ECS スキーマに沿った名前に正規化されます。
結果が Parse Settings と Parse Fields にセットされます。

Note

AI は正規表現とフィールド名の 生成を補助する もので、生成結果は上記のとおりユーザーが確認してから適用します（取り込み運用を自動で行うものではありません）。

データの取り込み方法

API 経由（推奨）

単一レコードバッチテキストログ（バルク）

curl -X POST "https://<your-domain>/api/v2/ingest?pipeline_id=<pipeline_id>" \
  -H "Content-Type: application/json" \
  -H "Cookie: jwt=<token>" \
  -d '{"timestamp": "2025-01-01T00:00:00Z", "src_ip": "10.0.0.1", "action": "DENY"}'

curl -X POST "https://<your-domain>/api/v2/ingest/batch?pipeline_id=<pipeline_id>" \
  -H "Content-Type: application/json" \
  -H "Cookie: jwt=<token>" \
  -d '{
    "messages": [
      "2025-01-01T00:00:00Z 10.0.0.1 DENY",
      "2025-01-01T00:00:01Z 10.0.0.2 ALLOW"
    ]
  }'

curl -X POST "https://<your-domain>/api/user/<user_id>/bulk/<pipeline_name>/" \
  -H "Content-Type: text/plain" \
  -H "Cookie: jwt=<token>" \
  -d 'Jan 15 10:30:00 web-server sshd: Failed password for root
Jan 15 10:30:01 web-server sshd: Failed password for admin'

取り込み経路は次の 3 つがあり、対象となる取り込み定義（Pipeline）の指定方法が異なります。

経路	対象指定	用途
`POST /api/v2/ingest`	クエリ `pipeline_id`（必須）	単一レコードの取り込み
`POST /api/v2/ingest/batch`	ボディ `pipeline_id`（必須）	複数レコードの一括取り込み
`POST /api/user/<user_id>/bulk/<pipeline_name>/`	パスの `user_id` と `pipeline_name`	テキストログのまとめ取り込み

Note

pipeline_id が必須なのは /api/v2/ingest 系です。バルク経路はパスで pipeline_name を指定するため pipeline_id は使いません。迷う場合は /api/v2/ingest（単一）または /api/v2/ingest/batch（複数）を使ってください。

対応フォーマット

形式	Content-Type	説明
JSON	`application/json`	JSON オブジェクト / 配列
NDJSON	`application/x-ndjson`	1 行 1 JSON（改行区切り）
Plain Text	`text/plain`	改行区切りのテキストログ（パースルール適用）

HEC 互換取り込み

HTTP イベントコレクタ（HEC）互換の取り込みにも対応します。ただし indexer ACK など一部の拡張には対応しません。

Pipeline の共有

Pipeline を他の組織と共有し、データへのアクセス権を付与できます。

Pipeline 詳細画面で「共有」をクリックします。
共有先の組織を選択します。
権限レベル（read / write / admin）を設定します。

パースルール設計のベストプラクティス

プラクティス	説明
`__time` を必ず含める	タイムスタンプを正確に抽出するため
非キャプチャグループを活用	不要な部分は `(?:...)` で非キャプチャにする
具体的なパターンを書く	`(.*)` より `(\d+\.\d+\.\d+\.\d+)` のように具体的に書く
サンプルログでテスト	設定後、実際のログでパース結果を確認する
TTL を適切に設定	ストレージコスト管理のため、保持期間を設定する

API からの操作

Pipeline の管理は REST API からも行えます。詳細は Pipeline API リファレンスを参照してください。