Elastic からの移行

このガイドは Elastic Security（Elasticsearch + Kibana）から CSIRT-Pro への移行方法を説明します。 KQL や Lucene から PRQL へのクエリ変換、概念の対応関係、Beats や Logstash からの移行手順を扱います。

---

概念マッピング

Elastic Stack と CSIRT-Pro の主要概念の対応関係です。

Elastic	CSIRT-Pro	説明
Index	Pipeline	ログの保存先。ログソースごとに作成
Index Template	Pipeline パースルール	フィールドマッピングとパース設定
Ingest Pipeline	Pipeline パースルール	ログ取り込み時の変換処理
Detection Rule	UEBA タスク / Playbook	脅威検知ルール
Alert	Case（ケース）	検知されたセキュリティイベント
Case（Elastic）	Case（CSIRT-Pro）	インシデント管理
Timeline	Search	ログの調査・分析
Dashboard (Kibana)	Dashboard	データの可視化
Machine Learning Job	UEBA タスク + 検知 Playbook	検知タスクのスケジュール実行
Beats / Elastic Agent	取り込み API（HTTP）/ フォワーダ経由	ログ収集エージェント
Logstash	Pipeline パースルール	ログの変換・正規化
KQL / Lucene	PRQL	クエリ言語
Connector	Integration	外部サービスとの連携
Spaces	マルチテナント（組織）	組織（テナント）単位の論理分離。Elastic の Spaces は 1 テナント内の区画だが、CSIRT-Pro では組織がテナント境界となる

Machine Learning Job の移行に関する注意

CSIRT-Pro の UEBA は、検知 Playbook タスクをスケジュール実行する仕組みです。 統計的ベースラインや機械学習による異常スコアリングのエンジンそのものは持ちません。 Elastic の Machine Learning Job を移行する場合は、検知ロジックを参照先の Playbook 側で表現するか、しきい値や集計ベースのルールに置き換える必要があります。

---

KQL / Lucene から PRQL へのクエリ変換

基本検索

Elastic (KQL)Elastic (Lucene)CSIRT-Pro (PRQL)

action: "DENY" and src_ip: "192.168.1.100"

action:"DENY" AND src_ip:"192.168.1.100"

from `firewall_logs`
filter action == "DENY"
filter src_ip == "192.168.1.100"

ワイルドカード検索

Elastic (KQL)Elastic (Lucene)CSIRT-Pro (PRQL)

src_ip: 192.168.*

src_ip:192.168.*

from `firewall_logs`
filter src_ip ~= "^192\\.168\\."

正規表現のエスケープ

~= の右辺は正規表現です。 PRQL の "..." 文字列はバックスラッシュをエスケープ処理するため、. などのメタ文字をリテラルとして扱うには \\. のように二重に書きます。

範囲検索

Elastic (KQL)Elastic (Lucene)CSIRT-Pro (PRQL)

status_code >= 400 and bytes > 1000

status_code:[400 TO *] AND bytes:{1000 TO *}

from `web_logs`
filter status_code >= 400
filter bytes > 1000

NOT 条件

Elastic (KQL)Elastic (Lucene)CSIRT-Pro (PRQL)

action: * and not action: "ALLOW"

NOT action:"ALLOW"

from `firewall_logs`
filter action != "ALLOW"

OR 条件

Elastic (KQL)Elastic (Lucene)CSIRT-Pro (PRQL)

action: "DENY" or action: "DROP"

action:"DENY" OR action:"DROP"

from `firewall_logs`
filter (action == "DENY" || action == "DROP")

集計（Aggregation）

Elastic (Aggregation DSL)CSIRT-Pro (PRQL)

GET firewall-logs/_search
{
  "size": 0,
  "query": {
    "term": { "action": "DENY" }
  },
  "aggs": {
    "by_src_ip": {
      "terms": {
        "field": "src_ip",
        "size": 10,
        "order": { "_count": "desc" }
      }
    }
  }
}

from `firewall_logs`
filter action == "DENY"
group {src_ip} (
  aggregate {cnt = count this}
)
sort {-cnt}
take 10

時系列集計（Date Histogram）

Elastic (Aggregation DSL)CSIRT-Pro (PRQL)

GET auth-logs/_search
{
  "size": 0,
  "query": {
    "term": { "event_type": "login_failed" }
  },
  "aggs": {
    "events_per_hour": {
      "date_histogram": {
        "field": "@timestamp",
        "calendar_interval": "hour"
      }
    }
  }
}

