317 lines
16 KiB
Python
317 lines
16 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""全队健康度监控: 汇总所有机器最新一小时的处理摘要, 跑异常检测, 产出报告。
|
||
配合 run_worker.py 里的 job_cos.run_hourly_digest_loop() 后台线程(每台机器每小时自动上传
|
||
一份小摘要到 COS)使用 —— 本脚本负责把这些小摘要拉回来汇总判断。
|
||
|
||
用法(项目根目录运行, 建议 crontab 每小时跑一次，比如 5分/10分 触发一次都行，脚本自己会
|
||
算准要看哪个整点):
|
||
python tools/fleet_monitor.py # 单小时报告 + 异常检测(原有行为)
|
||
python tools/fleet_monitor.py --history 24 # 全队历史趋势(最近24个小时, 不跑异常检测)
|
||
python tools/fleet_monitor.py --history # 不带数字 = 默认 24 小时
|
||
|
||
设计文档: docs/superpowers/specs/2026-07-08-fleet-monitoring-design.md
|
||
不接外部通知渠道(企业微信/钉钉/飞书webhook 是后续可插拔扩展点, 本次只落地本地+COS报告文件)。
|
||
"""
|
||
import os
|
||
import sys
|
||
import unicodedata
|
||
from datetime import datetime, timedelta
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
from tool import job_cos # noqa: E402
|
||
|
||
# ---- 异常检测阈值(先写死成常量, 要调直接改这里) ----
|
||
OFFLINE_HOURS = 2 # 机器掉线: 最新摘要距目标小时 >= 这么多小时
|
||
SUCCESS_RATE_MIN = 0.5 # 单小时成功率骤降阈值
|
||
SUCCESS_RATE_MIN_SAMPLE = 10 # 样本太少(成功+错误 < 这个数)不判断成功率, 避免噪音
|
||
MACHINE_DROP_RATIO = 0.3 # 单机吞吐骤降: 本小时产出 < 自己24h均值的这个比例
|
||
FLEET_DROP_RATIO = 0.5 # 全队吞吐骤降: 全队本小时产出 < 全队24h均值的这个比例
|
||
BASELINE_HOURS = 24 # 基线取多少小时的历史均值(不含目标小时本身)
|
||
KNOWN_MACHINES_LOOKBACK = 48 # "这台机器算不算存在"往回看多少小时
|
||
|
||
REPORT_DIR = "./monitor_reports"
|
||
|
||
DEFAULT_HISTORY_HOURS = 24 # --history 不带数字时默认看多少小时
|
||
|
||
|
||
def _dw(s):
|
||
"""字符串显示宽度(中文/全角算2), 用于历史趋势表格对齐。"""
|
||
return sum(2 if unicodedata.east_asian_width(c) in ("W", "F") else 1 for c in str(s))
|
||
|
||
|
||
def _padl(s, w):
|
||
"""按显示宽度左对齐(文字在前, 空格补在后面) —— 跟 hourly_stats.py 的同名函数保持一致约定。"""
|
||
s = str(s)
|
||
return s + " " * max(0, w - _dw(s))
|
||
|
||
|
||
def _padr(s, w):
|
||
"""按显示宽度右对齐(空格补在前面, 文字贴右边)。"""
|
||
return " " * max(0, w - _dw(s)) + str(s)
|
||
|
||
|
||
def target_hour(now=None) -> datetime:
|
||
"""确定要汇总的目标整点小时: 最新一个已经过完的整点; 如果当前分钟数<5(刚跨过整点没多久,
|
||
部分机器可能还没上传完), 再往回退一个小时留缓冲。"""
|
||
now = now or datetime.now()
|
||
completed = now.replace(minute=0, second=0, microsecond=0) - timedelta(hours=1)
|
||
if now.minute < 5:
|
||
completed -= timedelta(hours=1)
|
||
return completed
|
||
|
||
|
||
def fetch_range_digests(end_hour_dt: datetime, hours_back: int) -> dict:
|
||
"""拉取 [end_hour_dt - hours_back, end_hour_dt] 闭区间(按小时)所有机器的摘要,
|
||
返回 {hour_key(YYYYMMDDHH): {machine_id: digest_dict} | None}。
|
||
区分两种"没数据": 该小时【列COS目录本身失败】(比如COS暂时不可达)记为 None——这是需要
|
||
main() 判断是否整体报错退出的信号; 该小时【列成功但目录是空的】(单纯没人上报)记为 {}——
|
||
这是正常情况, 不代表故障。单台机器的摘要【下载】失败只跳过该机器、打印警告, 不影响这个
|
||
小时其它机器的数据, 也不算列目录失败。"""
|
||
result = {}
|
||
for i in range(hours_back + 1):
|
||
hdt = end_hour_dt - timedelta(hours=i)
|
||
hkey = hdt.strftime("%Y%m%d%H")
|
||
try:
|
||
files = job_cos.list_hourly_digest_files_for_hour(hdt)
|
||
except Exception as e:
|
||
print(f"[WARN] 列出 {hkey} 的摘要文件失败: {e}")
|
||
result[hkey] = None
|
||
continue
|
||
machines = {}
|
||
for machine_id, key in files:
|
||
try:
|
||
machines[machine_id] = job_cos.download_hourly_digest(key)
|
||
except Exception as e:
|
||
print(f"[WARN] 下载 {machine_id} 在 {hkey} 的摘要失败, 跳过: {e}")
|
||
result[hkey] = machines
|
||
return result
|
||
|
||
|
||
def compute_baseline(range_digests: dict, target_hdt: datetime) -> dict:
|
||
"""算每台机器过去 BASELINE_HOURS 小时(不含目标小时本身)的 success_hours 均值。
|
||
只用真实拉到的小时数据平均(缺的小时/列目录失败的小时都不计入, 不当0算, 避免拖低均值),
|
||
没有任何数据的机器不出现在结果里。"""
|
||
sums = {}
|
||
counts = {}
|
||
for i in range(1, BASELINE_HOURS + 1):
|
||
hkey = (target_hdt - timedelta(hours=i)).strftime("%Y%m%d%H")
|
||
for machine_id, digest in (range_digests.get(hkey) or {}).items():
|
||
sums[machine_id] = sums.get(machine_id, 0.0) + digest.get("success_hours", 0.0)
|
||
counts[machine_id] = counts.get(machine_id, 0) + 1
|
||
return {m: sums[m] / counts[m] for m in sums}
|
||
|
||
|
||
def known_machines_and_latest_hour(range_digests: dict) -> dict:
|
||
"""扫全部拉到的小时, 返回 {machine_id: 该机器出现过的最新 hour_key}。列目录失败(None)的
|
||
小时直接跳过, 不当"没有任何机器"处理(避免把偶发的单小时COS故障误判成全队一起掉线)。"""
|
||
latest = {}
|
||
for hkey, machines in range_digests.items():
|
||
if machines is None:
|
||
continue
|
||
for machine_id in machines:
|
||
if machine_id not in latest or hkey > latest[machine_id]:
|
||
latest[machine_id] = hkey
|
||
return latest
|
||
|
||
|
||
def aggregate_history(range_digests: dict, hours_back: int, end_hdt: datetime) -> list:
|
||
"""把 fetch_range_digests() 拉到的多小时摘要,按小时汇总成全队合计一行,按时间正序(从早到晚)
|
||
返回 [(hour_key, machine_count, count, success, no_subtitle, error, success_hours, ok), ...]。
|
||
ok=False 表示这个小时列COS目录本身失败(range_digests[hkey] is None), 其余字段填0/0.0;
|
||
ok=True 且 machine_count=0 是"列目录成功但没人上报"的正常空窗口(比如还没到那个整点),
|
||
跟 ok=False(拉取本身失败)是两码事,不能混在一起看成同一种"没数据"。"""
|
||
rows = []
|
||
for i in range(hours_back, -1, -1):
|
||
hdt = end_hdt - timedelta(hours=i)
|
||
hkey = hdt.strftime("%Y%m%d%H")
|
||
machines = range_digests.get(hkey)
|
||
if machines is None:
|
||
rows.append((hkey, 0, 0, 0, 0, 0, 0.0, False))
|
||
continue
|
||
count = sum(d.get("count", 0) for d in machines.values())
|
||
success = sum(d.get("success", 0) for d in machines.values())
|
||
no_sub = sum(d.get("no_subtitle", 0) for d in machines.values())
|
||
error = sum(d.get("error", 0) for d in machines.values())
|
||
succ_hours = sum(d.get("success_hours", 0.0) for d in machines.values())
|
||
rows.append((hkey, len(machines), count, success, no_sub, error, succ_hours, True))
|
||
return rows
|
||
|
||
|
||
def print_history(hours_back: int):
|
||
"""打印全队历史多个小时的汇总趋势(拉COS各机器已上传的小时摘要, 不依赖本机 数据.jsonl,
|
||
只要各机器在跑带小时摘要上传功能的新版 run_worker.py 就能看到)。跟 tools/hourly_stats.py
|
||
的单机版本是同一个"按process_start所在小时分桶计数"的逻辑, 只是数据源从本机 数据.jsonl
|
||
换成了COS上各机器上传的小时摘要, 且是全队加总, 不是单机。"""
|
||
thdt = target_hour()
|
||
print(f"目标小时: {thdt.strftime('%Y-%m-%d %H:00')} (往前拉 {hours_back} 小时, 共 {hours_back+1} 个小时)")
|
||
range_digests = fetch_range_digests(thdt, hours_back)
|
||
rows = aggregate_history(range_digests, hours_back, thdt)
|
||
|
||
# 每列宽度都比内容本身多留至少2个显示宽度的间隔, 避免"小时"这种定长字符串(如
|
||
# "2026-07-09 13:00"共16字符)紧贴着下一列数字、看起来像连在一起。
|
||
W = {"小时": 18, "机器数": 8, "视频数": 8, "成功": 6, "无字幕": 8, "错误": 6, "产出小时数": 12}
|
||
header = (_padl("小时", W["小时"]) + _padr("机器数", W["机器数"]) + _padr("视频数", W["视频数"]) +
|
||
_padr("成功", W["成功"]) + _padr("无字幕", W["无字幕"]) + _padr("错误", W["错误"]) +
|
||
_padr("产出小时数", W["产出小时数"]))
|
||
width = sum(W.values())
|
||
print("=" * width)
|
||
print(header)
|
||
print("-" * width)
|
||
|
||
tot = {"count": 0, "success": 0, "no_subtitle": 0, "error": 0, "succ_hours": 0.0}
|
||
fail_hours = []
|
||
for hkey, mcount, count, success, no_sub, error, succ_hours, ok in rows:
|
||
disp_hour = f"{hkey[:4]}-{hkey[4:6]}-{hkey[6:8]} {hkey[8:10]}:00"
|
||
if not ok:
|
||
print(_padl(disp_hour, W["小时"]) + _padr("(拉取失败)", width - W["小时"]))
|
||
fail_hours.append(hkey)
|
||
continue
|
||
print(_padl(disp_hour, W["小时"]) + _padr(mcount, W["机器数"]) + _padr(count, W["视频数"]) +
|
||
_padr(success, W["成功"]) + _padr(no_sub, W["无字幕"]) + _padr(error, W["错误"]) +
|
||
_padr(f"{succ_hours:.1f}", W["产出小时数"]))
|
||
tot["count"] += count
|
||
tot["success"] += success
|
||
tot["no_subtitle"] += no_sub
|
||
tot["error"] += error
|
||
tot["succ_hours"] += succ_hours
|
||
|
||
print("-" * width)
|
||
nh = len(rows) - len(fail_hours)
|
||
if nh > 0:
|
||
print(_padl("合计/时均", W["小时"]) + _padr(tot["count"], W["视频数"] + W["机器数"]) +
|
||
_padr(tot["success"], W["成功"]) + _padr(tot["no_subtitle"], W["无字幕"]) +
|
||
_padr(tot["error"], W["错误"]) + _padr(f"{tot['succ_hours']:.1f}", W["产出小时数"]))
|
||
print(f"\n共 {nh} 个有效小时 | 时均处理 {tot['count']/nh:.0f} 条 | "
|
||
f"时均产出 {tot['succ_hours']/nh:.1f} 小时视频内容")
|
||
if fail_hours:
|
||
print(f"\n({len(fail_hours)} 个小时COS拉取失败, 已跳过: {', '.join(fail_hours)})")
|
||
|
||
|
||
def detect_anomalies(target_hdt: datetime, target_digests: dict,
|
||
baseline: dict, latest_hour: dict) -> list:
|
||
"""跑4条异常检测规则, 返回 [(level, message), ...]，level ∈ {"严重","警告"}。"""
|
||
anomalies = []
|
||
|
||
# ① 机器掉线: 最新一次上报距目标小时 >= OFFLINE_HOURS
|
||
for machine_id, last_hkey in latest_hour.items():
|
||
last_hdt = datetime.strptime(last_hkey, "%Y%m%d%H")
|
||
gap_hours = (target_hdt - last_hdt).total_seconds() / 3600
|
||
if gap_hours >= OFFLINE_HOURS:
|
||
anomalies.append(("严重", f"机器 {machine_id} 已连续{gap_hours:.0f}小时无上报数据，可能停机"))
|
||
|
||
# ② 单小时成功率骤降
|
||
for machine_id, digest in target_digests.items():
|
||
succ, err = digest.get("success", 0), digest.get("error", 0)
|
||
sample = succ + err
|
||
if sample >= SUCCESS_RATE_MIN_SAMPLE:
|
||
rate = succ / sample
|
||
if rate < SUCCESS_RATE_MIN:
|
||
anomalies.append(("警告", f"机器 {machine_id} 本小时成功率仅{rate*100:.0f}%({succ}/{sample})"))
|
||
|
||
# ③ 单机吞吐骤降
|
||
for machine_id, digest in target_digests.items():
|
||
base = baseline.get(machine_id)
|
||
if base and base > 0:
|
||
cur = digest.get("success_hours", 0.0)
|
||
if cur < base * MACHINE_DROP_RATIO:
|
||
anomalies.append(("警告", f"机器 {machine_id} 本小时产出{cur:.1f}小时，远低于24h均值{base:.1f}小时"))
|
||
|
||
# ④ 全队吞吐骤降
|
||
fleet_cur = sum(d.get("success_hours", 0.0) for d in target_digests.values())
|
||
fleet_base = sum(baseline.values())
|
||
if fleet_base > 0 and fleet_cur < fleet_base * FLEET_DROP_RATIO:
|
||
anomalies.append(("严重", f"全队本小时总产出{fleet_cur:.1f}小时，远低于24h均值{fleet_base:.1f}小时，可能是全局性问题(比如OCR端点故障)"))
|
||
|
||
return anomalies
|
||
|
||
|
||
def render_report(target_hdt: datetime, target_digests: dict,
|
||
baseline: dict, anomalies: list) -> str:
|
||
"""生成 Markdown 报告文本。"""
|
||
lines = [f"# 全队监控报告 {target_hdt.strftime('%Y-%m-%d %H:00')}", ""]
|
||
lines.append("## 异常告警")
|
||
if anomalies:
|
||
for level, msg in anomalies:
|
||
icon = "🔴" if level == "严重" else "🟡"
|
||
lines.append(f"- {icon} [{level}] {msg}")
|
||
else:
|
||
lines.append("✅ 本小时一切正常")
|
||
lines.append("")
|
||
lines.append("## 本小时明细")
|
||
lines.append("| 机器 | 视频数 | 成功 | 无字幕 | 错误 | 产出小时数 |")
|
||
lines.append("|---|---|---|---|---|---|")
|
||
for machine_id in sorted(target_digests):
|
||
d = target_digests[machine_id]
|
||
lines.append(f"| {machine_id} | {d.get('count',0)} | {d.get('success',0)} | "
|
||
f"{d.get('no_subtitle',0)} | {d.get('error',0)} | {d.get('success_hours',0):.1f} |")
|
||
lines.append("")
|
||
lines.append("## 过去24小时全队趋势")
|
||
fleet_cur = sum(d.get("success_hours", 0.0) for d in target_digests.values())
|
||
fleet_base = sum(baseline.values())
|
||
lines.append(f"全队本小时产出小时数总和：{fleet_cur:.1f}，24小时均值：{fleet_base:.1f}")
|
||
return "\n".join(lines) + "\n"
|
||
|
||
|
||
def _parse_history_arg(argv):
|
||
"""解析 --history [N] —— 带上就返回要看多少小时(不带数字用默认值), 没带这个参数返回 None
|
||
(跑原来的单小时报告模式)。"""
|
||
if "--history" not in argv:
|
||
return None
|
||
i = argv.index("--history")
|
||
if i + 1 < len(argv):
|
||
try:
|
||
return int(argv[i + 1])
|
||
except ValueError:
|
||
pass
|
||
return DEFAULT_HISTORY_HOURS
|
||
|
||
|
||
def main():
|
||
hours_back = _parse_history_arg(sys.argv[1:])
|
||
if hours_back is not None:
|
||
print_history(hours_back)
|
||
return
|
||
|
||
thdt = target_hour()
|
||
print(f"目标小时: {thdt.strftime('%Y-%m-%d %H:00')}")
|
||
|
||
range_digests = fetch_range_digests(thdt, KNOWN_MACHINES_LOOKBACK)
|
||
target_hkey = thdt.strftime("%Y%m%d%H")
|
||
target_slot = range_digests.get(target_hkey)
|
||
if target_slot is None:
|
||
# 目标小时本身列目录都失败了(COS大概率整体不可达) -> 按设计文档"错误处理"要求直接报错退出,
|
||
# 不生成一份看起来正常、实际数据全空的假报告掩盖问题。
|
||
print(f"[FAIL] 目标小时 {target_hkey} 的摘要目录列取失败(COS可能不可达), 中止, 不生成报告")
|
||
sys.exit(1)
|
||
target_digests = target_slot
|
||
if not target_digests:
|
||
print(f"[WARN] 目标小时 {target_hkey} 没有拉到任何机器的摘要, 可能还没到点(正常空目录, 不是故障)")
|
||
|
||
baseline = compute_baseline(range_digests, thdt)
|
||
latest_hour = known_machines_and_latest_hour(range_digests)
|
||
anomalies = detect_anomalies(thdt, target_digests, baseline, latest_hour)
|
||
|
||
report = render_report(thdt, target_digests, baseline, anomalies)
|
||
|
||
os.makedirs(REPORT_DIR, exist_ok=True)
|
||
local_path = os.path.join(REPORT_DIR, f"{thdt.strftime('%Y%m%d_%H')}.md")
|
||
with open(local_path, "w", encoding="utf-8") as f:
|
||
f.write(report)
|
||
print(f"[OK] 报告已写入 {local_path}")
|
||
|
||
try:
|
||
dst = job_cos.upload_monitor_report(local_path, thdt)
|
||
if dst:
|
||
print(f"[OK] 报告已上传 -> {dst}")
|
||
except Exception as e:
|
||
print(f"[WARN] 报告上传COS失败(本地文件已保留): {e}")
|
||
|
||
if anomalies:
|
||
print(f"\n共检测到 {len(anomalies)} 条异常，详见报告。")
|
||
else:
|
||
print("\n本小时一切正常。")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|