feat: auto_heal.py + api/autoheal + F Tab auto-heal display

2026-07-18 12:28:47 +08:00
parent e4a9b7ff55
commit c43486a6f2
3 changed files with 312 additions and 0 deletions
@@ -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())
@@ -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 实际状态
 # ════════════════════════════════════════════════════════════
@@ -257,6 +257,20 @@ async function fHealth(){try{
      h+='<span class="tag '+(tk?'g':tn?'y':'r')+'">'+(tk?'正常':tn?'未部署':'异常')+'</span></div>';}
    h+='</div>';}

+  // --- 自动修复记录 ---
+  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+='<div class=section><h2 style=margin-bottom:8px>自动修复记录 <span style=font-size:18px;color:var(--dim)>'+esc(ahD.time.slice(0,16).replace('T',' '))+'</span></h2>';
+    for(var ai=0;ai<ahD.actions_taken.length;ai++){var a=ahD.actions_taken[ai];
+      var isOK=a.detail&&a.detail.indexOf('OK')>=0;
+      h+='<div style="display:flex;align-items:center;gap:8px;padding:4px 8px;margin:2px 0;background:var(--card);border-radius:4px">';
+      h+='<span class="sd '+(isOK?'g':'r')+'"></span>';
+      h+='<span style=flex:1>'+esc(a.action||'修复')+'</span>';
+      h+='<span style=color:var(--dim);font-size:18px>'+esc(a.service||'')+'</span>';
+      h+='<span class="tag '+(isOK?'g':'r')+'">'+(isOK?'成功':esc(a.detail||'失败'))+'</span></div>';}
+    h+='</div>';}}
+  catch(ahE){/* autoheal fetch failed, skip */}
+
  if(!h)h='<span class=dim>No data</span>';
  fill('health',h)}catch(e){fill('health','<span class=tag r>Error: '+esc(e.message)+'</span>')}}
 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:'';