v6ole
848f804169
- FastAPI 后端 + Vue 3 前端 - Docker Compose 一键部署 - Casdoor OAuth 认证集成 - LogHive 集中式日志 - 设备批量 CSV 导入/导出 - WebSocket 实时状态推送 - 企业微信告警通知 - fping 高性能并发 Ping 检测 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
320 lines
11 KiB
Python
320 lines
11 KiB
Python
"""
|
||
告警服务
|
||
|
||
职责：
|
||
1. 接收 Pinger 的状态变化事件
|
||
2. 判断是否需要发送告警
|
||
3. 上游心跳检测 + 全量离线抑制
|
||
4. 企业微信消息推送（聚合告警）
|
||
5. 记录告警事件到数据库
|
||
"""
|
||
|
||
import asyncio
|
||
import logging
|
||
from datetime import datetime
|
||
from typing import Optional
|
||
from collections import defaultdict
|
||
|
||
import httpx
|
||
from sqlalchemy import select, func
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.config import settings
|
||
from app.models.device import Device
|
||
from app.models.alert_event import AlertEvent, AlertTypeEnum
|
||
from app.services.pinger import DeviceStateChange
|
||
|
||
logger = logging.getLogger("pingwatch.alerter")
|
||
|
||
|
||
class PendingOfflineAlert:
|
||
"""等待发送的离线告警（用于聚合）"""
|
||
def __init__(self, device: Device, alert_time: datetime):
|
||
self.device = device
|
||
self.alert_time = alert_time
|
||
|
||
|
||
class Alerter:
|
||
"""
|
||
告警处理器。
|
||
|
||
核心逻辑：
|
||
- 设备 offline → 收集到待发送队列
|
||
- 每轮结束后检查：
|
||
a) 上游心跳是否正常？
|
||
b) 离线率是否 < 90%？
|
||
c) 满足条件 → 聚合所有待发告警 → 一条企业微信消息
|
||
d) 不满足 → 丢弃本轮告警，记录系统日志
|
||
- 设备 recovered → 单独发送恢复通知
|
||
"""
|
||
|
||
def __init__(self):
|
||
self._pending_alerts: list[PendingOfflineAlert] = []
|
||
self._lock = asyncio.Lock()
|
||
# 上游心跳状态
|
||
self._upstream_failures = 0
|
||
self._upstream_available = True
|
||
|
||
async def on_state_change(self, change: DeviceStateChange, db: AsyncSession):
|
||
"""Pinger 状态变化回调"""
|
||
if change.new_status == "offline":
|
||
async with self._lock:
|
||
self._pending_alerts.append(
|
||
PendingOfflineAlert(device=change.device, alert_time=datetime.now())
|
||
)
|
||
|
||
elif change.new_status == "online" and change.old_status == "offline":
|
||
# 设备恢复，立即记录并发送恢复通知
|
||
await self._handle_recovery(change.device, db)
|
||
|
||
async def flush_pending(self, db: AsyncSession, total_device_count: int):
|
||
"""
|
||
每轮结束时调用：处理待发送的离线告警。
|
||
判断是否应该抑制告警，然后发送或丢弃。
|
||
"""
|
||
async with self._lock:
|
||
if not self._pending_alerts:
|
||
return
|
||
pending = self._pending_alerts.copy()
|
||
self._pending_alerts.clear()
|
||
|
||
# 1. 检查上游心跳
|
||
upstream_ok = await self._check_upstream()
|
||
|
||
# 2. 计算离线率
|
||
offline_count = len(pending)
|
||
offline_ratio = offline_count / max(total_device_count, 1)
|
||
|
||
# 3. 抑制条件
|
||
suppressed = False
|
||
suppress_reason = ""
|
||
|
||
if not upstream_ok:
|
||
suppressed = True
|
||
suppress_reason = "上游网络不可达（监控节点可能断网）"
|
||
elif offline_ratio >= settings.OFFLINE_SUPPRESS_RATIO:
|
||
suppressed = True
|
||
suppress_reason = f"离线率 {offline_ratio:.0%} >= {settings.OFFLINE_SUPPRESS_RATIO:.0%}，疑似监控节点断网"
|
||
|
||
if suppressed:
|
||
logger.warning(
|
||
f"告警抑制: {suppress_reason}，"
|
||
f"本轮 {offline_count} 条告警已丢弃"
|
||
)
|
||
# 记录系统告警
|
||
db.add(AlertEvent(
|
||
device_id=0,
|
||
alert_type=AlertTypeEnum.system,
|
||
message=f"告警抑制: {suppress_reason}，丢弃 {offline_count} 条离线告警",
|
||
start_at=datetime.now(),
|
||
is_resolved=True,
|
||
notification_sent=False,
|
||
))
|
||
await db.commit()
|
||
return
|
||
|
||
# 4. 发送聚合告警
|
||
if pending:
|
||
await self._send_aggregated_alert(pending, db)
|
||
|
||
async def _check_upstream(self) -> bool:
|
||
"""
|
||
上游心跳检测。
|
||
连续 UPSTREAM_PING_THRESHOLD 次失败才判定为上游断网。
|
||
"""
|
||
try:
|
||
import subprocess
|
||
proc = await asyncio.create_subprocess_exec(
|
||
"ping", "-c", "1", "-W", "3",
|
||
settings.UPSTREAM_PING_TARGET,
|
||
stdout=subprocess.DEVNULL,
|
||
stderr=subprocess.DEVNULL,
|
||
)
|
||
await proc.wait()
|
||
if proc.returncode == 0:
|
||
self._upstream_failures = 0
|
||
self._upstream_available = True
|
||
return True
|
||
else:
|
||
self._upstream_failures += 1
|
||
if self._upstream_failures >= settings.UPSTREAM_PING_THRESHOLD:
|
||
self._upstream_available = False
|
||
return False
|
||
# 未达到阈值，认为上游还可用
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"上游心跳检测异常: {e}")
|
||
return True # 异常时保守地允许告警
|
||
|
||
async def _send_aggregated_alert(self, alerts: list[PendingOfflineAlert], db: AsyncSession):
|
||
"""发送聚合离线告警"""
|
||
now = datetime.now()
|
||
|
||
# 构建企业微信消息
|
||
if len(alerts) == 1:
|
||
a = alerts[0]
|
||
msg = self._build_offline_message_single(a.device, a.alert_time)
|
||
else:
|
||
msg = self._build_offline_message_batch(alerts)
|
||
|
||
# 发送
|
||
success = await self._send_wecom_message(msg)
|
||
|
||
# 记录告警事件
|
||
for a in alerts:
|
||
db.add(AlertEvent(
|
||
device_id=a.device.id,
|
||
alert_type=AlertTypeEnum.offline,
|
||
message=a.device.name,
|
||
start_at=a.alert_time,
|
||
is_resolved=False,
|
||
notification_sent=success,
|
||
))
|
||
|
||
await db.commit()
|
||
|
||
if success:
|
||
logger.info(f"已推送离线告警: {len(alerts)} 台设备")
|
||
else:
|
||
logger.error(f"企业微信推送失败: {len(alerts)} 台设备")
|
||
|
||
async def _handle_recovery(self, device: Device, db: AsyncSession):
|
||
"""处理设备恢复"""
|
||
now = datetime.now()
|
||
|
||
# 查找未解决的离线事件
|
||
result = await db.execute(
|
||
select(AlertEvent)
|
||
.where(AlertEvent.device_id == device.id)
|
||
.where(AlertEvent.alert_type == AlertTypeEnum.offline)
|
||
.where(AlertEvent.is_resolved == False)
|
||
.order_by(AlertEvent.created_at.desc())
|
||
.limit(1)
|
||
)
|
||
event = result.scalar_one_or_none()
|
||
|
||
duration_minutes = None
|
||
if event:
|
||
delta = now - event.start_at
|
||
duration_minutes = int(delta.total_seconds() / 60)
|
||
event.end_at = now
|
||
event.duration_minutes = duration_minutes
|
||
event.is_resolved = True
|
||
|
||
# 发送恢复通知
|
||
msg = self._build_recovery_message(device, now, duration_minutes)
|
||
success = await self._send_wecom_message(msg)
|
||
|
||
if event:
|
||
event.notification_sent = success
|
||
|
||
await db.commit()
|
||
|
||
if success:
|
||
logger.info(f"已推送恢复通知: {device.name}")
|
||
else:
|
||
logger.error(f"恢复通知推送失败: {device.name}")
|
||
|
||
# ---------- 消息格式化 ----------
|
||
|
||
def _build_offline_message_single(self, device: Device, alert_time: datetime) -> str:
|
||
"""单台设备离线消息"""
|
||
time_str = alert_time.strftime("%Y-%m-%d %H:%M:%S")
|
||
return (
|
||
f"⛔ 设备离线啦！\n"
|
||
f"地址：{device.location or '未知'}\n"
|
||
f"时间：{time_str}\n"
|
||
f"项目：{device.project_name or '未分组'}\n"
|
||
f"设备类型：{self._fmt_device_type(device.device_type)}\n"
|
||
f"IP地址：{device.ip}"
|
||
)
|
||
|
||
def _build_offline_message_batch(self, alerts: list[PendingOfflineAlert]) -> str:
|
||
"""多台设备聚合离线消息"""
|
||
now = alerts[0].alert_time
|
||
time_str = now.strftime("%Y-%m-%d %H:%M:%S")
|
||
lines = [f"⛔ 设备离线啦！（共 {len(alerts)} 台）\n"]
|
||
|
||
for a in alerts:
|
||
dev = a.device
|
||
lines.append(
|
||
f"地址：{dev.location or '未知'}\n"
|
||
f"时间：{a.alert_time.strftime('%Y-%m-%d %H:%M:%S')}\n"
|
||
f"项目：{dev.project_name or '未分组'}\n"
|
||
f"设备类型：{self._fmt_device_type(dev.device_type)}\n"
|
||
f"IP地址：{dev.ip}\n"
|
||
f"{'---' if len(alerts) > 1 else ''}"
|
||
)
|
||
|
||
return "\n".join(lines).rstrip("---\n")
|
||
|
||
def _build_recovery_message(self, device: Device, recover_time: datetime, duration: Optional[int]) -> str:
|
||
"""设备恢复消息"""
|
||
time_str = recover_time.strftime("%Y-%m-%d %H:%M:%S")
|
||
duration_str = f"{duration}分钟" if duration is not None else "未知"
|
||
return (
|
||
f"✅ 设备恢复在线！\n"
|
||
f"地址：{device.location or '未知'}\n"
|
||
f"时间：{time_str}\n"
|
||
f"项目：{device.project_name or '未分组'}\n"
|
||
f"设备类型：{self._fmt_device_type(device.device_type)}\n"
|
||
f"IP地址：{device.ip}\n"
|
||
f"离线时长：{duration_str}"
|
||
)
|
||
|
||
def _fmt_device_type(self, dtype) -> str:
|
||
mapping = {
|
||
"server": "服务器",
|
||
"olt": "OLT",
|
||
"switch": "交换机",
|
||
"firewall": "防火墙",
|
||
"other": "其他设备",
|
||
}
|
||
return mapping.get(str(dtype), str(dtype))
|
||
|
||
# ---------- 企业微信推送 ----------
|
||
|
||
async def _get_wecom_token(self) -> Optional[str]:
|
||
"""获取企业微信 access_token"""
|
||
url = (
|
||
f"https://qyapi.weixin.qq.com/cgi-bin/gettoken"
|
||
f"?corpid={settings.WECOM_CORP_ID}"
|
||
f"&corpsecret={settings.WECOM_APP_SECRET}"
|
||
)
|
||
async with httpx.AsyncClient(timeout=10) as client:
|
||
resp = await client.get(url)
|
||
data = resp.json()
|
||
if data.get("errcode") == 0:
|
||
return data["access_token"]
|
||
else:
|
||
logger.error(f"获取企业微信 token 失败: {data}")
|
||
return None
|
||
|
||
async def _send_wecom_message(self, content: str) -> bool:
|
||
"""发送企业微信应用消息"""
|
||
if not settings.WECOM_CORP_ID or not settings.WECOM_APP_SECRET:
|
||
logger.warning("企业微信未配置，跳过推送")
|
||
logger.info(f"[模拟推送] {content}")
|
||
return True # 开发模式
|
||
|
||
token = await self._get_wecom_token()
|
||
if not token:
|
||
return False
|
||
|
||
url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}"
|
||
payload = {
|
||
"touser": "@all",
|
||
"msgtype": "text",
|
||
"agentid": settings.WECOM_AGENT_ID,
|
||
"text": {"content": content},
|
||
"safe": 0,
|
||
}
|
||
|
||
async with httpx.AsyncClient(timeout=10) as client:
|
||
resp = await client.post(url, json=payload)
|
||
data = resp.json()
|
||
if data.get("errcode") != 0:
|
||
logger.error(f"发送企业微信消息失败: {data}")
|
||
return False
|
||
return True
|