yeuimu
9b130097f9
- wecom.py 重写: HTTP POST webhook(替代 aibot WebSocket 按需连接), 接口 send_markdown/send_text 不变, Secret Block wecom-webhook 存 URL - webhook markdown 支持标题/引用/font 三色, 报告改为 '## 标题 + 单行结论 + 引用详情' 紧凑排版, 结论行用 info/warning 色 - alert_failure 改标题式排版; device_check 结论用 font 色, 去掉负载行 - 移除 wecom-aibot-sdk 依赖与 wecom_capture_chatid.py(镜像下次构建生效, 无需立即重建)
464 lines
20 KiB
Python
464 lines
20 KiB
Python
"""设备/系统健康日报(每日 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'<font color="warning">**{pct:.0f}%** 🔴</font>'
|
||
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")
|
||
temp_txt = (f"温度峰 {temp_peak:.0f}°C" + (" 🔴过热" if temp_peak >= GPU_TEMP_WARN else " ✅")
|
||
if temp_peak is not None else "温度 无数据")
|
||
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:
|
||
docker_txt = f"运行 {docker['running']}/{docker.get('total', '?')} · 异常 {len(issues)}" + ("🔴" if issues else "")
|
||
if svc:
|
||
docker_txt += f" ｜ OCR {'✅' if svc.get('ocr') else '🔴'} ｜ 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"**失败任务(24h)** {len(failures)} 次 🔴")
|
||
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("**失败任务(24h)** 0 ✅")
|
||
|
||
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('<font color="warning">🔴 有项目需要处理，见上</font>' if alarms
|
||
else '<font color="info">✅ 一切正常</font>')
|
||
|
||
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 * * *")
|