예제 소스

daemon_omook_stack_watchdog.py

#!/usr/bin/env python3
"""Omook 서비스 스택 감시: Next(:3002) + VLM TCP(:9299).

죽으면 재기동. WS 브리지는 불필요(웹이 /api/vlm/move → :9299 직접 호출).

  .venv/bin/python scripts/daemon_omook_stack_watchdog.py
"""

from __future__ import annotations

import json
import os
import signal
import socket
import subprocess
import sys
import time
import urllib.error
import urllib.request
from datetime import datetime
from pathlib import Path
from zoneinfo import ZoneInfo

ROOT = Path(__file__).resolve().parents[1]
WEB = ROOT / "omook-web"
PY = str(ROOT / ".venv" / "bin" / "python")
OUT = ROOT / "logs" / "omook_stack_watch"
HEART = OUT / "HEARTBEAT.md"
LOG = OUT / "watchdog.log"
NEXT_LOG = OUT / "next.watchdog.log"
VLM_LOG = OUT / "vlm.watchdog.log"
RESTART_FLAG = OUT / "RESTART_REQUESTED"
KST = ZoneInfo("Asia/Seoul")

NEXT_PORT = 3002
VLM_PORT = 9299
POLL_S = 20
HTTP_TIMEOUT_S = 3.0
NEXT_RESTART_WAIT_S = 90
VLM_RESTART_WAIT_S = 600  # 모델 로드

NODE_BIN = Path.home() / ".nvm" / "versions" / "node" / "v24.14.1" / "bin"
NPM = str(NODE_BIN / "npm")

VLM_DEVICE = os.environ.get("OMOOK_VLM_DEVICE", "mps")
VLM_CMD = [
    PY,
    "vlm_pva_server.py",
    "--host",
    "127.0.0.1",
    "--port",
    str(VLM_PORT),
    "--device",
    VLM_DEVICE,
]
# {"python": "../go/.venv/bin/python", "args": ["--backend", "mlx", ...]}  (경로는 ROOT 기준)
VLM_PROFILE = ROOT / "vlm_profile.json"


def vlm_cmd() -> list[str]:
    """재시작할 때마다 vlm_profile.json 을 다시 읽는다. 없으면 기본 VLM_CMD."""
    try:
        prof = json.loads(VLM_PROFILE.read_text(encoding="utf-8"))
    except FileNotFoundError:
        return list(VLM_CMD)
    except (OSError, ValueError) as e:
        log(f"vlm_profile.json 무시 (읽기 실패: {e})")
        return list(VLM_CMD)
    # resolve() 는 venv python 심볼릭 링크를 원본 인터프리터로 풀어 venv site-packages 를 잃는다.
    py = os.path.normpath(ROOT / prof["python"]) if prof.get("python") else PY
    return [py, *VLM_CMD[1:], *[str(a) for a in prof.get("args", [])]]


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:
    """레거시 TCP 프로브 — VLM 헬스에는 쓰지 말 것(eof 로그 유발)."""
    try:
        with socket.create_connection((host, port), timeout=1.5):
            return True
    except OSError:
        return False


def http_ok(url: str, expect: int = 200) -> bool:
    try:
        req = urllib.request.Request(url, method="GET")
        with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT_S) as resp:
            return int(resp.status) == expect
    except (urllib.error.URLError, TimeoutError, OSError, ValueError):
        return False


def next_healthy() -> bool:
    return http_ok(f"http://127.0.0.1:{NEXT_PORT}/", 200) or http_ok(
        f"http://127.0.0.1:{NEXT_PORT}/play", 200
    )


def vlm_healthy() -> bool:
    # TCP connect 프로브는 vlm_pva_server 에 eof(메시지 0건) 경고를 남기므로
    # 리스너 존재만 확인한다.
    return bool(listeners_on(VLM_PORT))


def pgrep(pattern: str) -> list[int]:
    try:
        out = subprocess.check_output(["pgrep", "-f", pattern], text=True)
    except (subprocess.CalledProcessError, FileNotFoundError, PermissionError, OSError):
        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, PermissionError, OSError):
            pass
    time.sleep(2)
    for p in pids:
        try:
            os.kill(p, signal.SIGKILL)
        except (ProcessLookupError, PermissionError, OSError):
            pass


