hmo
1dd3f9aad5
- service_registry: xmpp_bot port 5802 -> 5807 (matches xmpp_agent_core), fix health_url/log - xmpp_watchdog: skip services without script or pid_file (dashboard self-managed, gateway_* remote) — previously crashed on open(None) - start.bat: launch watchdog after xxm bot so crash auto-recovery runs on boot
272 lines
9.9 KiB
Python
272 lines
9.9 KiB
Python
"""
|
|
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"]
|
|
# watchdog 只管理「本机可拉起」的进程服务：
|
|
# - script 为 None → 远程服务（gateway_mohe/zhiwei），由 246 crontab/systemd 管
|
|
# - pid_file 为 None → dashboard，自管进程
|
|
# 否则 start_service 会 open(None) 崩溃，且无 pid_file 无法追踪。
|
|
if not svc.get("script") or not svc.get("pid_file"):
|
|
print(f"[watchdog] skip {name}: remote/self-managed service", flush=True)
|
|
continue
|
|
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))
|