452 lines
19 KiB
Python
452 lines
19 KiB
Python
"""sanguo-data-monitor — VPS 数据面监控班次(spec §20.15.3, 23:50 evening / 08:05 morning).
|
|
|
|
只读体检不修数据; 告警写 data/alerts.db(单写者), 前端/晨检消费 health JSON。
|
|
日志源=akshare_static_download 自身 FileHandler(data/static/logs/, utf-8 确定性)
|
|
——migration_logs 是 PS 重定向副本身受编码坑, 不作解析源。
|
|
退出码: 0=绿/黄, 3=有红(wrapper 恒转 0, 红态由 alerts/health 承载)。
|
|
"""
|
|
import argparse
|
|
import datetime as dt
|
|
import glob
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
_HERE = os.path.dirname(os.path.abspath(__file__))
|
|
_REPO = os.path.dirname(os.path.dirname(_HERE))
|
|
for _p in (_HERE, _REPO):
|
|
if _p not in sys.path:
|
|
sys.path.insert(0, _p)
|
|
|
|
from sanguo_data import alerts_store as store # noqa: E402
|
|
import monitor_checks as chk # noqa: E402
|
|
import monitor_registry as reg # noqa: E402
|
|
|
|
_TODAY_OVERRIDE = None # 测试注入口
|
|
_HEALTH_DIR = "monitor"
|
|
_LOG_GLOB = os.path.join("static", "logs", "akshare_static_*.log")
|
|
_LOG_BACK_DAYS = {"evening": 1, "morning": 8} # morning 覆盖周六 03:00 周班
|
|
_OK_LAST_RESULTS = {"0", "267009", "267011"} # 成功/运行中/未跑过
|
|
|
|
|
|
def _now():
|
|
if _TODAY_OVERRIDE is not None:
|
|
return dt.datetime.combine(_TODAY_OVERRIDE, dt.time(23, 50))
|
|
return dt.datetime.now()
|
|
|
|
|
|
def _event(key, severity, title, detail, evidence, host="vps", source="data",
|
|
today=None):
|
|
return {"alert_id": f"{key}-{today}", "check_key": key, "source": source,
|
|
"host": host, "severity": severity, "title": title,
|
|
"detail": detail, "evidence": evidence}
|
|
|
|
|
|
def _ro_conn(path):
|
|
import sqlite3
|
|
return sqlite3.connect(f"file:{path}?mode=ro", uri=True, timeout=10)
|
|
|
|
|
|
# ---------- 结果层: 逐注册表项(log_only 归过程层, 不入此处) ----------
|
|
|
|
def check_entry(e, ctx):
|
|
root = os.path.join(ctx["data_root"], e.get("root", "static"))
|
|
name, kind = e["name"], e["kind"]
|
|
ev = {"kind": kind}
|
|
|
|
def done(check_kind, status, expected, actual, evidence=None):
|
|
return {"key": f"data-{name}-{check_kind}", "status": status,
|
|
"expected": expected, "actual": actual,
|
|
"evidence": evidence or ev}
|
|
|
|
if kind == "per_date":
|
|
exp = chk.expected_daily(ctx["days"], ctx["now"], e.get("ready", "20:30"),
|
|
e.get("lag", 0))
|
|
got = chk.latest_stem(os.path.join(root, e["subdir"]), e["pattern"])
|
|
latest = got[0].split("_")[0] if got else None
|
|
if exp is None: # 无日历(db 不可达): 只做存在性弱检查
|
|
return done("stale", "green" if got else "yellow", "n/a", latest)
|
|
if latest != exp:
|
|
return done("stale", e.get("severity", "red"), exp, latest,
|
|
{"producer": e.get("producer")})
|
|
rows = chk.parquet_rows(got[1]) if got else None
|
|
lo, hi = e["row_band"]
|
|
if rows is None or not (lo <= rows <= hi):
|
|
return done("rows", "yellow", f"{lo}~{hi}", rows)
|
|
return done("stale", "green", exp, latest)
|
|
|
|
if kind == "weekly_friday":
|
|
exp = chk.expected_weekly(ctx["days"], ctx["now"], e.get("ready", "06:00"))
|
|
got = chk.latest_stem(os.path.join(root, e["subdir"]), e["pattern"])
|
|
latest = got[0].split("_")[0] if got else None
|
|
if exp is None:
|
|
return done("stale", "green" if got else "yellow", "n/a", latest)
|
|
if latest != exp:
|
|
return done("stale", e.get("severity", "red"), exp, latest)
|
|
lo, hi = e["row_band"]
|
|
rows = chk.parquet_rows(got[1]) if got else None
|
|
if rows is None or not (lo <= rows <= hi):
|
|
return done("rows", "yellow", f"{lo}~{hi}", rows)
|
|
return done("stale", "green", exp, latest)
|
|
|
|
if kind == "month_bucket":
|
|
got = chk.latest_stem(os.path.join(root, e["subdir"]), e["pattern"])
|
|
if not got:
|
|
return done("stale", e.get("severity", "yellow"), "any", None)
|
|
ym = got[0].split("_")[0]
|
|
try:
|
|
d0 = dt.date(int(ym[:4]), int(ym[4:6]), 1)
|
|
except ValueError:
|
|
return done("stale", "yellow", "YYYYMM", ym)
|
|
age = (ctx["now"].date() - d0).days
|
|
if age > e["max_age_days"]:
|
|
return done("stale", "yellow", f"<={e['max_age_days']}d", f"{age}d")
|
|
lo, hi = e["row_band"]
|
|
rows = chk.parquet_rows(got[1])
|
|
if rows is not None and not (lo <= rows <= hi):
|
|
return done("rows", "yellow", f"{lo}~{hi}", rows)
|
|
return done("stale", "green", ym, ym)
|
|
|
|
if kind == "year_file":
|
|
year = str(ctx["now"].year)
|
|
p = os.path.join(root, e["subdir"], f"{year}.parquet")
|
|
if not os.path.exists(p):
|
|
return done("stale", e.get("severity", "yellow"), year, None)
|
|
mtime = dt.date.fromtimestamp(os.path.getmtime(p)).isoformat()
|
|
exp = chk.expected_daily(ctx["days"], ctx["now"], e.get("ready", "22:30"))
|
|
want = exp.replace("-", "") if exp else None
|
|
got_d = mtime.replace("-", "")
|
|
if want and got_d < want:
|
|
return done("stale", e.get("severity", "yellow"), want, got_d)
|
|
return done("stale", "green", want, got_d)
|
|
|
|
if kind == "dir_mtime":
|
|
d = os.path.join(root, e["subdir"])
|
|
if not os.path.isdir(d):
|
|
return done("stale", e.get("severity", "yellow"), "exists", None)
|
|
age_h = (ctx["now"] - dt.datetime.fromtimestamp(
|
|
chk.dir_latest_mtime(d))).total_seconds() / 3600
|
|
if age_h > e["max_age_days"] * 24:
|
|
return done("stale", "yellow", f"<={e['max_age_days']}d",
|
|
f"{age_h:.0f}h")
|
|
return done("stale", "green", f"<={e['max_age_days']}d", f"{age_h:.0f}h")
|
|
|
|
if kind == "db_sample":
|
|
exp = chk.expected_daily(ctx["days"], ctx["now"], e.get("ready", "22:30"),
|
|
e.get("lag", 0))
|
|
worst = None
|
|
for sym, exc in e["samples"]:
|
|
try:
|
|
conn = _ro_conn(os.path.join(ctx["data_root"], "quant_trading.db"))
|
|
row = conn.execute(
|
|
"SELECT MAX(datetime) FROM dbbardata WHERE symbol=? AND "
|
|
"exchange=? AND interval='d'", (sym, exc)).fetchone()
|
|
conn.close()
|
|
got_d = (row[0] or "").replace("-", "")
|
|
except Exception: # noqa: BLE001 只读探测, 库坏=黄红由调用方判
|
|
got_d = None
|
|
if exp and got_d != exp and worst is None:
|
|
worst = {"symbol": f"{sym}.{exc}", "expected": exp, "actual": got_d}
|
|
if exp and worst:
|
|
return done("stale", e.get("severity", "red"), exp,
|
|
worst["actual"], worst)
|
|
return done("stale", "green", exp, "ok")
|
|
|
|
if kind == "vintage_json":
|
|
p = os.path.join(root, e["subdir"])
|
|
try:
|
|
with open(p, encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
except (OSError, ValueError):
|
|
return done("verdict", "yellow", "readable", None)
|
|
checked = data.get("checked_at", "")
|
|
verdict = data.get("status", data.get("overall", "unknown"))
|
|
try:
|
|
stale = not checked or (ctx["now"] - dt.datetime.fromisoformat(
|
|
checked)).total_seconds() > 26 * 3600
|
|
except ValueError:
|
|
stale = True
|
|
if stale:
|
|
return done("verdict", "yellow", "<=26h", checked)
|
|
if verdict in ("ok", "green", 0):
|
|
return done("verdict", "green", "ok", verdict)
|
|
return done("verdict", "yellow", "ok", verdict)
|
|
|
|
if kind == "static_exists":
|
|
d = os.path.join(root, e["subdir"])
|
|
exists = os.path.isdir(d) and any(
|
|
f.endswith(".parquet") for f in os.listdir(d))
|
|
return done("exists", "green" if exists else e.get("severity", "yellow"),
|
|
"exists", exists)
|
|
|
|
if kind == "disk":
|
|
r = chk.check_disk(ctx["data_root"])
|
|
return {"key": "infra-disk-data", "status": r["status"],
|
|
"expected": f">={e['warn_gb']}G", "actual": f"{r['free_gb']}G",
|
|
"evidence": r}
|
|
|
|
return done("unknown", "yellow", "n/a", None)
|
|
|
|
|
|
# ---------- 过程层: 日志统计行+签名分诊 ----------
|
|
|
|
def collect_log_layer(data_root, now, back_days):
|
|
logs = sorted(glob.glob(os.path.join(data_root, _LOG_GLOB)))
|
|
keep = [p for p in logs
|
|
if (now - dt.datetime.fromtimestamp(os.path.getmtime(p))).days
|
|
<= back_days]
|
|
stats, sigs = {}, []
|
|
for p in keep:
|
|
try:
|
|
with open(p, encoding="utf-8", errors="replace") as f:
|
|
text = f.read()
|
|
except OSError:
|
|
continue
|
|
for t, s in chk.parse_stat_lines(text).items():
|
|
prev = stats.get(t)
|
|
stats[t] = s if prev is None or s.get("rows", 0) >= prev.get("rows", 0) \
|
|
else prev
|
|
sigs.extend(chk.scan_signatures(text))
|
|
return stats, sigs
|
|
|
|
|
|
def log_layer_results(stats, sigs, today):
|
|
"""统计行 failed>0 → 按 sig 分级告警; 断路器独立键; 其余 ok → passed."""
|
|
results, passed = [], []
|
|
worst = max((s["severity"] for s in sigs), default=None)
|
|
breaker = any(s["sig"] == "breaker" for s in sigs)
|
|
for t, s in sorted(stats.items()):
|
|
if s.get("failed", 0) > 0:
|
|
ev = {"stats": s, "signatures": sigs}
|
|
sev = "red" if (worst == "red" or breaker) else "yellow"
|
|
results.append(_event(f"data-{t}-failed", sev,
|
|
f"{t} 类型级 failed={s['failed']}",
|
|
f"rows={s.get('rows')}", ev, today=today))
|
|
else:
|
|
passed.append(f"data-{t}-failed")
|
|
if breaker:
|
|
results.append(_event("data-akshare-breaker", "red",
|
|
"断路器触发(熔断连累: 后续类型整批跳过)",
|
|
"exit code 2 会话", {"signatures": sigs}, today=today))
|
|
else:
|
|
passed.append("data-akshare-breaker")
|
|
return results, passed
|
|
|
|
|
|
def log_only_results(entries, stats, sigs, today):
|
|
"""valuation 等 log_only 项: 统计行存在且 failed=0(周班项平日无行属正常)."""
|
|
results, passed = [], []
|
|
red_sig = any(s["severity"] == "red" for s in sigs)
|
|
for e in entries:
|
|
key = f"data-{e['name']}-stat"
|
|
s = stats.get(e["name"])
|
|
if s is None:
|
|
if e["log_cadence"] == "weekly":
|
|
continue
|
|
results.append(_event(key, "yellow", f"{e['name']} 无当日统计行",
|
|
"任务未跑或日志缺失", {}, today=today))
|
|
elif s.get("failed", 0) > 0:
|
|
results.append(_event(key, "red" if red_sig else "yellow",
|
|
f"{e['name']} 统计 failed={s['failed']}",
|
|
f"rows={s.get('rows')}",
|
|
{"stats": s}, today=today))
|
|
else:
|
|
passed.append(key)
|
|
return results, passed
|
|
|
|
|
|
# ---------- 调度层: schtask 在岗(spec §20.15.1 表后注; retry 一次性死案即此类) ----------
|
|
|
|
def schtask_results(snapshot, now):
|
|
results = []
|
|
today = now.date().isoformat().replace("-", "")
|
|
limits = {"daily": 26 if now.weekday() < 5 else 80, "weekday": 26,
|
|
"weekly": 8 * 24, "monthly": 40 * 24, "always": None}
|
|
for name, cadence in reg.SCHTASKS:
|
|
key = f"infra-task-{name}"
|
|
info = snapshot.get(name)
|
|
if info is None:
|
|
results.append(_event(key, "red", f"schtask {name} 缺失",
|
|
"schtasks /query 无此任务", {}, source="infra",
|
|
today=today))
|
|
continue
|
|
next_run = info.get("next_run", "")
|
|
last_run = info.get("last_run", "")
|
|
last_result = info.get("last_result", "")
|
|
bad = []
|
|
if "N/A" in next_run and cadence != "always":
|
|
bad.append("next_run=N/A(一次性死)")
|
|
if last_result and last_result not in _OK_LAST_RESULTS:
|
|
bad.append(f"last_result={last_result}")
|
|
limit_h = limits[cadence]
|
|
if cadence == "weekday" and now.weekday() >= 5:
|
|
limit_h = None
|
|
age_h = _parse_schtask_age(last_run, now)
|
|
if limit_h is not None and age_h is not None and age_h > limit_h:
|
|
bad.append(f"last_run {age_h:.0f}h 前(限 {limit_h}h)")
|
|
if bad:
|
|
results.append(_event(key, "red" if "N/A" in next_run else "yellow",
|
|
f"schtask {name} 异常", "; ".join(bad),
|
|
{"next_run": next_run, "last_run": last_run,
|
|
"last_result": last_result}, source="infra",
|
|
today=today))
|
|
return results
|
|
|
|
|
|
def _parse_schtask_age(last_run, now):
|
|
if not last_run or "N/A" in last_run:
|
|
return None
|
|
for fmt in ("%Y/%m/%d %H:%M:%S", "%Y-%m-%d %H:%M"):
|
|
try:
|
|
t = dt.datetime.strptime(last_run.strip(), fmt)
|
|
return (now - t).total_seconds() / 3600
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
|
|
|
|
# ---------- 衍生层: NAS inbox 摄取(单写者纪律: NAS 不写远端 sqlite) ----------
|
|
|
|
def ingest_inbox(conn, data_root, now):
|
|
inbox = os.path.join(data_root, _HEALTH_DIR, "inbox")
|
|
latest_ts = None
|
|
if not os.path.isdir(inbox):
|
|
return latest_ts
|
|
processed = os.path.join(inbox, "processed")
|
|
os.makedirs(processed, exist_ok=True)
|
|
for p in sorted(glob.glob(os.path.join(inbox, "*.json"))):
|
|
try:
|
|
with open(p, encoding="utf-8") as f:
|
|
payload = json.load(f)
|
|
for ev in payload.get("events", []):
|
|
store.upsert_alert(conn, ev)
|
|
ts = payload.get("generated_at")
|
|
if ts:
|
|
latest_ts = max(latest_ts or "", ts)
|
|
os.replace(p, os.path.join(processed, os.path.basename(p)))
|
|
except (OSError, ValueError) as exc:
|
|
print(f"[inbox] skip bad file {p}: {exc}")
|
|
return latest_ts
|
|
|
|
|
|
# ---------- 主流程 ----------
|
|
|
|
def write_health(data_root, health):
|
|
d = os.path.join(data_root, _HEALTH_DIR)
|
|
os.makedirs(d, exist_ok=True)
|
|
for fn in (f"health_{health['date']}_{health['shift']}.json",
|
|
"health_latest.json"):
|
|
tmp = os.path.join(d, fn + ".tmp")
|
|
with open(tmp, "w", encoding="utf-8") as f:
|
|
json.dump(health, f, ensure_ascii=False, indent=1)
|
|
os.replace(tmp, os.path.join(d, fn))
|
|
|
|
|
|
def main(argv=None):
|
|
ap = argparse.ArgumentParser(description=__doc__)
|
|
ap.add_argument("--shift", choices=["evening", "morning"], default="evening")
|
|
ap.add_argument("--data-root", default="data")
|
|
ap.add_argument("--db", default=store.default_alerts_db_path())
|
|
ap.add_argument("--skip-tasks", action="store_true")
|
|
ap.add_argument("--skip-inbox", action="store_true")
|
|
args = ap.parse_args(argv)
|
|
|
|
now = _now()
|
|
today = now.date().isoformat().replace("-", "")
|
|
db_path = args.db
|
|
if args.db == "data/alerts.db" and args.data_root != "data":
|
|
db_path = os.path.join(args.data_root, "alerts.db")
|
|
|
|
days = chk.trading_days(
|
|
os.path.join(args.data_root, "quant_trading.db"), now.date())
|
|
if not days: # 日历降级: 近 50 日周一~五(节假日误报由带宽/黄级兜底)
|
|
d = now.date() - dt.timedelta(days=50)
|
|
while d <= now.date():
|
|
if d.weekday() < 5:
|
|
days.append(d.isoformat())
|
|
d += dt.timedelta(days=1)
|
|
ctx = {"data_root": args.data_root, "days": days, "now": now}
|
|
|
|
conn = store.connect(db_path)
|
|
checks, events, passed = [], [], []
|
|
|
|
for e in reg.REGISTRY:
|
|
if not e.get("enabled", True) or e["kind"] == "log_only":
|
|
continue # log_only 归过程层 log_only_results
|
|
r = check_entry(e, ctx)
|
|
checks.append(r)
|
|
if r["status"] == "green":
|
|
passed.append(r["key"])
|
|
else:
|
|
kindword = r["key"].rsplit("-", 1)[-1] # stale/rows/verdict/exists
|
|
events.append(_event(
|
|
r["key"], r["status"],
|
|
f"{e['name']} {kindword} 异常",
|
|
f"actual={r['actual']} expected={r['expected']}",
|
|
{"evidence": r.get("evidence", {}), "kind": e["kind"],
|
|
"producer": e.get("producer", "vps")},
|
|
source="infra" if r["key"].startswith("infra-") else "data",
|
|
today=today))
|
|
|
|
stats, sigs = collect_log_layer(args.data_root, now,
|
|
_LOG_BACK_DAYS[args.shift])
|
|
log_events, log_passed = log_layer_results(stats, sigs, today)
|
|
lo_events, lo_passed = log_only_results(
|
|
[e for e in reg.REGISTRY if e["kind"] == "log_only"
|
|
and e.get("enabled", True)], stats, sigs, today)
|
|
events += log_events + lo_events
|
|
passed += log_passed + lo_passed
|
|
|
|
if not args.skip_tasks and os.name == "nt":
|
|
out = subprocess.run(["schtasks", "/query", "/fo", "LIST", "/v"],
|
|
capture_output=True, text=True, errors="replace",
|
|
timeout=180).stdout
|
|
task_events = schtask_results(chk.parse_schtasks_list(out), now)
|
|
events += task_events
|
|
flagged = {ev["check_key"] for ev in task_events}
|
|
passed += [f"infra-task-{n}" for n, _ in reg.SCHTASKS
|
|
if f"infra-task-{n}" not in flagged]
|
|
|
|
nas_silent_key = "infra-nas-monitor-silent"
|
|
if not args.skip_inbox:
|
|
nas_ts = ingest_inbox(conn, args.data_root, now)
|
|
if nas_ts:
|
|
try:
|
|
age_h = (now - dt.datetime.fromisoformat(nas_ts)
|
|
).total_seconds() / 3600
|
|
except ValueError:
|
|
age_h = None
|
|
if age_h is not None and age_h > 30:
|
|
events.append(_event(nas_silent_key, "yellow",
|
|
"NAS 监控静默>30h", f"last={nas_ts}",
|
|
{"last_inbox_ts": nas_ts}, source="infra",
|
|
today=today))
|
|
else:
|
|
passed.append(nas_silent_key)
|
|
|
|
for ev in events:
|
|
store.upsert_alert(conn, ev)
|
|
store.resolve_passed(conn, passed)
|
|
store.purge_expired(conn)
|
|
conn.close()
|
|
|
|
counts = {s: sum(1 for c in checks if c["status"] == s)
|
|
for s in ("green", "yellow", "red")}
|
|
red_events = sum(1 for ev in events if ev["severity"] == "red")
|
|
yellow_events = sum(1 for ev in events if ev["severity"] == "yellow")
|
|
overall = "red" if (counts["red"] or red_events) else (
|
|
"yellow" if (counts["yellow"] or yellow_events) else "green")
|
|
health = {"date": now.date().isoformat(), "shift": args.shift, "host": "vps",
|
|
"generated_at": now.isoformat(timespec="seconds"),
|
|
"overall": overall, "counts": counts,
|
|
"alerts_emitted": len(events), "checks": checks}
|
|
write_health(args.data_root, health)
|
|
print(f"[data-monitor] {args.shift} overall={overall} checks={len(checks)}"
|
|
f" alerts={len(events)}")
|
|
return 3 if overall == "red" else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|