AgentsMeeting/gateway/scripts/xmpp_watchdog.py

"""
xmpp_watchdog.py — monitors xmpp_agent_core + article_processor,
auto-restarts on crash, reports status.
Runs alongside gateway processes as a separate 30s-loop daemon.
"""
import os, sys, time, subprocess, json, urllib.request, urllib.error

# ── Paths ────────────────────────────────────────────────────
PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))         # gateway/scripts/
GATEWAY_DIR   = os.path.dirname(PROJECT_ROOT)                     # gateway/
TEMP_DIR      = os.path.join(GATEWAY_DIR, "temp")
LOG_DIR       = os.path.join(GATEWAY_DIR, "logs")
os.makedirs(LOG_DIR, exist_ok=True)
os.makedirs(TEMP_DIR, exist_ok=True)

PYTHON        = r"C:\Users\hmo\AppData\Local\Programs\Python\Python310\python.exe"
CHECK_INTERVAL = 30          # seconds between health checks
HEALTH_TIMEOUT = 5           # HTTP health check timeout

WATCHDOG_LOG  = os.path.join(LOG_DIR, "watchdog.log")
PID_FILE      = os.path.join(TEMP_DIR, ".multi_watchdog.pid")


# ── Service registry（从共享 registry 导入）───────────────────
sys.path.insert(0, PROJECT_ROOT)
from service_registry import SERVICES as _SR

SERVICES = {}
for svc in _SR:
    name = svc["name"]
    SERVICES[name] = {
        "script": svc["script"],
        "args": svc["args"],
        "workdir": svc["workdir"],
        "pid_file": svc["pid_file"],
        "port": svc["port"],
        "health_url": svc["health_url"],
        "log_files": svc.get("log_files", lambda: [])(),
    }


# ── Logging ───────────────────────────────────────────────────
def wlog(msg: str):
    ts = time.strftime("%Y-%m-%d %H:%M:%S")
    line = f"{ts} [watchdog] {msg}\n"
    with open(WATCHDOG_LOG, "a", encoding="utf-8") as f:
        f.write(line)
    print(line, end="", flush=True)


# ── Log rotation ──────────────────────────────────────────────
def rotate_log(path: str, max_bytes: int = 5 * 1024 * 1024):
    try:
        if not os.path.exists(path):
            return
        if os.path.getsize(path) > max_bytes:
            bak2 = f"{path}.2"
            bak1 = f"{path}.1"
            if os.path.exists(bak2):
                os.remove(bak2)
            if os.path.exists(bak1):
                os.rename(bak1, bak2)
            os.rename(path, bak1)
            wlog(f"Rotated: {os.path.basename(path)}")
    except Exception:
        pass


# ── Process helpers ───────────────────────────────────────────
def is_process_alive(pid: int) -> bool:
    try:
        r = subprocess.run(
            ["tasklist", "/FI", f"PID eq {pid}", "/NH"],
            capture_output=True, text=True, timeout=5,
            creationflags=subprocess.CREATE_NO_WINDOW,
        )
        return str(pid) in r.stdout
    except:
        return False


def get_pid_from_file(path: str) -> int:
    try:
        with open(path) as f:
            return int(f.read().strip())
    except:
        return 0


def is_port_listening(port: int) -> bool:
    """Quick port liveness check via netstat."""
    try:
        r = subprocess.run(
            ["netstat", "-ano"],
            capture_output=True, text=True, timeout=5,
            creationflags=subprocess.CREATE_NO_WINDOW,
        )
        return any(f":{port} " in line and "LISTENING" in line
                   for line in r.stdout.splitlines())
    except:
        return False


def check_http_health(url: str, timeout: int = HEALTH_TIMEOUT) -> tuple[bool, str]:
    """Returns (ok, detail)."""
    try:
        r = urllib.request.urlopen(url, timeout=timeout)
        body = r.read().decode("utf-8", errors="replace")[:200]
        return (r.status == 200, body)
    except urllib.error.HTTPError as e:
        return (False, f"HTTP {e.code}")
    except urllib.error.URLError as e:
        return (False, str(e.reason)[:60])
    except Exception as e:
        return (False, str(e)[:60])


