268 lines
11 KiB
Python
268 lines
11 KiB
Python
"""设备/系统健康日报(每日一次):内存/磁盘/GPU 使用率 + 系统级错误检查。
|
||
|
||
对应 device-check deployment,每天 09:00(可调)。检查项:
|
||
- 内存/swap、宿主机磁盘(经 /:/host:ro 挂载)、GPU(nvidia-smi)——使用率 ≥ 阈值(默认 80%)标 🔴
|
||
- Docker 容器状态(经 /var/run/docker.sock):应运行但退出 / restarting / unhealthy / OOMKilled
|
||
- Prefect 近 24h FAILED/CRASHED flow runs、OCR 服务探活、PG 连通性、系统负载
|
||
|
||
依赖 docker-compose 给 prefect-worker 挂载 /:/host:ro、/var/run/docker.sock、GPU;
|
||
未挂载时对应检查项自动降级为"不可用",不影响其余检查。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import subprocess
|
||
from datetime import datetime, timedelta
|
||
|
||
import httpx
|
||
import psutil
|
||
from prefect import flow
|
||
|
||
from flows.shared import SyncDB, emit_sync_summary, send_markdown
|
||
|
||
# (容器内路径, 展示名);/host 为 compose 挂载的宿主机根目录
|
||
DISK_PATHS = [("/host", "/"), ("/host/home", "/home")]
|
||
DOCKER_SOCK = "/var/run/docker.sock"
|
||
|
||
_OCR_BASE = "http://192.168.124.92:9801"
|
||
|
||
|
||
def _gb(nbytes: float) -> str:
|
||
return f"{nbytes / (1 << 30):.1f}G"
|
||
|
||
|
||
def _check_memory() -> dict:
|
||
vm = psutil.virtual_memory()
|
||
sw = psutil.swap_memory()
|
||
return {
|
||
"pct": vm.percent, "used": _gb(vm.used), "total": _gb(vm.total),
|
||
"swapPct": sw.percent, "swapUsed": _gb(sw.used), "swapTotal": _gb(sw.total),
|
||
}
|
||
|
||
|
||
def _check_disks() -> list[dict]:
|
||
paths = DISK_PATHS if os.path.isdir(DISK_PATHS[0][0]) else [("/", "/")]
|
||
out = []
|
||
for raw, label in paths:
|
||
try:
|
||
du = psutil.disk_usage(raw)
|
||
out.append({"label": label, "pct": du.percent, "used": _gb(du.used), "total": _gb(du.total)})
|
||
except Exception as err: # noqa: BLE001
|
||
out.append({"label": label, "error": str(err)[:80]})
|
||
return out
|
||
|
||
|
||
def _check_gpus() -> list[dict] | str:
|
||
"""nvidia-smi 查 GPU 利用率与显存。容器无 GPU 访问时返回说明文字。"""
|
||
try:
|
||
proc = subprocess.run(
|
||
["nvidia-smi", "--query-gpu=index,name,utilization.gpu,memory.used,memory.total",
|
||
"--format=csv,noheader,nounits"],
|
||
capture_output=True, text=True, timeout=15,
|
||
)
|
||
if proc.returncode != 0:
|
||
return f"nvidia-smi 退出码 {proc.returncode}: {proc.stderr.strip()[:100]}"
|
||
gpus = []
|
||
for line in proc.stdout.strip().splitlines():
|
||
parts = [p.strip() for p in line.split(",")]
|
||
if len(parts) != 5:
|
||
continue
|
||
idx, name, util, mem_used, mem_total = parts
|
||
mem_pct = float(mem_used) / float(mem_total) * 100 if float(mem_total) > 0 else 0.0
|
||
gpus.append({"name": f"GPU{idx} {name[:24]}", "utilPct": float(util), "memPct": mem_pct,
|
||
"memUsed": f"{float(mem_used)/1024:.1f}G", "memTotal": f"{float(mem_total)/1024:.1f}G"})
|
||
return gpus or "未检测到 GPU"
|
||
except FileNotFoundError:
|
||
return "不可用(容器内无 nvidia-smi,需 compose 配置 GPU)"
|
||
except Exception as err: # noqa: BLE001
|
||
return f"检查失败: {str(err)[:80]}"
|
||
|
||
|
||
def _inspect_container(client: httpx.Client, cid: str) -> dict:
|
||
try:
|
||
resp = client.get(f"http://localhost/containers/{cid}/json", timeout=10)
|
||
resp.raise_for_status()
|
||
return resp.json()
|
||
except Exception: # noqa: BLE001
|
||
return {}
|
||
|
||
|
||
def _check_docker() -> dict:
|
||
"""容器状态汇总 + 异常容器列表(应运行但退出/restarting/unhealthy/OOMKilled)。"""
|
||
if not os.path.exists(DOCKER_SOCK):
|
||
return {"running": None, "issues": [{"name": "Docker", "issue": f"不可用(未挂载 {DOCKER_SOCK})"}]}
|
||
with httpx.Client(transport=httpx.HTTPTransport(uds=DOCKER_SOCK), timeout=10) as client:
|
||
resp = client.get("http://localhost/containers/json", params={"all": "1"})
|
||
resp.raise_for_status()
|
||
containers = resp.json()
|
||
running = sum(1 for c in containers if c.get("State") == "running")
|
||
issues = []
|
||
stopped = 0
|
||
for c in containers:
|
||
name = (c.get("Names") or ["?"])[0].lstrip("/")
|
||
state, status = c.get("State", ""), c.get("Status", "")
|
||
if state == "restarting":
|
||
issues.append({"name": name, "issue": f"反复重启 ({status})"})
|
||
elif state == "running" and "(unhealthy)" in status:
|
||
issues.append({"name": name, "issue": f"健康检查失败 ({status})"})
|
||
elif state in ("exited", "dead"):
|
||
detail = _inspect_container(client, c["Id"])
|
||
policy = ((detail.get("HostConfig") or {}).get("RestartPolicy") or {}).get("Name", "no")
|
||
oom = (detail.get("State") or {}).get("OOMKilled", False)
|
||
finished_at = (detail.get("State") or {}).get("FinishedAt", "")
|
||
if oom and _exited_recently(finished_at):
|
||
issues.append({"name": name, "issue": f"被 OOM 杀死 ({status})"})
|
||
elif policy in ("always", "unless-stopped"):
|
||
if _exited_recently(finished_at):
|
||
issues.append({"name": name, "issue": f"应常驻但刚退出 ({status}, restart={policy})"})
|
||
else:
|
||
stopped += 1 # 数周前人为停用的旧服务,不算异常
|
||
return {"running": running, "total": len(containers), "issues": issues, "stopped": stopped}
|
||
|
||
|
||
def _exited_recently(finished_at: str, hours: int = 24) -> bool:
|
||
try:
|
||
from datetime import timezone
|
||
dt = datetime.fromisoformat(finished_at.replace("Z", "+00:00"))
|
||
return (datetime.now(timezone.utc) - dt).total_seconds() < hours * 3600
|
||
except (ValueError, TypeError):
|
||
return True # 解析失败按新退出处理,宁可多报
|
||
|
||
|
||
def check_prefect_failures(hours: int = 24) -> list[dict]:
|
||
api_url = os.environ.get("PREFECT_API_URL", "http://localhost:4200/api")
|
||
since = (datetime.now().astimezone() - timedelta(hours=hours)).isoformat()
|
||
resp = httpx.post(
|
||
f"{api_url}/flow_runs/filter",
|
||
json={
|
||
"flow_runs": {"state": {"type": {"any_": ["FAILED", "CRASHED"]}}, "start_time": {"after_": since}},
|
||
"limit": 50,
|
||
"sort": "START_TIME_DESC",
|
||
},
|
||
timeout=15,
|
||
)
|
||
resp.raise_for_status()
|
||
out = []
|
||
for r in resp.json():
|
||
start = (r.get("start_time") or "")[:16].replace("T", " ")
|
||
out.append({"name": r.get("name", "?"), "state": (r.get("state") or {}).get("type", "?"),
|
||
"message": ((r.get("state") or {}).get("message") or "").strip()[:100], "start": start})
|
||
return out
|
||
|
||
|
||
def _check_services() -> dict:
|
||
from api.ocr import OcrClient
|
||
|
||
ocr_ok = False
|
||
try:
|
||
ocr_ok = OcrClient().health()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
pg_ok = False
|
||
try:
|
||
pg_ok = bool(SyncDB().ping())
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
return {"ocr": ocr_ok, "pg": pg_ok}
|
||
|
||
|
||
def _mark(pct: float | None, threshold: int) -> str:
|
||
"""百分比展示:超阈值加粗+🔴(WeCom aibot markdown 不支持 font 标签和表格)。"""
|
||
if pct is None:
|
||
return "—"
|
||
if pct >= threshold:
|
||
return f"**{pct:.0f}%**🔴"
|
||
return f"{pct:.0f}%"
|
||
|
||
|
||
@flow(name="device-check", log_prints=True)
|
||
def device_check(alarm_threshold: int = 80) -> dict:
|
||
"""设备/系统健康日报:检查并直接发企微群。单项检查失败不影响其余项。"""
|
||
checks: dict[str, object] = {}
|
||
for name, fn in [
|
||
("memory", _check_memory), ("disks", _check_disks), ("gpus", _check_gpus),
|
||
("docker", _check_docker), ("prefectFailures", check_prefect_failures), ("services", _check_services),
|
||
]:
|
||
try:
|
||
checks[name] = fn()
|
||
except Exception as err: # noqa: BLE001
|
||
checks[name] = None
|
||
print(f"[device-check] {name} 检查失败: {err}")
|
||
|
||
threshold = alarm_threshold
|
||
lines = [f"🖥 **设备健康日报** {datetime.now().strftime('%m-%d %H:%M')}", ""]
|
||
|
||
mem = checks.get("memory") or {}
|
||
if mem:
|
||
lines.append(f"**内存** {_mark(mem['pct'], threshold)}（{mem['used']}/{mem['total']}）"
|
||
f"｜swap {_mark(mem['swapPct'], threshold)}（{mem['swapUsed']}/{mem['swapTotal']}）")
|
||
else:
|
||
lines.append("**内存** 检查失败")
|
||
|
||
disks = checks.get("disks") or []
|
||
if disks:
|
||
lines.append("**磁盘**")
|
||
for d in disks:
|
||
if "error" in d:
|
||
lines.append(f"> {d['label']} 检查失败：{d['error']}")
|
||
else:
|
||
lines.append(f"> {d['label']} {_mark(d['pct'], threshold)}（{d['used']}/{d['total']}）")
|
||
|
||
gpus = checks.get("gpus")
|
||
if isinstance(gpus, list) and gpus:
|
||
lines.append("**GPU**")
|
||
for g in gpus:
|
||
lines.append(f"> {g['name']}｜利用率 {_mark(g['utilPct'], threshold)}"
|
||
f"｜显存 {_mark(g['memPct'], threshold)}（{g['memUsed']}/{g['memTotal']}）")
|
||
else:
|
||
lines.append(f"**GPU** {gpus if isinstance(gpus, str) else '未检测到'}")
|
||
|
||
docker = checks.get("docker") or {}
|
||
issues = docker.get("issues") or []
|
||
if docker.get("running") is None:
|
||
lines.append(f"**Docker** {issues[0]['issue'] if issues else '不可用'}")
|
||
else:
|
||
docker_line = f"**Docker** 运行 {docker['running']}/{docker.get('total', '?')}"
|
||
if docker.get("stopped"):
|
||
docker_line += f" · 已停用 {docker['stopped']}"
|
||
if issues:
|
||
docker_line += f" · 异常 **{len(issues)}**🔴"
|
||
else:
|
||
docker_line += " · 异常 0"
|
||
lines.append(docker_line)
|
||
for it in issues[:8]:
|
||
lines.append(f"> 🔴 {it['name']}：{it['issue']}")
|
||
|
||
failures = checks.get("prefectFailures") or []
|
||
lines.append(f"**Prefect 失败(24h)** {len(failures)} 个" + (" ✅" if not failures else ""))
|
||
for f in failures[:5]:
|
||
lines.append(f"> 🔴 {f['name']}（{f['state']}）{f['start']}")
|
||
|
||
svc = checks.get("services") or {}
|
||
lines.append(f"**探活** OCR {'✅' if svc.get('ocr') else '🔴'}｜PG {'✅' if svc.get('pg') else '🔴'}"
|
||
f"｜负载 {os.getloadavg()[0]:.1f}")
|
||
|
||
alarms = (
|
||
(mem and (mem["pct"] >= threshold or mem["swapPct"] >= threshold))
|
||
or any(d.get("pct", 0) >= threshold for d in (checks.get("disks") or []))
|
||
or (isinstance(gpus, list) and any(g["utilPct"] >= threshold or g["memPct"] >= threshold for g in gpus))
|
||
or bool(issues) or bool(failures) or not svc.get("ocr") or not svc.get("pg")
|
||
)
|
||
lines += ["", ("🔴 **有项目需要处理，见上**" if alarms else "✅ **一切正常**")]
|
||
|
||
content = "\n".join(lines)
|
||
sent = send_markdown(content)
|
||
emit_sync_summary("设备健康日报", [
|
||
("内存", f"{_mark(mem['pct'], threshold)}" if mem else "—"),
|
||
("Docker 异常", len(issues)),
|
||
("Prefect 失败(24h)", len(failures)),
|
||
("OCR", "✅" if svc.get("ocr") else "🔴"), ("PG", "✅" if svc.get("pg") else "🔴"),
|
||
("已发企微", "是" if sent else "否"),
|
||
])
|
||
return {"alarms": bool(alarms), "sent": sent, "dockerIssues": len(issues), "prefectFailures": len(failures)}
|
||
|
||
|
||
if __name__ == "__main__":
|
||
device_check.serve(name="device-check", cron="0 9 * * *")
|