障害直後の証拠を1コマンドで集める方法――修復しないスナップショットの実測

落ちた直後に再起動へ手を伸ばす前に、消える証拠を1つの場所へ残せるでしょうか。定期実行が動かなかった、自動処理が失敗した、プロセスの状態が怪しい――そんな場面で、対象を止めず直さず、journal・プロセス・設定ファイルを採取する最小の道具を作りました。

2026-08-18 09:46:21 JSTの実走では、cronのjournal本文20行、プロセス24件、パス属性4件、設定ファイルのSHA-256を3件採取しました。証拠payloadは5ファイルで、最後にmanifestを加えたバンドルは6ファイルです。

これは自動修復器ではありません。異常名を外側から渡されると、新しい証拠バンドルを作って停止します。障害時に「何を先に見るべきか」と「修復機能まで同じ道具へ入れるべきか」を判断したい、自宅ラボの運用者向けの工作です。

修復の前にしか残らない状態がある

所長の事故記録を読み返すと、復旧操作の前後で形が変わる材料が繰り返し現れました。

• 2026-07-29のcron事故では、20:30にjournalへ発火記録が残った一方、本体とログの実行痕跡は0でした。
• 管理外プロセスが制御ソケットを占有した事故では、停止するとクライアント再接続によって再生成され、プロセス断面が変化しました。正規起動は6回失敗し、最終的には停止から1秒後の正規起動で復旧しています。
• 実行指定を必須にした直後の事故では、async便6件が起動時に失敗し、completion eventのverificationに残った原因文字列が入口になりました。

journalだけでも、アプリケーションログだけでも、設定ファイルだけでも答えが出ないことがあります。以前作ったレーンの番犬もjournalとmtimeを複数系統で読みますが、対象を再起動・停止・削除する経路は持ちません。今回の道具は、その観測の直後に「いまの断面」を固定するための小さな続きです。

採取と修復を分けるように設計した

選択肢は、収集した証拠からそのまま再起動まで進む設計と、通知だけ出す設計、証拠を固定して人の判断へ渡す設計でした。今回は最後を選びました。

異常かどうかは判定しません。--anomalyで短い名前を受け取り、外側の検知器が付けた文脈としてmanifestへ記録するだけです。通知、signal、再起動、削除、修復の関数やサブコマンドはありません。

また、プロセスは引数を読まず、PID・UID・状態・名前だけを採ります。事故解析の情報量は減りますが、実行中の機密文字列を複製しにくい側へ倒しました。

採取の順番は次のとおりです。

図1で重要なのは、manifestが最後に作られる点です。5つのpayloadを一度だけ書き、そのハッシュと採取件数をmanifestへ記録します。manifest自身をハッシュ一覧へ入れると循環するため、物理ファイルは6つでも、記録対象のartifactは5つです。

各ファイルはO_EXCLで作ります。これは同じアルバムのページへ上書きしないための留め具です。ランダムな新規ディレクトリを作るため、既存バンドルを開いたり追記したりしません。

実走では6ファイルを作り、既存バンドルを変えなかった

Linuxホストでcronを対象に単発実走しました。外部コマンドはread-onlyのjournalctl 1本だけで、プロセス断面、mtime、容量、ハッシュはPython標準ライブラリで読みました。

$ mkdir -p run-bundles
$ python3 e3-artifact.py --anomaly theme-e3-material-recovery --output-root run-bundles --config rules.md --config lanes/watchdog/watchdog.cron --config lanes/publish/evidence_tick.py --journal-unit cron
captured_at: 2026-08-18T09:46:21+09:00
bundle: run-bundles/snapshot-20260818T094621+0900-f2a0743e
anomaly: theme-e3-material-recovery
captured: journal_lines=20 processes=24 paths=4 config_hashes=3 artifacts=5
journal_returncode: 0
boundary: evidence captured; no process signal, restart, deletion, notification, or repair was attempted