def listeners_on(port: int) -> list[int]:
    try:
        out = subprocess.check_output(
            ["lsof", "-nP", f"-iTCP:{port}", "-sTCP:LISTEN", "-t"],
            text=True,
        )
    except (subprocess.CalledProcessError, FileNotFoundError, PermissionError, OSError):
        return []
    pids: list[int] = []
    for tok in out.split():
        try:
            pids.append(int(tok))
        except ValueError:
            pass
    return pids


def free_port(port: int) -> None:
    kill_pids(listeners_on(port), f"port:{port}")


def spawn(cmd: list[str], cwd: Path, logfile: Path) -> subprocess.Popen:
    env = os.environ.copy()
    env["PYTHONUNBUFFERED"] = "1"
    env["TOKENIZERS_PARALLELISM"] = "********"
    path_parts = [str(NODE_BIN), "/opt/homebrew/bin", "/usr/local/bin"]
    if env.get("PATH"):
        path_parts.append(env["PATH"])
    env["PATH"] = ":".join(path_parts)
    env["PORT"] = str(NEXT_PORT)
    logfile.parent.mkdir(parents=True, exist_ok=True)
    fp = open(logfile, "a", encoding="utf-8")
    fp.write(f"\n===== spawn {now_kst():%Y-%m-%d %H:%M:%S} =====\n")
    fp.write(" ".join(cmd) + "\n")
    fp.flush()
    return subprocess.Popen(
        cmd,
        cwd=str(cwd),
        env=env,
        stdout=fp,
        stderr=subprocess.STDOUT,
        start_new_session=True,
    )


def ensure_caffeinate() -> None:
    me = str(os.getpid())
    try:
        if pgrep(f"caffeinate -dims -w {me}"):
            return
        spawn(["caffeinate", "-dims", "-w", me], ROOT, OUT / "caffeinate.log")
        log("caffeinate started (tied to this watchdog pid)")
    except (PermissionError, OSError) as e:
        log(f"caffeinate skip: {e}")


def proc_cmdline(pid: int) -> str:
    try:
        return subprocess.check_output(
            ["ps", "-p", str(pid), "-o", "command="], text=True
        )
    except (subprocess.CalledProcessError, FileNotFoundError, PermissionError, OSError):
        return ""


def next_is_dev() -> bool:
    for pid in listeners_on(NEXT_PORT):
        out = proc_cmdline(pid).lower()
        if "next dev" in out or "dev-with-ts.mjs dev" in out:
            return True
    return False


def ensure_next_build() -> bool:
    marker = WEB / ".next" / "BUILD_ID"
    if marker.exists():
        return True
    log("Next .next missing → npm run build")
    if not Path(NPM).exists():
        log(f"npm missing at {NPM}")
        return False
    try:
        subprocess.run(
            [NPM, "run", "build"],
            cwd=str(WEB),
            check=True,
            env={
                **os.environ,
                "PATH": f"{NODE_BIN}:{os.environ.get('PATH', '')}",
                "PORT": str(NEXT_PORT),
            },
        )
    except subprocess.CalledProcessError as e:
        log(f"build failed: {e}")
        return False
    return marker.exists()


def restart_next(n: int) -> bool:
    log(f"Next :{NEXT_PORT} DOWN/dev → service restart #{n}")
    kill_pids(
        pgrep(r"/omook/omook-web|/omook-web.*(next|npm|pnpm)"),
        "omook-next",
    )
    free_port(NEXT_PORT)
    if not Path(NPM).exists():
        log(f"npm missing at {NPM}")
        return False
    if not ensure_next_build():
        return False
    spawn([NPM, "run", "start"], WEB, NEXT_LOG)
    deadline = time.time() + NEXT_RESTART_WAIT_S
    while time.time() < deadline:
        if next_healthy() and not next_is_dev():
            log(f"Next service UP after restart #{n}")
            return True
        time.sleep(2)
    log(f"Next STILL DOWN after restart #{n}")
    return False


