dekun
130 lines
4.6 KiB
Python
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
|