この出力のartifacts=5はpayload数です。manifestを含めると物理ファイル数は6です。実走後、manifest内の5つのSHA-256と各payloadを再計算した値は全件一致しました。バンドルディレクトリはmode 700、6ファイルはすべてmode 600でした。

同じ出力rootへ2回目のプローブを実行すると、バンドル数は2になりました。1回目の6ファイルを実行前後で再計算したハッシュは全件不変です。これはファイルシステムの不変属性ではなく、このプログラムが新規ディレクトリとO_EXCLで既存物の上書きを拒む、という範囲の保証です。実行後の別主体による変更まで防ぐものではないため、必要ならmanifestのハッシュを別途照合します。

図2の「変化しない側」は、2回目の実行後も1回目の6ファイルが変わらなかった点です。ただし、常設運転の証拠ではありません。現時点で確かめたのは、手で渡した1つの異常名から1バンドルを作り、停止できたことだけです。

観測した事実

journal 20行、プロセス24件、パス属性4件、設定ハッシュ3件を採取し、5つのpayloadとmanifestを持つ6ファイルのバンドルを作りました。5つのSHA-256は再計算値と一致しました。

解釈

修復前の状態を、個別のログや端末出力に散らしたままにせず、後から読み直せる単位へまとめる用途には足ります。

未確定

cron、timer、異常検知器へ常設接続した実績はありません。どの異常に接続するかも、まだ固定していません。

証拠を読めることを完了条件にする

AIや自動処理の「設定した」「採取した」という自己申告だけでは、障害時の判断材料として弱いものです。自己申告を検証した記録で扱ったように、ここではmanifestとSHA-256を完了根拠にします。

また、失敗理由を見えないまま成功扱いにすると、次の切り分けが難しくなります。except passが隠したサイレント失敗で見た問題と同じく、journalを読めない場合もreturn codeとstderrをjournal.txtへ残し、ほかの採取は続けます。今回の実走はreturn code 0だったため、その退化ケースではありません。

使う前に知っておきたい境界

• --journal-lines 20で作られるjournal.txtは29行でした。採取時刻、unit、return code、件数、stdout/stderr見出しの9行が加わるためで、journal本文は指定どおり20行です。
• この道具は異常を検知しません。--anomalyは外側で検知した異常の短い名前です。
• プロセス一覧は全コマンドラインを保存しません。そのため、引数が必要な解析には別の、意図的に設計した採取手段が必要です。
• いまは手動起動の単発実走です。すぐに監視へ常設接続するより、まず隔離した出力rootで自分の設定ファイルを1つ指定して、保存される内容と権限を確認するほうがよいでしょう。

次は二つのバンドルの差だけを読む

今回の現在地は、壊れた直後の状態を6ファイルの証拠バンドルへ残せたところです。再起動も削除もしていません。一方で、異常検知器との接続は未実装です。

次に作りたいのは、二つのバンドルを横に並べ、設定ハッシュ、mtime、プロセス名の差だけを読める比較器です。修復ボタンは、まだ作業台の外に置きます。

コピペで再現

前提条件 Linuxのprocfs、journalctl、Python 3が必要です。追加ライブラリは不要です。--output-rootには、あらかじめ存在するディレクトリを渡してください。--configは1件以上の通常ファイルを指定します。

次の内容をe3-artifact.pyとして保存します。

#!/usr/bin/env python3
"""異常直後の証拠を新しいバンドルへ保存し、何も直さず止まる。

読むものは journal の末尾、procfs のプロセス断面、ファイル属性、設定ファイル
の内容。外部コマンドは read-only の journalctl だけ。既存バンドルは上書きせず、
新規バンドル内のファイルも一度だけ作る。通知・signal・再起動・削除・修復はしない。
"""
from __future__ import annotations

import argparse
import hashlib
import json
import os
import subprocess
import sys
import uuid
from datetime import datetime
from pathlib import Path

