297 lines
12 KiB
Python
297 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""cron_health_monitor.py — 全局cron job健康监控 v2
|
||
|
||
每10分钟跑一次，检查：
|
||
1. 所有启用cron job的last_status是否为failed
|
||
2. 关键job的last_run_at是否超过预期空闲时间
|
||
3. price_monitor等核心管道是否活着
|
||
4. 自己是否正常跑（自检）
|
||
|
||
发现新问题 → 推XMPP告警（同类问题2小时内不重复推）
|
||
"""
|
||
import json, os, sys, re
|
||
from pathlib import Path
|
||
from datetime import datetime, timezone, timedelta
|
||
from urllib.request import Request, urlopen
|
||
|
||
XMPP_BRIDGE = "http://127.0.0.1:5805/"
|
||
XMPP_USER = "hmo@yoin.fun"
|
||
STATE_FILE = Path.home() / ".hermes" / ".cron_health_state.json"
|
||
COOLDOWN_HOURS = 2 # 同类告警2小时内不重复
|
||
|
||
# 关键job及其最大允许空闲分钟
|
||
# 只监控实时管道（每分~每小时跑的关键任务）
|
||
# 日/周任务不在此列（它们不耽误实时推荐也不需要秒级响应）
|
||
CRITICAL_JOBS = {
|
||
"price_monitor": {"max_idle_min": 15}, # 每2分
|
||
"MoFin盘前中监控": {"max_idle_min": 90}, # 每25分（盘中）
|
||
"自选买入区提醒": {"max_idle_min": 90}, # 每30分
|
||
"自选买入区提醒-盘前午间尾盘": {"max_idle_min": 185}, # 每3小时(9/12/15点)，允许185分
|
||
"宏观风险信号消费": {"max_idle_min": 40}, # 每15分
|
||
"重评管道审计": {"max_idle_min": 90}, # 每30分
|
||
"全局cron健康监控": {"max_idle_min": 30}, # 每10分 ← 自己
|
||
"实时消息中继": {"max_idle_min": 20},
|
||
"宏观上下文刷新": {"max_idle_min": 90}, # 每5分
|
||
}
|
||
|
||
def xmpp_push(text):
|
||
try:
|
||
payload = json.dumps({"to": XMPP_USER, "body": text, "type": "chat"}).encode()
|
||
req = Request(XMPP_BRIDGE, data=payload, headers={"Content-Type": "application/json"})
|
||
urlopen(req, timeout=5)
|
||
return True
|
||
except Exception as e:
|
||
print(f"[XMPP推送失败] {e}", file=sys.stderr)
|
||
return False
|
||
|
||
|
||
def _create_cron_todo(tname, desc):
|
||
"""为故障cron创建TODO，fix_action带自动重试+修复逻辑"""
|
||
try:
|
||
# 找这个cron对应的脚本
|
||
script_path = None
|
||
try:
|
||
import json as _j
|
||
with open("/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json") as _f:
|
||
_crons = _j.load(_f).get("jobs", [])
|
||
for _c in _crons:
|
||
if _c.get("name") == tname:
|
||
_s = _c.get("script", "")
|
||
if _s:
|
||
_sp = f"/home/hmo/.hermes/profiles/position-analyst/scripts/{_s.split()[0]}"
|
||
if __import__('os').path.exists(_sp):
|
||
script_path = _sp
|
||
break
|
||
except Exception:
|
||
pass
|
||
|
||
if script_path:
|
||
fix = (
|
||
f"cd /home/hmo/MoFin && timeout 60 python3 {script_path} 2>&1 "
|
||
f"| tail -5; if [ $? -eq 0 ]; then "
|
||
f"python3 -c \"import json; p='/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json'; "
|
||
f"d=json.load(open(p)); "
|
||
f"for j in d['jobs']: "
|
||
f" if j.get('name')=='{tname}': j['last_status']='ok'; "
|
||
f"json.dump(d,open(p,'w'),ensure_ascii=False,indent=2)\" "
|
||
f"&& echo 'AUTO_FIXED'; "
|
||
f"else echo 'AUTO_FAIL'; fi"
|
||
)
|
||
else:
|
||
fix = (
|
||
f"cd /home/hmo/MoFin && echo 'check cron: {tname}' && "
|
||
f"python3 scripts/preflight_verify.py --cron-check"
|
||
)
|
||
|
||
_c = __import__('sqlite3').connect("/home/hmo/MoFin/data/mofin.db", timeout=10)
|
||
_c.execute(
|
||
"INSERT OR REPLACE INTO todos (id, title, description, status, priority, source, fix_action, created_at) "
|
||
"VALUES (?, ?, ?, 'pending', 'high', 'cron_health', ?, datetime('now','localtime'))",
|
||
(f"cron_err_{tname.replace(' ','_')}", f"[CRON_ERROR] {tname}", desc, fix)
|
||
)
|
||
_c.commit()
|
||
_c.close()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def parse_time(ts_str):
|
||
if not ts_str:
|
||
return None
|
||
try:
|
||
dt = datetime.fromisoformat(ts_str)
|
||
if dt.tzinfo is None:
|
||
dt = dt.replace(tzinfo=timezone.utc)
|
||
return dt
|
||
except Exception:
|
||
return None
|
||
|
||
def get_idle_minutes(last_run_str):
|
||
"""计算上次运行到现在过了多少分钟"""
|
||
last = parse_time(last_run_str)
|
||
if not last:
|
||
return None
|
||
now = datetime.now(timezone.utc)
|
||
return (now - last).total_seconds() / 60
|
||
|
||
def scan_jobs_files():
|
||
"""扫描所有cron jobs.json，合并返回全量job列表"""
|
||
jobs = []
|
||
seen = set()
|
||
for jf in [
|
||
"/home/hmo/.hermes/cron/jobs.json",
|
||
"/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json",
|
||
]:
|
||
try:
|
||
data = json.load(open(jf))
|
||
for job in data.get("jobs", []):
|
||
jid = job.get("id", "")
|
||
if jid in seen:
|
||
continue
|
||
seen.add(jid)
|
||
jobs.append({
|
||
"id": jid,
|
||
"name": job.get("name", jid[:12]) or jid[:12],
|
||
"status": job.get("last_status", ""),
|
||
"last_run": job.get("last_run_at", ""),
|
||
"enabled": job.get("enabled", True),
|
||
"no_agent": job.get("no_agent", False),
|
||
"schedule": job.get("schedule", {}),
|
||
})
|
||
except Exception:
|
||
pass
|
||
return jobs
|
||
|
||
def check_idle(jobs):
|
||
"""检查关键job是否超过最大空闲时间"""
|
||
stale = []
|
||
for j in jobs:
|
||
if not j["enabled"]:
|
||
continue
|
||
name = j["name"]
|
||
rule = None
|
||
for key, r in CRITICAL_JOBS.items():
|
||
if key.lower() in name.lower():
|
||
rule = r
|
||
break
|
||
if not rule:
|
||
continue
|
||
idle = get_idle_minutes(j["last_run"])
|
||
if idle is None:
|
||
stale.append((j, "无运行记录"))
|
||
elif idle > rule["max_idle_min"]:
|
||
idle_str = f"{idle:.0f}分"
|
||
stale.append((j, f"超过{rule['max_idle_min']}分未运行(已{idle_str})"))
|
||
return stale
|
||
|
||
def load_state():
|
||
try:
|
||
with open(STATE_FILE) as f:
|
||
return json.load(f)
|
||
except Exception:
|
||
return {}
|
||
|
||
def save_state(state):
|
||
STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
|
||
with open(STATE_FILE, "w") as f:
|
||
json.dump(state, f, indent=2)
|
||
|
||
def is_new_alert(alert_key, state):
|
||
"""检查同类告警是否在cooldown期内"""
|
||
now = datetime.now().timestamp()
|
||
last = state.get(alert_key, {}).get("last_alerted", 0)
|
||
return (now - last) > COOLDOWN_HOURS * 3600
|
||
|
||
def main():
|
||
now_str = datetime.now().strftime("%H:%M")
|
||
jobs = scan_jobs_files()
|
||
state = load_state()
|
||
alerts = []
|
||
|
||
# ── 检查1：last_status=failed ──
|
||
failed_jobs = [j for j in jobs if j["status"] == "failed" and j["enabled"]]
|
||
if failed_jobs:
|
||
for j in failed_jobs:
|
||
tag = "no_agent" if j["no_agent"] else "LLM"
|
||
key = f"failed_{j['id']}"
|
||
if is_new_alert(key, state):
|
||
alerts.append(f"❌ {j['name']}({j['id'][:8]}) [{tag}] {j['status']} @ {j['last_run'][:19]}")
|
||
state[key] = {"last_alerted": datetime.now().timestamp()}
|
||
# 同时写TODO供自愈系统处理
|
||
_create_cron_todo(j['name'], f"[CRON_ERROR] {j['name']} 状态={j['status']} 最后运行={j['last_run'][:19]}")
|
||
|
||
# ── 检查2：关键job空闲超时 ──
|
||
stale_jobs = check_idle(jobs)
|
||
if stale_jobs:
|
||
for j, reason in stale_jobs:
|
||
key = f"stale_{j['id']}"
|
||
if is_new_alert(key, state):
|
||
alerts.append(f"⏰ {j['name']}({j['id'][:8]}) {reason}")
|
||
state[key] = {"last_alerted": datetime.now().timestamp()}
|
||
|
||
# ── 检查3：XMPP bridge是否在线（TCP端口检测，不发消息到Dad） ─
|
||
bridge_key = "bridge_down"
|
||
try:
|
||
import socket
|
||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
sock.settimeout(3)
|
||
result = sock.connect_ex(("127.0.0.1", 5805))
|
||
sock.close()
|
||
bridge_ok = (result == 0)
|
||
if not bridge_ok and is_new_alert(bridge_key, state):
|
||
alerts.append("🔴 XMPP bridge(5805)端口无响应")
|
||
state[bridge_key] = {"last_alerted": datetime.now().timestamp()}
|
||
except Exception as e:
|
||
if is_new_alert(bridge_key, state):
|
||
alerts.append(f"🔴 XMPP bridge检测失败: {e}")
|
||
state[bridge_key] = {"last_alerted": datetime.now().timestamp()}
|
||
|
||
# ── 检查4：gateway LLM推理是否正常（发真实请求测响应时间） ─
|
||
gateway_key = "gateway_down"
|
||
try:
|
||
import urllib.request, json, time
|
||
payload = json.dumps({
|
||
"model": "position-analyst",
|
||
"messages": [{"role": "user", "content": "ok"}],
|
||
"max_tokens": 5
|
||
}).encode()
|
||
req = urllib.request.Request("http://127.0.0.1:8643/v1/chat/completions",
|
||
data=payload,
|
||
headers={"Authorization": "Bearer hermes123", "Content-Type": "application/json"})
|
||
t0 = time.time()
|
||
resp = urllib.request.urlopen(req, timeout=30)
|
||
elapsed = time.time() - t0
|
||
data = json.loads(resp.read())
|
||
usage = data.get("usage", {})
|
||
tok = usage.get("total_tokens", 0)
|
||
if elapsed > 25 and is_new_alert(gateway_key, state):
|
||
alerts.append(f"🔴 gateway响应慢: {elapsed:.0f}s ({tok}tok)")
|
||
state[gateway_key] = {"last_alerted": datetime.now().timestamp()}
|
||
except Exception as e:
|
||
if is_new_alert(gateway_key, state):
|
||
alerts.append(f"🔴 gateway推理失败: {e}")
|
||
state[gateway_key] = {"last_alerted": datetime.now().timestamp()}
|
||
|
||
# ── 检查5：price_monitor数据新鲜度（直接查DB，不依赖job记录）─
|
||
price_key = "price_stale"
|
||
try:
|
||
from mofin_db import get_conn
|
||
conn = get_conn()
|
||
lp = conn.execute("SELECT MAX(updated_at) FROM live_prices").fetchone()[0]
|
||
conn.close()
|
||
if lp:
|
||
lp_dt = datetime.fromisoformat(lp) if isinstance(lp, str) else lp
|
||
if hasattr(lp_dt, 'tzinfo') and lp_dt.tzinfo is None:
|
||
mins = (datetime.now() - lp_dt).total_seconds() / 60
|
||
if mins > 15 and is_new_alert(price_key, state):
|
||
alerts.append(f"🔴 price_monitor {mins:.0f}分未更新数据")
|
||
state[price_key] = {"last_alerted": datetime.now().timestamp()}
|
||
except Exception as e:
|
||
if is_new_alert(price_key, state):
|
||
alerts.append(f"🔴 price_monitor查询失败: {e}")
|
||
state[price_key] = {"last_alerted": datetime.now().timestamp()}
|
||
|
||
# ── 本脚本自检：检查自己是否在job列表里且有正常last_run ──
|
||
my_name = "全局cron健康监控"
|
||
myself = [j for j in jobs if my_name in j["name"]]
|
||
if myself:
|
||
m = myself[0]
|
||
idle = get_idle_minutes(m["last_run"])
|
||
if idle is None or idle > 30:
|
||
# 自己异常但可能还在跑？记录但不推（否则死循环）
|
||
print(f"[SELF_CHECK] 自身last_run={m.get('last_run','?')} {idle:.0f}分前", file=sys.stderr)
|
||
|
||
save_state(state)
|
||
|
||
if not alerts:
|
||
print("[SILENT]")
|
||
return
|
||
|
||
# 推告警
|
||
text = f"🔴 {len(alerts)}条健康告警 ({now_str}):\n" + "\n".join(alerts)
|
||
print(text, file=sys.stderr)
|
||
xmpp_push(text)
|
||
|
||
if __name__ == "__main__":
|
||
main()
|