ログ取り込み API

共通仕様

認証・エラー形式・レート制限・共通ヘッダーは API 概要・認証を参照のこと。

ログデータを CSIRT-Pro に送信するための API である。高スループット向けのコア API（/api/v2/）と、互換目的の管理 API の 2 系統を提供する。

コア API による取り込み（推奨）

高スループットのログ取り込みに最適化したエンドポイントである。取り込みエンドポイントは pipeline_id を必須とする。

単一レコード取り込み

POST /api/v2/ingest?pipeline_id=<pipeline_id>

クエリパラメータ:

パラメータ	型	必須	説明
`pipeline_id`	string	Yes	取り込み先の Pipeline ID

リクエストヘッダー:

Content-Type: application/json
Cookie: jwt=<jwt_token>

リクエストボディ:

任意の JSON オブジェクトを送る。取り込んだ生ログは messages 列に保持され、フィールド抽出は検索時に行われる（schema-on-read）。抽出の挙動は Pipeline の取り込み定義に従う。

ログにはパース可能なタイムスタンプを含める

取り込み時に、ログ本文のタイムスタンプから __time を決定する。タイムスタンプが見つからないログは parsefailure_* テーブルへ振り分けられ、対象 Pipeline の通常検索には現れない（実投入で確認済み）。ISO 8601 / Syslog / Unix epoch など多数の形式に対応し、フィールド名は問わない（本文中に時刻文字列があればよい）。例の timestamp のように必ず時刻を含めること。

{
  "timestamp": "2025-01-15T10:30:00Z",
  "src_ip": "192.168.1.100",
  "dst_ip": "10.0.0.1",
  "action": "DENY",
  "bytes": 1024,
  "message": "Firewall blocked connection"
}

レスポンス:

{
  "status": "ok"
}

curl 例:

curl -X POST "https://<your-domain>/api/v2/ingest?pipeline_id=abc-123" \
  -H "Content-Type: application/json" \
  -H "Cookie: jwt=eyJhbGciOi..." \
  -d '{
    "timestamp": "2025-01-15T10:30:00Z",
    "src_ip": "192.168.1.100",
    "action": "DENY"
  }'

バッチ取り込み

複数のレコードを一度に送る。 pipeline_id はクエリパラメータで指定し、ボディには messages（文字列の配列）を渡す。配列の各要素が 1 レコードになる。

POST /api/v2/ingest/batch?pipeline_id=<pipeline_id>

リクエストボディ:

{
  "messages": [
    "2025-01-15T10:30:00Z DENY 192.168.1.100 10.0.0.50 443 1024",
    "2025-01-15T10:30:01Z ALLOW 192.168.1.101 10.0.0.51 80 2048"
  ]
}

レスポンス:

{
  "status": "ok",
  "count": 2
}

管理 API による取り込み

バルク取り込み

POST /api/user/<user_id>/bulk/<pipeline_name>/

パスパラメータ:

パラメータ	型	説明
`user_id`	string	ユーザー ID
`pipeline_name`	string	Pipeline 名

リクエストボディ:

JSON 配列形式で送る。

[
  {"timestamp": "2025-01-15T10:30:00Z", "message": "log1"},
  {"timestamp": "2025-01-15T10:30:01Z", "message": "log2"}
]

Splunk HEC 互換エンドポイント

Splunk HTTP Event Collector (HEC) 形式で POST する送信元（アプリ、SDK、ログエージェント、curl 等）から直接投入できる互換エンドポイントを提供する。既存の HEC 送信元は、送信先 URL とトークンの変更だけで接続できる。

Splunk フォワーダ（UF / HF）はここではない

Splunk の Universal / Heavy Forwarder は HEC を話さず、outputs.conf の出力は S2S（Splunk 間）と syslog のみである。フォワーダからの取り込みは本エンドポイントではなく、UF は raw TCP → Vector、HF は syslog → Vector のブリッジ経由で行う（Splunk からの移行参照）。

エンドポイント	メソッド	用途
`/services/collector/event`（`/services/collector`、`/event/1.0`）	POST	構造化イベント（`event` フィールド）
`/services/collector/raw`（`/raw/1.0`）	POST	生データ。`?index=` で投入先を指定
`/services/collector/health`（`/health/1.0`）	GET	ヘルスチェック（HEC code 17）

認証: Authorization: Splunk <token>（Authorization: Bearer <token> / X-API-Key: <token> も可）。トークンには ingest スコープが必要。

投入先の決定（index → Pipeline）: イベントの index フィールド（raw は ?index=）で振り分ける。index には CSIRT-Pro の Pipeline ID を指定する（例: "index": "49"）。Pipeline 名や元の Splunk index 名では解決できず、HEC code 7「Incorrect index」を返す（実投入で確認済み）。

対応: 連続 JSON（{"event":"a"}{"event":"b"}）/ 改行区切り / 配列、Content-Encoding: gzip。time、host、source、sourcetype 等のメタは現状取り込み時に保持されない。

# event エンドポイント（index には Pipeline ID を指定）
# event 内にタイムスタンプを含めること（無いと parsefailure_ 行きになり検索に出ない）
curl "https://<your-domain>/services/collector/event" \
  -H "Authorization: Splunk <api_key>" \
  -d '{"event":{"timestamp":"2025-01-15T10:30:00Z","action":"DENY","src_ip":"192.168.1.100"},"index":"49"}'

# raw エンドポイント（?index= に Pipeline ID を指定）
curl "https://<your-domain>/services/collector/raw?index=49" \
  -H "Authorization: Splunk <api_key>" \
  --data-binary @access.log

