465 lines
20 KiB
Python
465 lines
20 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:
|
||
sev = "yellow" if exp not in ctx["confirmed_days"] \
|
||
else e.get("severity", "red")
|
||
return done("stale", sev, exp, latest,
|
||
{"producer": e.get("producer"),
|
||
"unconfirmed_day": exp not in ctx["confirmed_days"]})
|
||
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:
|
||
sev = "yellow" if exp not in ctx["confirmed_days"] \
|
||
else e.get("severity", "red")
|
||
return done("stale", sev, 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:
|
||
sev = "yellow" if exp not in ctx["confirmed_days"] \
|
||
else e.get("severity", "red")
|
||
return done("stale", sev, 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())
|
||
# 日历=确认集(dbbardata 000300 实证) ∪ 近 15 日工作日假定集(09-25 首跑实锤的
|
||
# 循环依赖修: dbbardata 晚间源滞后/漏日会把全体期望塌一天, 误红鲜活域).
|
||
# 语义: 缺失遇「未确认日」(不在 dbbardata) 一律降黄——节假日/合并窗/源滞后
|
||
# 三态容忍; 确认日缺失才红(跨域互证: 某域有该日数据=真是交易日).
|
||
confirmed = {d.replace("-", "") for d in days}
|
||
union = set(days)
|
||
d0 = now.date() - dt.timedelta(days=15)
|
||
while d0 <= now.date():
|
||
if d0.weekday() < 5:
|
||
union.add(d0.isoformat())
|
||
d0 += dt.timedelta(days=1)
|
||
days = sorted(union)
|
||
ctx = {"data_root": args.data_root, "days": days, "now": now,
|
||
"confirmed_days": confirmed}
|
||
|
||
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())
|