287 lines
10 KiB
Python
287 lines
10 KiB
Python
#!/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"]
|
||
raw_key = entry.get("key", "")
|
||
# 从复合 key（如 "agent-002:hermes_gateway"）提取服务名用于查表
|
||
service_key = raw_key.split(":")[-1] if ":" in raw_key else raw_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(service_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'}"
|
||
container = DOCKER_MAP.get(service_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(service_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())
|