"""设备/系统健康日报(每日 09:00):内存/swap/GPU 平均·峰值·最低 + 磁盘余量与目录增长 + 系统级错误。
对应 device-check deployment。数据来源:
- 24h 指标:metrics-agent 侧车容器每 5 分钟采样,写 data/metrics/metrics-YYYY-MM-DD.jsonl(worker 经
/data/metrics:ro 只读挂载读取;本地调试直接读仓库 data/metrics)
- 磁盘增长:metrics-agent 每日 du 快照 data/metrics/disk-*.json,日报对比最近两份得"增长最快目录"
- 采样缺失时降级为 psutil/nvidia-smi 瞬时值,不阻塞报告
另检查:Docker 容器异常、Prefect 近 24h 失败明细、OCR/PG 探活。超阈值(默认 80%)标 🔴。
依赖 compose 挂载:/data/metrics:ro、/:/host:ro、/var/run/docker.sock、GPU;缺失项自动降级。
"""
from __future__ import annotations
import json
import os
import subprocess
from datetime import datetime, timedelta
from pathlib import Path
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"
METRICS_DIR_CANDIDATES = ("/data/metrics", str(Path(__file__).resolve().parents[2] / "data" / "metrics"))
GPU_TEMP_WARN = 85 # °C
_OCR_BASE = "http://192.168.124.92:9801"
def _gb(nbytes: float) -> str:
return f"{nbytes / (1 << 30):.1f}G"
def _fmt_growth(nbytes: float) -> str:
if abs(nbytes) >= 1 << 30:
return f"{nbytes / (1 << 30):+.1f}G"
return f"{nbytes / (1 << 20):+.0f}M"
def _metrics_dir() -> str | None:
cands = [os.environ.get("METRICS_DIR"), *METRICS_DIR_CANDIDATES]
for p in cands:
if p and os.path.isdir(p):
return p
return None
# ---------- 24h 采样聚合 ----------
def read_samples(metrics_dir: str | None, hours: int = 24) -> list[dict]:
"""读取 metrics-agent 的 JSONL 采样,返回窗口内记录(ts 升序)。"""
if not metrics_dir:
return []
now = datetime.now()
cutoff = now - timedelta(hours=hours)
out: list[dict] = []
for i in range(hours // 24 + 1):
day = (now - timedelta(days=i)).strftime("%Y-%m-%d")
path = Path(metrics_dir) / f"metrics-{day}.jsonl"
if not path.is_file():
continue
try:
rows = [json.loads(l) for l in path.read_text(encoding="utf-8").splitlines() if l.strip()]
except (json.JSONDecodeError, OSError):
continue
for r in rows:
try:
if datetime.fromisoformat(r["ts"]) >= cutoff:
out.append(r)
except (KeyError, ValueError, TypeError):
continue
out.sort(key=lambda r: r["ts"])
return out
def _stats(values: list[float]) -> dict | None:
if not values:
return None
return {"avg": sum(values) / len(values), "peak": max(values), "min": min(values)}
def aggregate_samples(samples: list[dict]) -> dict:
"""聚合采样:内存/swap 统计 + 按 GPU 的利用率/显存/温度统计。"""
agg: dict = {"count": len(samples)}
if not samples:
return agg
agg["firstTs"] = samples[0]["ts"]
mem = [s["memPct"] for s in samples if isinstance(s.get("memPct"), (int, float))]
swap = [s["swapPct"] for s in samples if isinstance(s.get("swapPct"), (int, float))]
agg["mem"], agg["swap"] = _stats(mem), _stats(swap)
agg["memTotal"] = next((s.get("memTotal") for s in reversed(samples) if s.get("memTotal")), None)
agg["swapTotal"] = next((s.get("swapTotal") for s in reversed(samples) if s.get("swapTotal")), None)
gpus: dict[int, dict] = {}
for s in samples:
for g in s.get("gpus") or []:
slot = gpus.setdefault(g.get("index", 0), {"name": g.get("name", "?"), "util": [], "memUsed": [], "memTotal": g.get("memTotal"), "temp": []})
for key in ("utilPct", "memUsed", "tempC"):
if isinstance(g.get(key), (int, float)):
slot["util" if key == "utilPct" else "memUsed" if key == "memUsed" else "temp"].append(g[key])
agg["gpus"] = [
{"index": idx, "name": slot["name"], "util": _stats(slot["util"]),
"memPeakUsed": max(slot["memUsed"]) if slot["memUsed"] else None,
"memTotal": slot["memTotal"], "tempPeak": max(slot["temp"]) if slot["temp"] else None}
for idx, slot in sorted(gpus.items())
]
return agg
# ---------- 磁盘目录增长(du 快照对比) ----------
def disk_growth(metrics_dir: str | None) -> dict:
"""对比最近两份 disk-*.json 快照,返回各挂载点 {dir, growth, snapshotTs}。"""
out: dict = {}
if not metrics_dir:
return out
snaps = sorted(Path(metrics_dir).glob("disk-*.json"))
if len(snaps) < 2:
return out
try:
new = json.loads(snaps[-1].read_text(encoding="utf-8"))
old = json.loads(snaps[-2].read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
return out
for key in ("root", "home"):
nd = (new.get(key) or {}).get("dirs") or {}
od = (old.get(key) or {}).get("dirs") or {}
best: tuple[str, int] | None = None
for name, nb in nd.items():
ob = od.get(name)
if ob is None:
continue
if best is None or nb - ob > best[1]:
best = (name, nb - ob)
out[key] = {"dir": best[0] if best else None,
"growth": best[1] if best else 0,
"snapshotTs": new.get("ts", "")}
return out
# ---------- 瞬时兜底检查(采样缺失时用) ----------
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]:
"""近 N 小时 FAILED/CRASHED 的 flow run,附 workflow 名(供日报/设备报告)。"""
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()
runs = resp.json()
flow_names: dict[str, str] = {}
flow_ids = list({r.get("flow_id") for r in runs if r.get("flow_id")})
if flow_ids:
fresp = httpx.post(f"{api_url}/flows/filter", json={"flows": {"id": {"any_": flow_ids}}}, timeout=15)
fresp.raise_for_status()
flow_names = {f["id"]: f["name"] for f in fresp.json()}
out = []
for r in runs:
start = (r.get("start_time") or "")[:16].replace("T", " ")
out.append({"flow": flow_names.get(r.get("flow_id"), "?"), "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:
"""百分比展示:超阈值用 warning 色加粗。"""
if pct is None:
return "—"
if pct >= threshold:
return f'**{pct:.0f}%**'
return f"{pct:.0f}%"
def _live_disks() -> list[dict]:
"""实时磁盘(含设备去重汇总),供 总体/根目录/home 行。"""
paths = DISK_PATHS if os.path.isdir(DISK_PATHS[0][0]) else [("/", "/")]
mounts = []
seen: set[tuple[int, int]] = set()
total_used = total_avail = 0
for raw, label in paths:
try:
du = psutil.disk_usage(raw)
except Exception: # noqa: BLE001
continue
key = (du.total, du.free)
if key not in seen:
seen.add(key)
total_used += du.used
total_avail += du.free
mounts.append({"label": label, "pct": du.percent, "used": du.used,
"free": du.free, "total": du.total})
overall = {"used": total_used, "free": total_avail,
"pct": total_used / (total_used + total_avail) * 100 if total_used + total_avail else 0.0}
return mounts, overall
@flow(name="device-check", log_prints=True)
def device_check(alarm_threshold: int = 80) -> dict:
"""设备/系统健康日报:按 24h 采样报 平均/峰值/最低,直接发企微群。单项失败不影响其余。"""
mdir = _metrics_dir()
agg = aggregate_samples(read_samples(mdir))
growth = disk_growth(mdir)
checks: dict[str, object] = {}
for name, fn in [("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}")
try:
mounts, overall = _live_disks()
except Exception as err: # noqa: BLE001
mounts, overall = [], None
print(f"[device-check] 磁盘检查失败: {err}")
threshold = alarm_threshold
header = f"## 设备日报 {datetime.now():%m-%d %H:%M}（24h 采样 {agg['count']} 次"
if agg.get("firstTs"):
first = datetime.fromisoformat(agg["firstTs"])
if first > datetime.now() - timedelta(hours=23):
header += f" · 自 {first:%m-%d %H:%M} 起"
lines = [header + "）"]
# ---- 内存 / swap ----
mem, swap = agg.get("mem"), agg.get("swap")
if mem:
total_txt = f"（共 {_gb((agg.get('memTotal') or 0) * (1 << 30))}）" if agg.get("memTotal") else ""
lines.append(f"**内存** 均 {mem['avg']:.0f}% · 峰 {_mark(mem['peak'], threshold)} · 低 {mem['min']:.0f}%{total_txt}")
else:
try:
now_mem = _check_memory()
lines.append(f"**内存** 当前 {_mark(now_mem['pct'], threshold)}（{now_mem['used']}/{now_mem['total']}，无采样）")
mem = {"peak": now_mem["pct"]}
except Exception: # noqa: BLE001
lines.append("**内存** 检查失败")
mem = None
if swap:
total_txt = f"（共 {_gb((agg.get('swapTotal') or 0) * (1 << 30))}）" if agg.get("swapTotal") else ""
lines.append(f"**交换内存** 均 {swap['avg']:.0f}% · 峰 {_mark(swap['peak'], threshold)} · 低 {swap['min']:.0f}%{total_txt}")
else:
lines.append("**交换内存** 无采样数据")
swap = None
# ---- 磁盘:总体 / 根目录 / home ----
if overall:
lines.append(f"**磁盘** 总体 已用 {_gb(overall['used'])} · 剩 {_gb(overall['free'])}"
f"（{_mark(overall['pct'], threshold)}）")
else:
lines.append("**磁盘** 检查不可用")
label_map = {"/": "root", "/home": "home"}
for m in mounts:
g = growth.get(label_map.get(m["label"], ""))
growth_txt = f" · 增长最快 {g['dir']} {_fmt_growth(g['growth'])}" if g and g.get("dir") else ""
lines.append(f"{m['label']} {_mark(m['pct'], threshold)} · 已用 {_gb(m['used'])} · 剩 {_gb(m['free'])}{growth_txt}")
# ---- GPU ----
gpu_stats = agg.get("gpus") or []
if gpu_stats:
for g in gpu_stats:
name = g.get("name") or f"GPU{g['index']}"
util = g.get("util")
util_txt = (f"利用率 均 {util['avg']:.0f}% · 峰 {util['peak']:.0f}% · 低 {util['min']:.0f}%"
if util else "利用率 无数据")
mem_peak, mem_total = g.get("memPeakUsed"), g.get("memTotal")
if mem_peak is not None and mem_total:
mem_txt = (f"显存峰 {_gb(mem_peak * (1 << 30))}/{_gb(mem_total * (1 << 30))}"
f"（{_mark(mem_peak / mem_total * 100, threshold)}）")
else:
mem_txt = "显存 无数据"
temp_peak = g.get("tempPeak")
if temp_peak is None:
temp_txt = "温度 无数据"
elif temp_peak >= GPU_TEMP_WARN:
temp_txt = f'温度峰 {temp_peak:.0f}°C 过热'
else:
temp_txt = f"温度峰 {temp_peak:.0f}°C 正常"
lines.append(f"**GPU** {name} {util_txt} ｜ {mem_txt} ｜ {temp_txt}")
else:
gpus_now = _check_gpus()
if isinstance(gpus_now, list) and gpus_now:
for g in gpus_now:
lines.append(f"**GPU** {g['name']} 利用率 {g['utilPct']:.0f}% · 显存 {g['memUsed']}/{g['memTotal']}（无采样）")
else:
lines.append(f"**GPU** {gpus_now if isinstance(gpus_now, str) else '未检测到'}")
# ---- Docker / 探活 / 失败任务 ----
docker = checks.get("docker") or {}
issues = docker.get("issues") or []
svc = checks.get("services") or {}
if docker.get("running") is None:
docker_txt = issues[0]["issue"] if issues else "不可用"
else:
n_issues = len(issues)
docker_txt = "运行 " + f"{docker['running']}/{docker.get('total', '?')} · 异常 "
docker_txt += f'{n_issues}' if n_issues else "0"
if svc:
docker_txt += " · OCR " + ("正常" if svc.get("ocr") else '异常')
docker_txt += " · PG " + ("正常" if svc.get("pg") else '异常')
lines.append(f"**Docker** {docker_txt}")
for it in issues[:8]:
lines.append(f"{it['name']}：{it['issue']}")
failures = checks.get("prefectFailures") or []
if failures:
lines.append(f'**任务** 失败 {len(failures)} 次（24h）')
for f in failures[:5]:
lines.append(f"{f['flow']}：{f['name']} {f['state']} {(f.get('start') or '')[5:]}")
if len(failures) > 5:
lines.append(f"…共 {len(failures)} 次")
else:
lines.append("**任务** 失败 0 次（24h）")
mem_peak_for_alarm = mem.get("peak") if mem else None
swap_peak_for_alarm = swap.get("peak") if swap else None
alarms = (
(mem_peak_for_alarm is not None and mem_peak_for_alarm >= threshold)
or (swap_peak_for_alarm is not None and swap_peak_for_alarm >= threshold)
or any(m["pct"] >= threshold for m in mounts)
or any((g.get("memPeakUsed") or 0) / (g.get("memTotal") or 1) * 100 >= threshold for g in gpu_stats)
or any((g.get("tempPeak") or 0) >= GPU_TEMP_WARN for g in gpu_stats)
or bool(issues) or bool(failures) or not svc.get("ocr") or not svc.get("pg")
)
lines.append('有项目需要处理，见上' if alarms
else '一切正常')
content = "\n".join(lines)
sent = send_markdown(content)
emit_sync_summary("设备健康日报", [
("采样次数", agg["count"]),
("内存峰值", f"{mem_peak_for_alarm:.0f}%" if mem_peak_for_alarm is not None else "—"),
("Docker 异常", len(issues)),
("失败任务(24h)", len(failures)),
("OCR", "✅" if svc.get("ocr") else "🔴"), ("PG", "✅" if svc.get("pg") else "🔴"),
("已发企微", "是" if sent else "否"),
])
return {"alarms": bool(alarms), "sent": sent, "samples": agg["count"],
"dockerIssues": len(issues), "prefectFailures": len(failures)}
if __name__ == "__main__":
device_check.serve(name="device-check", cron="0 9 * * *")