diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..894f25d --- /dev/null +++ b/.dockerignore @@ -0,0 +1,6 @@ +data/ +.git/ +.venv/ +__pycache__/ +*.log +.env diff --git a/.gitignore b/.gitignore index c08faad..7dd2b42 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,5 @@ __pycache__/ external_projects/ *.log .prefect/ +data/ +.zcode/ diff --git a/AGENTS.md b/AGENTS.md index 7495b5f..41d1670 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -84,9 +84,10 @@ uv run prefect # 运行 prefect CLI - 发送层 `flows/shared/wecom.py`:`send_markdown/send_text`,wecom-aibot-sdk 按需连接(连接→认证→发送→断开),全容错。 凭据在 `Secret` Block `wecom-aibot-credentials`(JSON:`bot_id/secret/chat_id`);chat_id 用 `scripts/wecom_capture_chatid.py` 捕获。 - `flows/monitor/alert_failure.py`(wecom-alert):prefect.yaml 里声明的事件触发器,任意 flow run `Failed/Crashed` 自动运行并发群报警,零代码侵入,新 flow 自动覆盖。 -- `flows/monitor/daily_report.py`(daily-report,18:30):近 7 天 RIIN/SDS 核验(subflow 复用 verify)+商品完整性(`db.get_goods_integrity`)+OCR+失败回顾。 -- `flows/monitor/device_check.py`(device-check,09:00):内存/swap/宿主机磁盘(`/` 挂载在容器 `/host`)/GPU(nvidia-smi)阈值 80% 报警 + Docker 容器异常(docker.sock)+Prefect 失败+OCR/PG 探活。 -- worker 容器已挂载 `/:/host:ro`、`/var/run/docker.sock`、GPU(见 docker-compose.yml),勿移除,否则对应检查降级。 +- `flows/monitor/daily_report.py`(daily-report,18:30):近 7 天 RIIN/SDS 核验(subflow 复用 verify,先报订单数量/商品行数是否一致,再报状态不符单数"已自动刷新修复",verify 有差异时修复后重数即 `verifyReport.post`)+OCR 当日处理量(按 `Orders.ocrProcessedAt` 盖章,`db.count_ocr_today`)+近 24h 失败任务明细(workflow/run/次数/时间)。 +- `flows/monitor/device_check.py`(device-check,09:00):按 24h 采样报内存/swap/GPU 的平均·峰值·最低 + 磁盘 总体/根目录/home 余量与"增长最快目录"(du 快照对比)+ GPU 温度;另查 Docker 容器异常(docker.sock)、Prefect 失败、OCR/PG 探活。阈值 80% 报警。 +- **`metrics-agent` 侧车容器**(docker-compose 服务,非 Prefect flow):`scripts/metrics_agent.py` 每 5 分钟采样内存/swap/GPU 写 `data/metrics/metrics-YYYY-MM-DD.jsonl`,每日 7 点后 du 快照 `data/metrics/disk-*.json`(实际占块;容错个别文件消失的 rc=1);保留采样 14 天/快照 7 份。device-check 经 `/data/metrics:ro` 读取,采样缺失自动降级为瞬时值。 +- worker 容器已挂载 `/:/host:ro`、`/var/run/docker.sock`、GPU、`./data/metrics:ro`(见 docker-compose.yml),勿移除,否则对应检查降级;镜像含 tzdata(`ENV TZ=Asia/Shanghai`)。 ## 参考文档 diff --git a/Dockerfile b/Dockerfile index c5e61fa..61f80d0 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,8 +1,10 @@ FROM python:3.12-slim -RUN apt-get update && apt-get install -y --no-install-recommends git curl ca-certificates \ +RUN apt-get update && apt-get install -y --no-install-recommends git curl ca-certificates tzdata \ && rm -rf /var/lib/apt/lists/* +ENV TZ=Asia/Shanghai + RUN curl -LsSf https://astral.sh/uv/install.sh | sh ENV PATH="/root/.local/bin:$PATH" diff --git a/docker-compose.yml b/docker-compose.yml index 79bef4d..81db5f6 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -21,9 +21,10 @@ services: depends_on: - prefect-server volumes: - # 设备监控(device-check flow):宿主机磁盘使用率 + Docker 容器状态 + # 设备监控(device-check flow):宿主机磁盘使用率 + Docker 容器状态 + 指标采样 - /var/run/docker.sock:/var/run/docker.sock - /:/host:ro + - ./data/metrics:/data/metrics:ro deploy: resources: reservations: @@ -34,5 +35,26 @@ services: capabilities: [gpu] restart: unless-stopped + # 宿主机指标采样侧车:每 5 分钟写内存/swap/GPU,每日 du 快照(供 device-check 日报) + metrics-agent: + build: . + container_name: metrics-agent + command: python /opt/metrics_agent.py + environment: + TZ: Asia/Shanghai + METRICS_DIR: /data/metrics + volumes: + - /:/host:ro + - ./data/metrics:/data/metrics + - ./scripts/metrics_agent.py:/opt/metrics_agent.py:ro + deploy: + resources: + reservations: + devices: + - driver: nvidia + count: all + capabilities: [gpu] + restart: unless-stopped + volumes: prefect-data: diff --git a/flows/monitor/daily_report.py b/flows/monitor/daily_report.py index a618893..83eeb4c 100644 --- a/flows/monitor/daily_report.py +++ b/flows/monitor/daily_report.py @@ -1,12 +1,12 @@ -"""数据核验日报(每日 18:30):近 N 天 RIIN/SDS 订单+商品数据准确性核验并上报企微。 +"""数据核验日报(每日 18:30):RIIN/SDS 数量一致性 + 状态差异 + OCR 当日 + 失败任务回顾。 -对应 daily-report deployment,cron 30 18 * * *。内容: -1. 复用 riin-verify / sds-verify 逻辑(subflow 调用):远程 vs 本地逐天逐状态计数比对, - 有差异自动拉取修复——与早 6/7 点的每日核验互补,二次确认白天同步质量。 -2. 商品数据完整性:窗口内零商品订单 / 商品无名称 / 全库孤儿商品(本地 SQL)。 -3. OCR 处理状态与当日失败 flow runs 回顾(事件报警的兜底汇总)。 +对应 daily-report deployment,cron 30 18 * * *。报告结构: +1. RIIN/SDS:先报订单数量(远程 vs 本地,有差异修复后重数)与商品行数是否一致; + 再报核验时发现多少单状态不符(已自动刷新修复)。subflow 复用 riin/sds-verify。 +2. OCR:今天处理了多少订单、识别失败多少(按 ocrProcessedAt 盖章统计)。 +3. 失败任务:近 24h 哪些 workflow、哪些 run、失败几次、何时失败(事件报警的兜底汇总)。 -消息排版遵循 WeCom markdown 子集(无表格,引用块+彩色字体)。 +消息排版遵循 WeCom aibot markdown 子集(无表格、无 font 标签,引用块 + emoji)。 """ from __future__ import annotations @@ -21,70 +21,79 @@ from flows.sds.verify import sds_verify from flows.shared import SyncDB, emit_sync_summary, send_markdown, start_of_day - -def _short_date(date_str: str) -> str: - return date_str[5:] if len(date_str) == 10 and date_str[:4].isdigit() else date_str - - def _affected_units(diff: dict) -> int: """状态不符单数 ≈ 本地比远程多的状态计数之和(远程改了状态的单)。""" return sum(v for v in (diff or {}).values() if v > 0) +def _system_lines(name: str, report: dict, goods: dict, lines: list[str]) -> bool: + """单系统的核验两段:①数量是否一致(订单+商品) ②状态不符多少单。返回是否需关注。""" + vr = (report.get("verifyReport") or {}) if isinstance(report, dict) else {} + days_r = vr.get("days") or [] + post = vr.get("post") + if post: + remote_t, local_t, ok = post["remote"], post["local"], post.get("daysWithDiff", 0) == 0 + else: + remote_t = sum(d.get("remote", 0) for d in days_r) + local_t = sum(d.get("local", 0) for d in days_r) + ok = all(d.get("ok") for d in days_r) + + if ok: + lines.append(f"**{name} 订单** 远程 {remote_t:,} · 本地 {local_t:,} ✅") + else: + lines.append(f"**{name} 订单** 远程 {remote_t:,} · 本地 {local_t:,} ⚠️ 不一致") + + diff_days = [d for d in days_r if not d.get("ok")] + diff_units = sum(_affected_units(d.get("diff")) for d in diff_days) + if diff_units: + lines.append(f"> 核验时发现约 {diff_units:,} 单状态不符（{len(diff_days)} 天），已自动刷新修复") + + g = goods or {} + zero = g.get("zeroGoodsOrders", 0) + if zero: + lines.append(f"**{name} 商品** {g.get('goods', 0):,} 行 · {zero:,} 单缺商品 ⚠️") + return True + lines.append(f"**{name} 商品** {g.get('goods', 0):,} 行 · 无缺漏 ✅") + return not ok + + def build_daily_message( days: int, window: str, riin: dict, sds: dict, - goods: dict, ocr: dict, failures: list[dict], + goods: dict, ocr_today: dict, ocr_pending: int, failures: list[dict], ) -> tuple[str, bool]: - """拼装日报 markdown,返回 (消息, 是否需关注)。结论式极简排版,不暴露内部术语。""" + """拼装日报 markdown,返回 (消息, 是否需关注)。""" lines = [f"📊 **数据核验日报** {datetime.now().strftime('%m-%d %H:%M')}" f"（近 {days} 天，{window}）", ""] attention = False - diff_names: list[str] = [] for name, report in (("RIIN", riin), ("SDS", sds)): try: - days_r = (report.get("verifyReport") or {}).get("days") or [] - summary = (report.get("verifyReport") or {}).get("summary") or {} - diff_days = [d for d in days_r if not d.get("ok")] - total = summary.get("totalDays", len(days_r)) - if not diff_days: - lines.append(f"**{name}** {total}/{total} 天与远程一致 ✅") - else: - diff_names.append(name) - attention = True - lines.append(f"**{name}** {total - len(diff_days)}/{total} 天与远程一致 ⚠️") - if len(diff_days) <= 4: - per_day = " · ".join( - f"{_short_date(d['date'])} {_affected_units(d.get('diff'))}单" for d in diff_days) - lines.append(f"> {per_day}状态不符，已自动重新同步") - else: - total_affected = sum(_affected_units(d.get("diff")) for d in diff_days) - lines.append(f"> {len(diff_days)} 天共约 {total_affected:,} 单状态不符，已自动重新同步") + attention |= _system_lines(name, report, (goods or {}).get(name), lines) except Exception as err: # noqa: BLE001 lines.append(f"**{name}** 核验结果解析失败：{err}") attention = True + lines.append("") + + processed, failed = (ocr_today or {}).get("processed", 0), (ocr_today or {}).get("failed", 0) + lines.append(f"**OCR 今日** 已处理 {processed:,} 单 · 识别失败 {failed:,} 单" + (" ✅" if not failed else "")) + if ocr_pending: + lines.append(f"> 累计待处理 {ocr_pending:,} 单") + lines.append("") - zero_total = sum(g["zeroGoodsOrders"] for g in goods.values()) - lines += ["", f"**商品** RIIN {goods['RIIN']['orders']:,} 单/{goods['RIIN']['goods']:,} 行" - f" · SDS {goods['SDS']['orders']:,} 单/{goods['SDS']['goods']:,} 行" - + (" ⚠️" if zero_total else " ✅")] - if zero_total: - lines.append(f"> 有 {zero_total} 单缺商品数据") - attention = True - lines.append(f"**OCR** 已完成 {ocr.get('done', 0):,} · 待处理 {ocr.get('pending', 0):,}" - f" · 失败 {ocr.get('error', 0):,}") if failures: attention = True - names = "、".join(f"{f['name']}（{f['state']}）" for f in failures[:4]) - lines.append(f"**失败任务(24h)** {len(failures)} 个：{names}") + lines.append(f"**失败任务(24h)** 共 {len(failures)} 次 🔴") + for f in failures[:6]: + start = (f.get("start") or "")[5:] # MM-DD HH:MM + lines.append(f"> {f.get('flow', '?')} · run「{f.get('name', '?')}」{f.get('state', '?')} · {start}") + if len(failures) > 6: + lines.append(f"> …等共 {len(failures)} 次") else: - lines.append("**失败任务(24h)** 0 个 ✅") + lines.append("**失败任务(24h)** 0 次 ✅") lines.append("") if attention: lines.append("⚠️ **存在需关注项，见上**") - if diff_names: - lines.append("> 每天都出现的相同差异属远程统计口径问题，非数据丢失") else: lines.append("✅ **数据全部一致，无异常**") return "\n".join(lines), attention @@ -102,11 +111,13 @@ def daily_report(days: int = 7) -> dict: db = SyncDB() goods = {sys: db.get_goods_integrity(sys, window_start, window_end) for sys in ("RIIN", "SDS")} + ocr_today = db.count_ocr_today(today) ocr = db.count_ocr_status() failures = check_prefect_failures(hours=24) window = f"{window_start:%m-%d}~{window_end:%m-%d}" - content, attention = build_daily_message(days, window, riin, sds, goods, ocr, failures) + content, attention = build_daily_message(days, window, riin, sds, goods, + ocr_today, ocr.get("pending", 0), failures) sent = send_markdown(content) emit_sync_summary("数据核验日报", [ @@ -116,11 +127,12 @@ def daily_report(days: int = 7) -> dict: ("SDS 零商品订单", goods["SDS"]["zeroGoodsOrders"]), ("SDS 商品无名称", goods["SDS"]["goodsNoName"]), ("全库孤儿商品", goods["RIIN"]["orphanGoods"]), + ("OCR 今日处理", ocr_today.get("processed", 0)), ("OCR 今日失败", ocr_today.get("failed", 0)), ("OCR 待处理", ocr.get("pending", 0)), ("失败 runs(24h)", len(failures)), ("已发企微", "是" if sent else "否"), ]) return {"attention": attention, "sent": sent, - "goods": goods, "ocr": ocr, "failures24h": len(failures)} + "goods": goods, "ocrToday": ocr_today, "failures24h": len(failures)} if __name__ == "__main__": diff --git a/flows/monitor/device_check.py b/flows/monitor/device_check.py index f4afd21..afe2a6b 100644 --- a/flows/monitor/device_check.py +++ b/flows/monitor/device_check.py @@ -1,19 +1,22 @@ -"""设备/系统健康日报(每日一次):内存/磁盘/GPU 使用率 + 系统级错误检查。 +"""设备/系统健康日报(每日 09:00):内存/swap/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 连通性、系统负载 +对应 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%)标 🔴。 -依赖 docker-compose 给 prefect-worker 挂载 /:/host:ro、/var/run/docker.sock、GPU; -未挂载时对应检查项自动降级为"不可用",不影响其余检查。 +依赖 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 @@ -24,6 +27,8 @@ 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" @@ -32,6 +37,115 @@ 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() @@ -131,6 +245,7 @@ def _exited_recently(finished_at: str, hours: int = 24) -> bool: 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( @@ -143,10 +258,18 @@ def check_prefect_failures(hours: int = 24) -> list[dict]: 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 resp.json(): + for r in runs: start = (r.get("start_time") or "")[:16].replace("T", " ") - out.append({"name": r.get("name", "?"), "state": (r.get("state") or {}).get("type", "?"), + 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 @@ -176,48 +299,124 @@ def _mark(pct: float | None, threshold: int) -> str: 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 [ - ("memory", _check_memory), ("disks", _check_disks), ("gpus", _check_gpus), - ("docker", _check_docker), ("prefectFailures", check_prefect_failures), ("services", _check_services), - ]: + 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 - lines = [f"🖥 **设备健康日报** {datetime.now().strftime('%m-%d %H:%M')}", ""] + 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 + "）", ""] - mem = checks.get("memory") or {} + # ---- 内存 / swap ---- + mem, swap = agg.get("mem"), agg.get("swap") if mem: - lines.append(f"**内存** {_mark(mem['pct'], threshold)}（{mem['used']}/{mem['total']}）" - f"｜swap {_mark(mem['swapPct'], threshold)}（{mem['swapUsed']}/{mem['swapTotal']}）") + lines.append(f"**内存** 平均 {mem['avg']:.0f}% · 峰值 {_mark(mem['peak'], threshold)} · 最低 {mem['min']:.0f}%" + + (f"（共 {_gb((agg.get('memTotal') or 0) * (1 << 30))}）" if agg.get("memTotal") else "")) else: - lines.append("**内存** 检查失败") + 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: + lines.append(f"**交换内存** 平均 {swap['avg']:.0f}% · 峰值 {_mark(swap['peak'], threshold)} · 最低 {swap['min']:.0f}%" + + (f"（共 {_gb((agg.get('swapTotal') or 0) * (1 << 30))}）" if agg.get("swapTotal") else "")) + else: + lines.append("**交换内存** 无采样数据") + swap = None - disks = checks.get("disks") or [] - if disks: - lines.append("**磁盘**") - for d in disks: - if "error" in d: - lines.append(f"> {d['label']} 检查失败：{d['error']}") + # ---- 磁盘:总体 / 根目录 / home ---- + lines.append("**磁盘**") + if overall: + lines.append(f"> 总体 已用 {_gb(overall['used'])} · 剩余 {_gb(overall['free'])}（{_mark(overall['pct'], threshold)}）") + label_map = {"/": "root", "/home": "home"} + for m in mounts: + g = growth.get(label_map.get(m["label"], "")) + growth_txt = "" + if g and g.get("dir"): + growth_txt = f"｜增长最快 {g['dir']} {_fmt_growth(g['growth'])}" + elif m["label"] in ("/", "/home"): + growth_txt = "｜增长数据不足" if not g else "" + lines.append(f"> {m['label']} 已用 {_gb(m['used'])} · 剩余 {_gb(m['free'])}" + f"（{_mark(m['pct'], threshold)}）{growth_txt}") + if not mounts: + lines.append("> 磁盘检查不可用") + + # ---- GPU ---- + gpu_stats = agg.get("gpus") or [] + if gpu_stats: + for g in gpu_stats: + name = f"{g['name']}（GPU{g['index']}）" if g.get("name") else 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 = g.get("memPeakUsed") + mem_total = g.get("memTotal") + if mem_peak is not None and mem_total: + mem_pct = mem_peak / mem_total * 100 + mem_txt = f"显存峰值 {_gb(mem_peak * (1 << 30))}/{_gb(mem_total * (1 << 30))}（{_mark(mem_pct, threshold)}）" + else: + mem_txt = "显存 无数据" + temp_peak = g.get("tempPeak") + if temp_peak is not None: + temp_txt = f"温度峰值 {temp_peak:.0f}°C" + ("🔴 过热" if temp_peak >= GPU_TEMP_WARN else " 正常 ✅") + else: + temp_txt = "温度 无数据" + 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']}" + f"（无采样数据）") 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 '未检测到'}") + lines.append(f"**GPU** {gpus_now if isinstance(gpus_now, str) else '未检测到'}") + # ---- Docker / 失败任务 / 探活 ---- docker = checks.get("docker") or {} issues = docker.get("issues") or [] if docker.get("running") is None: @@ -226,27 +425,31 @@ def device_check(alarm_threshold: int = 80) -> dict: 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" + docker_line += f" · 异常 **{len(issues)}**🔴" if issues else " · 异常 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 "")) + lines.append(f"**失败任务(24h)** {len(failures)} 次" + (" ✅" if not failures else "🔴")) for f in failures[:5]: - lines.append(f"> 🔴 {f['name']}（{f['state']}）{f['start']}") + lines.append(f"> 🔴 {f['flow']} ·「{f['name']}」{f['state']} · {f['start']}") + if len(failures) > 5: + lines.append(f"> …共 {len(failures)} 次") 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}") + if svc: + lines.append(f"**探活** OCR {'✅' if svc.get('ocr') else '🔴'}｜PG {'✅' if svc.get('pg') else '🔴'}" + f"｜负载 {os.getloadavg()[0]:.1f}") + mem_peak_for_alarm = mem.get("peak") if mem else None + swap_peak_for_alarm = swap.get("peak") if swap else None 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)) + (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 += ["", ("🔴 **有项目需要处理，见上**" if alarms else "✅ **一切正常**")] @@ -254,13 +457,15 @@ def device_check(alarm_threshold: int = 80) -> dict: content = "\n".join(lines) sent = send_markdown(content) emit_sync_summary("设备健康日报", [ - ("内存", f"{_mark(mem['pct'], threshold)}" if mem else "—"), + ("采样次数", agg["count"]), + ("内存峰值", f"{mem_peak_for_alarm:.0f}%" if mem_peak_for_alarm is not None else "—"), ("Docker 异常", len(issues)), - ("Prefect 失败(24h)", len(failures)), + ("失败任务(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)} + return {"alarms": bool(alarms), "sent": sent, "samples": agg["count"], + "dockerIssues": len(issues), "prefectFailures": len(failures)} if __name__ == "__main__": diff --git a/flows/riin/verify.py b/flows/riin/verify.py index e3e9f7c..ec035b8 100644 --- a/flows/riin/verify.py +++ b/flows/riin/verify.py @@ -105,21 +105,39 @@ def riin_verify(mode: str = "recent", recent_days: int = 1, date_range: str | No {"time_type": 3, "start_time": format_date(day_start), "end_time": format_date(day_end), "page_size": 200}, ) stats = sync_riin_records(raw_orders, client, db) - for k in ("ordersCreated", "ordersUpdated", "goodsCreated", "goodsUpdated"): - pass totals["ordersAdded"] += stats["ordersCreated"] totals["ordersUpdated"] += stats["ordersUpdated"] totals["goodsAdded"] += stats["goodsCreated"] totals["goodsUpdated"] += stats["goodsUpdated"] day_reports.append({"date": day_str, "remote": remote_total, "local": sum(local_counts.values()), "diff": diff, "ok": False, "orders": stats["orderCount"], "goods": stats["goodsCount"]}) + + # 修复后重数:数量 API + 本地 SQL 很便宜,给出修复后的数量一致性结论(口径与 _diff 相同) + post = None + if days_with_diff: + post_days, post_diff_days = [], 0 + for day in days: + day_start, day_end = start_of_day(day), start_of_day(day) + timedelta(days=1) + count_resp = client.get_order_count( + OrderPageRequest(time_type=3, start_time=format_date(day_start), end_time=format_date(day_end)) + ) + remote_map = _count_to_status_map(count_resp.data.model_dump(by_alias=False)) + local_counts = db.get_order_count_by_status("RIIN", day_start, day_end) + diff, has_diff = _diff(remote_map, local_counts) + post_diff_days += int(has_diff) + post_days.append({"date": format_date(day).split(" ")[0], "remote": sum(remote_map.values()), + "local": sum(local_counts.values()), "ok": not has_diff}) + post = {"days": post_days, "daysWithDiff": post_diff_days, + "remote": sum(d["remote"] for d in post_days), "local": sum(d["local"] for d in post_days)} + print(f"[核验重数] 修复后 远程 {post['remote']} 本地 {post['local']} / 有差异 {post_diff_days} 天") finally: client.close() print(f"[核验汇总] 共 {len(days)} 天 / 有差异 {days_with_diff} 天 / {totals}") emit_verify_report("RIIN 数据核验", day_reports, totals, len(days), days_with_diff) return { - "verifyReport": {"days": day_reports, "summary": {"totalDays": len(days), "daysWithDiff": days_with_diff, **totals}}, + "verifyReport": {"days": day_reports, "post": post, + "summary": {"totalDays": len(days), "daysWithDiff": days_with_diff, **totals}}, "syncRange": {"from": format_date(start_of_day(days[0])) if days else "", "to": format_date(start_of_day(days[-1])) if days else ""}, } diff --git a/flows/sds/verify.py b/flows/sds/verify.py index 693ab9b..b11ac3a 100644 --- a/flows/sds/verify.py +++ b/flows/sds/verify.py @@ -48,6 +48,14 @@ def _diff(remote_stats: list, local: dict[str, int]) -> tuple[dict[str, int], bo return diff, has_diff +def _remote_map(remote_stats: list) -> dict[str, int]: + """远程状态→数量映射(排除"售后"和无名状态,与本地口径对称)。""" + out = {getattr(s, "name", None): (getattr(s, "orderNum", 0) or 0) for s in remote_stats} + out.pop("售后", None) + out.pop(None, None) + return out + + @flow(name="sds-verify", log_prints=True) def sds_verify(mode: str = "recent", recent_days: int = 1, date_range: str | None = None) -> dict: days = _parse_days(mode, recent_days, date_range) @@ -68,7 +76,8 @@ def sds_verify(mode: str = "recent", recent_days: int = 1, date_range: str | Non remote_stats = client.get_status( OrderQueryParams(pay_date_begin=format_date(day_start), pay_date_end=format_date(day_end)) ) - remote_total = sum(getattr(s, "orderNum", 0) or 0 for s in remote_stats) + remote_map = _remote_map(remote_stats) + remote_total = sum(remote_map.values()) local_counts = db.get_order_count_by_status("SDS", day_start, day_end) diff, has_diff = _diff(remote_stats, local_counts) @@ -91,13 +100,37 @@ def sds_verify(mode: str = "recent", recent_days: int = 1, date_range: str | Non totals["goodsUpdated"] += stats["goodsUpdated"] day_reports.append({"date": day_str, "remote": remote_total, "local": sum(local_counts.values()), "diff": diff, "ok": False, "orders": stats["orderCount"], "goods": stats["goodsCount"]}) + + # 修复后重数:数量接口 + 本地 SQL 很便宜,给出修复后的数量一致性结论(口径与 _diff 相同,排除"售后") + post = None + if days_with_diff: + post_days, post_diff_days = [], 0 + for day in days: + day_start, day_end = start_of_day(day), start_of_day(day) + timedelta(days=1) + remote_stats = client.get_status( + OrderQueryParams(pay_date_begin=format_date(day_start), pay_date_end=format_date(day_end)) + ) + remote_map = _remote_map(remote_stats) + local_counts = db.get_order_count_by_status("SDS", day_start, day_end) + statuses = set(remote_map) | set(local_counts) + statuses.discard("售后") + has_diff = any(remote_map.get(st, 0) != local_counts.get(st, 0) for st in statuses) + post_diff_days += int(has_diff) + post_days.append({"date": format_date(day).split(" ")[0], + "remote": sum(remote_map.values()), + "local": sum(v for k, v in local_counts.items() if k != "售后"), + "ok": not has_diff}) + post = {"days": post_days, "daysWithDiff": post_diff_days, + "remote": sum(d["remote"] for d in post_days), "local": sum(d["local"] for d in post_days)} + print(f"[核验重数] 修复后 远程 {post['remote']} 本地 {post['local']} / 有差异 {post_diff_days} 天") finally: client.close() print(f"[核验汇总] 共 {len(days)} 天 / 有差异 {days_with_diff} 天 / {totals}") emit_verify_report("SDS 数据核验", day_reports, totals, len(days), days_with_diff) return { - "verifyReport": {"days": day_reports, "summary": {"totalDays": len(days), "daysWithDiff": days_with_diff, **totals}}, + "verifyReport": {"days": day_reports, "post": post, + "summary": {"totalDays": len(days), "daysWithDiff": days_with_diff, **totals}}, "syncRange": {"from": format_date(start_of_day(days[0])) if days else "", "to": format_date(start_of_day(days[-1])) if days else ""}, } diff --git a/flows/shared/db.py b/flows/shared/db.py index 060dcf5..02c1e00 100644 --- a/flows/shared/db.py +++ b/flows/shared/db.py @@ -200,9 +200,10 @@ class SyncDB: def update_ocr_fields( self, order_no: str, created_at: datetime, fields: dict[str, str | None] ) -> None: - """更新单个订单的 ocr_* 字段。fields 为 {ocr_xxx: value} 字典。""" + """更新单个订单的 ocr_* 字段。fields 为 {ocr_xxx: value} 字典。自动盖处理时间戳。""" if not fields: return + fields = {"ocrProcessedAt": datetime.now(), **fields} assignments = ", ".join(f'"{k}"=%s' for k in fields) params = list(fields.values()) + [order_no, created_at] with self._conn() as conn: @@ -228,6 +229,7 @@ class SyncDB: for order_no, created_at, fields in updates: if not fields: continue + fields = {"ocrProcessedAt": datetime.now(), **fields} assignments = ", ".join(f'"{k}"=%s' for k in fields) params = list(fields.values()) + [order_no, created_at] cur.execute( @@ -258,6 +260,21 @@ class SyncDB: row = cur.fetchone() return {"done": row[0], "error": row[1], "unsupported": row[2], "pending": row[3]} if row else {} + def count_ocr_today(self, day_start: datetime) -> dict[str, int]: + """今日 OCR 处理量(按 ocrProcessedAt 盖章时间):processed=成功识别, failed=识别失败/不支持。""" + with self._conn() as conn: + with conn.cursor() as cur: + cur.execute( + '''SELECT + COUNT(*) FILTER (WHERE "ocrLayout" IS NOT NULL + AND "ocrLayout" NOT IN ('ERROR','UNSUPPORTED')) AS processed, + COUNT(*) FILTER (WHERE "ocrLayout" IN ('ERROR','UNSUPPORTED')) AS failed + FROM "Orders" WHERE "ocrProcessedAt">=%s''', + (day_start,), + ) + row = cur.fetchone() + return {"processed": row[0], "failed": row[1]} if row else {"processed": 0, "failed": 0} + # ---------- 主 upsert ---------- def sync_orders_and_goods( diff --git a/scripts/metrics_agent.py b/scripts/metrics_agent.py new file mode 100644 index 0000000..eff6ba8 --- /dev/null +++ b/scripts/metrics_agent.py @@ -0,0 +1,175 @@ +"""metrics-agent:宿主机指标采样侧车(独立于 Prefect,常驻循环,compose 服务运行)。 + +每 5 分钟追加一行 JSONL:data/metrics/metrics-YYYY-MM-DD.jsonl + {"ts","memPct","memUsed","memTotal","swapPct","swapUsed","swapTotal", + "gpus":[{"index","name","utilPct","memPct","memUsed","memTotal","tempC"}]} +每日 7 点后做一次 du 快照:data/metrics/disk-YYYYMMDD-HHMM.json + (根目录 / home 各自的总量 + 顶层目录字节数,device-check 对比最近两份算"增长最快目录")。 +保留:采样 14 天、快照 7 份。 + +compose 中挂载 /:/host:ro(du 扫描)、GPU(nvidia-smi)、./data/metrics(writable)。 +调试单次采样:uv run python scripts/metrics_agent.py --once +""" + +from __future__ import annotations + +import argparse +import json +import os +import subprocess +import threading +import time +from datetime import datetime +from pathlib import Path + +INTERVAL = int(os.environ.get("METRICS_INTERVAL", "300")) +DATA_DIR = Path(os.environ.get("METRICS_DIR", "data/metrics")) +HOST_ROOT = os.environ.get("METRICS_HOST_ROOT", "/host") +SNAPSHOT_AFTER_HOUR = 7 +KEEP_SAMPLE_DAYS = 14 +KEEP_SNAPSHOTS = 7 +# 虚拟文件系统与不关心的挂载点,du 快照跳过 +DU_EXCLUDE = {"proc", "sys", "dev", "run", "lost+found"} + + +def _gb(nbytes: float) -> float: + return round(nbytes / (1 << 30), 2) + + +def sample() -> dict: + import psutil + + vm, sw = psutil.virtual_memory(), psutil.swap_memory() + return { + "ts": datetime.now().isoformat(timespec="seconds"), + "memPct": vm.percent, "memUsed": _gb(vm.used), "memTotal": _gb(vm.total), + "swapPct": sw.percent, "swapUsed": _gb(sw.used), "swapTotal": _gb(sw.total), + "gpus": _gpus(), + } + + +def _gpus() -> list[dict]: + try: + proc = subprocess.run( + ["nvidia-smi", "--query-gpu=index,name,utilization.gpu,memory.used,memory.total,temperature.gpu", + "--format=csv,noheader,nounits"], + capture_output=True, text=True, timeout=20, + ) + if proc.returncode != 0: + return [] + out = [] + for line in proc.stdout.strip().splitlines(): + p = [x.strip() for x in line.split(",")] + if len(p) != 6: + continue + idx, name, util, mem_used, mem_total, temp = p + mt = float(mem_total) + out.append({ + "index": int(idx), "name": name[:24], "utilPct": float(util), + "memPct": round(float(mem_used) / mt * 100, 1) if mt > 0 else 0.0, + "memUsed": _gb(float(mem_used) * (1 << 20)), "memTotal": _gb(mt * (1 << 20)), + "tempC": float(temp), + }) + return out + except Exception: # noqa: BLE001 + return [] + + +def write_sample(rec: dict) -> None: + DATA_DIR.mkdir(parents=True, exist_ok=True) + path = DATA_DIR / f"metrics-{rec['ts'][:10]}.jsonl" + with path.open("a", encoding="utf-8") as f: + f.write(json.dumps(rec, ensure_ascii=False) + "\n") + + +def _du_bytes(path: str) -> int: + # 实际占块(非表面大小,避免 docker 稀疏文件虚增);rc=1 时是个别文件消失,stdout 仍有效 + proc = subprocess.run(["du", "-sB1", path], capture_output=True, text=True, timeout=7200) + if not proc.stdout: + return -1 + try: + return int(proc.stdout.split()[0]) + except (IndexError, ValueError): + return -1 + + +def _usage_entry(base: str) -> dict: + import shutil + + st = shutil.disk_usage(base) + return {"total": st.total, "used": st.used, "free": st.free} + + +def disk_snapshot() -> Path: + """du 扫描根目录/home 顶层目录,写 disk-YYYYMMDD-HHMM.json(原子写)。""" + now = datetime.now() + out: dict = {"ts": now.isoformat(timespec="seconds")} + for key, base in (("root", HOST_ROOT), ("home", f"{HOST_ROOT}/home")): + if not os.path.isdir(base): + continue + entry = _usage_entry(base) + dirs: dict[str, int] = {} + for name in sorted(os.listdir(base)): + if name in DU_EXCLUDE: + continue + p = os.path.join(base, name) + if os.path.isdir(p) and not os.path.islink(p) and not os.path.ismount(p): + size = _du_bytes(p) + if size >= 0: + dirs[name] = size + entry["dirs"] = dirs + out[key] = entry + DATA_DIR.mkdir(parents=True, exist_ok=True) + path = DATA_DIR / f"disk-{now:%Y%m%d-%H%M}.json" + tmp = path.with_suffix(".tmp") + tmp.write_text(json.dumps(out, ensure_ascii=False), encoding="utf-8") + tmp.rename(path) + print(f"[metrics-agent] 磁盘快照完成: {path.name}", flush=True) + return path + + +def snapshot_today_exists() -> bool: + today = datetime.now().strftime("%Y%m%d") + return any(DATA_DIR.glob(f"disk-{today}-*.json")) + + +def prune() -> None: + cutoff = time.time() - KEEP_SAMPLE_DAYS * 86400 + for p in DATA_DIR.glob("metrics-*.jsonl"): + if p.stat().st_mtime < cutoff: + p.unlink(missing_ok=True) + snaps = sorted(DATA_DIR.glob("disk-*.json")) + for p in snaps[:-KEEP_SNAPSHOTS] if len(snaps) > KEEP_SNAPSHOTS else []: + p.unlink(missing_ok=True) + + +def _maybe_snapshot_async() -> None: + """7 点后补当日快照;du 可能耗时数分钟,放后台线程不阻塞采样。""" + if datetime.now().hour < SNAPSHOT_AFTER_HOUR or snapshot_today_exists(): + return + threading.Thread(target=lambda: _safe(disk_snapshot), daemon=True).start() + + +def _safe(fn, *args): + try: + fn(*args) + except Exception as err: # noqa: BLE001 + print(f"[metrics-agent] {type(err).__name__}: {err}", flush=True) + + +def main() -> None: + ap = argparse.ArgumentParser(description="宿主机指标采样侧车") + ap.add_argument("--once", action="store_true", help="只采样一次即退出(调试)") + args = ap.parse_args() + print(f"[metrics-agent] 启动: interval={INTERVAL}s dir={DATA_DIR} host_root={HOST_ROOT}", flush=True) + while True: + _safe(lambda: write_sample(sample())) + _maybe_snapshot_async() + _safe(prune) + if args.once: + break + time.sleep(INTERVAL) + + +if __name__ == "__main__": + main()