def restart_vlm(n: int) -> bool:
    log(f"VLM :{VLM_PORT} DOWN → restart #{n}")
    kill_pids(pgrep("vlm_pva_server\\.py"), "omook-vlm")
    free_port(VLM_PORT)
    cmd = vlm_cmd()
    if not Path(cmd[0]).exists():
        log(f"python missing at {cmd[0]}")
        return False
    spawn(cmd, ROOT, VLM_LOG)
    deadline = time.time() + VLM_RESTART_WAIT_S
    while time.time() < deadline:
        if vlm_healthy():
            log(f"VLM UP after restart #{n}")
            return True
        time.sleep(3)
    log(f"VLM STILL DOWN after restart #{n}")
    return False


def write_heartbeat(
    *,
    next_ok: bool,
    vlm_ok: bool,
    next_restarts: int,
    vlm_restarts: int,
    note: str,
) -> None:
    mode = "service" if next_ok and not next_is_dev() else ("dev" if next_ok else "down")
    lines = [
        f"# Omook stack watchdog — {now_kst():%Y-%m-%d %H:%M:%S KST}",
        "",
        f"- **Next :{NEXT_PORT}**: `{'up' if next_ok else 'DOWN'}` ({mode})",
        f"- **VLM :{VLM_PORT}**: `{'up' if vlm_ok else 'DOWN'}` (vlm_pva_server)",
        f"- **next_restarts**: `{next_restarts}`",
        f"- **vlm_restarts**: `{vlm_restarts}`",
        f"- **device**: `{VLM_DEVICE}`",
        f"- **note**: {note}",
        "",
        "로그: `logs/omook_stack_watch/watchdog.log`",
        "",
    ]
    HEART.write_text("\n".join(lines), encoding="utf-8")


def acquire_singleton() -> object:
    """중복 워치독이 VLM을 동시에 재시작/프로브하지 않도록 락."""
    OUT.mkdir(parents=True, exist_ok=True)
    lock_path = OUT / "watchdog.lock"
    fh = lock_path.open("a+", encoding="utf-8")
    try:
        import fcntl

        fcntl.flock(fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
    except OSError as e:
        fh.close()
        raise SystemExit(f"another omook watchdog holds {lock_path}: {e}") from e
    fh.seek(0)
    fh.truncate()
    fh.write(f"{os.getpid()}\n")
    fh.flush()
    return fh


def main() -> int:
    OUT.mkdir(parents=True, exist_ok=True)
    lock_fh = acquire_singleton()
    log(
        f"omook stack watchdog start "
        f"(Next:{NEXT_PORT} service + VLM:{VLM_PORT} device={VLM_DEVICE})"
    )
    ensure_caffeinate()
    next_restarts = 0
    vlm_restarts = 0
    migrated_to_service = False

    while True:
        if RESTART_FLAG.exists():
            reason = RESTART_FLAG.read_text(encoding="utf-8").strip()
            try:
                RESTART_FLAG.unlink()
            except OSError:
                pass
            log(f"ops requested restart: {reason}")
            low = reason.lower()
            if "vlm" in low:
                vlm_restarts += 1
                vlm = restart_vlm(vlm_restarts)
            else:
                next_restarts += 1
                nxt = restart_next(next_restarts)

        nxt = next_healthy()
        if nxt and next_is_dev() and not migrated_to_service:
            log("Next is dev mode → migrate to service (next start)")
            next_restarts += 1
            nxt = restart_next(next_restarts)
            migrated_to_service = True
        elif not nxt:
            next_restarts += 1
            nxt = restart_next(next_restarts)
            migrated_to_service = True

        vlm = vlm_healthy()
        if not vlm:
            vlm_restarts += 1
            vlm = restart_vlm(vlm_restarts)

        write_heartbeat(
            next_ok=nxt,
            vlm_ok=vlm,
            next_restarts=next_restarts,
            vlm_restarts=vlm_restarts,
            note="polling",
        )
        time.sleep(POLL_S)


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except KeyboardInterrupt:
        log("watchdog interrupted")
        raise SystemExit(0)
    except Exception:
        logging = __import__("logging")
        log("watchdog crashed")
        import traceback

        OUT.mkdir(parents=True, exist_ok=True)
        with LOG.open("a", encoding="utf-8") as f:
            f.write(traceback.format_exc() + "\n")
        raise