JOURNAL_LIMIT = 20
PROCESS_LIMIT = 24
BOUNDARY = (
    "evidence captured; no process signal, restart, deletion, notification, "
    "or repair was attempted"
)

def now_iso() -> str:
    return datetime.now().astimezone().isoformat(timespec="seconds")

def mtime_iso(path: Path) -> str:
    return datetime.fromtimestamp(path.stat().st_mtime).astimezone().isoformat(timespec="seconds")

def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        while True:
            block = stream.read(1024 * 1024)
            if not block:
                break
            digest.update(block)
    return digest.hexdigest()

def capture_journal(unit: str, limit: int) -> tuple[str, int, int]:
    command = ("journalctl", "--unit", unit, "--lines", str(limit), "--no-pager", "--output", "short-iso-precise")
    completed = subprocess.run(command, check=False, capture_output=True, text=True, timeout=20)
    stdout = completed.stdout.rstrip("\n")
    stderr = completed.stderr.rstrip("\n")
    lines = len(stdout.splitlines()) if stdout else 0
    rendered = [f"captured_at: {now_iso()}", f"unit: {unit}", f"returncode: {completed.returncode}", f"lines: {lines}", "", "stdout:", stdout or "(empty)", "", "stderr:", stderr or "(empty)"]
    return "\n".join(rendered) + "\n", completed.returncode, lines

def proc_root() -> Path:
    return Path(os.sep) / "proc"

def capture_processes(limit: int) -> tuple[str, int]:
    rows = ["pid\tuid\tstate\tname"]
    process_dirs = [entry for entry in proc_root().iterdir() if entry.name.isdigit()]
    process_dirs.sort()
    for directory in process_dirs:
        try:
            fields: dict[str, str] = {}
            for raw in (directory / "status").read_text(encoding="utf-8", errors="replace").splitlines():
                if ":" not in raw:
                    continue
                label, value = raw.split(":", 1)
                if label in {"Name", "State", "Uid"}:
                    fields[label] = " ".join(value.split())
            if not {"Name", "State", "Uid"}.issubset(fields):
                continue
            uid = fields["Uid"].split()[0]
            state = fields["State"].split()[0]
            name = fields["Name"].replace("\t", " ")
            rows.append(f"{directory.name}\t{uid}\t{state}\t{name}")
        except OSError:
            continue
        if len(rows) - 1 >= limit:
            break
    return "\n".join(rows) + "\n", len(rows) - 1

def capture_paths(output_root: Path, configs: list[Path]) -> tuple[str, str, int]:
    paths = [output_root, *configs]
    rows = ["role\tpath\tsize_bytes\tmtime"]
    for number, path in enumerate(paths):
        role = "bundle-root" if number == 0 else "config"
        info = path.stat()
        rows.append(f"{role}\t{path}\t{info.st_size}\t{mtime_iso(path)}")
    disk = os.statvfs(output_root)
    unit = disk.f_frsize
    disk_rows = ["path\ttotal_bytes\tfree_bytes\tavailable_bytes", f"{output_root}\t{disk.f_blocks * unit}\t{disk.f_bfree * unit}\t{disk.f_bavail * unit}"]
    return "\n".join(rows) + "\n", "\n".join(disk_rows) + "\n", len(paths)

def capture_config_hashes(configs: list[Path]) -> tuple[str, int]:
    rows = ["path\tsize_bytes\tmtime\tsha256"]
    for path in configs:
        info = path.stat()
        rows.append(f"{path}\t{info.st_size}\t{mtime_iso(path)}\t{sha256_file(path)}")
    return "\n".join(rows) + "\n", len(configs)

def exclusive_write(path: Path, content: bytes) -> None:
    descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    try:
        with os.fdopen(descriptor, "wb") as stream:
            stream.write(content)
    except BaseException:
        try:
            os.close(descriptor)
        except OSError:
            pass
        raise

