yeuimu
1d0da2a00b
- flows/shared/wecom.py: WSClient 按需连接(连接→认证→发送→断开),全容错,超长分段 - flows/monitor/alert_failure.py: Prefect 事件触发(Failed/Crashed→run deployment),发群报警 - flows/monitor/daily_report.py: 每日 18:30 近7天 RIIN/SDS 核验(复用 verify)+商品完整性+OCR+失败回顾 - flows/monitor/device_check.py: 每日 09:00 内存/磁盘/GPU(80%阈值)+Docker 容器异常+服务探活 - db.py: 新增 get_goods_integrity/ping; compose worker 挂载 docker.sock+/:/host:ro+GPU - scripts/wecom_capture_chatid.py: 一次性捕获群 chat_id 工具
263 lines
11 KiB
Python
263 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:
|
|
if pct is None:
|
|
return "—"
|
|
return f"{pct:.0f}%" + (" 🔴" if pct >= threshold else "")
|
|
|
|
|
|
@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('%Y-%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("**内存**: 检查失败")
|
|
|
|
lines.append("**磁盘**:")
|
|
lines.append("| 挂载点 | 使用率 | 已用/总量 |")
|
|
lines.append("| --- | --- | --- |")
|
|
for d in checks.get("disks") or []:
|
|
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']} |")
|
|
|
|
lines.append("**GPU**:")
|
|
gpus = checks.get("gpus")
|
|
if isinstance(gpus, list) and gpus:
|
|
lines.append("| GPU | 利用率 | 显存 |")
|
|
lines.append("| --- | --- | --- |")
|
|
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"> {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:
|
|
lines.append(f"**Docker 容器**: 运行中 {docker['running']}/{docker.get('total', '?')}"
|
|
+ (f",已停用 {docker['stopped']}" if docker.get("stopped") else "")
|
|
+ (f",⚠️ 异常 {len(issues)} 个" if issues else ",全部正常"))
|
|
for it in issues[:10]:
|
|
lines.append(f"- 🔴 {it['name']}: {it['issue']}")
|
|
|
|
failures = checks.get("prefectFailures") or []
|
|
lines.append(f"**Prefect 近24h 失败**: {len(failures)} 个 flow run"
|
|
+ ("" if failures else " ✅"))
|
|
for f in failures[:8]:
|
|
lines.append(f"- 🔴 {f['name']}({f['state']}) {f['start']} {f['message'][:60]}")
|
|
|
|
svc = checks.get("services") or {}
|
|
lines.append(f"**服务探活**: OCR {'✅' if svc.get('ocr') else '🔴'} | "
|
|
f"PG {'✅' if svc.get('pg') else '🔴'} | 负载 {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 * * *")
|