Runbook（運用手順書）で定義すべきこと
障害ゼロ秒対応を支える書き方と Python 自動収集 Tips

## 手順

### 1. 現在のプライマリを確認する
```bash
# どのノードがプライマリかを確認（pg_is_in_recovery = f がプライマリ）
psql -h db-cluster.internal -U admin -c "SELECT pg_is_in_recovery();"
```
**期待される出力:**
```
 pg_is_in_recovery
-------------------
 f
(1 row)
```
⚠️ プライマリが応答しない場合は手順 3 へ進む

### 2. スタンバイの同期状態を確認する
```bash
psql -h db-standby.internal -U admin -c "SELECT pg_last_wal_receive_lsn(), pg_last_wal_replay_lsn();"
```
**確認ポイント:** receive_lsn と replay_lsn が一致していること（ラグが 0 に近いこと）

### 3. フェイルオーバーを実行する
```bash
# pg_promote() でスタンバイをプライマリに昇格
psql -h db-standby.internal -U admin -c "SELECT pg_promote();"
```

## ロールバック手順

**実行判断基準:** 手順 3 の完了後 5 分経過してもヘルスチェックが通らない場合

### 1. フェイルオーバー前のノードを復帰させる
```bash
# 旧プライマリを再起動
ssh db-primary.internal "sudo systemctl restart postgresql"
```

### 2. スタンバイに戻す設定を適用する
```bash
psql -h db-primary.internal -U admin -c "SELECT pg_ctl_promote();"
# ※ 新プライマリ（元スタンバイ）がある状態でのみ実行
```

### 3. アプリの接続先を元に戻す
```bash
# 環境変数を旧プライマリに変更して再起動
export DB_HOST=db-primary.internal
sudo systemctl restart app-service
```

"""
Runbook 実行前の事前確認用スクリプト。
psutil で CPU・メモリ・ディスク・プロセスを取得して表示する。
インストール: pip install psutil
"""
import psutil
from datetime import datetime

def system_snapshot() -> dict:
    cpu    = psutil.cpu_percent(interval=1)
    mem    = psutil.virtual_memory()
    disk   = psutil.disk_usage("/")
    uptime = datetime.fromtimestamp(psutil.boot_time()).strftime("%Y-%m-%d %H:%M:%S")

    return {
        "timestamp": datetime.now().isoformat(),
        "uptime_since": uptime,
        "cpu_percent":  cpu,
        "mem_total_gb": round(mem.total / 1024**3, 1),
        "mem_used_pct": mem.percent,
        "disk_total_gb": round(disk.total / 1024**3, 1),
        "disk_used_pct": disk.percent,
    }

def find_process(name: str) -> list[dict]:
    """指定プロセス名が動いているか確認する。"""
    result = []
    for proc in psutil.process_iter(["pid", "name", "status", "create_time"]):
        if name.lower() in proc.info["name"].lower():
            result.append({
                "pid":    proc.info["pid"],
                "name":   proc.info["name"],
                "status": proc.info["status"],
                "since":  datetime.fromtimestamp(proc.info["create_time"]).strftime("%H:%M:%S"),
            })
    return result

if __name__ == "__main__":
    snap = system_snapshot()
    print("=== システム状態スナップショット ===")
    for k, v in snap.items():
        print(f"  {k}: {v}")

    print("\n=== PostgreSQL プロセス確認 ===")
    procs = find_process("postgres")
    if procs:
        for p in procs:
            print(f"  PID {p['pid']} | {p['name']} | {p['status']} | 起動: {p['since']}")
    else:
        print("  ⚠️  PostgreSQL プロセスが見つかりません")

"""
指定ログファイルの末尾 N 行から ERROR / WARN を抽出する。
Runbook のヘルスチェック・障害初動で使う。
"""
import re
from pathlib import Path
from collections import Counter

LOG_FILE = Path("/var/log/app/application.log")
TAIL_LINES = 200
PATTERN = re.compile(r"\b(ERROR|WARN|CRITICAL)\b", re.IGNORECASE)

def tail(path: Path, n: int) -> list[str]:
    """ファイルの末尾 n 行を返す。大きなファイルでも効率的。"""
    with path.open("rb") as f:
        f.seek(0, 2)
        size = f.tell()
        buf = bytearray()
        lines = 0
        pos = size - 1
        while pos >= 0 and lines < n:
            f.seek(pos)
            byte = f.read(1)
            if byte == b"\n" and pos != size - 1:
                lines += 1
            buf.extend(byte)
            pos -= 1
        return buf[::-1].decode("utf-8", errors="replace").splitlines()

def analyze_log(lines: list[str]) -> None:
    level_counter = Counter()
    print(f"{'レベル':<10} {'件数':>5}   メッセージ（最初の3件）")
    print("-" * 60)
    collected = {"ERROR": [], "WARN": [], "CRITICAL": []}

    for line in lines:
        m = PATTERN.search(line)
        if m:
            level = m.group(1).upper()
            level_counter[level] += 1
            if len(collected.get(level, [])) < 3:
                collected.setdefault(level, []).append(line.strip())

    for level, count in sorted(level_counter.items()):
        print(f"{level:<10} {count:>5}")
        for msg in collected.get(level, [])[:3]:
            print(f"           → {msg[:80]}")

if __name__ == "__main__":
    if not LOG_FILE.exists():
        print(f"ログファイルが見つかりません: {LOG_FILE}")
    else:
        lines = tail(LOG_FILE, TAIL_LINES)
        print(f"=== {LOG_FILE} 末尾 {TAIL_LINES} 行の解析結果 ===\n")
        analyze_log(lines)

"""
Runbook の確認手順で使うヘルスチェックスクリプト。
エンドポイント一覧に対して HTTP GET を実行して結果を出力する。
"""
import urllib.request
import urllib.error
import time

ENDPOINTS = [
    ("order-api   ", "http://localhost:8080/health"),
    ("payment-api ", "http://localhost:8081/health"),
    ("DB proxy    ", "http://localhost:5432/"),  # pg_proxy など
]
TIMEOUT = 5

def check(label: str, url: str) -> dict:
    start = time.monotonic()
    try:
        with urllib.request.urlopen(url, timeout=TIMEOUT) as resp:
            elapsed = (time.monotonic() - start) * 1000
            return {"label": label, "status": resp.status, "ms": round(elapsed, 1), "ok": True}
    except urllib.error.HTTPError as e:
        return {"label": label, "status": e.code, "ms": -1, "ok": False}
    except Exception as e:
        return {"label": label, "status": str(e)[:30], "ms": -1, "ok": False}

if __name__ == "__main__":
    print(f"{'サービス':<14} {'Status':>6}  {'応答(ms)':>9}  結果")
    print("-" * 45)
    all_ok = True
    for label, url in ENDPOINTS:
        r = check(label, url)
        icon = "✅" if r["ok"] else "❌"
        ms = f"{r['ms']} ms" if r["ms"] >= 0 else "timeout"
        print(f"{r['label']:<14} {str(r['status']):>6}  {ms:>9}  {icon}")
        if not r["ok"]:
            all_ok = False
    print()
    print("🟢 全サービス正常" if all_ok else "🔴 異常あり — Runbook のトラブルシューティングを参照")