PingWatch 网络设备离线监控系统
- FastAPI 后端 + Vue 3 前端 - Docker Compose 一键部署 - Casdoor OAuth 认证集成 - LogHive 集中式日志 - 设备批量 CSV 导入/导出 - WebSocket 实时状态推送 - 企业微信告警通知 - fping 高性能并发 Ping 检测 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,319 @@
|
||||
"""
|
||||
告警服务
|
||||
|
||||
职责：
|
||||
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
|
||||
@@ -0,0 +1,38 @@
|
||||
"""数据清理服务"""
|
||||
import logging
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from sqlalchemy import delete
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.config import settings
|
||||
from app.models.ping_record import PingRecord
|
||||
from app.models.alert_event import AlertEvent
|
||||
from app.core.deps import async_session
|
||||
|
||||
logger = logging.getLogger("pingwatch.cleanup")
|
||||
|
||||
|
||||
async def cleanup_old_data():
|
||||
"""清理超过保留期限的旧数据"""
|
||||
now = datetime.now()
|
||||
|
||||
# 清理 ping_records
|
||||
ping_cutoff = now - timedelta(days=settings.PING_RECORD_RETENTION_DAYS)
|
||||
async with async_session() as db:
|
||||
result = await db.execute(
|
||||
delete(PingRecord).where(PingRecord.created_at < ping_cutoff)
|
||||
)
|
||||
await db.commit()
|
||||
if result.rowcount > 0:
|
||||
logger.info(f"清理 ping_records: {result.rowcount} 条")
|
||||
|
||||
# 清理 alert_events
|
||||
alert_cutoff = now - timedelta(days=settings.ALERT_RETENTION_DAYS)
|
||||
async with async_session() as db:
|
||||
result = await db.execute(
|
||||
delete(AlertEvent).where(AlertEvent.created_at < alert_cutoff)
|
||||
)
|
||||
await db.commit()
|
||||
if result.rowcount > 0:
|
||||
logger.info(f"清理 alert_events: {result.rowcount} 条")
|
||||
@@ -0,0 +1,242 @@
|
||||
"""
|
||||
异步 Ping 引擎
|
||||
|
||||
核心逻辑：
|
||||
1. 每轮从数据库加载所有启用设备，批量 fping
|
||||
2. 记录每台设备本轮 ping 结果（存活/响应时间）
|
||||
3. 状态机管理设备状态，判断是否从 online→offline 或 offline→online
|
||||
4. 结果通过回调或队列通知 alerter
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import subprocess
|
||||
import time
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from typing import Optional, Callable, Awaitable
|
||||
from collections import defaultdict
|
||||
|
||||
from sqlalchemy import select, update
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.config import settings
|
||||
from app.models.device import Device, DeviceTypeEnum
|
||||
from app.models.ping_record import PingRecord
|
||||
|
||||
logger = logging.getLogger("pingwatch.pinger")
|
||||
|
||||
|
||||
class PingResult:
|
||||
"""单台设备一轮 ping 的结果"""
|
||||
def __init__(self, device_id: int, is_alive: bool, response_time_ms: Optional[float] = None):
|
||||
self.device_id = device_id
|
||||
self.is_alive = is_alive
|
||||
self.response_time_ms = response_time_ms
|
||||
|
||||
|
||||
class DeviceStateChange:
|
||||
"""设备状态变化事件"""
|
||||
def __init__(self, device: Device, old_status: str, new_status: str, consecutive_failures: int):
|
||||
self.device = device
|
||||
self.old_status = old_status
|
||||
self.new_status = new_status
|
||||
self.consecutive_failures = consecutive_failures
|
||||
|
||||
|
||||
class Pinger:
|
||||
"""
|
||||
Ping 引擎，使用 fping 批量并发检测。
|
||||
对所有设备进 ping，返回存活状态和响应时间。
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self._round_num = 0
|
||||
self._on_state_change: Optional[Callable[[DeviceStateChange], Awaitable[None]]] = None
|
||||
|
||||
def on_state_change(self, callback: Callable[[DeviceStateChange], Awaitable[None]]):
|
||||
"""注册状态变化回调"""
|
||||
self._on_state_change = callback
|
||||
|
||||
async def run_one_round(self, db: AsyncSession) -> list[PingResult]:
|
||||
"""
|
||||
执行一轮 ping 检测：
|
||||
1. 加载所有启用设备
|
||||
2. 批量 fping
|
||||
3. 记录结果
|
||||
4. 更新设备状态
|
||||
"""
|
||||
self._round_num += 1
|
||||
round_num = self._round_num
|
||||
|
||||
# 1. 加载启用设备
|
||||
result = await db.execute(
|
||||
select(Device).where(Device.is_enabled == True)
|
||||
)
|
||||
devices = list(result.scalars().all())
|
||||
|
||||
if not devices:
|
||||
logger.info(f"[Round {round_num}] 没有启用的设备")
|
||||
return []
|
||||
|
||||
logger.info(f"[Round {round_num}] 开始检测 {len(devices)} 台设备")
|
||||
|
||||
# 2. 批量 ping
|
||||
start_time = time.time()
|
||||
ip_to_device = {d.ip: d for d in devices}
|
||||
ip_list = list(ip_to_device.keys())
|
||||
|
||||
ping_results_map = await self._batch_ping(ip_list)
|
||||
|
||||
# 3. 构造结果
|
||||
results: list[PingResult] = []
|
||||
for ip, dev in ip_to_device.items():
|
||||
is_alive, rtt = ping_results_map.get(ip, (False, None))
|
||||
results.append(PingResult(device_id=dev.id, is_alive=is_alive, response_time_ms=rtt))
|
||||
|
||||
elapsed = time.time() - start_time
|
||||
alive_count = sum(1 for r in results if r.is_alive)
|
||||
|
||||
# 4. 批量写入 ping_records
|
||||
now = datetime.now()
|
||||
records = [
|
||||
PingRecord(
|
||||
device_id=r.device_id,
|
||||
is_alive=r.is_alive,
|
||||
response_time_ms=r.response_time_ms,
|
||||
round_num=round_num,
|
||||
created_at=now,
|
||||
)
|
||||
for r in results
|
||||
]
|
||||
db.add_all(records)
|
||||
await db.flush()
|
||||
|
||||
# 5. 更新设备状态（状态机）
|
||||
device_map = {d.id: d for d in devices}
|
||||
for r in results:
|
||||
dev = device_map.get(r.device_id)
|
||||
if not dev:
|
||||
continue
|
||||
|
||||
old_status = dev.current_status
|
||||
if r.is_alive:
|
||||
dev.consecutive_failures = 0
|
||||
dev.last_ping_time = now
|
||||
dev.last_online_time = now
|
||||
dev.current_status = "online"
|
||||
else:
|
||||
dev.consecutive_failures = (dev.consecutive_failures or 0) + 1
|
||||
dev.last_ping_time = now
|
||||
if dev.consecutive_failures >= dev.alert_threshold:
|
||||
if dev.current_status != "offline":
|
||||
dev.current_status = "offline"
|
||||
dev.last_offline_time = now
|
||||
else:
|
||||
if dev.current_status == "online":
|
||||
dev.current_status = "checking"
|
||||
|
||||
# 状态变化回调
|
||||
if old_status != dev.current_status and self._on_state_change:
|
||||
change = DeviceStateChange(
|
||||
device=dev,
|
||||
old_status=old_status,
|
||||
new_status=dev.current_status,
|
||||
consecutive_failures=dev.consecutive_failures,
|
||||
)
|
||||
await self._on_state_change(change)
|
||||
|
||||
await db.commit()
|
||||
|
||||
logger.info(
|
||||
f"[Round {round_num}] 完成: {alive_count}/{len(devices)} 在线, "
|
||||
f"耗时 {elapsed:.2f}s"
|
||||
)
|
||||
return results
|
||||
|
||||
async def _batch_ping(self, ip_list: list[str]) -> dict[str, tuple[bool, Optional[float]]]:
|
||||
"""
|
||||
使用 fping 批量 ping
|
||||
返回: { ip: (is_alive, response_time_ms) }
|
||||
"""
|
||||
if not ip_list:
|
||||
return {}
|
||||
|
||||
try:
|
||||
# fping 一次性 ping 多个 IP
|
||||
# -c 1: 每个 IP 发 1 个包
|
||||
# -t: 超时毫秒
|
||||
timeout_ms = int(settings.PING_TIMEOUT_SECONDS * 1000)
|
||||
cmd = [
|
||||
settings.FPING_PATH,
|
||||
"-c", "1",
|
||||
"-t", str(timeout_ms),
|
||||
"-e", # 显示响应时间
|
||||
] + ip_list
|
||||
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*cmd,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
)
|
||||
stdout, stderr = await proc.communicate()
|
||||
|
||||
result_map: dict[str, tuple[bool, Optional[float]]] = {}
|
||||
|
||||
# fping 标准输出逐行: "IP : xmt/rcv/%loss = 1/1/0%, rtt min/avg/max = 0.12/0.12/0.12"
|
||||
# 或 "IP : xmt/rcv/%loss = 1/0/100%"
|
||||
for line in stdout.decode("utf-8", errors="replace").splitlines():
|
||||
line = line.strip()
|
||||
if ":" not in line:
|
||||
continue
|
||||
ip = line.split(":")[0].strip()
|
||||
# 解析响应时间
|
||||
if "rtt" in line:
|
||||
try:
|
||||
# 提取 avg rtt
|
||||
rtt_part = line.split("rtt")[1]
|
||||
# 格式: min/avg/max = 0.12/0.12/0.12
|
||||
if "=" in rtt_part:
|
||||
avg_rtt_str = rtt_part.split("=")[1].strip().split("/")[1]
|
||||
rtt_ms = float(avg_rtt_str)
|
||||
else:
|
||||
rtt_ms = None
|
||||
except (IndexError, ValueError):
|
||||
rtt_ms = None
|
||||
result_map[ip] = (True, rtt_ms)
|
||||
else:
|
||||
result_map[ip] = (False, None)
|
||||
|
||||
return result_map
|
||||
|
||||
except FileNotFoundError:
|
||||
logger.warning("fping 未找到，回退到系统 ping (串行)")
|
||||
return await self._fallback_ping(ip_list)
|
||||
except Exception as e:
|
||||
logger.error(f"fping 异常: {e}")
|
||||
return await self._fallback_ping(ip_list)
|
||||
|
||||
async def _fallback_ping(self, ip_list: list[str]) -> dict[str, tuple[bool, Optional[float]]]:
|
||||
"""回退方案：使用系统 ping，并发执行"""
|
||||
async def ping_one(ip: str) -> tuple[str, bool, Optional[float]]:
|
||||
try:
|
||||
timeout = settings.PING_TIMEOUT_SECONDS
|
||||
cmd = ["ping", "-c", "1", "-W", str(int(timeout)), ip]
|
||||
start = time.time()
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
||||
)
|
||||
await proc.wait()
|
||||
elapsed = (time.time() - start) * 1000
|
||||
return ip, proc.returncode == 0, round(elapsed, 2)
|
||||
except Exception:
|
||||
return ip, False, None
|
||||
|
||||
tasks = [ping_one(ip) for ip in ip_list]
|
||||
sem = asyncio.Semaphore(settings.PING_CONCURRENCY)
|
||||
|
||||
async def bounded_ping(ip: str):
|
||||
async with sem:
|
||||
return await ping_one(ip)
|
||||
|
||||
results = await asyncio.gather(*[bounded_ping(ip) for ip in ip_list])
|
||||
return {ip: (alive, rtt) for ip, alive, rtt in results}
|
||||
@@ -0,0 +1,94 @@
|
||||
"""
|
||||
定时任务调度器
|
||||
|
||||
使用 asyncio 循环驱动 Ping 引擎，协调 Pinger 和 Alerter。
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.config import settings
|
||||
from app.services.pinger import Pinger
|
||||
from app.services.alerter import Alerter
|
||||
from app.core.deps import async_session
|
||||
|
||||
logger = logging.getLogger("pingwatch.scheduler")
|
||||
|
||||
|
||||
class PingScheduler:
|
||||
"""
|
||||
调度器职责：
|
||||
1. 按间隔驱动 Ping 引擎
|
||||
2. 每轮结束后触发 Alerter 处理待发送告警
|
||||
3. 控制并发和清理
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self._pinger = Pinger()
|
||||
self._alerter = Alerter()
|
||||
self._running = False
|
||||
self._task: asyncio.Task | None = None
|
||||
|
||||
# 注册状态变化回调
|
||||
self._pinger.on_state_change(self._on_state_change)
|
||||
|
||||
async def _on_state_change(self, change):
|
||||
"""收到设备状态变化，转给 alerter"""
|
||||
async with async_session() as db:
|
||||
try:
|
||||
await self._alerter.on_state_change(change, db)
|
||||
except Exception as e:
|
||||
logger.error(f"告警处理异常: {e}", exc_info=True)
|
||||
|
||||
async def _run_loop(self):
|
||||
"""主循环"""
|
||||
logger.info("Ping 调度器已启动")
|
||||
self._running = True
|
||||
|
||||
while self._running:
|
||||
try:
|
||||
async with async_session() as db:
|
||||
# 执行一轮 ping
|
||||
results = await self._pinger.run_one_round(db)
|
||||
|
||||
if results:
|
||||
# 直接用启用的设备数
|
||||
total = len(results)
|
||||
|
||||
# 处理待发送告警
|
||||
await self._alerter.flush_pending(db, total)
|
||||
|
||||
except asyncio.CancelledError:
|
||||
break
|
||||
except Exception as e:
|
||||
logger.error(f"调度器异常: {e}", exc_info=True)
|
||||
|
||||
# 等待下一轮
|
||||
await asyncio.sleep(settings.PING_INTERVAL_SECONDS)
|
||||
|
||||
logger.info("Ping 调度器已停止")
|
||||
|
||||
def start(self):
|
||||
"""启动调度器（后台任务）"""
|
||||
if self._running:
|
||||
logger.warning("调度器已在运行")
|
||||
return
|
||||
self._task = asyncio.create_task(self._run_loop())
|
||||
|
||||
async def stop(self):
|
||||
"""停止调度器"""
|
||||
self._running = False
|
||||
if self._task:
|
||||
self._task.cancel()
|
||||
try:
|
||||
await self._task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
self._task = None
|
||||
|
||||
|
||||
# 全局调度器实例
|
||||
scheduler = PingScheduler()
|
||||
Reference in New Issue
Block a user