/api/v2/ingest との違い

高スループット用の /api/v2/ingest は pipeline_id の指定が必須だが、HEC エンドポイント（/services/collector/*）は index で振り分けるため pipeline_id は不要。indexer ACK には対応しない。

syslog からの取り込み（Vector ブリッジ）

CSIRT-Pro は HTTP で取り込むため、syslog（RFC3164 / RFC5424、UDP/TCP）を直接は受信しない。既存の syslog 送信元（ネットワーク機器、ファイアウォール、Linux ホスト等）は、間に Vector を 1 台置き、syslog を受信して CSIRT-Pro へ HTTP 転送する。

[syslog 送信元] --UDP/TCP--> [Vector: syslog source] --HTTP--> [/api/v2/ingest] --> 検索

Vector 設定（vector.toml）:

[sources.syslog_in]
type = "syslog"
mode = "udp"            # tcp も可
address = "0.0.0.0:514" # 任意のポート（特権ポートを避けるなら 5514 等）

[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                 # 単一レコード取り込み（1リクエスト＝1 JSON）
[sinks.csirt.request.headers]
X-API-Key = "<your-api-key>"         # ingest スコープの API キー

Vector の syslog source が RFC3164 / RFC5424 をパースし、timestamp / hostname / appname / severity / facility / message などの構造化フィールドに展開する。timestamp から __time が決まる。
既存の rsyslog / syslog-ng から直接 HTTP 転送することも可能だが、いずれも HTTP 出力モジュールの導入が前提である（rsyslog の omhttp は標準ディストリのリポジトリには含まれず Adiscon 配布リポジトリやソースビルドが必要、syslog-ng は http() ドライバ）。さらに JSON テンプレートでの整形（RSYSLOG_FileFormat のようなプレーンテキストではなく Content-Type: application/json に合う本文）も必要である。導入の手間を避けるなら、syslog source を内蔵する Vector が最も簡単である（本ページの設定は Vector で実機検証済み）。
スループットを上げる場合は batch.max_events を増やし、/api/v2/ingest/batch（messages 配列）に切り替える。

syslog → Vector → CSIRT-Pro は実機検証済み

RFC3164 の syslog（UDP）を Vector に送出 → Vector が /api/v2/ingest?pipeline_id=<id> へ HTTP 200 で転送 → CSIRT-Pro の検索（messages ~= "..."）で参照できることを確認した。Vector の syslog source が構造化した messages の例:

{"appname":"firewall","facility":"auth","hostname":"fw01","severity":"crit","source_type":"syslog","timestamp":"2026-06-24T13:00:01Z","message":"DENY src=203.0.113.77 dst=10.0.0.9 dpt=22"}

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

ローカルの Docker で syslog → Vector → CSIRT-Pro を実際に流して確認できる。<your-domain> <pipeline_id> <your-api-key>（Ingest スコープ）と <pipeline 名> を自分の値に置き換える。

mkdir -p syslog-lab && cd syslog-lab

# ① Vector 設定（syslog 受信 → CSIRT-Pro へ HTTP 転送）
cat > vector.toml <<'EOF'
[sources.syslog_in]
type = "syslog"
mode = "udp"
address = "0.0.0.0:5514"

[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>"
EOF

# ② Vector 起動（--config を明示。付けないと既定の demo 設定が読まれる）
docker run -d --name syslog_vector -p 5514:5514/udp \
  -v "$(pwd)/vector.toml:/etc/vector/vector.toml" \
  timberio/vector:0.41.1-alpine --config /etc/vector/vector.toml
docker logs -f syslog_vector 2>&1 | grep -m1 "Listening"

# ③ RFC3164 の syslog を UDP 送信（<PRI>TIMESTAMP HOST TAG: MSG）
printf '<34>Jun 24 13:00:01 fw01 firewall: DENY src=203.0.113.77 dst=10.0.0.9 migrate-check-1\n' > /dev/udp/127.0.0.1/5514
# logger があれば: logger -n 127.0.0.1 -P 5514 -d "firewall: ... migrate-check-1" でも可

数秒後（取り込みは非同期）、CSIRT-Pro の 検索（Search） で確認:

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

appname / hostname / severity / timestamp 等に構造化された 1 件が返れば成立である。

# 後始末
docker rm -f syslog_vector

送信先について

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 を付ける。

データフロー

ログ取り込みは次の流れで処理される。

受信: クライアントが取り込みエンドポイントにログデータを送る。
取り込み定義の取得: Pipeline の取り込み定義（正規表現とフィールド名、または JSON モード設定）を参照する。
キューイング: ログデータをメッセージバス（ストリーミング取り込み基盤）へ非同期で渡し、クライアントへ応答を返す。
書き込み: 取り込みワーカーがメッセージバスから取り出し、列指向ログ分析基盤へ書き込む。
抽出: フィールド抽出は保存時ではなく検索時に行う（schema-on-read）。

非同期処理とバッチ送信

取り込みは非同期で処理されるため、API レスポンスは速やかに返る。大量のログを送る場合はバッチ取り込みを使うとよい。

対応データフォーマット

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

取り込み定義の正規表現が空の場合は JSON モードとして扱う。このとき messages を JSON とみなして抽出し、型注釈があれば型別に抽出する。不正な JSON は NULL になる。正規表現が空でない場合は正規表現モードとして扱う。

ヘルスチェック

GET /api/v2/health

レスポンス:

{
  "status": "ok"
}