117 lines
3.8 KiB
Python
117 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""cron_health_monitor.py — 全局cron job健康监控
|
|
|
|
每10分钟跑一次，检查所有cron job的last_status。
|
|
发现failed → 立即推XMPP告警到老爸。
|
|
"""
|
|
import json, os, sys, sqlite3, time
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
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"
|
|
|
|
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 scan_all_jobs():
|
|
"""扫描所有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)
|
|
last_status = job.get("last_status", "")
|
|
if not last_status:
|
|
continue
|
|
jobs.append({
|
|
"id": jid,
|
|
"name": job.get("name", jid[:12]),
|
|
"status": last_status,
|
|
"last_run": job.get("last_run_at", "?"),
|
|
"enabled": job.get("enabled", True),
|
|
"no_agent": job.get("no_agent", False),
|
|
})
|
|
except Exception:
|
|
pass
|
|
return jobs
|
|
|
|
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 main():
|
|
now = datetime.now().strftime("%H:%M")
|
|
jobs = scan_all_jobs()
|
|
state = load_state()
|
|
|
|
failed_jobs = [j for j in jobs if j["status"] == "failed" and j["enabled"]]
|
|
|
|
if not failed_jobs:
|
|
# 全部正常
|
|
print("[SILENT]")
|
|
# 记录全部分辨率，用于跟踪从failed→ok的恢复
|
|
save_state({j["id"]: {"status": j["status"], "notified": False} for j in jobs})
|
|
return
|
|
|
|
# 有failed job → 过滤出未通知过的（避免重复推送）
|
|
new_failures = []
|
|
ongoing_failures = []
|
|
for j in failed_jobs:
|
|
prev = state.get(j["id"], {})
|
|
prev_status = prev.get("status", "")
|
|
if prev_status != "failed":
|
|
# 新发现失败（之前正常或未记录）
|
|
new_failures.append(j)
|
|
else:
|
|
ongoing_failures.append(j)
|
|
|
|
if not new_failures:
|
|
# 没有新增失败 → 静默（上次已通知过）
|
|
print("[SILENT] 已有失败未恢复")
|
|
save_state({j["id"]: {"status": j["status"], "notified": True} for j in jobs})
|
|
return
|
|
|
|
# 新发现的失败 → 推XMPP
|
|
msg_lines = [f"🔴 {len(new_failures)}个cron job失败 ({now}):"]
|
|
for j in new_failures:
|
|
tag = "no_agent" if j["no_agent"] else "LLM"
|
|
msg_lines.append(f" ❌ {j['name']}({j['id'][:8]}) [{tag}] last_run={j['last_run']}")
|
|
if ongoing_failures:
|
|
msg_lines.append(f" (另有{len(ongoing_failures)}个上次已报过)")
|
|
|
|
text = "\n".join(msg_lines)
|
|
print(text)
|
|
xmpp_push(text)
|
|
|
|
# 记录状态
|
|
save_state({j["id"]: {"status": j["status"], "notified": True} for j in jobs})
|
|
|
|
if __name__ == "__main__":
|
|
main()
|