コンテンツにスキップ

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 出力 → VectorUF は 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/ingestpipeline_id 指定が必須ですが、これは HEC エンドポイント(/services/collector/*index で振り分け)とは別系統です。indexer ACK など一部機能には非対応。

UEBA の位置づけと移行先

CSIRT-Pro の UEBA は、検知 Playbook タスクをスケジュール実行する仕組みです。 統計的ベースライン、機械学習による異常スコアリング、ジオロケーション異常検知のエンジンそのものは持ちません。 したがって Splunk UBA やアラートの移行は、検知ロジックを Playbook として作成し、それを UEBA でスケジュール実行する形になります(「UEBA 機能に設定を移す」のではなく、ロジックは Playbook を作る作業です)。


SPL から PRQL へのクエリ変換

基本検索

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

時間範囲指定

index=firewall earliest=-24h latest=now
from `firewall_logs`
filter __time > @2025-01-14
filter __time < @2025-01-15

時間範囲セレクタ

CSIRT-Pro の UI では、時間範囲セレクタから「Last 24 hours」などのプリセットを選択できます。 クエリに時間条件を書かなくても、UI 側で時間範囲を制御できます。

stats(集計)

index=firewall action=DENY
| stats count by src_ip
| sort -count
| head 10
from `firewall_logs`
filter action == "DENY"
group {src_ip} (
  aggregate {cnt = count this}
)
sort {-cnt}
take 10

stats(複数集計関数)

index=firewall action=DENY
| stats count, sum(bytes) as total_bytes, avg(bytes) as avg_bytes by src_ip
from `firewall_logs`
filter action == "DENY"
group {src_ip} (
  aggregate {
    cnt = count this,
    total_bytes = sum bytes,
    avg_bytes = average bytes
  }
)

timechart(時系列集計)

index=firewall action=DENY
| timechart span=1h count by src_ip
from `firewall_logs`
filter action == "DENY"
group {hour = to_start_of_hour __time, src_ip} (
  aggregate {cnt = count this}
)
sort {hour}

eval(フィールド計算)

index=proxy
| eval response_time_ms = response_time * 1000
| eval status_category = case(
    status_code >= 500, "Server Error",
    status_code >= 400, "Client Error",
    status_code >= 300, "Redirect",
    status_code >= 200, "Success")
from `proxy_logs`
derive {
  response_time_ms = response_time * 1000,
  status_category = case [
    status_code >= 500 => "Server Error",
    status_code >= 400 => "Client Error",
    status_code >= 300 => "Redirect",
    status_code >= 200 => "Success",
  ]
}

rex(正規表現によるフィールド抽出)

index=web
| rex field=url "\/(?<path_segment>[^\/]+)\/"
| rex field=user_agent "(?<browser>Chrome|Firefox|Safari|Edge)"

CSIRT-Pro ではパースルールを Pipeline レベルで設定するため、クエリ内でのフィールド抽出は不要です。

# Pipeline のパースルールで抽出されたフィールドをそのまま参照
from `web_logs`
select {__time, path_segment, browser, url, user_agent}

パースルールでの対応

Splunk の rex に相当する処理は、CSIRT-Pro では Pipeline のパースルール(正規表現 + フィールド名)で表現します。 抽出は検索時に行われます(schema-on-read 方式)。 数値として比較・集計するフィールドは、Parse Fields に status_code(Int32)bytes(Int64) のように型を併記します(未指定は文字列。正規表現・JSON 両モードで有効)。

lookup(ルックアップ結合)

index=firewall
| lookup geo_ip_lookup ip as src_ip OUTPUT country, city
| stats count by country

Splunk の lookup(CSV ルックアップ / KV Store)は、参照データを Pipeline として取り込み、PRQL の join で突き合わせます。

# geo_ipIP→国・都市の参照データ)を Pipeline 化しておき、src_ip で結合
from `firewall_logs`
join side:left `geo_ip` (this.src_ip == that.ip)
group {country} (
  aggregate {cnt = count this}
)
sort {-cnt}

ルックアップ=参照データを Pipeline として持つ

Splunk の CSV ルックアップや KV Store に相当する参照テーブルは、CSIRT-Pro では通常の Pipeline として取り込んで保持し、検索時に join で結合します(結合条件の書き方は join を参照)。 脅威インテリジェンスの IOC との照合は、脅威ハンティング機能を併用することもできます。

where(条件フィルタ)

index=auth
| where like(username, "admin%")
| where login_count > 10
from `auth_logs`
filter username ~= "^admin"
filter login_count > 10

dedup(重複排除)

index=firewall
| dedup src_ip, dst_ip
from `firewall_logs`
group {src_ip, dst_ip} (
  sort {-__time}
  take 1
)

rename / fields(フィールド名変更・選択)

index=firewall
| rename src_ip as source_address, dst_ip as dest_address
| fields source_address, dest_address, action
from `firewall_logs`
derive {
  source_address = src_ip,
  dest_address = dst_ip
}
select {source_address, dest_address, action}

transaction(トランザクション)

index=web
| transaction session_id maxspan=30m
| stats count, avg(duration) as avg_duration
from `web_logs`
group {session_id} (
  sort {__time}
  aggregate {
    event_cnt = count this,
    session_start = min __time,
    session_end = max __time
  }
)
derive {duration_sec = session_end - session_start}

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 → syslogVector → CSIRT-Pro HF はサードパーティ向けの汎用 HTTP 出力を持たない。ネイティブの syslog 出力Vectorsyslog source)で受けて HTTP 転送する。設定は下記「Heavy Forwarder(HF)→ syslog → Vector」を参照
Universal Forwarder (UF) UF → raw TCPVector → CSIRT-Pro UF は S2S/raw TCP 出力のため、間に Vectorsocket 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-Prosyslog → 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件が出ます:

from `<pipeline >`
filter messages ~= "smoke-test"

実際に格納・検索できた messages の中身(このとおり1行ヒット):

{"action":"DENY","src_ip":"198.51.100.77","timestamp":"2026-01-15T10:30:00Z","via":"smoke-test"}

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.confhttpout は存在しますが、これは「S2S を HTTP トランスポートでくるんだ Splunk 間通信」(HEC 入力+トークンを持つ Splunk 受信側向け)であり、CSIRT-Pro の取り込み API が期待する形式とは異なります(Splunk 公式 outputs.conf 仕様)。 したがって HF からの移行は、HF のネイティブな syslog 出力を使い、UF と同じ要領で Vector で受けて HTTP 転送します。なお「HEC ドロップイン」が当てはまるのは、標準 HEC 形式で POST するアプリ / SDK / エージェントであって Splunk フォワーダではありません。

[HF] --syslog (TCP/UDP)--> [Vector: syslog source] --HTTP--> [CSIRT-Pro /api/v2/ingest] --> 検索

① 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 転送します。

[UF] --raw TCP--> [Vector: socket source] --HTTP--> [CSIRT-Pro /api/v2/ingest] --> 検索

① 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/batchmessages 配列)に切り替えます。
  • rsyslog / fluent-bit など他のシッパーでも、raw TCP 受信 → HTTP 転送ができれば同様に利用できます。
  • syslog(RFC3164 / RFC5424)送信元は、Vector の syslog source で受けて同様に転送できます(ログ取り込み 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) で確認:

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

{"raw":"Jan 15 10:30:00 fw01 kernel: ACTION=DENY src=203.0.113.50 … migrate-check-1"} のように返れば移行経路は成立です。

# 後始末
docker rm -f uf_vector

送信先について

vector.tomluri は 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 を付けてください。

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 で再構築します。

  1. 既存の Splunk Dashboard の各パネルのクエリを SPL から PRQL に変換
  2. CSIRT-Pro の Dashboard で新しいパネルを作成
  3. PRQL クエリと可視化タイプを設定
  4. レイアウトを調整

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 への変換

ブルートフォース検知

index=auth action=failure
| stats count by src_ip, user
| where count > 5
| sort -count
from `auth_logs`
filter action == "failure"
group {src_ip, user} (
  aggregate {cnt = count this}
)
filter cnt > 5
sort {-cnt}

異常なデータ転送量の検知

index=proxy
| stats sum(bytes_out) as total_bytes by src_ip
| where total_bytes > 1073741824
| eval total_gb = round(total_bytes/1073741824, 2)
| sort -total_gb
from `proxy_logs`
group {src_ip} (
  aggregate {total_bytes = sum bytes_out}
)
filter total_bytes > 1073741824
derive {total_gb = total_bytes / 1073741824}
sort {-total_gb}

複数ソースの横断検索

index=firewall OR index=auth
| stats count by index, src_ip

CSIRT-Pro では、Pipeline ごとに個別にクエリを実行し、Dashboard で統合表示します。

# Pipeline 1
from `firewall_logs`
group {src_ip} (
  aggregate {cnt = count this}
)

# Pipeline 2(別クエリ)
from `auth_logs`
group {src_ip} (
  aggregate {cnt = count this}
)

次のステップ