예제 소스
바둑 on-policy 학습: 대국 로그를 KataGo 라벨로 바꾸기 · 파일 169개
daemon_mariago_external_probe.py
#!/usr/bin/env python3
"""MariaGO 외부 접속 주기 점검 데몬.
외부 사용자와 같은 경로(HTTPS)로 페이지·WS 프록시가 살아 있는지 확인.
.venv/bin/python scripts/daemon_mariago_external_probe.py
결과:
logs/mariago_external_probe/probes.jsonl
logs/mariago_external_probe/HEARTBEAT.md
"""
from __future__ import annotations
import argparse
import json
import os
import signal
import socket
import ssl
import subprocess
import time
import urllib.error
import urllib.request
from collections import deque
from datetime import datetime, timezone
from pathlib import Path
from zoneinfo import ZoneInfo
ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "logs" / "mariago_external_probe"
TUNNEL_OUT = ROOT / "logs" / "mariago_ssh_tunnel"
RESTART_FLAG = TUNNEL_OUT / "RESTART_REQUESTED"
HEART = OUT / "HEARTBEAT.md"
LOG = OUT / "probe.log"
JSONL = OUT / "probes.jsonl"
KST = ZoneInfo("Asia/Seoul")
PY = str(ROOT / ".venv" / "bin" / "python")
SSH_HOST = "[REDACTED]"
LOCAL_WS = 9320
GATEWAY_ERRORS = {501, 502, 503}
RESTART_COOLDOWN_S = 90.0
DEFAULT_TARGETS = [
{"name": "mariago-root", "url": "https://mariago.kr/", "kind": "http"},
{"name": "baduk-play", "url": "https://baduk.mariago.kr/play", "kind": "http"},
{"name": "stoku-home", "url": "https://stoku.mariago.kr/", "kind": "http"},
{"name": "omook-home", "url": "https://omook.mariago.kr/", "kind": "http"},
{"name": "omook-play", "url": "https://omook.mariago.kr/play", "kind": "http"},
{"name": "othello-home", "url": "https://othello.mariago.kr/", "kind": "http"},
{"name": "othello-play", "url": "https://othello.mariago.kr/play", "kind": "http"},
{"name": "tsp-home", "url": "https://tsp.mariago.kr/", "kind": "http"},
{"name": "tsp-play", "url": "https://tsp.mariago.kr/play", "kind": "http"},
{"name": "viruswar-home", "url": "https://viruswar.mariago.kr/", "kind": "http"},
{"name": "viruswar-play", "url": "https://viruswar.mariago.kr/play", "kind": "http"},
{"name": "janggy-home", "url": "https://janggy.mariago.kr/", "kind": "http"},
{"name": "janggy-play", "url": "https://janggy.mariago.kr/play", "kind": "http"},
{"name": "chess-home", "url": "https://chess.mariago.kr/", "kind": "http"},
{"name": "chess-play", "url": "https://chess.mariago.kr/play", "kind": "http"},
{"name": "crossword-home", "url": "https://crossword.mariago.kr/", "kind": "http"},
{"name": "crossword-play", "url": "https://crossword.mariago.kr/play", "kind": "http"},
{"name": "baduk-ws", "url": "https://baduk.mariago.kr/vlm-ws", "kind": "ws"},
]
def now_kst() -> datetime:
return datetime.now(KST)
def log(msg: str) -> None:
line = f"[{now_kst():%Y-%m-%d %H:%M:%S}] {msg}"
print(line, flush=True)
OUT.mkdir(parents=True, exist_ok=True)
with LOG.open("a", encoding="utf-8") as f:
f.write(line + "\n")
def port_open(port: int, host: str = "127.0.0.1") -> bool:
try:
with socket.create_connection((host, port), timeout=1.5):
return True
except OSError:
return False
def pgrep(pattern: str) -> list[int]:
try:
out = subprocess.check_output(["pgrep", "-f", pattern], text=True)
except subprocess.CalledProcessError:
return []
me = os.getpid()
pids: list[int] = []
for tok in out.split():
try:
p = int(tok)
except ValueError:
continue
if p != me:
pids.append(p)
return pids
def kill_pids(pids: list[int], label: str) -> None:
if not pids:
return
log(f"kill {label}: {pids}")
for p in pids:
try:
os.kill(p, signal.SIGTERM)
except ProcessLookupError:
pass
time.sleep(1)
for p in pids:
try:
os.kill(p, signal.SIGKILL)
except ProcessLookupError:
pass
def restart_bridge() -> None:
log("bridge DOWN → restart bridge")
kill_pids(pgrep("mariago_ws_bridge.py"), "bridge")
TUNNEL_OUT.mkdir(parents=True, exist_ok=True)
bridge_log = TUNNEL_OUT / "bridge.probe-restart.log"
with bridge_log.open("a", encoding="utf-8") as fp:
fp.write(f"\n===== probe restart {now_kst():%Y-%m-%d %H:%M:%S} =====\n")
fp.flush()
subprocess.Popen(
[PY, "scripts/mariago_ws_bridge.py", "--ws-port", str(LOCAL_WS)],
cwd=str(ROOT),
stdout=fp,
stderr=subprocess.STDOUT,
start_new_session=True,
)
def request_tunnel_restart(reason: str) -> None:
"""터널 watchdog에 재시작 요청 + SSH 세션 끊기."""
TUNNEL_OUT.mkdir(parents=True, exist_ok=True)
RESTART_FLAG.write_text(f"{now_kst().isoformat()} {reason}\n", encoding="utf-8")
log(f"tunnel restart requested: {reason}")
kill_pids(pgrep(f"ssh .*{SSH_HOST}.*127.0.0.1:18080"), "ssh-tunnel")
kill_pids(pgrep(f"ssh -N.*{SSH_HOST}"), "ssh-tunnel-n")
def needs_recovery(results: list[dict], bridge_ok: bool) -> tuple[bool, list[str], list[str]]:
"""반환: (recover 필요, 터널 재시작 사유, 브리지-only 사유).
WS 단독 실패로 SSH 터널 전체를 끊으면 omook/baduk HTTP 도 잠깐 502 가 난다.
HTTP 게이트웨이 오류일 때만 터널을 재시작하고, WS/브리지는 브리지 재기동으로 충분.
"""
tunnel_reasons: list[str] = []
bridge_reasons: list[str] = []
for r in results:
st = r.get("status")
kind = r.get("kind", "http")
name = r["name"]
if st in GATEWAY_ERRORS:
tunnel_reasons.append(f"{name}:HTTP{st}")
elif not r.get("ok"):
if kind == "ws":
bridge_reasons.append(f"{name}:down")
else:
tunnel_reasons.append(f"{name}:down")
if not bridge_ok:
bridge_reasons.append("bridge:9320_down")
need = bool(tunnel_reasons or bridge_reasons)
return need, tunnel_reasons, bridge_reasons
def maybe_recover(
results: list[dict],
*,
enabled: bool,
last_restart_at: float,
) -> float:
if not enabled:
return last_restart_at
bridge_ok = port_open(LOCAL_WS)
need, tunnel_reasons, bridge_reasons = needs_recovery(results, bridge_ok)
if not need:
return last_restart_at
now = time.monotonic()
if now - last_restart_at < RESTART_COOLDOWN_S:
log(
f"recover skipped (cooldown) tunnel={tunnel_reasons} bridge={bridge_reasons}"
)
return last_restart_at
if bridge_reasons or not bridge_ok:
restart_bridge()
if tunnel_reasons:
request_tunnel_restart(", ".join(tunnel_reasons))
return now
log(f"bridge-only recovery (no tunnel kill): {', '.join(bridge_reasons)}")
return now
def probe_http(url: str, timeout_s: float) -> dict:
t0 = time.monotonic()
ctx = ssl.create_default_context()
req = urllib.request.Request(
url,
method="GET",
headers={
"User-Agent": "MariaGO-external-probe/1.0",
"Accept": "text/html,*/*",
"Connection": "close",
},
)
try:
with urllib.request.urlopen(req, timeout=timeout_s, context=ctx) as resp:
body = resp.read(2048)
code = int(resp.status)
ok = 200 <= code < 400
return {
"ok": ok,
"status": code,
"ms": round((time.monotonic() - t0) * 1000, 1),
"bytes": len(body),
"error": None,
}
except urllib.error.HTTPError as e:
code = int(e.code)
ok = 200 <= code < 400
return {
"ok": ok,
"status": code,
"ms": round((time.monotonic() - t0) * 1000, 1),
"bytes": 0,
"error": None if ok else f"HTTPError: {e}",
}
except Exception as e:
return {
"ok": False,
"status": None,
"ms": round((time.monotonic() - t0) * 1000, 1),
"bytes": 0,
"error": f"{type(e).__name__}: {e}",
}
def probe_ws(url: str, timeout_s: float) -> dict:
"""WS 엔드포인트: 101/400/426 등은 살아 있음, 502/503/000 은 장애."""
t0 = time.monotonic()
ctx = ssl.create_default_context()
req = urllib.request.Request(
url,
method="GET",
headers={
"User-Agent": "MariaGO-external-probe/1.0",
"Connection": "Upgrade",
"Upgrade": "websocket",
"Sec-WebSocket-Version": "13",
"Sec-WebSocket-Key": "dGhlIHNhbXBsZSBub25jZQ==",
},
)
try:
with urllib.request.urlopen(req, timeout=timeout_s, context=ctx) as resp:
code = int(resp.status)
ok = code in (101, 200, 400, 426)
return {
"ok": ok,
"status": code,
"ms": round((time.monotonic() - t0) * 1000, 1),
"bytes": 0,
"error": None if ok else f"unexpected status {code}",
}
except urllib.error.HTTPError as e:
code = int(e.code)
ok = code in (400, 426, 101, 200)
return {
"ok": ok,
"status": code,
"ms": round((time.monotonic() - t0) * 1000, 1),
"bytes": 0,
"error": None if ok else f"HTTPError: {e}",
}
except Exception as e:
return {
"ok": False,
"status": None,
"ms": round((time.monotonic() - t0) * 1000, 1),
"bytes": 0,
"error": f"{type(e).__name__}: {e}",
}
def probe_target(target: dict, timeout_s: float) -> dict:
kind = target.get("kind", "http")
if kind == "ws":
body = probe_ws(target["url"], timeout_s)
else:
body = probe_http(target["url"], timeout_s)
return {
"ts": datetime.now(timezone.utc).isoformat(),
"name": target["name"],
"url": target["url"],
"kind": kind,
**body,
}
def write_heartbeat(
*,
interval_s: float,
results: list[dict],
history: dict[str, deque],
fail_streak: int,
total_rounds: int,
ok_rounds: int,
last_recovery: str | None = None,
) -> None:
lines = [
f"# MariaGO external probe — {now_kst():%Y-%m-%d %H:%M:%S KST}",
"",
f"- **interval_s**: `{interval_s}`",
f"- **rounds**: `{total_rounds}` (all-ok `{ok_rounds}` = `{100.0 * ok_rounds / max(1, total_rounds):.1f}%`)",
f"- **fail_streak**: `{fail_streak}` (연속 전체 실패 회차)",
"",
"## last check",
"",
]
all_ok = True
for r in results:
tag = "OK" if r["ok"] else "FAIL"
if not r["ok"]:
all_ok = False
err = f" err=`{r['error']}`" if r["error"] else ""
lines.append(
f"- **{r['name']}**: `{tag}` status=`{r['status']}` {r['ms']}ms{err}"
)
lines.append("")
lines.append(f"- **overall**: `{'OK' if all_ok else 'FAIL'}`")
if last_recovery:
lines.append(f"- **last_recovery**: `{last_recovery}`")
lines.append("")
lines.append("## per-target window (recent ok rate)")
lines.append("")
for name, win in history.items():
ok_n = sum(1 for x in win if x)
lines.append(f"- **{name}**: `{ok_n}/{len(win)}`")
lines.append("")
lines.append("로그: `logs/mariago_external_probe/probe.log`")
lines.append("")
HEART.write_text("\n".join(lines), encoding="utf-8")
def main() -> int:
p = argparse.ArgumentParser(description="MariaGO external reachability probe daemon")
p.add_argument("--interval", type=float, default=60.0, help="초 단위")
p.add_argument("--timeout", type=float, default=12.0)
p.add_argument("--window", type=int, default=60, help="타깃별 최근 N회 요약")
p.add_argument(
"--recover",
action=argparse.BooleanOptionalAction,
default=True,
help="502/503/브리지 다운 시 터널·브리지 자동 복구",
)
args = p.parse_args()
OUT.mkdir(parents=True, exist_ok=True)
history: dict[str, deque] = {
t["name"]: deque(maxlen=max(10, args.window)) for t in DEFAULT_TARGETS
}
fail_streak = 0
total_rounds = ok_rounds = 0
last_restart_at = 0.0
last_recovery_note: str | None = None
log(
f"external probe start interval={args.interval}s recover={args.recover} "
f"targets={[t['name'] for t in DEFAULT_TARGETS]}"
)
while True:
results = [probe_target(t, args.timeout) for t in DEFAULT_TARGETS]
all_ok = all(r["ok"] for r in results)
total_rounds += 1
if all_ok:
ok_rounds += 1
fail_streak = 0
round_tag = "OK"
else:
fail_streak += 1
round_tag = "FAIL"
prev = last_restart_at
last_restart_at = maybe_recover(
results,
enabled=args.recover,
last_restart_at=last_restart_at,
)
if last_restart_at != prev:
last_recovery_note = now_kst().strftime("%Y-%m-%d %H:%M:%S KST")
for r in results:
history[r["name"]].append(r["ok"])
with JSONL.open("a", encoding="utf-8") as f:
f.write(json.dumps(r, ensure_ascii=False) + "\n")
write_heartbeat(
interval_s=args.interval,
results=results,
history=history,
fail_streak=fail_streak,
total_rounds=total_rounds,
ok_rounds=ok_rounds,
last_recovery=last_recovery_note,
)
parts = []
for r in results:
tag = "OK" if r["ok"] else "FAIL"
parts.append(f"{r['name']}={tag}({r['status']},{r['ms']}ms)")
log(f"[{round_tag}] " + " · ".join(parts))
time.sleep(max(5.0, args.interval))
if __name__ == "__main__":
try:
raise SystemExit(main())
except KeyboardInterrupt:
log("external probe stopped")
raise SystemExit(0)