hyowons
231 lines
8.4 KiB
Python
Executable File
231 lines
8.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""자동화 헬스체크 — 조용히 죽은 잡을 찾아낸다 (LLM 미경유).
|
|
|
|
배경: 2026-09-10 점검에서 **하드 실패는 0건인데** 다른 것들이 썩고 있었다.
|
|
`holiday_sync` 가 investing 403 이후 2개월간 stale, `codex_fallback_monitor` 가
|
|
trajectory 의 sqlite 이관으로 조용히 무력화, TASKS.md 가 실제 트리거와 6곳 어긋남.
|
|
공통점은 **프로세스가 정상 종료(exit 0)하는데 결과가 옛것**이라 아무 신호가 없다는 것.
|
|
|
|
그래서 exit code 가 아니라 **결과물의 mtime** 을 본다. 감시 대상은
|
|
`TASKS.md` §4 표에서 읽는다 — 표가 곧 설정이라, 트리거를 바꾸고 표를 안 고치면
|
|
알림이 떠서 문서 드리프트가 스스로 드러난다.
|
|
|
|
- 증거가 `daemon:<label>` 이면 mtime 대신 **PID 생존**을 본다(KeepAlive 데몬).
|
|
⚠️ exit code 는 보지 않는다 — KeepAlive 데몬은 재시작 직후 옛 종료코드(-15 등)가
|
|
남아 있어 매번 오탐이 난다. 지금 떠 있으면 정상이다.
|
|
- 이상이 없으면 **아무것도 보내지 않는다**(정상 알림은 노이즈라 무시하게 된다).
|
|
|
|
⚠️ **같은 이상을 매일 다시 보내지 않는다** — 일간으로 도는 이상, 연휴 오탐 하나가
|
|
닷새 연속 울리면 알림 자체를 무시하게 된다. 직전과 이상 목록이 같으면 침묵하되,
|
|
REPEAT_SEC 이 지나면 "아직 안 고쳐졌다"는 뜻으로 한 번 더 보낸다.
|
|
|
|
CLI: python3 health_check.py [--dry-run] [--verbose]
|
|
launchd: ai.openclaw.health-check (매일 09:00)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
KST = timezone(timedelta(hours=9))
|
|
ROOT = os.path.expanduser("~/.openclaw")
|
|
TASKS_MD = os.path.join(ROOT, "TASKS.md")
|
|
LOG_DIR = os.path.join(ROOT, "logs")
|
|
STATE_PATH = os.path.join(ROOT, "workspace", "state", "health_check.json")
|
|
LOG_LIMIT_MB = 200 # 로그 로테이션이 도는 한 넘을 일이 없다
|
|
REPEAT_SEC = 7 * 86400 # 같은 이상을 다시 알릴 최소 간격
|
|
|
|
sys.path.insert(0, os.path.join(ROOT, "workspace", "scripts"))
|
|
|
|
|
|
def _load_state() -> dict:
|
|
try:
|
|
with open(STATE_PATH, encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def _save_state(state: dict) -> None:
|
|
try:
|
|
os.makedirs(os.path.dirname(STATE_PATH), exist_ok=True)
|
|
tmp = STATE_PATH + ".tmp"
|
|
with open(tmp, "w", encoding="utf-8") as f:
|
|
json.dump(state, f, ensure_ascii=False, indent=2)
|
|
os.replace(tmp, STATE_PATH)
|
|
except Exception as e:
|
|
print(f"[state] 저장 실패: {e}", file=sys.stderr)
|
|
|
|
|
|
def _parse_period(text: str) -> float | None:
|
|
"""'2h' / '100h' / '8d' / '-' → 초. 데몬('-')은 None."""
|
|
text = text.strip()
|
|
m = re.fullmatch(r"(\d+)([hd])", text)
|
|
if not m:
|
|
return None
|
|
n = int(m.group(1))
|
|
return n * 3600 if m.group(2) == "h" else n * 86400
|
|
|
|
|
|
def parse_targets() -> list[tuple[str, str, float | None, str]]:
|
|
"""TASKS.md §4 표 → [(작업명, 증거, 기대초, 기대표기)]."""
|
|
with open(TASKS_MD, encoding="utf-8") as f:
|
|
doc = f.read()
|
|
m = re.search(r"^## 4\. 헬스체크.*?(?=^## 5\.)", doc, re.M | re.S)
|
|
if not m:
|
|
raise RuntimeError("TASKS.md §4 헬스체크 섹션을 찾지 못했습니다")
|
|
|
|
targets = []
|
|
for line in m.group(0).splitlines():
|
|
if not line.startswith("|"):
|
|
continue
|
|
cells = [c.strip() for c in line.strip("|").split("|")]
|
|
if len(cells) != 3:
|
|
continue
|
|
name, ev, period = cells
|
|
if name == "작업" or set(name) <= set("-: "):
|
|
continue
|
|
targets.append((name, ev.strip("`"), _parse_period(period), period))
|
|
return targets
|
|
|
|
|
|
def _daemon_pid(label: str) -> str | None:
|
|
try:
|
|
out = subprocess.run(["launchctl", "list"], capture_output=True,
|
|
text=True, timeout=20).stdout
|
|
except Exception:
|
|
return None
|
|
for line in out.splitlines():
|
|
parts = line.split("\t")
|
|
if len(parts) >= 3 and parts[2].strip() == label:
|
|
return parts[0] if parts[0] != "-" else None
|
|
return None
|
|
|
|
|
|
def _fmt_age(sec: float) -> str:
|
|
if sec < 3600:
|
|
return f"{sec/60:.0f}분"
|
|
if sec < 86400:
|
|
return f"{sec/3600:.1f}시간"
|
|
return f"{sec/86400:.1f}일"
|
|
|
|
|
|
def run_checks(verbose: bool) -> list[str]:
|
|
now = time.time()
|
|
issues: list[str] = []
|
|
|
|
for name, ev, expect, expect_raw in parse_targets():
|
|
if ev.startswith("daemon:"):
|
|
label = ev.split(":", 1)[1]
|
|
if _daemon_pid(label) is None:
|
|
issues.append(f"❌ {name}: 데몬이 떠 있지 않음 ({label})")
|
|
elif verbose:
|
|
print(f"[ok] {name}: 실행 중")
|
|
continue
|
|
|
|
path = os.path.join(ROOT, ev)
|
|
if not os.path.exists(path):
|
|
issues.append(f"❌ {name}: 증거 파일 없음 ({ev})")
|
|
continue
|
|
if expect is None:
|
|
continue
|
|
age = now - os.path.getmtime(path)
|
|
if age > expect:
|
|
issues.append(f"⚠️ {name}: {_fmt_age(age)}째 갱신 없음 "
|
|
f"(기대 {expect_raw}) — {ev}")
|
|
elif verbose:
|
|
print(f"[ok] {name}: {_fmt_age(age)} 전")
|
|
|
|
# 로그 총량 — 로테이션이 멈추면 여기서 드러난다
|
|
if os.path.isdir(LOG_DIR):
|
|
total = sum(os.path.getsize(os.path.join(LOG_DIR, f))
|
|
for f in os.listdir(LOG_DIR)
|
|
if os.path.isfile(os.path.join(LOG_DIR, f)))
|
|
mb = total / 1048576
|
|
if mb > LOG_LIMIT_MB:
|
|
issues.append(f"⚠️ 로그 용량 {mb:.0f}MB (상한 {LOG_LIMIT_MB}MB) "
|
|
f"— 로그 로테이션 점검 필요")
|
|
elif verbose:
|
|
print(f"[ok] 로그 용량: {mb:.0f}MB")
|
|
|
|
return issues
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--dry-run", action="store_true",
|
|
help="텔레그램 발송 없이 결과만 출력")
|
|
ap.add_argument("--verbose", "-v", action="store_true")
|
|
args = ap.parse_args()
|
|
|
|
try:
|
|
issues = run_checks(args.verbose)
|
|
except Exception as e:
|
|
print(f"[error] 점검 실패: {e}", file=sys.stderr)
|
|
return 1
|
|
|
|
now = time.time()
|
|
today = datetime.now(KST).strftime("%Y-%m-%d")
|
|
state = _load_state()
|
|
|
|
# 매월 1일 — 코디 월간 점검 리마인더. 스크립트로는 못 잡는 외부 드리프트
|
|
# (모델 deprecated·본체 업그레이드·데이터소스 차단)를 사람이 짚어야 해서 둔다.
|
|
monthly = (datetime.now(KST).day == 1
|
|
and state.get("last_monthly") != today)
|
|
|
|
if not issues:
|
|
# 이상이 해소되면 서명을 지운다 — 안 지우면 같은 이상이 재발했을 때
|
|
# 쿨다운에 걸려 조용히 넘어간다.
|
|
if state.pop("last_signature", None) is not None:
|
|
state.pop("last_sent", None)
|
|
_save_state(state)
|
|
if not monthly:
|
|
print("[ok] 전체 정상")
|
|
return 0
|
|
|
|
parts = []
|
|
if issues:
|
|
parts.append("🩺 OpenClaw 헬스체크\n" + "\n".join(issues))
|
|
if monthly:
|
|
parts.append("🗓 월간 점검일입니다 — 코디에게 "
|
|
"`MAINTENANCE.md` 체크리스트를 맡겨주세요.")
|
|
text = "\n\n".join(parts)
|
|
print(text)
|
|
if args.dry_run:
|
|
return 0
|
|
|
|
# 같은 이상을 매일 다시 보내지 않는다(월간 리마인더가 낀 날은 그대로 보낸다)
|
|
sig = "\n".join(sorted(issues))
|
|
if issues and not monthly:
|
|
if (sig == state.get("last_signature")
|
|
and now - state.get("last_sent", 0) < REPEAT_SEC):
|
|
print("[skip] 직전과 같은 이상 — 재발송 안 함 "
|
|
f"(최소 간격 {REPEAT_SEC//86400}일)")
|
|
return 0
|
|
|
|
try:
|
|
from codex_fallback_monitor import _send_telegram
|
|
sent = _send_telegram(text)
|
|
except Exception as e:
|
|
print(f"[tg] 발송 실패: {e}", file=sys.stderr)
|
|
return 0
|
|
|
|
if sent:
|
|
if issues:
|
|
state["last_signature"] = sig
|
|
state["last_sent"] = now
|
|
if monthly:
|
|
state["last_monthly"] = today
|
|
_save_state(state)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|