from `auth_logs`
filter event_type == "login_failed"
group {hour = to_start_of_hour __time} (
  aggregate {cnt = count this}
)
sort {hour}

複数集計関数

Elastic (Aggregation DSL)CSIRT-Pro (PRQL)

GET proxy-logs/_search
{
  "size": 0,
  "aggs": {
    "by_host": {
      "terms": { "field": "host", "size": 10 },
      "aggs": {
        "total_bytes": { "sum": { "field": "bytes" } },
        "avg_response": { "avg": { "field": "response_time" } },
        "max_response": { "max": { "field": "response_time" } }
      }
    }
  }
}

from `proxy_logs`
group {host} (
  aggregate {
    cnt = count this,
    total_bytes = sum bytes,
    avg_response = average response_time,
    max_response = max response_time
  }
)
sort {-cnt}
take 10

サブ集計（Nested Aggregation）

Elastic (Aggregation DSL)CSIRT-Pro (PRQL)

GET firewall-logs/_search
{
  "size": 0,
  "aggs": {
    "by_action": {
      "terms": { "field": "action" },
      "aggs": {
        "by_src_ip": {
          "terms": { "field": "src_ip", "size": 5 }
        }
      }
    }
  }
}

from `firewall_logs`
group {action, src_ip} (
  aggregate {cnt = count this}
)
sort {action, -cnt}

親バケットごとの上位 N について

Elastic のサブ集計は親バケットごとに size 件へ絞れますが、上記の group {action, src_ip} は全組み合わせを返します。 親ごとの上位 N に絞り込みたい場合は、PRQL 側で追加の絞り込みが必要です。

フィールド計算（Script Field）

Elastic (Painless Script)CSIRT-Pro (PRQL)

GET proxy-logs/_search
{
  "script_fields": {
    "response_time_ms": {
      "script": {
        "source": "doc['response_time'].value * 1000"
      }
    }
  }
}

from `proxy_logs`
derive {response_time_ms = response_time * 1000}

---

KQL / Lucene から PRQL への変換早見表

Elastic (KQL/Lucene)	PRQL 相当	備考
field: "value"	filter field == "value"	完全一致
field: value*	filter field ~= "^value"	前方一致（正規表現）
field >= N	filter field >= N	範囲条件
NOT field: "value"	filter field != "value"	否定条件
field: *	filter field != null	存在チェック
field1: A and field2: B	filter field1 == "A" + filter field2 == "B"	AND 条件
field1: A or field2: B	filter (field1 == "A" \|\| field2 == "B")	OR 条件
terms aggregation	group {field} (aggregate {...})	バケット集計（値ごとのグループ化）
date_histogram	group {to_start_of_hour __time}	時系列集計
sum/avg/max/min agg	aggregate {sum/average/max/min}	集計関数
top_hits	group + sort + take	上位レコード
Script field	derive	フィールド計算

---

Beats / Logstash からの移行

Splunk との非対称性: Elasticsearch 互換の取り込み口はない

Splunk は HEC 互換エンドポイントでほぼドロップイン移行できますが、CSIRT-Pro には Elasticsearch の _bulk / _doc 互換エンドポイントはありません。そのため Beats / Logstash / Elastic Agent の output.elasticsearch（ES プロトコル）を CSIRT-Pro に直接向けることはできず、HTTP で取り込み API（/api/v2/ingest）へ送る経路に置き換える必要があります。

送信元	移行方法
Logstash	output { http { ... } } で取り込み API へ送る（grok は Pipeline パースルール / schema-on-read に移行）
Beats（Filebeat / Winlogbeat / Metricbeat 等）	Beats はネイティブに汎用 HTTP 出力を持たないため、Logstash 経由にするか、HTTP 出力を持つ汎用シッパー（fluent-bit / Vector / OpenTelemetry Collector / rsyslog omhttp）へ置き換える
Elastic Agent / Fleet	ES / Logstash 出力前提のため、Logstash 経由 or シッパー置換
汎用シッパー（fluentd / fluent-bit / Vector / OTel / rsyslog）	各 HTTP 出力プラグインで取り込み API（?pipeline_id=... + X-API-Key）へ送る

Logstash → CSIRT-Pro の出力設定例（/api/v2/ingest への HTTP 出力）:

output {
  http {
    url => "https://<your-domain>/api/v2/ingest?pipeline_id=<pipeline_id>"
    http_method => "post"
    format => "json"
    content_type => "application/json"
    headers => { "X-API-Key" => "<your-api-key>" }
  }
}

