一覧SP-07-1 · SIEM：ログの集約と相関

SIEMのログ収集と解析

原記録から検索結果、相関の根拠までをたどり、欠落や誤った結合を確認します。

説明

SIEMは複数のログを取得し、検索できる形へ整えて保存します。検索に出ない理由は、送られていない、届いていない、解析に失敗した、保存期間を過ぎたなど段階ごとに異なります。

4行の固定ログを転送記録に従って取り込みます。実際のJSON解析エラー、再転送の重複、時刻の正規化、保持期間による削除を確認します。

原記録から検索結果、相関の根拠までをたどり、欠落や誤った結合を確認します。

1原記録と受信状態を開く

行3は未受信、行4はJSONが途中で切れています。

from sp07 import *
from sp02 import show
db=connect()
for i,line in enumerate(

RAW原記録と受信状態を開く

,1): print(i,line)
show(db,"SELECT * FROM transport ORDER BY line")

原記録の件数と受信数が一致していません。検索件数だけを見る前に、転送段階を調べます。

2届いた行を解析して保存する

受信済みの行だけを取り込みます。

for line, in db.execute("SELECT line FROM transport WHERE received=1 ORDER BY line").fetchall():
    print(line,

ingest届いた行を解析して保存する(db,RAW[line-1],line))

1と2は保存、4は解析拒否です。構文エラーの原文は拒否用テーブルへ残ります。

3保存と解析失敗を別々に検索する

通常の検索結果と拒否記録を比較します。

show保存と解析失敗を別々に検索する

(db,"SELECT source,id,user,kind FROM event ORDER BY id")
show(db,"SELECT line,reason,raw FROM rejected")

行4はJSONDecodeErrorです。拒否された原文を通常イベントへ黙って読み替えず、解析失敗として追跡します。

4遅れて届いた行を再送する

未受信の行3を取り込み、同じ行をもう一度送ります。

print(ingest遅れて届いた行を再送する

(db,RAW[2],3))
print(ingest(db,RAW[2],3))
show(db,"SELECT COUNT(*) AS stored FROM event")

stored、duplicateとなり保存件数は3です。sourceとidの複合キーで再送を識別しています。IDの再利用がある資料にはこの条件をそのまま使えません。

5異なるタイムゾーンをUTCへそろえる

保存した秒数をUTC文字列で表示します。

show異なるタイムゾーンをUTCへそろえる(db,"SELECT id,datetime(t,'unixepoch') AS utc FROM event ORDER BY t")

日本時間09:00はUTC00:00へ対応します。文字列の見た目だけで9時間の差があると判断すると、相関窓から外れてしまいます。

6保持期限による削除を確認する

00:01より前を通常検索の保存対象から除きます。

cutoff=int(datetime.datetime.fromisoformat('2026-11-13T00:01:00Z').timestamp())
db.execute("

DELETE保持期限による削除を確認する

 FROM event WHERE t<?",(cutoff,))
show(db,"SELECT id,kind FROM event ORDER BY id")

e1が消えました。解析済みだったログでも保持期限後はこの検索テーブルに残りません。保管庫や原記録の保持は別に確認します。

7発生・配送・保存の件数を照合する

転送台帳と最終保存数を表示します。

show発生・配送・保存の件数を照合する

(db,"SELECT SUM(sent) AS sent,SUM(received) AS received_at_first FROM transport")
show(db,"SELECT COUNT(*) AS retained FROM event")
show(db,"SELECT COUNT(*) AS parse_errors FROM rejected")

初回受信3、現在保存2、解析失敗1です。再送と保持期限の履歴を含めて説明すると、件数の違いが追えます。

まとめ

転送・解析・保持の段階を分けて確認します。

再送とIDの一意性を確認します。

発生時刻はタイムゾーンをそろえて検索します。

操作履歴と解説を振り返れます。

Python 3.15 / SQLite / 固定の認証ログ

from sp07 import *
from sp02 import show
db=connect()
for i,line in enumerate(RAW,1): print(i,line)
show(db,"SELECT * FROM transport ORDER BY line")

1 {"id":"e1","source":"auth","time":"2026-11-13T09:00:00+09:00","user":"alice","kind":"fail"}
2 {"id":"e2","source":"auth","time":"2026-11-13T00:01:00Z","user":"alice","kind":"fail"}
3 {"id":"e3","source":"auth","time":"2026-11-13T00:02:00Z","user":"alice","kind":"success"}
4 {"id":
line | sent | received
1 | 1 | 1
2 | 1 | 1
3 | 1 | 0
4 | 1 | 1

for line, in db.execute("SELECT line FROM transport WHERE received=1 ORDER BY line").fetchall():
    print(line,ingest(db,RAW[line-1],line))

1 stored
2 stored
4 rejected

show(db,"SELECT source,id,user,kind FROM event ORDER BY id")
show(db,"SELECT line,reason,raw FROM rejected")

source | id | user | kind
auth | e1 | alice | fail
auth | e2 | alice | fail
line | reason | raw
4 | JSONDecodeError | {"id":

print(ingest(db,RAW[2],3))
print(ingest(db,RAW[2],3))
show(db,"SELECT COUNT(*) AS stored FROM event")

stored
duplicate
stored
3

show(db,"SELECT id,datetime(t,'unixepoch') AS utc FROM event ORDER BY t")

id | utc
e1 | 2026-11-13 00:00:00
e2 | 2026-11-13 00:01:00
e3 | 2026-11-13 00:02:00

cutoff=int(datetime.datetime.fromisoformat('2026-11-13T00:01:00Z').timestamp())
db.execute("DELETE FROM event WHERE t<?",(cutoff,))
show(db,"SELECT id,kind FROM event ORDER BY id")

id | kind
e2 | fail
e3 | success

show(db,"SELECT SUM(sent) AS sent,SUM(received) AS received_at_first FROM transport")
show(db,"SELECT COUNT(*) AS retained FROM event")
show(db,"SELECT COUNT(*) AS parse_errors FROM rejected")

sent | received_at_first
4 | 3
retained
2
parse_errors
1

演習完了