diff --git a/gateway/scripts/auto_heal.py b/gateway/scripts/auto_heal.py new file mode 100644 index 0000000..c4860e4 --- /dev/null +++ b/gateway/scripts/auto_heal.py @@ -0,0 +1,285 @@ +#!/usr/bin/env python3 +""" +auto_heal.py — 自动诊断修复引擎 (Linux/246) +============================================= +1. 调用 Dashboard /api/expected 获取健康状态 +2. 对每个 critical 异常服务进行诊断 +3. 对可修复的服务执行修复（systemctl restart / docker restart） +4. 记录修复日志到 health_todos.jsonl + last_auto_heal.json +5. 供 Dashboard F Tab 展示修复记录 + +用法: + python3 auto_heal.py # 完整运行 + python3 auto_heal.py --dry-run # 只诊断不修复 + +定时触发（crontab）: + */5 * * * * cd /home/hmo/AgentsMeeting/gateway/scripts && /usr/bin/python3 auto_heal.py >> ../logs/auto_heal.log 2>&1 +""" +import json, os, sys, subprocess, time, logging +from datetime import datetime +from pathlib import Path +from urllib.request import urlopen, Request +from urllib.error import URLError, HTTPError + +# ── 路径 ──────────────────────────────────────────────────── +SCRIPT_DIR = Path(__file__).resolve().parent # gateway/scripts/ +GATEWAY_DIR = SCRIPT_DIR.parent # gateway/ +LOGS_DIR = GATEWAY_DIR / "logs" +TEMP_DIR = GATEWAY_DIR / "temp" +HEAL_LOG = LOGS_DIR / "auto_heal.log" +TODO_FILE = TEMP_DIR / "health_todos.jsonl" +SUMMARY_FILE = TEMP_DIR / "last_auto_heal.json" + +DASHBOARD_URL = os.environ.get("DASHBOARD_URL", "http://127.0.0.1:5803") +DRY_RUN = "--dry-run" in sys.argv + +os.makedirs(str(LOGS_DIR), exist_ok=True) +os.makedirs(str(TEMP_DIR), exist_ok=True) + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + handlers=[ + logging.FileHandler(str(HEAL_LOG), encoding="utf-8"), + logging.StreamHandler(), + ], +) +log = logging.getLogger("auto_heal") + +# ── 系统服务映射：service_type → systemd unit name ──────────── +# 从 agents.yaml 和实际 systemd 配置推导 +SERVICE_UNIT_MAP = { + "hermes_gateway": { + "mohe": "hermes-gateway@mohe", + "zhiwei": "hermes-gateway@zhiwei", + }, + "xmpp_bot": { + "xiaoguo": "xmpp-bot@xiaoguo", + "zhiwei": "xmpp-bot@zhiwei", + }, + "dashboard": "agentsmeeting-dashboard", +} + +# Docker 容器映射 +DOCKER_MAP = { + "wechat_bridge": "wxBotWebhook", +} + +# ── API 调用 ──────────────────────────────────────────────── + +def api_get(path, timeout=8): + """调用 Dashboard API，返回 JSON 或 None。""" + url = f"{DASHBOARD_URL}{path}" + try: + req = Request(url) + with urlopen(req, timeout=timeout) as resp: + return json.loads(resp.read().decode()) + except (URLError, HTTPError, json.JSONDecodeError, OSError) as e: + log.error(f"API GET {url} failed: {e}") + return None + + +# ── 修复执行 ──────────────────────────────────────────────── + +def systemctl_restart(unit_name): + """重启 systemd 服务单元。""" + if DRY_RUN: + log.info(f"[DRY-RUN] systemctl restart {unit_name}") + return True + try: + r = subprocess.run( + ["sudo", "systemctl", "restart", unit_name], + capture_output=True, text=True, timeout=20, + ) + if r.returncode == 0: + log.info(f"systemctl restart {unit_name} → OK") + return True + else: + log.warning(f"systemctl restart {unit_name} → FAIL ({r.stderr.strip()[:80]})") + return False + except Exception as e: + log.error(f"systemctl restart {unit_name} → EXCEPTION: {e}") + return False + + +def docker_restart(container_name): + """重启 Docker 容器。""" + if DRY_RUN: + log.info(f"[DRY-RUN] docker restart {container_name}") + return True + try: + r = subprocess.run( + ["docker", "restart", container_name], + capture_output=True, text=True, timeout=30, + ) + if r.returncode == 0: + log.info(f"docker restart {container_name} → OK") + return True + else: + log.warning(f"docker restart {container_name} → FAIL ({r.stderr.strip()[:80]})") + return False + except Exception as e: + log.error(f"docker restart {container_name} → EXCEPTION: {e}") + return False + + +def is_local_host(host): + """判断是否是本机（同 dashboard.py 的 _is_local_host）。""" + if host in ("127.0.0.1", "localhost", "::1"): + return True + if host == "192.168.1.246": + return True + try: + import socket + local_name = socket.gethostname() + local_ip = socket.gethostbyname(local_name) + if host in (local_name, local_ip): + return True + except: + pass + return False + + +def diagnose_and_heal(entry, actual_status): + """ + 对单个异常服务进行诊断并执行修复。 + 返回动作描述字符串，None 表示无操作。 + """ + name = entry["name"] + key = entry.get("key", "") + host = entry.get("host", "127.0.0.1") + check = entry.get("check", "") + expected = entry.get("expected", "") + port = entry.get("port") + + # ── 非 running 期望，跳过 ── + if expected != "running": + return None + + # ── 远程服务，无法自动修复 ── + if not is_local_host(host): + log.info(f"[SKIP] {name}: 远程服务 ({host})，跳过自动修复") + return None + + actual = actual_status or "unknown" + + # ── 本机端口不通 → 重启服务 ── + if check == "tcp" and port and actual != "running": + # 尝试匹配 systemd 单元 + unit = SERVICE_UNIT_MAP.get(key) + if isinstance(unit, dict): + # 需要从 name 中提取标识（"mohe", "zhiwei" 等） + for ident, u in unit.items(): + if ident in name.lower(): + ok = systemctl_restart(u) + return f"restart systemd {u}: {'OK' if ok else 'FAIL'}" + # 没匹配到具体标识，尝试第一个 + first_u = list(unit.values())[0] + ok = systemctl_restart(first_u) + return f"restart systemd {first_u}: {'OK' if ok else 'FAIL'}" + elif isinstance(unit, str): + ok = systemctl_restart(unit) + return f"restart systemd {unit}: {'OK' if ok else 'FAIL'}" + # Docker 容器 + container = DOCKER_MAP.get(key) + if container: + ok = docker_restart(container) + return f"restart docker {container}: {'OK' if ok else 'FAIL'}" + + log.warning(f"[UNKNOWN] {name}: 端口 {port} 不通，但无已知修复方式") + return f"port {port} down (无已知systemd单元)" + + # ── XMPP 离线 → 重启 xmpp bot ── + if check == "xmpp" and actual == "offline": + for ident, u in SERVICE_UNIT_MAP.get("xmpp_bot", {}).items(): + if ident in name.lower(): + ok = systemctl_restart(u) + return f"restart systemd {u}: {'OK' if ok else 'FAIL'}" + return f"xmpp offline (无对应 systemd 单元)" + + # ── health_url 不通 → 重启对应服务 ── + if check == "health_url" and actual != "running": + container = DOCKER_MAP.get(key) + if container: + ok = docker_restart(container) + return f"restart docker {container}: {'OK' if ok else 'FAIL'}" + + return None + + +def write_todo(action, entry_name, detail=""): + """写入 health_todos.jsonl（dashboard /api/todos 可读）。""" + record = { + "time": datetime.now().isoformat(), + "action": action, + "service": entry_name, + "detail": detail, + } + try: + with open(str(TODO_FILE), "a", encoding="utf-8") as f: + f.write(json.dumps(record, ensure_ascii=False) + "\n") + except Exception as e: + log.error(f"写 TODO 失败: {e}") + return record + + +def main(): + log.info("=" * 50) + log.info(f"auto_heal 启动 (dry_run={DRY_RUN})") + + # 1. 获取期望矩阵 + data = api_get("/api/expected") + if not data: + log.error("无法获取 /api/expected，退出") + return 1 + + expected_list = data.get("expected", []) + actual_dict = data.get("actual", {}) + log.info(f"期望条目: {len(expected_list)}") + + # 2. 诊断关键异常 + actions = [] + for entry in expected_list: + if not entry.get("critical"): + continue + expected = entry.get("expected", "") + if expected != "running": + continue + name = entry["name"] + actual_status = actual_dict.get(name, "unknown") + + # 健康（running / online / remote）→ 跳过 + if actual_status in ("running", "online") or (actual_status and actual_status.startswith("remote")): + continue + + log.warning(f"[ANOMALY] {name}: expected={expected}, actual={actual_status}") + action = diagnose_and_heal(entry, actual_status) + if action: + record = write_todo(action, name, actual_status) + actions.append(record) + + # 3. 写摘要 + summary = { + "time": datetime.now().isoformat(), + "dry_run": DRY_RUN, + "total_expected": len(expected_list), + "anomalies_found": len(actions), + "actions_taken": actions, + } + try: + with open(str(SUMMARY_FILE), "w", encoding="utf-8") as f: + json.dump(summary, f, ensure_ascii=False, indent=2) + except Exception as e: + log.error(f"写摘要失败: {e}") + + # 4. 报告 + if actions: + log.warning(f"发现 {len(actions)} 个异常，已执行 {len([a for a in actions if 'FAIL' not in a.get('detail','')])} 次修复") + else: + log.info("无异常，系统健康") + log.info("=" * 50) + return 0 if not actions or all("FAIL" not in a.get("detail", "") for a in actions) else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/gateway/scripts/dashboard.py b/gateway/scripts/dashboard.py index 9c23e98..93eb87d 100644 --- a/gateway/scripts/dashboard.py +++ b/gateway/scripts/dashboard.py @@ -943,6 +943,19 @@ def api_todos(): return jsonify({"todos": todos, "executor_log_tail": executor_tail, "count": len(todos)}) +@app.route("/api/autoheal") +def api_autoheal(): + """最近一次自动修复记录（auto_heal.py 写入的摘要）。""" + sf = TEMP_DIR / "last_auto_heal.json" + if sf.exists(): + try: + with open(str(sf)) as f: + return jsonify(json.load(f)) + except: + pass + return jsonify({"time": None, "dry_run": False, "total_expected": 0, "anomalies_found": 0, "actions_taken": []}) + + # ════════════════════════════════════════════════════════════ # F — 期望状态 vs 实际状态 # ════════════════════════════════════════════════════════════ diff --git a/gateway/scripts/templates/dashboard.html b/gateway/scripts/templates/dashboard.html index d984fed..cec6b21 100644 --- a/gateway/scripts/templates/dashboard.html +++ b/gateway/scripts/templates/dashboard.html @@ -257,6 +257,20 @@ async function fHealth(){try{ h+=''+(tk?'正常':tn?'未部署':'异常')+'';} h+='';} + // --- 自动修复记录 --- + try{var ahR=await fetch('/api/autoheal');var ahD=await ahR.json(); + if(ahD&&ahD.time&&ahD.actions_taken&&ahD.actions_taken.length>0){ + h+='

自动修复记录 '+esc(ahD.time.slice(0,16).replace('T',' '))+'

'; + for(var ai=0;ai=0; + h+='
'; + h+=''; + h+=''+esc(a.action||'修复')+''; + h+=''+esc(a.service||'')+''; + h+=''+(isOK?'成功':esc(a.detail||'失败'))+'
';} + h+='
';}} + catch(ahE){/* autoheal fetch failed, skip */} + if(!h)h='No data'; fill('health',h)}catch(e){fill('health','Error: '+esc(e.message)+'')}} async function fK(){try{var r=await fetch('/api/kanban'),d=await r.json();var tasks=d.tasks||[];var s=document.getElementById('kf-status');var a=document.getElementById('kf-assignee');var fs=s?s.value:'';var fa=a?a.value:'';