API キーは Organization → API キー（Ingest スコープ） で発行し、pipeline_id に送信先 Pipeline の ID を指定します。format => "json" は 1 イベントを 1 JSON オブジェクトとして送信し、/api/v2/ingest（単一取り込み）が受け付けます。

実機検証済み（このとおり検索に出ました）

上記設定で Logstash から実際に送信し、CSIRT-Pro の検索で参照できることを確認しました。Logstash は format => "json" 時に @timestamp を自動付与するため、本文にタイムスタンプが無いログでも __time がパースされ、対象 Pipeline の検索に出ます（タイムスタンプの無いログをそのまま送ると parsefailure_* 行きになり検索に出ない点に注意。詳細は Splunk 移行ガイドの注記）。

実際に格納・検索できた messages の例（Logstash イベント全体が JSON で入る）:

{"@timestamp":"2026-06-24T03:04:20.475Z","@version":"1","event":{"original":"{\"action\":\"DENY\",\"src_ip\":\"203.0.113.200\"}"},"host":{"name":"..."}}

検索で確認:

from `<pipeline 名>`
filter messages ~= "203.0.113.200"

Beats エージェントの移行（Filebeat ほか）

Elastic Beats（Filebeat, Winlogbeat, Metricbeat, Packetbeat, Auditbeat）はネイティブの汎用 HTTP 出力を持たないため、間に Logstash を 1 台置き、Beats の出力先を Elasticsearch から Logstash へ向け替えます。Filebeat 側は出力（output）を差し替えるだけで、inputs やモジュール設定はそのまま使えます。

[Filebeat] --beats--> [Logstash: beats input] --HTTP--> [CSIRT-Pro /api/v2/ingest] --> 検索

① Filebeat（filebeat.yml） — 出力を Elasticsearch から Logstash に変更（inputs はそのまま）:

filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /var/log/syslog

# 変更前:
#   output.elasticsearch:
#     hosts: ["elasticsearch:9200"]
#     index: "syslog-%{+yyyy.MM.dd}"
# 変更後: 出力を Logstash に向けるだけ
output.logstash:
  hosts: ["<logstash-host>:5044"]

② Logstash（logstash.conf） — Beats を受けて CSIRT-Pro へ HTTP 転送:

input {
  beats { port => 5044 }
}
output {
  http {
    url => "https://<your-domain>/api/v2/ingest?pipeline_id=<pipeline_id>"
    http_method => "post"
    format => "json"
    content_type => "application/json"
    headers => { "X-API-Key" => "<your-api-key>" }
  }
}

• Filebeat は @timestamp を自動付与するため __time がパースされ、対象 Pipeline の検索に出ます。元のログ行は message / event.original に保持されます（フィールド抽出は Pipeline のパースルールで検索時に実施）。
• Winlogbeat / Metricbeat なども同様に output.logstash へ向けるだけで移行できます。fluent-bit / Vector / OpenTelemetry Collector / rsyslog omhttp など、Beats を受けて HTTP 出力できる他のシッパーでも代替可能です。

Filebeat → Logstash → CSIRT-Pro は実機検証済み

Filebeat 8.13 にログファイルを読ませ、output.logstash → Logstash（beats input）→ http で /api/v2/ingest?pipeline_id=<id> へ転送 → CSIRT-Pro の検索（messages ~= "..."）で参照できることを確認しました。実際に格納された messages（Filebeat の ECS イベント全体が JSON で入る）:

{"@timestamp":"2026-06-24T05:35:07.744Z","@version":"1","agent":{"type":"filebeat","version":"8.13.4"},"event":{"original":"Jun 24 12:30:01 host01 sshd[1001]: … filebeat-e2e-1"},"log":{"file":{"path":"/logs/test.log"}},"message":"Jun 24 12:30:01 host01 sshd[1001]: … filebeat-e2e-1"}

動作確認ラボ（コピペで実行）

ローカルの Docker で Filebeat → Logstash → CSIRT-Pro を実際に流して確認できます。<your-domain> <pipeline_id> <your-api-key>（Ingest スコープ）と <pipeline 名> を自分の値に置き換えて実行してください。

mkdir -p fb-lab/logs && cd fb-lab

# ① Logstash 設定（beats 入力 → CSIRT-Pro へ HTTP 転送）
cat > logstash.conf <<'EOF'
input { beats { port => 5044 } }
output {
  http {
    url => "https://<your-domain>/api/v2/ingest?pipeline_id=<pipeline_id>"
    http_method => "post"
    format => "json"
    content_type => "application/json"
    headers => { "X-API-Key" => "<your-api-key>" }
  }
}
EOF