def validate_inputs(output_root: Path, configs: list[Path]) -> None:
    if not output_root.is_dir():
        raise ValueError("--output-root must be an existing directory")
    if not configs:
        raise ValueError("at least one --config is required")
    for path in configs:
        if not path.is_file():
            raise ValueError(f"config is not a regular file: {path}")

def main() -> int:
    parser = argparse.ArgumentParser(description="異常直後の証拠を集め、修復せず停止する")
    parser.add_argument("--anomaly", required=True, help="外側で検知した異常の短い名前")
    parser.add_argument("--output-root", type=Path, required=True, help="新規バンドルを作る既存ディレクトリ")
    parser.add_argument("--config", type=Path, action="append", default=[], help="mtime と SHA-256 を採る設定ファイル（複数指定可）")
    parser.add_argument("--journal-unit", default="cron")
    parser.add_argument("--journal-lines", type=int, default=JOURNAL_LIMIT)
    parser.add_argument("--process-limit", type=int, default=PROCESS_LIMIT)
    args = parser.parse_args()
    if not 1 <= args.journal_lines <= 200:
        parser.error("--journal-lines must be between 1 and 200")
    if not 1 <= args.process_limit <= 200:
        parser.error("--process-limit must be between 1 and 200")
    try:
        validate_inputs(args.output_root, args.config)
        captured_at = now_iso()
        journal, journal_returncode, journal_lines = capture_journal(args.journal_unit, args.journal_lines)
        processes, process_count = capture_processes(args.process_limit)
        paths, disk, path_count = capture_paths(args.output_root, args.config)
        config_hashes, config_count = capture_config_hashes(args.config)
        bundle_id = datetime.now().astimezone().strftime("snapshot-%Y%m%dT%H%M%S%z-") + uuid.uuid4().hex[:8]
        bundle = args.output_root / bundle_id
        bundle.mkdir(mode=0o700)
        payloads = {"journal.txt": journal.encode("utf-8"), "processes.tsv": processes.encode("utf-8"), "paths.tsv": paths.encode("utf-8"), "disk.tsv": disk.encode("utf-8"), "config-sha256.tsv": config_hashes.encode("utf-8")}
        artifacts = []
        for name in payloads:
            content = payloads[name]
            exclusive_write(bundle / name, content)
            artifacts.append({"name": name, "bytes": len(content), "sha256": hashlib.sha256(content).hexdigest()})
        manifest = {"schema": "incident-evidence-snapshot/v1", "bundle_id": bundle_id, "captured_at": captured_at, "anomaly": args.anomaly, "journal": {"unit": args.journal_unit, "returncode": journal_returncode, "lines": journal_lines}, "processes": process_count, "paths": path_count, "config_hashes": config_count, "artifacts": artifacts, "boundary": BOUNDARY}
        exclusive_write(bundle / "manifest.json", (json.dumps(manifest, ensure_ascii=False, indent=2) + "\n").encode("utf-8"))
    except (OSError, ValueError, subprocess.SubprocessError) as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2
    print(f"captured_at: {captured_at}")
    print(f"bundle: {bundle}")
    print(f"anomaly: {args.anomaly}")
    print(f"captured: journal_lines={journal_lines} processes={process_count} paths={path_count} config_hashes={config_count} artifacts={len(payloads)}")
    print(f"journal_returncode: {journal_returncode}")
    print(f"boundary: {BOUNDARY}")
    return 0

if __name__ == "__main__":
    sys.exit(main())

保存後、既存の設定ファイルを1件以上指定して実行します。

$ mkdir -p run-bundles
$ python3 e3-artifact.py --anomaly manual-check --output-root run-bundles --config ./example.conf --journal-unit cron

ハマり所 --output-rootはスクリプトが作成しません。先に作成してください。journalctlを読めない環境では、return codeとstderrをjournal.txtへ残して他の採取を続けます。またartifacts=5はmanifestを除くpayload数であり、バンドルの物理ファイル数は6です。