Files
crypto_okx_hub/lib/health_watch_lib.py
T

130 lines
4.6 KiB
Python

"""客户端在线健康巡检与企业微信告警(冷却去重)."""
from __future__ import annotations
import threading
import time
from typing import Any
from lib.wechat_lib import (
hub_alert_cooldown_sec,
hub_health_fail_threshold,
hub_wechat_enabled,
send_hub_wechat,
)
_lock = threading.Lock()
# instance_id -> {online, fail_count, last_alert_at, last_reason, name}
_state: dict[Any, dict[str, Any]] = {}
def _now() -> float:
return time.time()
def reset_health_state() -> None:
with _lock:
_state.clear()
def inspect_overview_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""根据总览拉取结果更新状态并按需告警.返回本次新发出的告警摘要."""
if not rows:
return []
if not hub_wechat_enabled():
# 仍更新在线状态,便于开关打开后立刻有基线
with _lock:
for row in rows:
iid = row.get("id")
if iid is None:
continue
online = bool(row.get("online"))
st = _state.get(iid) or {}
st.update(
{
"online": online,
"fail_count": 0 if online else int(st.get("fail_count") or 0) + 1,
"name": row.get("name") or st.get("name") or str(iid),
"last_reason": "" if online else (row.get("error") or "离线"),
}
)
_state[iid] = st
return []
threshold = hub_health_fail_threshold()
cooldown = hub_alert_cooldown_sec()
sent: list[dict[str, Any]] = []
now = _now()
with _lock:
for row in rows:
iid = row.get("id")
if iid is None:
continue
name = str(row.get("name") or iid)
online = bool(row.get("online"))
err = str(row.get("error") or "").strip()
st = _state.get(iid) or {
"online": None,
"fail_count": 0,
"last_alert_at": 0.0,
"last_reason": "",
"name": name,
}
prev_online = st.get("online")
st["name"] = name
if online:
# 从离线恢复
if prev_online is False:
if now - float(st.get("last_alert_at") or 0) >= cooldown:
msg = f"客户端恢复在线\n名称:{name}\nID:{iid}"
result = send_hub_wechat(msg)
if result.get("ok"):
st["last_alert_at"] = now
sent.append({"instance_id": iid, "kind": "online", "msg": msg})
st["online"] = True
st["fail_count"] = 0
st["last_reason"] = ""
else:
reason = err or "离线"
st["fail_count"] = int(st.get("fail_count") or 0) + 1
st["online"] = False
st["last_reason"] = reason
# 首次已知状态若就是离线,也累计;达到阈值才告警
if st["fail_count"] >= threshold:
if now - float(st.get("last_alert_at") or 0) >= cooldown:
kind = "fault" if reason and reason not in ("离线", "超时") else "offline"
title = "客户端故障" if kind == "fault" else "客户端离线"
msg = "\n".join(
[
title,
f"名称:{name}",
f"ID:{iid}",
f"原因:{reason}",
f"连续失败:{st['fail_count']} 次",
]
)
result = send_hub_wechat(msg)
if result.get("ok"):
st["last_alert_at"] = now
sent.append({"instance_id": iid, "kind": kind, "msg": msg})
_state[iid] = st
return sent
def health_snapshot() -> list[dict[str, Any]]:
with _lock:
out = []
for iid, st in _state.items():
out.append(
{
"id": iid,
"name": st.get("name"),
"online": st.get("online"),
"fail_count": st.get("fail_count") or 0,
"last_reason": st.get("last_reason") or "",
"last_alert_at": st.get("last_alert_at") or 0,
}
)
return out