# ② Filebeat 設定（ファイルを読み Logstash へ送る。Elastic 構成からは output だけ変更）
cat > filebeat.yml <<'EOF'
filebeat.inputs:
  - type: log
    enabled: true
    paths: ["/logs/test.log"]
output.logstash:
  hosts: ["ls:5044"]
EOF

# ③ ネットワーク作成 + Logstash 起動
docker network create fblab 2>/dev/null
docker run -d --name ls --network fblab -e XPACK_MONITORING_ENABLED=false \
  -v "$(pwd)/logstash.conf:/usr/share/logstash/pipeline/logstash.conf" \
  docker.elastic.co/logstash/logstash:8.13.4

# beats 入力が開くまで待つ（この行が表示されたら次へ）
docker logs -f ls 2>&1 | grep -m1 "Starting server on port: 5044"

# ④ Filebeat 起動（--strict.perms=false はマウントした設定の権限チェック回避）
docker run -d --name fb --network fblab \
  -v "$(pwd)/filebeat.yml:/usr/share/filebeat/filebeat.yml:ro" \
  -v "$(pwd)/logs:/logs:ro" \
  docker.elastic.co/beats/filebeat:8.13.4 -e --strict.perms=false

# ⑤ タイムスタンプ付きのログ行を追記（Filebeat が追従して送信）
echo "Jun 24 12:30:01 host01 sshd[1001]: login success migrate-check-1" >> logs/test.log

数秒後（取り込みは非同期のため）、CSIRT-Pro の 検索（Search） で確認します。

from `<pipeline 名>`
filter messages ~= "migrate-check"

投入した1件が、元ログ行を message / event.original に持つ ECS イベントとして返れば、移行経路は成立です。

# 後始末
docker rm -f fb ls && docker network rm fblab

ローカルの CSIRT-Pro に向けて試す場合

クラウド/リモートの CSIRT-Pro（https://<your-domain>）にはコンテナからそのまま到達できます。同一ホスト上のローカル CSIRT-Pro（例: localhost:3000）に送る場合は、logstash.conf の URL のホストを host.docker.internal に変え、Docker Desktop 以外（Linux の Docker）では Logstash 起動コマンドに --add-host=host.docker.internal:host-gateway を付けてください。

Logstash パイプラインの移行

Logstash の filter 設定を CSIRT-Pro の Pipeline パースルールに変換します。

Logstash grok フィルタ:

# logstash.conf
filter {
  grok {
    match => {
      "message" => "%{IPORHOST:client_ip} - - \[%{HTTPDATE:timestamp}\] \"%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:http_version}\" %{NUMBER:status_code} %{NUMBER:bytes}"
    }
  }
  date {
    match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
    target => "@timestamp"
  }
}

CSIRT-Pro の Pipeline パースルール:

Parse Settings（正規表現）:

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

Parse Fields（フィールド名）:

client_ip, __time, method, request, status_code(Int32), bytes(Int64)

数値フィールドは フィールド名(型) と型を併記すると、検索時に数値として比較・集計できます（未指定は文字列）。

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

Logstash の grok パターンを手作業で正規表現に変換する代わりに、サンプルログを CSIRT-Pro の「AI 生成」機能に入力すると、パースルールの候補が提示されます。 候補は内容を確認してから適用してください。

Logstash から CSIRT-Pro へのフィルタ対応表

Logstash フィルタ	CSIRT-Pro の対応
grok	Pipeline パースルール（正規表現 + フィールド名）
date	__time フィールドの指定
mutate (rename)	Parse Fields でフィールド名を直接指定
mutate (add_field)	Pipeline パースルール or PRQL derive
mutate (convert)	Parse Fields で フィールド名(型) と型指定（例: status_code(Int32), bytes(Int64)。正規表現・JSON 両モードで有効）
geoip	脅威インテリジェンス連携
dns	Playbook での DNS 照会
drop	Pipeline パースルール（マッチしないログは parsefailure テーブルへ）
output (elasticsearch)	取り込み API

---

Detection Rule の移行

Elastic Security の Detection Rule を CSIRT-Pro の UEBA タスクまたは Playbook に移行します。

Threshold Rule

Elastic (Detection Rule)CSIRT-Pro (UEBA タスク / Playbook)

{
  "type": "threshold",
  "query": "event.action: \"login_failed\"",
  "threshold": {
    "field": ["source.ip"],
    "value": 10
  },
  "interval": "5m"
}

検知 Playbook タスクとして 5 分間隔でスケジュール実行します。