# ── Process lifecycle ────────────────────────────────────────
def kill_service(name: str, script_match: str):
    """Kill ALL processes whose command line contains script_match."""
    killed = 0
    try:
        r = subprocess.run(
            ["tasklist", "/FO", "CSV", "/NH", "/FI", "IMAGENAME eq python.exe"],
            capture_output=True, text=True, timeout=10
        )
        for line in r.stdout.splitlines():
            parts = line.strip('"').split('","')
            if len(parts) < 2 or parts[0] != "python.exe":
                continue
            pid_str = parts[1].strip()
            try:
                wmi = subprocess.run(
                        ["wmic", "process", "where", f"ProcessId={pid_str}",
                         "get", "CommandLine", "/format:list"],
                        capture_output=True, text=True, timeout=5,
                        creationflags=subprocess.CREATE_NO_WINDOW,
                    )
                if script_match in wmi.stdout:
                    subprocess.run(["taskkill", "/f", "/pid", pid_str],
                                   capture_output=True, timeout=5)
                    killed += 1
                    wlog(f"Killed old {name} (PID {pid_str})")
            except:
                pass
        if killed > 0:
            time.sleep(3)
    except:
        pass
    return killed


def start_service(name: str, cfg: dict) -> int:
    """Start service and return PID. Kills old instances first."""
    kill_service(name, os.path.basename(cfg["script"]))
    wlog(f"Starting {name}...")
    proc = subprocess.Popen(
        [PYTHON, cfg["script"]] + cfg["args"],
        cwd=cfg["workdir"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        creationflags=subprocess.CREATE_NO_WINDOW,
    )
    pid = proc.pid
    with open(cfg["pid_file"], "w") as f:
        f.write(str(pid))
    wlog(f"{name} started (PID {pid})")
    return pid


# ── Service health check ─────────────────────────────────────
def check_service(name: str, cfg: dict, pid: int) -> tuple[bool, int, str]:
    """
    Returns (is_healthy, new_pid, status_msg).
    If dead, auto-restarts.
    """
    alive = is_process_alive(pid)

    if not alive:
        wlog(f"{name} PID {pid} is DEAD. Restarting...")
        new_pid = start_service(name, cfg)
        time.sleep(5)
        return (True, new_pid, "restarted")

    # Port-level check
    port_ok = is_port_listening(cfg["port"])
    if not port_ok:
        wlog(f"{name} PID {pid} alive but port {cfg['port']} not listening. Restarting...")
        new_pid = start_service(name, cfg)
        time.sleep(5)
        return (True, new_pid, "restarted (port dead)")

    # HTTP health check
    http_ok, detail = check_http_health(cfg["health_url"])
    if not http_ok:
        # 401/403 = service alive but requires auth → treat as healthy
        if "HTTP 401" in detail or "HTTP 403" in detail:
            wlog(f"{name} PID {pid} /health {detail} (auth required, service alive)")
        else:
            wlog(f"{name} PID {pid} /health FAILED: {detail}. Restarting...")
            new_pid = start_service(name, cfg)
            time.sleep(5)
            return (True, new_pid, f"restarted (/health: {detail})")

    return (True, pid, "ok")


# ── Main ──────────────────────────────────────────────────────
if __name__ == "__main__":
    wlog("=== Multi-Watchdog started ===")

    # Discover / start all services
    pids = {}
    statuses = {}
    for name, cfg in SERVICES.items():
        pid = get_pid_from_file(cfg["pid_file"])
        if pid and is_process_alive(pid) and is_port_listening(cfg["port"]):
            wlog(f"{name}: found running (PID {pid})")
        else:
            pid = start_service(name, cfg)
        pids[name] = pid
        statuses[name] = "starting"
        time.sleep(2)

    # Track "last good" per-service for activity monitoring
    last_http_ok = {name: time.time() for name in SERVICES}

    log_rotate_counter = 0
    cycle = 0

    while True:
        time.sleep(CHECK_INTERVAL)
        cycle += 1
        log_rotate_counter += 1

        any_restart = False

        for name, cfg in SERVICES.items():
            ok, new_pid, msg = check_service(name, cfg, pids[name])
            if new_pid != pids[name]:
                any_restart = True
                pids[name] = new_pid
                statuses[name] = msg
            else:
                statuses[name] = msg
            if ok:
                last_http_ok[name] = time.time()

        # Log rotation every 30 checks ≈ 15 min
        if log_rotate_counter >= 30:
            log_rotate_counter = 0
            for name, cfg in SERVICES.items():
                for log_path in cfg["log_files"]:
                    rotate_log(log_path)
            rotate_log(WATCHDOG_LOG)

        # Status report every 10 cycles (5 min)
        if cycle % 10 == 0 or any_restart:
            parts = []
            for name in SERVICES:
                p = pids[name]
                s = statuses[name]
                parts.append(f"{name} PID {p} [{s}]")
            wlog(" | ".join(parts))