Splunk からの移行
このガイドは Splunk Enterprise や Splunk Cloud から CSIRT-Pro への移行方法を説明します。 SPL から PRQL へのクエリ変換、概念の対応関係、データ移行の手順を扱います。
概念マッピング
Splunk と CSIRT-Pro の主要概念の対応関係です。
| Splunk | CSIRT-Pro | 説明 |
|---|---|---|
| Index | Pipeline | ログの取り込み先。ログソースごとに作成 |
| Source / Sourcetype | Pipeline のパースルール | ログ形式の定義(正規表現 + フィールド名。フィールド名(型) で型指定可) |
| Saved Search | 保存済みの可視化(Saved Visualization)/ Playbook | 繰り返し使うクエリやアラート条件。検索画面で可視化として保存し Dashboard で再利用 |
| Alert | Playbook(検知)+ UEBA(スケジュール)/ Case | 検知ロジックを Playbook 化し、UEBA で定期実行。ヒット時に Case 化・通知 |
| Dashboard | Dashboard | データの可視化パネル |
| Notable Event | Case(ケース) | セキュリティイベントのインシデント管理 |
| Splunk SOAR | Playbook | 対応ワークフロー(承認を経て適用) |
| Splunk UBA | Playbook(検知ロジックを作成)+ UEBA(スケジュール実行) | 行動分析の検知ロジックは Playbook として実装し、UEBA で定期実行する。UEBA 自体は検知エンジンではなくスケジューラ |
| HEC (HTTP Event Collector) | HEC 互換エンドポイント(/services/collector/*)/ 取り込み API |
HTTP 経由のログ取り込み。既存 HEC 送信元はほぼそのまま接続可 |
| Universal / Heavy Forwarder | Vector ブリッジ経由 | フォワーダは汎用 HTTP 出力を持たないため、HF は syslog 出力 → Vector、UF は raw TCP → Vector で受けて HTTP 転送(Step 2) |
| SPL | PRQL | クエリ言語 |
| App / Add-on | Integration | 外部連携 |
| Lookup(CSV)/ KV Store | Pipeline(参照データ)+ join |
参照データを Pipeline として取り込み、PRQL の join で照合する。小さな設定値の保持は Playbook 環境変数も利用可 |
HEC からの移行(実機検証済み)
CSIRT-Pro は Splunk HEC 互換エンドポイント /services/collector/event・/raw・/health を実装しています。既存の HEC 送信元は、送信先 URL・トークン・index を変更するだけで CSIRT-Pro に向けられます。ただし HEC の index には CSIRT-Pro の Pipeline ID を指定します(Splunk の元の index 名のままでは code 7 Incorrect index になります。実投入で確認済み)。詳細は Step 2: ログ転送の切り替え を参照。
高スループット用の /api/v2/ingest は pipeline_id 指定が必須ですが、これは HEC エンドポイント(/services/collector/*、index で振り分け)とは別系統です。indexer ACK など一部機能には非対応。
UEBA の位置づけと移行先
CSIRT-Pro の UEBA は、検知 Playbook タスクをスケジュール実行する仕組みです。 統計的ベースライン、機械学習による異常スコアリング、ジオロケーション異常検知のエンジンそのものは持ちません。 したがって Splunk UBA やアラートの移行は、検知ロジックを Playbook として作成し、それを UEBA でスケジュール実行する形になります(「UEBA 機能に設定を移す」のではなく、ロジックは Playbook を作る作業です)。
SPL から PRQL へのクエリ変換
基本検索
時間範囲指定
時間範囲セレクタ
CSIRT-Pro の UI では、時間範囲セレクタから「Last 24 hours」などのプリセットを選択できます。 クエリに時間条件を書かなくても、UI 側で時間範囲を制御できます。
stats(集計)
stats(複数集計関数)
timechart(時系列集計)
eval(フィールド計算)
rex(正規表現によるフィールド抽出)
パースルールでの対応
Splunk の rex に相当する処理は、CSIRT-Pro では Pipeline のパースルール(正規表現 + フィールド名)で表現します。
抽出は検索時に行われます(schema-on-read 方式)。
数値として比較・集計するフィールドは、Parse Fields に status_code(Int32)、bytes(Int64) のように型を併記します(未指定は文字列。正規表現・JSON 両モードで有効)。
lookup(ルックアップ結合)
ルックアップ=参照データを Pipeline として持つ
Splunk の CSV ルックアップや KV Store に相当する参照テーブルは、CSIRT-Pro では通常の Pipeline として取り込んで保持し、検索時に join で結合します(結合条件の書き方は join を参照)。
脅威インテリジェンスの IOC との照合は、脅威ハンティング機能を併用することもできます。
where(条件フィルタ)
dedup(重複排除)
rename / fields(フィールド名変更・選択)
transaction(トランザクション)
SPL から PRQL への変換早見表
| SPL コマンド | PRQL 相当 | 備考 |
|---|---|---|
search / index= |
from + filter |
Pipeline 名を from で指定 |
where |
filter |
条件式で絞り込み |
stats count by X |
group {X} (aggregate {cnt = count this}) |
集計 |
stats sum(Y) by X |
group {X} (aggregate {total = sum Y}) |
合計 |
stats avg(Y) |
aggregate {avg_val = average Y} |
平均 |
timechart span=1h |
group {hour = to_start_of_hour __time} |
時系列集計 |
eval |
derive |
フィールド計算・条件分岐 |
rex |
Pipeline パースルール | 検索時にフィールド抽出 |
lookup |
join(参照 Pipeline) |
参照データを Pipeline 化して結合 |
sort |
sort |
ソート(降順は - プレフィックス) |
head / tail |
take |
件数制限 |
dedup |
group + take 1 |
重複排除 |
rename |
derive |
フィールド名変更 |
fields |
select |
フィールド選択 |
table |
select |
表示フィールドの指定 |
top |
group + sort + take |
上位 N 件 |
rare |
group + sort + take |
下位 N 件(昇順ソート) |
transaction |
group + aggregate |
セッション分析 |
データ移行手順
Step 1: Pipeline の設計
Splunk の index(データソース)ごとに、CSIRT-Pro の Pipeline を 1 つ作成します。
Pipeline 名は任意で、Splunk の index 名に合わせる必要はありません(検索では from \で参照します)。
パースルールはログ形式に応じて設定します(props.conf/transforms.conf` の作り直しは不要です。schema-on-read で検索時に抽出します)。
AI によるパースルール生成
各 Pipeline のサンプルログを「AI 生成」ボタンに入力すると、正規表現とフィールド名の候補が提示されます。
候補は内容を確認してから適用してください。
Splunk の props.conf で定義していたパース設定を作り直す手間を抑えられます。
Step 2: ログ転送の切り替え
CSIRT-Pro は Splunk HEC 互換エンドポイントを備えるため、既存の HEC 送信元はほぼそのまま接続できます。フォワーダ構成に応じて方法を選びます。
フォワーダ構成別の移行方法(いずれも実機検証済み):
CSIRT-Pro は HTTP での取り込み(HEC 互換 /services/collector/* と /api/v2/ingest)を入口とします。Splunk のフォワーダは種類ごとに次の経路で接続します。
| 送信元 | 経路 | 移行方法 |
|---|---|---|
| HEC 送信元 | → CSIRT-Pro HEC | 送信先 URL・トークン・index を変えるだけ(ドロップイン)。/services/collector/event に向ける |
| Heavy Forwarder (HF) | HF → syslog → Vector → CSIRT-Pro | HF はサードパーティ向けの汎用 HTTP 出力を持たない。ネイティブの syslog 出力を Vector(syslog source)で受けて HTTP 転送する。設定は下記「Heavy Forwarder(HF)→ syslog → Vector」を参照 |
| Universal Forwarder (UF) | UF → raw TCP → Vector → CSIRT-Pro | UF は S2S/raw TCP 出力のため、間に Vector(socket source で raw TCP を受信)を置き、Vector から CSIRT-Pro へ HTTP 転送する。設定は UF → Vector ブリッジ を参照 |
| 新規アプリ | → HTTP | HEC(/services/collector)または /api/v2/ingest に直接送信 |
本ガイドの経路は実データで検証済み
上記 3 経路すべてを、稼働中の CSIRT-Pro に実際にデータを投入して検証しています。
HEC 送信(index = Pipeline ID)、HTTP クライアント(HEC 送信元・各種シッパー相当)の送信、UF 相当の raw TCP → Vector → CSIRT-Pro、syslog → Vector → CSIRT-Pro のいずれも、投入したログが CSIRT-Pro の検索で参照できることを確認しました。
ログにはパース可能なタイムスタンプを含める(実機確認済み)
取り込んだログは、本文中のタイムスタンプから __time を決定します。タイムスタンプが見つからないログは取り込み時に parsefailure_* テーブルへ振り分けられ、対象 Pipeline の通常検索には出てきません(実投入で確認)。
対応形式は ISO 8601(2026-01-15T10:30:00Z)、Syslog(Jan 15 10:30:00)、Unix epoch、YYYY/MM/DD HH:MM:SS など多数に対応します。フィールド名は問わず、本文のどこかにタイムスタンプ文字列があればパースされます。HEC/Ingest の JSON では timestamp フィールド等に必ず時刻を入れてください。
props.conf / transforms.conf は移行不要
Splunk の index 時フィールド抽出(props.conf / transforms.conf)は、CSIRT-Pro の schema-on-read では不要です。フィールドは検索時に抽出されるため、抽出ルールは Pipeline のパースルール側に集約します。
HEC エンドポイント(ドロップイン):
# 変更前(Splunk HEC)
curl -X POST "https://splunk-server:8088/services/collector/event" \
-H "Authorization: Splunk <HEC_TOKEN>" \
-d '{"event": {"action":"DENY","src_ip":"192.168.1.100"}, "index": "firewall"}'
# 変更後(CSIRT-Pro の HEC 互換エンドポイント)
# URL・トークンを変更し、index には CSIRT-Pro の Pipeline ID を指定する
# event 内にタイムスタンプを入れる(無いと parsefailure 行きになり検索に出ない)
curl -X POST "https://<your-domain>/services/collector/event" \
-H "Authorization: Splunk <your-api-key>" \
-d '{"event": {"timestamp":"2026-01-15T10:30:00Z","action":"DENY","src_ip":"192.168.1.100"}, "index": "<pipeline_id>"}'
- 認証:
Authorization: Splunk <token>(Bearer/X-API-Keyも可)。トークン(API キー)にはingestスコープが必要。 - 投入先(index → Pipeline): HEC の
indexフィールド(raw は?index=)には CSIRT-Pro の Pipeline ID を指定します(例:"index": "49")。Splunk の元の index 名のままでは解決できず HEC code 7「Incorrect index」を返します(実投入で確認済み)。移行時は 「Splunk index → CSIRT-Pro Pipeline ID」の対応表を作り、HEC 送信元のindexを Pipeline ID に書き換えます。 - 対応: 連続 JSON / 配列 / 改行区切り、
Content-Encoding: gzip、/raw(生データ)、/health。 - 並行運用: HF の出力を Splunk と CSIRT-Pro の両方に向ければ、切り替え前に並行運用できます。
高スループット用に pipeline_id を明示する /api/v2/ingest を使う方法は ログ取り込み API を参照。
疎通確認(コピペで実行・検証済み)
本格的な移行の前に、まず1件だけ送って検索に出ることを確認します。以下は稼働中の CSIRT-Pro で実際に通したコマンドと結果です(Pipeline ID・ドメイン・API キーは自分の値に置き換えてください)。
# ① タイムスタンプ付きイベントを1件送信(ingest API / X-API-Key)
curl -X POST "https://<your-domain>/api/v2/ingest?pipeline_id=<pipeline_id>" \
-H "X-API-Key: <your-api-key>" \
-H "Content-Type: application/json" \
-d '{"timestamp":"2026-01-15T10:30:00Z","action":"DENY","src_ip":"198.51.100.77","via":"smoke-test"}'
// → 200 OK。messages_count が 1 なら受理されたことを示します
{"status":"ok","table_name":"<pipeline 名>","messages_count":1}
数秒後、CSIRT-Pro の 検索(Search)画面で次の PRQL を実行すると、送った1件が出ます:
実際に格納・検索できた messages の中身(このとおり1行ヒット):
HEC で確認する場合
同じ確認を HEC で行うなら、-H "Authorization: Splunk <your-api-key>" で /services/collector/event に {"event":{"timestamp":"...","via":"smoke-test"},"index":"<pipeline_id>"} を送ります(応答は {"code":0,"text":"Success"})。こちらも実投入で検索に出ることを確認済みです。
Heavy Forwarder(HF)→ syslog → Vector
Splunk フォワーダにサードパーティ向けの汎用 HTTP/HEC 出力はない
Splunk の UF / HF が outputs.conf で出せるのは S2S(Splunk 間)と syslog のみで、任意の HTTP エンドポイントへ HEC 形式で POST する機能はありません。outputs.conf の httpout は存在しますが、これは「S2S を HTTP トランスポートでくるんだ Splunk 間通信」(HEC 入力+トークンを持つ Splunk 受信側向け)であり、CSIRT-Pro の取り込み API が期待する形式とは異なります(Splunk 公式 outputs.conf 仕様)。
したがって HF からの移行は、HF のネイティブな syslog 出力を使い、UF と同じ要領で Vector で受けて HTTP 転送します。なお「HEC ドロップイン」が当てはまるのは、標準 HEC 形式で POST するアプリ / SDK / エージェントであって Splunk フォワーダではありません。
① Splunk HF 側(outputs.conf): syslog 出力を Vector に向ける
[syslog]
defaultGroup = csirt_syslog
[syslog:csirt_syslog]
server = <vector-host>:514
type = tcp # udp も可
# Splunk の syslog 出力は RFC3164 形式で送出します
② Vector 側(vector.toml): syslog を受けて CSIRT-Pro へ HTTP 転送(設定と動作確認ラボは ログ取り込み API の「syslog からの取り込み」 を参照)
[sources.syslog_in]
type = "syslog"
mode = "tcp" # HF の type に合わせる(udp なら "udp")
address = "0.0.0.0:514"
[sinks.csirt]
type = "http"
inputs = ["syslog_in"]
uri = "https://<your-domain>/api/v2/ingest?pipeline_id=<pipeline_id>"
method = "post"
encoding.codec = "json"
framing.method = "newline_delimited"
batch.max_events = 1
[sinks.csirt.request.headers]
X-API-Key = "<your-api-key>"
検証状況
syslog → Vector → CSIRT-Pro の区間は実データで検証済み(動作確認ラボ 参照)。HF → syslog の区間は Splunk のネイティブ syslog 出力(outputs.conf [syslog])を使います。
UF → Vector ブリッジ
Universal Forwarder(UF)は raw TCP(tcpout)でログを送出します。CSIRT-Pro は HTTP 取り込みのため、間に Vector を1台置き、raw TCP を受信して CSIRT-Pro へ HTTP 転送します。
① Splunk UF 側(outputs.conf): Vector の受信ポートに raw データを送る
[tcpout]
defaultGroup = csirt_vector
[tcpout:csirt_vector]
server = <vector-host>:9997
sendCookedData = false # Splunk の S2S 整形を行わず raw で送出
② Vector 側(vector.toml): raw TCP を受信し、CSIRT-Pro へ HTTP 転送
[sources.uf_tcp]
type = "socket"
mode = "tcp"
address = "0.0.0.0:9997"
decoding.codec = "bytes"
framing.method = "newline_delimited"
[transforms.to_obj]
type = "remap"
inputs = ["uf_tcp"]
source = '''
# 受信した1行を messages として包む(フィールド抽出は Pipeline 側 schema-on-read で行う)
. = { "raw": to_string(.message) ?? "" }
'''
[sinks.csirt]
type = "http"
inputs = ["to_obj"]
uri = "https://<your-domain>/api/v2/ingest?pipeline_id=<pipeline_id>"
method = "post"
encoding.codec = "json"
framing.method = "newline_delimited"
batch.max_events = 1 # 単一レコード取り込み(1リクエスト=1 JSON)
[sinks.csirt.request.headers]
X-API-Key = "<your-api-key>" # ingest スコープの API キー
- 受信した raw 行は
messages列に格納され、フィールド抽出は Pipeline のパースルールに従って検索時に行われます(schema-on-read)。 - スループットを上げる場合は
batch.max_eventsを増やし、送信先を/api/v2/ingest/batch(messages配列)に切り替えます。 rsyslog/fluent-bitなど他のシッパーでも、raw TCP 受信 → HTTP 転送ができれば同様に利用できます。- syslog(RFC3164 / RFC5424)送信元は、Vector の
syslogsource で受けて同様に転送できます(ログ取り込み API の「syslog からの取り込み」を参照。実機検証済み)。
UF → Vector → CSIRT-Pro は実機検証済み
稼働中の CSIRT-Pro で raw TCP 3 行を送出 → Vector が /api/v2/ingest?pipeline_id=<id> へ HTTP 200 ×3 で転送 → 検索で全件参照できることを確認しました。格納された messages(Vector が raw 行を包んだ形): {"raw":"Jan 15 10:30:00 fw01 kernel: ACTION=DENY src=203.0.113.50 …"}
動作確認ラボ(コピペで実行)
ローカルの Docker で raw TCP(UF 相当)→ Vector → CSIRT-Pro を実際に流して確認できます。<your-domain> <pipeline_id> <your-api-key>(Ingest スコープ)と <pipeline 名> を自分の値に置き換えてください。
mkdir -p uf-lab && cd uf-lab
# ① Vector 設定(raw TCP :9997 受信 → CSIRT-Pro へ HTTP 転送)
cat > vector.toml <<'EOF'
[sources.uf_tcp]
type = "socket"
mode = "tcp"
address = "0.0.0.0:9997"
decoding.codec = "bytes"
framing.method = "newline_delimited"
[transforms.to_obj]
type = "remap"
inputs = ["uf_tcp"]
source = '''
. = { "raw": to_string(.message) ?? "" }
'''
[sinks.csirt]
type = "http"
inputs = ["to_obj"]
uri = "https://<your-domain>/api/v2/ingest?pipeline_id=<pipeline_id>"
method = "post"
encoding.codec = "json"
framing.method = "newline_delimited"
batch.max_events = 1
[sinks.csirt.request.headers]
X-API-Key = "<your-api-key>"
EOF
# ② Vector 起動(--config を明示。付けないと既定の demo 設定が読まれる)
docker run -d --name uf_vector -p 9997:9997 \
-v "$(pwd)/vector.toml:/etc/vector/vector.toml" \
timberio/vector:0.41.1-alpine --config /etc/vector/vector.toml
# 受信待機を確認("Listening" が出れば OK。Ctrl-C で抜ける)
docker logs -f uf_vector 2>&1 | grep -m1 "Listening"
# ③ UF を模した raw TCP を送信(タイムスタンプ付きの行であること)
exec 3<>/dev/tcp/127.0.0.1/9997
printf 'Jan 15 10:30:00 fw01 kernel: ACTION=DENY src=203.0.113.50 dst=10.0.0.9 dpt=22 migrate-check-1\n' >&3
exec 3>&-
# netcat があれば: printf '...\n' | nc 127.0.0.1 9997 でも可
数秒後(取り込みは非同期)、CSIRT-Pro の 検索(Search) で確認:
{"raw":"Jan 15 10:30:00 fw01 kernel: ACTION=DENY src=203.0.113.50 … migrate-check-1"} のように返れば移行経路は成立です。
送信先について
vector.toml の uri は CSIRT-Pro の URL(https://<your-domain>)です。同一ホスト上のローカル CSIRT-Pro(例: localhost:3000)に送る場合は host.docker.internal を使い、Docker Desktop 以外(Linux の Docker)では Vector 起動コマンドに --add-host=host.docker.internal:host-gateway を付けてください。
Step 3: Saved Search の移行
Splunk の Saved Search を CSIRT-Pro の保存済みの可視化(Saved Visualization)または Playbook に移行します。
| Splunk 設定 | CSIRT-Pro の移行先 | 用途 |
|---|---|---|
| 定期レポート用 Saved Search | 保存済みの可視化 + Dashboard | 定期的な可視化 |
| アラート用 Saved Search | Playbook(検知ロジック)+ UEBA でスケジュール | 検知・自動アラート。UEBA は実行スケジュールを担う |
| 自動対応付き Alert | Playbook | 検知から通知、対応申請の自動化 |
Step 4: Dashboard の再構築
Splunk Dashboard を CSIRT-Pro Dashboard で再構築します。
- 既存の Splunk Dashboard の各パネルのクエリを SPL から PRQL に変換
- CSIRT-Pro の Dashboard で新しいパネルを作成
- PRQL クエリと可視化タイプを設定
- レイアウトを調整
Step 5: 並行運用と切り替え
移行期間中は、Splunk と CSIRT-Pro を並行運用することを推奨します。
[移行フェーズ]
Phase 1: 準備(1-2 週間)
- Pipeline 作成、パースルール設定
- ログの並行送信開始
Phase 2: 検証(2-4 週間)
- CSIRT-Pro での検索・分析を検証
- Saved Search / Alert の移行
- Dashboard の再構築
Phase 3: 切り替え(1 週間)
- CSIRT-Pro を主系に切り替え
- Splunk への送信を停止
Phase 4: 運用安定化(2 週間)
- モニタリングと微調整
- チームトレーニング
移行中の注意
並行運用期間中は両方のシステムにログが送信されるため、ストレージコストが一時的に増加します。 Splunk 側の保持期間を短縮するなど、コスト管理に注意してください。
よくある SPL パターンと PRQL への変換
ブルートフォース検知
異常なデータ転送量の検知
複数ソースの横断検索
次のステップ
- クイックスタート CSIRT-Pro の基本操作を体験
- 検索・クエリの詳細 PRQL クエリ言語の完全ガイド
- Pipeline の詳細 パースルール設計のベストプラクティス
- Playbook の詳細 対応ワークフローの構築