from `auth_logs`
filter action == "login_failed"
filter __time > (now) - (to_interval_minute 5)
group {src_ip} (
  aggregate {cnt = count this}
)
filter cnt > 10

EQL (Event Query Language)

Elastic (EQL)CSIRT-Pro (PRQL)

sequence by source.ip with maxspan=5m
  [authentication where event.action == "login_failed"] with runs=5
  [authentication where event.action == "login_success"]

# ログインの失敗後に成功したパターンを検出
from `auth_logs`
filter action == "login_failed" || action == "login_success"
sort {src_ip, __time}
group {src_ip} (
  aggregate {
    failed_count = sum (cond_if (action == "login_failed") 1 0),
    success_count = sum (cond_if (action == "login_success") 1 0),
    first_event = min __time,
    last_event = max __time
  }
)
filter failed_count >= 5
filter success_count > 0

シーケンス検知の近似について

上記は「失敗が一定回数以上、かつ成功がある」を集計で近似したものです。 EQL の sequence（イベントの順序付きの並び）そのものは表現していません。 厳密な順序検知が必要な場合は、検知 Playbook 側でロジックを実装してください。

---

データ移行手順

Step 1: Pipeline の作成

Elastic の index（データソース）ごとに、CSIRT-Pro の Pipeline を 1 つ作成します。 Pipeline 名は任意で、Elastic の index 名や Beats の種類（filebeat / winlogbeat 等）に合わせる必要はありません（検索では from \`` で参照します）。 パースルールはログ形式に応じて設定します。Beats のモジュールですでに構造化された JSON は、JSON モード（パース定義を空にする）でそのまま取り込めます。

Step 2: ログ転送の切り替え

[移行フェーズ]

Phase 1: 準備（1-2 週間）
  - CSIRT-Pro Pipeline の作成
  - パースルールの設定・テスト
  - 取り込み API の接続確認

Phase 2: 並行運用（2-4 週間）
  - Beats / Logstash から CSIRT-Pro への並行送信開始
  - Detection Rule を Playbook（検知ロジック）として作成し、UEBA でスケジュール実行
  - Dashboard の再構築

Phase 3: 切り替え（1 週間）
  - CSIRT-Pro を主系に切り替え
  - Elastic への送信を停止

Phase 4: 運用安定化（2 週間）
  - モニタリングと微調整
  - チームトレーニング

過去ログの移行（バックフィル）

本手順は新規ログの転送切り替えが中心です。 Elasticsearch 内の過去ログを移行する場合は、_search（scroll / PIT）でエクスポートし、取り込み API へ再投入してください。 並行運用期間を設け、過去分は一定期間 Elastic を参照系として残す方法もあります。

Step 3: Detection Rule の移行

Elastic Rule タイプ	CSIRT-Pro 移行先	方法
Custom Query	UEBA タスク	KQL を PRQL に変換して定期実行
Threshold	UEBA タスク	集計クエリ + しきい値判定
EQL	Playbook	シーケンス検知を Playbook で実装
ML Job	UEBA タスク + Playbook	しきい値・集計ベースのルールに置き換え
Indicator Match	脅威インテリジェンス連携	IOC マッチングで検知

ML Job の置き換え

CSIRT-Pro の UEBA は機械学習による異常スコアリングのエンジンを持ちません。 Elastic の ML Job は、集計としきい値で近似できる検知ルールに置き換えるか、参照先 Playbook 側に検知ロジックを実装します。

Step 4: Dashboard の再構築

1. Kibana Dashboard の各 Visualization のクエリを PRQL に変換
2. CSIRT-Pro Dashboard で新しいパネルを作成
3. 可視化タイプ（Timeline, Bar, Pie 等）を設定
4. レイアウトを調整して保存

移行中の注意

Elastic のライセンスレベルによっては、一部の機能（ML Jobs, Alert 等）が使えない場合があります。 CSIRT-Pro では UEBA と Playbook を追加ライセンスなしで利用できます。 ただし UEBA は検知 Playbook タスクのスケジュール実行を担う仕組みであり、機械学習エンジンを内蔵するものではありません。

---

次のステップ

• クイックスタート CSIRT-Pro の基本操作を体験
• 検索・クエリの詳細 PRQL クエリ言語の完全ガイド
• Pipeline の詳細 パースルール設計のベストプラクティス
• UEBA の詳細 検知 Playbook タスクのスケジュール実行
• 動作前提・制約 取り込み方法・保持期間・レート制限
• プラン・提供機能・上限 プランごとの機能と上限
• 責任分界点 S&J と顧客の責任範囲