feat: deliver alert events through persistent outbox
This commit is contained in:
+114
-297
@@ -1,319 +1,136 @@
|
||||
"""
|
||||
告警服务
|
||||
"""Create durable alert events and their notification outbox entries."""
|
||||
|
||||
职责：
|
||||
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 import select
|
||||
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.config import Settings, settings
|
||||
from app.models import (
|
||||
AlertEvent,
|
||||
AlertTypeEnum,
|
||||
NotificationOutbox,
|
||||
)
|
||||
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:
|
||||
"""
|
||||
告警处理器。
|
||||
"""Translate a persisted device transition into one atomic outbox event."""
|
||||
|
||||
核心逻辑：
|
||||
- 设备 offline → 收集到待发送队列
|
||||
- 每轮结束后检查：
|
||||
a) 上游心跳是否正常？
|
||||
b) 离线率是否 < 90%？
|
||||
c) 满足条件 → 聚合所有待发告警 → 一条企业微信消息
|
||||
d) 不满足 → 丢弃本轮告警，记录系统日志
|
||||
- 设备 recovered → 单独发送恢复通知
|
||||
"""
|
||||
def __init__(self, runtime_settings: Settings | None = None):
|
||||
self._settings = runtime_settings or settings
|
||||
|
||||
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())
|
||||
async def record_transition(
|
||||
self,
|
||||
change: DeviceStateChange,
|
||||
db: AsyncSession,
|
||||
) -> AlertEvent:
|
||||
"""Stage an event and notification without committing the caller session."""
|
||||
occurred_at = change.device.last_ping_time or datetime.now()
|
||||
event_type = AlertTypeEnum(change.event_type or change.new_status)
|
||||
prior_event = await self._find_prior_open_event(change, db)
|
||||
duration_minutes = self._close_prior_event(prior_event, occurred_at)
|
||||
event = AlertEvent(
|
||||
device_id=change.device.id,
|
||||
alert_type=event_type,
|
||||
message=change.reason,
|
||||
start_at=occurred_at,
|
||||
is_resolved=event_type == AlertTypeEnum.recovered,
|
||||
duration_minutes=(
|
||||
duration_minutes
|
||||
if event_type == AlertTypeEnum.recovered
|
||||
else None
|
||||
),
|
||||
related_event_id=(prior_event.id if prior_event is not None else None),
|
||||
previous_status=change.old_status,
|
||||
current_status=change.new_status,
|
||||
)
|
||||
db.add(event)
|
||||
await db.flush()
|
||||
|
||||
elif change.new_status == "online" and change.old_status == "offline":
|
||||
# 设备恢复，立即记录并发送恢复通知
|
||||
await self._handle_recovery(change.device, db)
|
||||
if prior_event is not None:
|
||||
prior_event.related_event_id = event.id
|
||||
|
||||
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(
|
||||
NotificationOutbox(
|
||||
alert_event_id=event.id,
|
||||
message_content=self._render_notification(
|
||||
change,
|
||||
occurred_at,
|
||||
duration_minutes,
|
||||
),
|
||||
delivery_scope=(
|
||||
self._settings.WECOM_TO_PARTY.strip() or "@all"
|
||||
),
|
||||
)
|
||||
# 记录系统告警
|
||||
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
|
||||
await db.flush()
|
||||
return event
|
||||
|
||||
@staticmethod
|
||||
async def _find_prior_open_event(
|
||||
change: DeviceStateChange,
|
||||
db: AsyncSession,
|
||||
) -> AlertEvent | None:
|
||||
"""Find only the incident that this escalation or recovery supersedes."""
|
||||
event_type = change.event_type or change.new_status
|
||||
if event_type == AlertTypeEnum.offline.value and change.old_status == "degraded":
|
||||
alert_types = (AlertTypeEnum.degraded,)
|
||||
elif event_type == AlertTypeEnum.recovered.value:
|
||||
alert_types = (AlertTypeEnum.offline, AlertTypeEnum.degraded)
|
||||
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 # 开发模式
|
||||
return await db.scalar(
|
||||
select(AlertEvent)
|
||||
.where(AlertEvent.device_id == change.device.id)
|
||||
.where(AlertEvent.alert_type.in_(alert_types))
|
||||
.where(AlertEvent.is_resolved.is_(False))
|
||||
.order_by(AlertEvent.start_at.desc(), AlertEvent.id.desc())
|
||||
.limit(1)
|
||||
)
|
||||
|
||||
token = await self._get_wecom_token()
|
||||
if not token:
|
||||
return False
|
||||
@staticmethod
|
||||
def _close_prior_event(
|
||||
event: AlertEvent | None,
|
||||
occurred_at: datetime,
|
||||
) -> int | None:
|
||||
"""Resolve one incident and return a non-negative minute duration."""
|
||||
if event is None:
|
||||
return None
|
||||
duration_minutes = max(
|
||||
0,
|
||||
int((occurred_at - event.start_at).total_seconds() / 60),
|
||||
)
|
||||
event.end_at = occurred_at
|
||||
event.duration_minutes = duration_minutes
|
||||
event.is_resolved = True
|
||||
return duration_minutes
|
||||
|
||||
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,
|
||||
@staticmethod
|
||||
def _render_notification(
|
||||
change: DeviceStateChange,
|
||||
occurred_at: datetime,
|
||||
duration_minutes: int | None = None,
|
||||
) -> str:
|
||||
"""Render the bounded device and transition details needed by operators."""
|
||||
labels = {
|
||||
"offline": "设备离线",
|
||||
"degraded": "链路质量故障",
|
||||
"recovered": "设备恢复",
|
||||
"system": "监控系统异常",
|
||||
}
|
||||
|
||||
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
|
||||
event_type = change.event_type or change.new_status
|
||||
device = change.device
|
||||
lines = [
|
||||
f"【{labels.get(event_type, event_type)}】",
|
||||
f"设备：{device.name}",
|
||||
f"IP：{device.ip}",
|
||||
f"位置：{device.location or '未配置'}",
|
||||
f"项目：{device.project_name or '未分组'}",
|
||||
f"丢包摘要：{change.reason or '无'}",
|
||||
f"时间：{occurred_at:%Y-%m-%d %H:%M:%S}",
|
||||
]
|
||||
if duration_minutes is not None:
|
||||
lines.append(f"持续时间：{duration_minutes} 分钟")
|
||||
return "\n".join(lines)
|
||||
|
||||
@@ -0,0 +1,322 @@
|
||||
"""Deliver pending notifications from the durable outbox."""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Callable, Protocol
|
||||
|
||||
import httpx
|
||||
from sqlalchemy import or_, select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.config import Settings, settings
|
||||
from app.models import AlertEvent, NotificationOutbox, NotificationStatus
|
||||
|
||||
|
||||
logger = logging.getLogger("pingwatch.notification_dispatcher")
|
||||
|
||||
# A small count caps one WeCom text request while still aggregating alert bursts.
|
||||
MAX_EVENTS_PER_MESSAGE = 5
|
||||
_MESSAGE_SEPARATOR = "\n\n---\n\n"
|
||||
_TOKEN_EXPIRY_MARGIN_SECONDS = 60
|
||||
_HTTP_TIMEOUT = httpx.Timeout(connect=5.0, read=10.0, write=10.0, pool=5.0)
|
||||
_TOKEN_URL = "https://qyapi.weixin.qq.com/cgi-bin/gettoken"
|
||||
_MESSAGE_URL = "https://qyapi.weixin.qq.com/cgi-bin/message/send"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class DeliveryResult:
|
||||
"""Sanitized delivery outcome returned by a notification transport."""
|
||||
|
||||
success: bool
|
||||
error: str | None = None
|
||||
retryable: bool = False
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class DispatchSummary:
|
||||
"""Counts produced by one bounded dispatcher pass."""
|
||||
|
||||
sent: int = 0
|
||||
rescheduled: int = 0
|
||||
failed: int = 0
|
||||
|
||||
|
||||
class TextNotificationClient(Protocol):
|
||||
"""Minimal transport boundary used by the persistent dispatcher."""
|
||||
|
||||
async def send_text(self, content: str) -> DeliveryResult:
|
||||
"""Deliver one text message without exposing credentials."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _TokenResult:
|
||||
token: str | None
|
||||
error: str | None = None
|
||||
retryable: bool = False
|
||||
|
||||
|
||||
class WeComClient:
|
||||
"""Minimal WeCom application client with an in-process token cache."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
runtime_settings: Settings | None = None,
|
||||
*,
|
||||
http_client: httpx.AsyncClient | None = None,
|
||||
now: Callable[[], datetime] | None = None,
|
||||
):
|
||||
self._settings = runtime_settings or settings
|
||||
self._http = http_client or httpx.AsyncClient(timeout=_HTTP_TIMEOUT)
|
||||
self._owns_http_client = http_client is None
|
||||
self._now = now or datetime.now
|
||||
self._access_token: str | None = None
|
||||
self._access_token_expires_at: datetime | None = None
|
||||
self._token_lock = asyncio.Lock()
|
||||
|
||||
async def aclose(self) -> None:
|
||||
"""Close only the HTTP client created by this service instance."""
|
||||
if self._owns_http_client:
|
||||
await self._http.aclose()
|
||||
|
||||
async def send_text(self, content: str) -> DeliveryResult:
|
||||
"""Deliver text while returning only sanitized failure evidence."""
|
||||
if not self._settings.wecom_notification_enabled:
|
||||
return DeliveryResult(False, "delivery disabled", False)
|
||||
|
||||
token_result = await self._get_access_token()
|
||||
if token_result.token is None:
|
||||
return DeliveryResult(
|
||||
False,
|
||||
token_result.error,
|
||||
token_result.retryable,
|
||||
)
|
||||
|
||||
payload = {
|
||||
"msgtype": "text",
|
||||
"agentid": self._settings.WECOM_AGENT_ID,
|
||||
"text": {"content": content},
|
||||
"safe": 0,
|
||||
}
|
||||
to_party = self._settings.WECOM_TO_PARTY.strip()
|
||||
if to_party:
|
||||
payload["toparty"] = to_party
|
||||
else:
|
||||
payload["touser"] = "@all"
|
||||
|
||||
try:
|
||||
response = await self._http.post(
|
||||
_MESSAGE_URL,
|
||||
params={"access_token": token_result.token},
|
||||
json=payload,
|
||||
)
|
||||
except (httpx.TimeoutException, httpx.TransportError):
|
||||
logger.warning("WeCom message transport failure")
|
||||
return DeliveryResult(False, "transport error", True)
|
||||
|
||||
if response.status_code == 429 or response.status_code >= 500:
|
||||
logger.warning(
|
||||
"WeCom message transient HTTP failure status=%s",
|
||||
response.status_code,
|
||||
)
|
||||
return DeliveryResult(False, f"HTTP {response.status_code}", True)
|
||||
if response.status_code >= 400:
|
||||
logger.warning(
|
||||
"WeCom message HTTP failure status=%s",
|
||||
response.status_code,
|
||||
)
|
||||
return DeliveryResult(False, f"HTTP {response.status_code}", False)
|
||||
|
||||
try:
|
||||
data = response.json()
|
||||
if not isinstance(data, dict):
|
||||
raise TypeError("response JSON must be an object")
|
||||
error_code = int(data.get("errcode", -1))
|
||||
except (TypeError, ValueError):
|
||||
logger.warning("WeCom message returned invalid JSON")
|
||||
return DeliveryResult(False, "invalid response", True)
|
||||
|
||||
if error_code == 0:
|
||||
return DeliveryResult(True)
|
||||
|
||||
if error_code in {40014, 42001}:
|
||||
self._access_token = None
|
||||
self._access_token_expires_at = None
|
||||
retryable = error_code in {-1, 40014, 42001, 45009}
|
||||
logger.warning("WeCom message rejected errcode=%s", error_code)
|
||||
return DeliveryResult(
|
||||
False,
|
||||
f"WeCom errcode {error_code}",
|
||||
retryable,
|
||||
)
|
||||
|
||||
async def _get_access_token(self) -> _TokenResult:
|
||||
"""Return a cached token or refresh it once under a process-local lock."""
|
||||
if self._token_is_valid():
|
||||
return _TokenResult(self._access_token)
|
||||
|
||||
async with self._token_lock:
|
||||
if self._token_is_valid():
|
||||
return _TokenResult(self._access_token)
|
||||
try:
|
||||
response = await self._http.get(
|
||||
_TOKEN_URL,
|
||||
params={
|
||||
"corpid": self._settings.WECOM_CORP_ID,
|
||||
"corpsecret": self._settings.WECOM_APP_SECRET,
|
||||
},
|
||||
)
|
||||
except (httpx.TimeoutException, httpx.TransportError):
|
||||
logger.warning("WeCom token transport failure")
|
||||
return _TokenResult(None, "token transport error", True)
|
||||
|
||||
if response.status_code == 429 or response.status_code >= 500:
|
||||
logger.warning(
|
||||
"WeCom token transient HTTP failure status=%s",
|
||||
response.status_code,
|
||||
)
|
||||
return _TokenResult(None, f"token HTTP {response.status_code}", True)
|
||||
if response.status_code >= 400:
|
||||
logger.warning(
|
||||
"WeCom token HTTP failure status=%s",
|
||||
response.status_code,
|
||||
)
|
||||
return _TokenResult(None, f"token HTTP {response.status_code}", False)
|
||||
|
||||
try:
|
||||
data = response.json()
|
||||
if not isinstance(data, dict):
|
||||
raise TypeError("response JSON must be an object")
|
||||
error_code = int(data.get("errcode", -1))
|
||||
access_token = data.get("access_token")
|
||||
expires_in = int(data.get("expires_in", 0))
|
||||
except (TypeError, ValueError):
|
||||
logger.warning("WeCom token endpoint returned invalid JSON")
|
||||
return _TokenResult(None, "invalid token response", True)
|
||||
|
||||
if error_code != 0 or not access_token or expires_in <= 0:
|
||||
logger.warning("WeCom token request rejected errcode=%s", error_code)
|
||||
return _TokenResult(
|
||||
None,
|
||||
f"WeCom token errcode {error_code}",
|
||||
error_code == -1,
|
||||
)
|
||||
|
||||
self._access_token = str(access_token)
|
||||
cache_seconds = max(0, expires_in - _TOKEN_EXPIRY_MARGIN_SECONDS)
|
||||
self._access_token_expires_at = self._now() + timedelta(
|
||||
seconds=cache_seconds
|
||||
)
|
||||
return _TokenResult(self._access_token)
|
||||
|
||||
def _token_is_valid(self) -> bool:
|
||||
return (
|
||||
self._access_token is not None
|
||||
and self._access_token_expires_at is not None
|
||||
and self._now() < self._access_token_expires_at
|
||||
)
|
||||
|
||||
|
||||
class NotificationDispatcher:
|
||||
"""Claim due outbox rows and persist their delivery outcome."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
client: TextNotificationClient,
|
||||
runtime_settings: Settings | None = None,
|
||||
):
|
||||
self._client = client
|
||||
self._settings = runtime_settings or settings
|
||||
|
||||
async def dispatch_due(
|
||||
self,
|
||||
db: AsyncSession,
|
||||
now: datetime,
|
||||
) -> DispatchSummary:
|
||||
"""Attempt each currently due message once without committing the session."""
|
||||
rows = list(
|
||||
(
|
||||
await db.execute(
|
||||
select(NotificationOutbox)
|
||||
.where(NotificationOutbox.status == NotificationStatus.pending)
|
||||
.where(
|
||||
or_(
|
||||
NotificationOutbox.next_attempt_at.is_(None),
|
||||
NotificationOutbox.next_attempt_at <= now,
|
||||
)
|
||||
)
|
||||
.order_by(NotificationOutbox.created_at, NotificationOutbox.id)
|
||||
)
|
||||
)
|
||||
.scalars()
|
||||
.all()
|
||||
)
|
||||
|
||||
event_ids = [row.alert_event_id for row in rows]
|
||||
events = {}
|
||||
if event_ids:
|
||||
events = {
|
||||
event.id: event
|
||||
for event in (
|
||||
(
|
||||
await db.execute(
|
||||
select(AlertEvent).where(AlertEvent.id.in_(event_ids))
|
||||
)
|
||||
)
|
||||
.scalars()
|
||||
.all()
|
||||
)
|
||||
}
|
||||
|
||||
sent = rescheduled = failed = 0
|
||||
for start in range(0, len(rows), MAX_EVENTS_PER_MESSAGE):
|
||||
batch = rows[start : start + MAX_EVENTS_PER_MESSAGE]
|
||||
for row in batch:
|
||||
row.status = NotificationStatus.sending
|
||||
row.locked_at = now
|
||||
try:
|
||||
result = await self._client.send_text(
|
||||
_MESSAGE_SEPARATOR.join(row.message_content for row in batch)
|
||||
)
|
||||
except Exception as exc: # Transport implementations must not strand rows.
|
||||
logger.warning(
|
||||
"Notification client raised type=%s",
|
||||
type(exc).__name__,
|
||||
)
|
||||
result = DeliveryResult(False, "notification client error", True)
|
||||
|
||||
for row in batch:
|
||||
row.attempt_count += 1
|
||||
event = events.get(row.alert_event_id)
|
||||
if result.success:
|
||||
row.status = NotificationStatus.sent
|
||||
row.sent_at = now
|
||||
row.next_attempt_at = None
|
||||
row.last_error = None
|
||||
sent += 1
|
||||
elif (
|
||||
result.retryable
|
||||
and row.attempt_count
|
||||
< self._settings.wecom_notification_max_attempts
|
||||
):
|
||||
delay = self._settings.wecom_retry_base_seconds * (
|
||||
2 ** (row.attempt_count - 1)
|
||||
)
|
||||
row.status = NotificationStatus.pending
|
||||
row.next_attempt_at = now + timedelta(seconds=delay)
|
||||
row.last_error = (result.error or "delivery failed")[:512]
|
||||
rescheduled += 1
|
||||
else:
|
||||
row.status = NotificationStatus.failed
|
||||
row.next_attempt_at = None
|
||||
row.last_error = (result.error or "delivery failed")[:512]
|
||||
failed += 1
|
||||
|
||||
if event is not None:
|
||||
event.notification_sent = result.success
|
||||
event.notification_attempts = row.attempt_count
|
||||
event.last_notification_error = row.last_error
|
||||
|
||||
await db.flush()
|
||||
return DispatchSummary(sent=sent, rescheduled=rescheduled, failed=failed)
|
||||
@@ -5,7 +5,6 @@ import time
|
||||
from collections import defaultdict
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from typing import Awaitable, Callable
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
@@ -38,16 +37,6 @@ class Pinger:
|
||||
def __init__(self, runtime_settings: Settings | None = None):
|
||||
self._settings = runtime_settings or settings
|
||||
self._round_num = 0
|
||||
self._on_state_change: (
|
||||
Callable[[DeviceStateChange], Awaitable[None]] | None
|
||||
) = None
|
||||
|
||||
def on_state_change(
|
||||
self,
|
||||
callback: Callable[[DeviceStateChange], Awaitable[None]],
|
||||
) -> None:
|
||||
"""Register the legacy transition callback during scheduler migration."""
|
||||
self._on_state_change = callback
|
||||
|
||||
async def run_one_round(
|
||||
self,
|
||||
@@ -128,11 +117,9 @@ class Pinger:
|
||||
if change is not None:
|
||||
changes.append(change)
|
||||
|
||||
await db.commit()
|
||||
|
||||
if self._on_state_change is not None:
|
||||
for change in changes:
|
||||
await self._on_state_change(change)
|
||||
# The scheduler records transition events and commits the complete
|
||||
# probe/state/outbox unit. A direct caller retains the same ownership.
|
||||
await db.flush()
|
||||
|
||||
elapsed = time.monotonic() - started_at
|
||||
logger.info(
|
||||
|
||||
@@ -1,94 +1,142 @@
|
||||
"""
|
||||
定时任务调度器
|
||||
|
||||
使用 asyncio 循环驱动 Ping 引擎，协调 Pinger 和 Alerter。
|
||||
"""
|
||||
"""Single-task scheduler for probe, event recording, and outbox delivery."""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import time
|
||||
from datetime import datetime
|
||||
from typing import Callable
|
||||
|
||||
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.config import Settings, settings
|
||||
from app.core.deps import async_session
|
||||
from app.services.alerter import Alerter
|
||||
from app.services.notification_dispatcher import (
|
||||
NotificationDispatcher,
|
||||
WeComClient,
|
||||
)
|
||||
from app.services.pinger import Pinger
|
||||
|
||||
|
||||
logger = logging.getLogger("pingwatch.scheduler")
|
||||
_DEFAULT_DISPATCH_INTERVAL_SECONDS = 5.0
|
||||
|
||||
|
||||
class PingScheduler:
|
||||
"""
|
||||
调度器职责：
|
||||
1. 按间隔驱动 Ping 引擎
|
||||
2. 每轮结束后触发 Alerter 处理待发送告警
|
||||
3. 控制并发和清理
|
||||
"""
|
||||
"""Serialize monitoring cycles and retain one cancellable background task."""
|
||||
|
||||
def __init__(self):
|
||||
self._pinger = Pinger()
|
||||
self._alerter = Alerter()
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
pinger=None,
|
||||
alerter=None,
|
||||
dispatcher=None,
|
||||
session_factory: Callable = async_session,
|
||||
runtime_settings: Settings | None = None,
|
||||
interval_seconds: float | None = None,
|
||||
dispatch_interval_seconds: float = _DEFAULT_DISPATCH_INTERVAL_SECONDS,
|
||||
):
|
||||
self._settings = runtime_settings or settings
|
||||
self._pinger = pinger or Pinger(self._settings)
|
||||
self._alerter = alerter or Alerter(self._settings)
|
||||
self._notification_client = None
|
||||
self._dispatch_enabled = (
|
||||
dispatcher is not None or self._settings.wecom_notification_enabled
|
||||
)
|
||||
if dispatcher is None:
|
||||
self._notification_client = WeComClient(self._settings)
|
||||
dispatcher = NotificationDispatcher(
|
||||
self._notification_client,
|
||||
self._settings,
|
||||
)
|
||||
self._dispatcher = dispatcher
|
||||
self._session_factory = session_factory
|
||||
self._probe_interval_seconds = max(
|
||||
1.0,
|
||||
float(
|
||||
interval_seconds
|
||||
if interval_seconds is not None
|
||||
else self._settings.PING_INTERVAL_SECONDS
|
||||
),
|
||||
)
|
||||
self._dispatch_interval_seconds = min(
|
||||
max(1.0, float(dispatch_interval_seconds)),
|
||||
30.0,
|
||||
)
|
||||
self._running = False
|
||||
self._task: asyncio.Task | None = None
|
||||
self._cycle_lock = asyncio.Lock()
|
||||
|
||||
# 注册状态变化回调
|
||||
self._pinger.on_state_change(self._on_state_change)
|
||||
|
||||
async def _on_state_change(self, change):
|
||||
"""收到设备状态变化，转给 alerter"""
|
||||
async with async_session() as db:
|
||||
async def _run_cycle(self, *, run_probe: bool = True) -> None:
|
||||
"""Run one serialized transaction cycle and close its session on cancel."""
|
||||
async with self._cycle_lock:
|
||||
async with self._session_factory() as db:
|
||||
try:
|
||||
await self._alerter.on_state_change(change, db)
|
||||
except Exception as e:
|
||||
logger.error(f"告警处理异常: {e}", exc_info=True)
|
||||
if run_probe:
|
||||
changes = await self._pinger.run_one_round(db)
|
||||
for change in changes:
|
||||
await self._alerter.record_transition(change, db)
|
||||
await db.commit()
|
||||
|
||||
async def _run_loop(self):
|
||||
"""主循环"""
|
||||
logger.info("Ping 调度器已启动")
|
||||
self._running = True
|
||||
if self._dispatch_enabled:
|
||||
await self._dispatcher.dispatch_due(db, datetime.now())
|
||||
await db.commit()
|
||||
except BaseException:
|
||||
await db.rollback()
|
||||
raise
|
||||
|
||||
async def _run_loop(self) -> None:
|
||||
"""Probe at its configured cadence and service retries every few seconds."""
|
||||
logger.info("Ping scheduler started")
|
||||
next_probe_at = 0.0
|
||||
while self._running:
|
||||
now = time.monotonic()
|
||||
run_probe = now >= next_probe_at
|
||||
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)
|
||||
|
||||
await self._run_cycle(run_probe=run_probe)
|
||||
if run_probe:
|
||||
next_probe_at = time.monotonic() + self._probe_interval_seconds
|
||||
except asyncio.CancelledError:
|
||||
break
|
||||
except Exception as e:
|
||||
logger.error(f"调度器异常: {e}", exc_info=True)
|
||||
except Exception as exc:
|
||||
logger.error(
|
||||
"Scheduler cycle failed type=%s",
|
||||
type(exc).__name__,
|
||||
exc_info=True,
|
||||
)
|
||||
if run_probe:
|
||||
next_probe_at = time.monotonic() + self._probe_interval_seconds
|
||||
|
||||
# 等待下一轮
|
||||
await asyncio.sleep(settings.PING_INTERVAL_SECONDS)
|
||||
seconds_until_probe = max(0.0, next_probe_at - time.monotonic())
|
||||
sleep_seconds = min(
|
||||
self._dispatch_interval_seconds,
|
||||
seconds_until_probe or self._dispatch_interval_seconds,
|
||||
)
|
||||
try:
|
||||
await asyncio.sleep(sleep_seconds)
|
||||
except asyncio.CancelledError:
|
||||
break
|
||||
logger.info("Ping scheduler stopped")
|
||||
|
||||
logger.info("Ping 调度器已停止")
|
||||
|
||||
def start(self):
|
||||
"""启动调度器（后台任务）"""
|
||||
if self._running:
|
||||
logger.warning("调度器已在运行")
|
||||
def start(self) -> None:
|
||||
"""Start exactly one background scheduler task."""
|
||||
if self._task is not None and not self._task.done():
|
||||
logger.warning("Ping scheduler is already running")
|
||||
return
|
||||
self._running = True
|
||||
self._task = asyncio.create_task(self._run_loop())
|
||||
|
||||
async def stop(self):
|
||||
"""停止调度器"""
|
||||
async def stop(self) -> None:
|
||||
"""Cancel and await the active cycle before releasing its HTTP client."""
|
||||
self._running = False
|
||||
if self._task:
|
||||
if self._task is not None:
|
||||
self._task.cancel()
|
||||
try:
|
||||
await self._task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
self._task = None
|
||||
if self._notification_client is not None:
|
||||
await self._notification_client.aclose()
|
||||
self._notification_client = None
|
||||
|
||||
|
||||
# 全局调度器实例
|
||||
scheduler = PingScheduler()
|
||||
|
||||
@@ -0,0 +1,427 @@
|
||||
"""Transactional alert-event and notification-outbox workflow tests."""
|
||||
|
||||
import asyncio
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import (
|
||||
AsyncSession,
|
||||
async_sessionmaker,
|
||||
create_async_engine,
|
||||
)
|
||||
|
||||
from app.config import Settings
|
||||
from app.models import (
|
||||
AlertEvent,
|
||||
AlertTypeEnum,
|
||||
Base,
|
||||
Device,
|
||||
DeviceTypeEnum,
|
||||
NotificationOutbox,
|
||||
NotificationStatus,
|
||||
PingRecord,
|
||||
)
|
||||
from app.services.alerter import Alerter
|
||||
from app.services.fping_runner import ProbeResult
|
||||
from app.services.pinger import DeviceStateChange, Pinger
|
||||
from app.services.scheduler import PingScheduler
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def db_session() -> AsyncSession:
|
||||
"""Provide a complete isolated persistence boundary for alert tests."""
|
||||
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
||||
try:
|
||||
async with engine.begin() as connection:
|
||||
await connection.run_sync(Base.metadata.create_all)
|
||||
async with AsyncSession(engine, expire_on_commit=False) as session:
|
||||
yield session
|
||||
finally:
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def offline_change(db_session: AsyncSession) -> DeviceStateChange:
|
||||
"""Persist a device and expose a real offline state-change value."""
|
||||
observed_at = datetime(2026, 8, 4, 9, 30)
|
||||
device = Device(
|
||||
name="核心交换机",
|
||||
ip="10.0.0.8",
|
||||
device_type=DeviceTypeEnum.switch,
|
||||
location="一楼机房",
|
||||
project_name="园区网",
|
||||
current_status="offline",
|
||||
last_ping_time=observed_at,
|
||||
)
|
||||
db_session.add(device)
|
||||
await db_session.commit()
|
||||
return DeviceStateChange(
|
||||
device=device,
|
||||
old_status="online",
|
||||
new_status="offline",
|
||||
consecutive_failures=2,
|
||||
event_type="offline",
|
||||
reason="2 consecutive full-loss rounds",
|
||||
)
|
||||
|
||||
|
||||
async def test_transition_creates_event_and_pending_outbox_in_one_transaction(
|
||||
db_session: AsyncSession,
|
||||
offline_change: DeviceStateChange,
|
||||
):
|
||||
"""Removing either insert breaks the durable event-delivery contract."""
|
||||
event = await Alerter().record_transition(offline_change, db_session)
|
||||
|
||||
outbox = await db_session.scalar(
|
||||
select(NotificationOutbox).where(
|
||||
NotificationOutbox.alert_event_id == event.id
|
||||
)
|
||||
)
|
||||
|
||||
assert outbox is not None
|
||||
assert outbox.status == NotificationStatus.pending
|
||||
assert "核心交换机" in outbox.message_content
|
||||
assert "10.0.0.8" in outbox.message_content
|
||||
assert "一楼机房" in outbox.message_content
|
||||
assert "园区网" in outbox.message_content
|
||||
assert "2 consecutive full-loss rounds" in outbox.message_content
|
||||
assert "2026-08-04 09:30:00" in outbox.message_content
|
||||
|
||||
|
||||
async def test_event_and_outbox_rollback_together(
|
||||
db_session: AsyncSession,
|
||||
offline_change: DeviceStateChange,
|
||||
):
|
||||
"""A caller rollback cannot retain an event without its notification."""
|
||||
await Alerter().record_transition(offline_change, db_session)
|
||||
|
||||
await db_session.rollback()
|
||||
|
||||
assert await db_session.scalar(select(AlertEvent)) is None
|
||||
assert await db_session.scalar(select(NotificationOutbox)) is None
|
||||
|
||||
|
||||
async def test_offline_escalation_closes_and_links_open_degraded_event(
|
||||
db_session: AsyncSession,
|
||||
offline_change: DeviceStateChange,
|
||||
):
|
||||
"""Escalating degraded to offline must not leave two open incidents."""
|
||||
device = offline_change.device
|
||||
device.last_ping_time = datetime(2026, 8, 4, 9, 20)
|
||||
degraded = await Alerter().record_transition(
|
||||
DeviceStateChange(
|
||||
device=device,
|
||||
old_status="online",
|
||||
new_status="degraded",
|
||||
consecutive_failures=0,
|
||||
event_type="degraded",
|
||||
reason="window packet loss 20.00%",
|
||||
),
|
||||
db_session,
|
||||
)
|
||||
await db_session.commit()
|
||||
device.last_ping_time = datetime(2026, 8, 4, 9, 30)
|
||||
|
||||
offline = await Alerter().record_transition(
|
||||
DeviceStateChange(
|
||||
device=device,
|
||||
old_status="degraded",
|
||||
new_status="offline",
|
||||
consecutive_failures=2,
|
||||
event_type="offline",
|
||||
reason="2 consecutive full-loss rounds",
|
||||
),
|
||||
db_session,
|
||||
)
|
||||
|
||||
assert degraded.is_resolved is True
|
||||
assert degraded.end_at == datetime(2026, 8, 4, 9, 30)
|
||||
assert degraded.duration_minutes == 10
|
||||
assert degraded.related_event_id == offline.id
|
||||
assert offline.related_event_id == degraded.id
|
||||
|
||||
|
||||
async def test_recovery_closes_open_fault_and_notifies_with_duration(
|
||||
db_session: AsyncSession,
|
||||
offline_change: DeviceStateChange,
|
||||
):
|
||||
"""Recovery closes one active incident and carries its duration to operators."""
|
||||
opened = await Alerter().record_transition(offline_change, db_session)
|
||||
await db_session.commit()
|
||||
device = offline_change.device
|
||||
device.last_ping_time = datetime(2026, 8, 4, 10, 1)
|
||||
|
||||
recovered = await Alerter().record_transition(
|
||||
DeviceStateChange(
|
||||
device=device,
|
||||
old_status="offline",
|
||||
new_status="online",
|
||||
consecutive_failures=0,
|
||||
event_type="recovered",
|
||||
reason="3 consecutive clean rounds",
|
||||
),
|
||||
db_session,
|
||||
)
|
||||
outbox = await db_session.scalar(
|
||||
select(NotificationOutbox).where(
|
||||
NotificationOutbox.alert_event_id == recovered.id
|
||||
)
|
||||
)
|
||||
|
||||
assert opened.is_resolved is True
|
||||
assert opened.end_at == datetime(2026, 8, 4, 10, 1)
|
||||
assert opened.duration_minutes == 31
|
||||
assert recovered.alert_type == AlertTypeEnum.recovered
|
||||
assert recovered.is_resolved is True
|
||||
assert recovered.duration_minutes == 31
|
||||
assert recovered.related_event_id == opened.id
|
||||
assert "持续时间：31 分钟" in outbox.message_content
|
||||
|
||||
|
||||
async def test_probe_state_event_and_outbox_share_the_caller_transaction(
|
||||
monkeypatch,
|
||||
):
|
||||
"""A failed caller commit cannot persist a state transition without its event."""
|
||||
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
||||
observed_at = datetime.now() - timedelta(seconds=30)
|
||||
try:
|
||||
async with engine.begin() as connection:
|
||||
await connection.run_sync(Base.metadata.create_all)
|
||||
async with AsyncSession(engine, expire_on_commit=False) as setup:
|
||||
device = Device(
|
||||
name="edge-atomic",
|
||||
ip="10.0.0.18",
|
||||
device_type=DeviceTypeEnum.switch,
|
||||
current_status="online",
|
||||
is_enabled=True,
|
||||
offline_consecutive_rounds=2,
|
||||
)
|
||||
setup.add(device)
|
||||
await setup.flush()
|
||||
device_id = device.id
|
||||
setup.add(
|
||||
PingRecord(
|
||||
device_id=device_id,
|
||||
is_alive=False,
|
||||
response_time_ms=None,
|
||||
round_num=1,
|
||||
sent_count=3,
|
||||
received_count=0,
|
||||
packet_loss_percent=100.0,
|
||||
average_rtt_ms=None,
|
||||
is_valid=True,
|
||||
created_at=observed_at,
|
||||
)
|
||||
)
|
||||
await setup.commit()
|
||||
|
||||
async def fake_run_fping_count(*args, **kwargs):
|
||||
return {
|
||||
"10.0.0.18": ProbeResult(
|
||||
"10.0.0.18",
|
||||
3,
|
||||
0,
|
||||
None,
|
||||
True,
|
||||
)
|
||||
}
|
||||
|
||||
monkeypatch.setattr(
|
||||
"app.services.pinger.run_fping_count",
|
||||
fake_run_fping_count,
|
||||
)
|
||||
async with AsyncSession(engine, expire_on_commit=False) as session:
|
||||
changes = await Pinger().run_one_round(session)
|
||||
await Alerter().record_transition(changes[0], session)
|
||||
await session.rollback()
|
||||
|
||||
async with AsyncSession(engine, expire_on_commit=False) as verification:
|
||||
persisted_device = await verification.get(Device, device_id)
|
||||
records = list(
|
||||
(
|
||||
await verification.execute(
|
||||
select(PingRecord).where(PingRecord.device_id == device_id)
|
||||
)
|
||||
)
|
||||
.scalars()
|
||||
.all()
|
||||
)
|
||||
assert persisted_device.current_status == "online"
|
||||
assert len(records) == 1
|
||||
assert await verification.scalar(select(AlertEvent)) is None
|
||||
assert await verification.scalar(select(NotificationOutbox)) is None
|
||||
finally:
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
class RecordingSession:
|
||||
"""Small session boundary exposing transaction and close ordering."""
|
||||
|
||||
def __init__(self, events: list[str]):
|
||||
self.events = events
|
||||
self.closed = asyncio.Event()
|
||||
|
||||
async def __aenter__(self):
|
||||
self.events.append("session-enter")
|
||||
return self
|
||||
|
||||
async def __aexit__(self, exc_type, exc_value, traceback):
|
||||
self.events.append("session-exit")
|
||||
self.closed.set()
|
||||
|
||||
async def commit(self):
|
||||
self.events.append("commit")
|
||||
|
||||
async def rollback(self):
|
||||
self.events.append("rollback")
|
||||
|
||||
|
||||
class RecordingPinger:
|
||||
def __init__(self, events: list[str], changes=None):
|
||||
self.events = events
|
||||
self.changes = list(changes or ["transition"])
|
||||
|
||||
async def run_one_round(self, db):
|
||||
self.events.append("probe")
|
||||
return self.changes
|
||||
|
||||
|
||||
class RecordingAlerter:
|
||||
def __init__(self, events: list[str]):
|
||||
self.events = events
|
||||
|
||||
async def record_transition(self, change, db):
|
||||
self.events.append(f"alert:{change}")
|
||||
|
||||
|
||||
class RecordingDispatcher:
|
||||
def __init__(self, events: list[str]):
|
||||
self.events = events
|
||||
|
||||
async def dispatch_due(self, db, now):
|
||||
assert isinstance(now, datetime)
|
||||
self.events.append("dispatch")
|
||||
|
||||
|
||||
async def test_scheduler_persists_transitions_before_dispatching_due_messages():
|
||||
"""Changing scheduler order cannot expose uncommitted outbox rows to dispatch."""
|
||||
events: list[str] = []
|
||||
session = RecordingSession(events)
|
||||
scheduler = PingScheduler(
|
||||
pinger=RecordingPinger(events),
|
||||
alerter=RecordingAlerter(events),
|
||||
dispatcher=RecordingDispatcher(events),
|
||||
session_factory=lambda: session,
|
||||
)
|
||||
|
||||
await scheduler._run_cycle()
|
||||
|
||||
assert events == [
|
||||
"session-enter",
|
||||
"probe",
|
||||
"alert:transition",
|
||||
"commit",
|
||||
"dispatch",
|
||||
"commit",
|
||||
"session-exit",
|
||||
]
|
||||
|
||||
|
||||
async def test_scheduler_lock_prevents_overlapping_cycles():
|
||||
"""Even concurrent triggers cannot overlap probe or notification sessions."""
|
||||
events: list[str] = []
|
||||
|
||||
class ConcurrencyPinger:
|
||||
active = 0
|
||||
maximum = 0
|
||||
|
||||
async def run_one_round(self, db):
|
||||
self.active += 1
|
||||
self.maximum = max(self.maximum, self.active)
|
||||
await asyncio.sleep(0.01)
|
||||
self.active -= 1
|
||||
return []
|
||||
|
||||
pinger = ConcurrencyPinger()
|
||||
scheduler = PingScheduler(
|
||||
pinger=pinger,
|
||||
alerter=RecordingAlerter(events),
|
||||
dispatcher=RecordingDispatcher(events),
|
||||
session_factory=lambda: RecordingSession(events),
|
||||
)
|
||||
|
||||
await asyncio.gather(scheduler._run_cycle(), scheduler._run_cycle())
|
||||
|
||||
assert pinger.maximum == 1
|
||||
|
||||
|
||||
async def test_scheduler_stop_waits_for_active_session_to_close():
|
||||
"""Cancellation cannot return while a probe session remains open."""
|
||||
events: list[str] = []
|
||||
session = RecordingSession(events)
|
||||
probe_started = asyncio.Event()
|
||||
never_complete = asyncio.Event()
|
||||
|
||||
class BlockingPinger:
|
||||
async def run_one_round(self, db):
|
||||
probe_started.set()
|
||||
await never_complete.wait()
|
||||
|
||||
scheduler = PingScheduler(
|
||||
pinger=BlockingPinger(),
|
||||
alerter=RecordingAlerter(events),
|
||||
dispatcher=RecordingDispatcher(events),
|
||||
session_factory=lambda: session,
|
||||
interval_seconds=1,
|
||||
)
|
||||
scheduler.start()
|
||||
first_task = scheduler._task
|
||||
scheduler.start()
|
||||
|
||||
await probe_started.wait()
|
||||
await scheduler.stop()
|
||||
|
||||
assert first_task is not None
|
||||
assert session.closed.is_set()
|
||||
assert scheduler._task is None
|
||||
assert events[-2:] == ["rollback", "session-exit"]
|
||||
|
||||
|
||||
async def test_scheduler_leaves_pending_outbox_untouched_when_delivery_disabled():
|
||||
"""Disabling WeCom preserves queued notifications for a later enablement."""
|
||||
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
||||
session_factory = async_sessionmaker(engine, expire_on_commit=False)
|
||||
|
||||
class EmptyPinger:
|
||||
async def run_one_round(self, db):
|
||||
return []
|
||||
|
||||
try:
|
||||
async with engine.begin() as connection:
|
||||
await connection.run_sync(Base.metadata.create_all)
|
||||
async with session_factory() as setup:
|
||||
setup.add(
|
||||
NotificationOutbox(
|
||||
alert_event_id=42,
|
||||
message_content="保留待发送事件",
|
||||
)
|
||||
)
|
||||
await setup.commit()
|
||||
|
||||
runtime_settings = Settings(wecom_notification_enabled=False)
|
||||
scheduler = PingScheduler(
|
||||
pinger=EmptyPinger(),
|
||||
alerter=Alerter(runtime_settings),
|
||||
session_factory=session_factory,
|
||||
runtime_settings=runtime_settings,
|
||||
)
|
||||
await scheduler._run_cycle()
|
||||
|
||||
async with session_factory() as verification:
|
||||
outbox = await verification.scalar(select(NotificationOutbox))
|
||||
assert outbox.status == NotificationStatus.pending
|
||||
assert outbox.attempt_count == 0
|
||||
await scheduler.stop()
|
||||
finally:
|
||||
await engine.dispose()
|
||||
@@ -0,0 +1,327 @@
|
||||
"""Persistent notification dispatch and retry behavior tests."""
|
||||
|
||||
import json
|
||||
import logging
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
|
||||
|
||||
from app.config import Settings
|
||||
from app.models import (
|
||||
AlertEvent,
|
||||
AlertTypeEnum,
|
||||
Base,
|
||||
NotificationOutbox,
|
||||
NotificationStatus,
|
||||
)
|
||||
from app.services.notification_dispatcher import (
|
||||
DeliveryResult,
|
||||
MAX_EVENTS_PER_MESSAGE,
|
||||
NotificationDispatcher,
|
||||
WeComClient,
|
||||
)
|
||||
|
||||
|
||||
class FakeWeComClient:
|
||||
"""Return one controlled outcome without any external network access."""
|
||||
|
||||
def __init__(self, *results: DeliveryResult):
|
||||
self._results = list(results)
|
||||
self.contents: list[str] = []
|
||||
|
||||
async def send_text(self, content: str) -> DeliveryResult:
|
||||
self.contents.append(content)
|
||||
return self._results.pop(0)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def db_session() -> AsyncSession:
|
||||
"""Provide an isolated outbox database."""
|
||||
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
||||
try:
|
||||
async with engine.begin() as connection:
|
||||
await connection.run_sync(Base.metadata.create_all)
|
||||
async with AsyncSession(engine, expire_on_commit=False) as session:
|
||||
yield session
|
||||
finally:
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
async def test_retryable_failure_reschedules(
|
||||
db_session: AsyncSession,
|
||||
):
|
||||
"""A transient failure must remain pending with exponential retry evidence."""
|
||||
frozen_time = datetime(2026, 8, 4, 10, 0)
|
||||
pending_message = NotificationOutbox(
|
||||
alert_event_id=1,
|
||||
message_content="设备离线",
|
||||
)
|
||||
db_session.add(pending_message)
|
||||
await db_session.commit()
|
||||
client = FakeWeComClient(DeliveryResult(False, "timeout", True))
|
||||
|
||||
await NotificationDispatcher(client).dispatch_due(
|
||||
db_session,
|
||||
frozen_time,
|
||||
)
|
||||
|
||||
assert pending_message.status == NotificationStatus.pending
|
||||
assert pending_message.attempt_count == 1
|
||||
assert pending_message.next_attempt_at == datetime(2026, 8, 4, 10, 0, 30)
|
||||
|
||||
|
||||
async def test_success_marks_outbox_and_event_delivered(
|
||||
db_session: AsyncSession,
|
||||
):
|
||||
"""A successful transport result updates both delivery evidence records."""
|
||||
now = datetime(2026, 8, 4, 10, 0)
|
||||
event = AlertEvent(
|
||||
device_id=7,
|
||||
alert_type=AlertTypeEnum.offline,
|
||||
message="full loss",
|
||||
start_at=now,
|
||||
is_resolved=False,
|
||||
)
|
||||
db_session.add(event)
|
||||
await db_session.flush()
|
||||
message = NotificationOutbox(
|
||||
alert_event_id=event.id,
|
||||
message_content="设备 7 离线",
|
||||
)
|
||||
db_session.add(message)
|
||||
await db_session.commit()
|
||||
|
||||
summary = await NotificationDispatcher(
|
||||
FakeWeComClient(DeliveryResult(True)),
|
||||
_enabled_settings(),
|
||||
).dispatch_due(db_session, now)
|
||||
|
||||
assert summary.sent == 1
|
||||
assert message.status == NotificationStatus.sent
|
||||
assert message.sent_at == now
|
||||
assert message.attempt_count == 1
|
||||
assert event.notification_sent is True
|
||||
assert event.notification_attempts == 1
|
||||
assert event.last_notification_error is None
|
||||
|
||||
|
||||
async def test_non_retryable_or_exhausted_failure_is_terminal(
|
||||
db_session: AsyncSession,
|
||||
):
|
||||
"""The configured attempt ceiling prevents an infinite retry loop."""
|
||||
now = datetime(2026, 8, 4, 10, 0)
|
||||
message = NotificationOutbox(
|
||||
alert_event_id=99,
|
||||
message_content="bad recipient",
|
||||
attempt_count=1,
|
||||
)
|
||||
db_session.add(message)
|
||||
await db_session.commit()
|
||||
runtime_settings = _enabled_settings(wecom_notification_max_attempts=2)
|
||||
|
||||
summary = await NotificationDispatcher(
|
||||
FakeWeComClient(DeliveryResult(False, "HTTP 503", True)),
|
||||
runtime_settings,
|
||||
).dispatch_due(db_session, now)
|
||||
|
||||
assert summary.failed == 1
|
||||
assert message.status == NotificationStatus.failed
|
||||
assert message.attempt_count == 2
|
||||
assert message.next_attempt_at is None
|
||||
|
||||
|
||||
async def test_only_due_messages_are_dispatched_in_fixed_safe_batches(
|
||||
db_session: AsyncSession,
|
||||
):
|
||||
"""Large bursts are split while future retries remain untouched."""
|
||||
now = datetime(2026, 8, 4, 10, 0)
|
||||
due = [
|
||||
NotificationOutbox(alert_event_id=index, message_content=f"事件 {index}")
|
||||
for index in range(MAX_EVENTS_PER_MESSAGE + 1)
|
||||
]
|
||||
future = NotificationOutbox(
|
||||
alert_event_id=100,
|
||||
message_content="未来重试",
|
||||
next_attempt_at=now + timedelta(minutes=1),
|
||||
)
|
||||
db_session.add_all([*due, future])
|
||||
await db_session.commit()
|
||||
client = FakeWeComClient(DeliveryResult(True), DeliveryResult(True))
|
||||
|
||||
summary = await NotificationDispatcher(
|
||||
client,
|
||||
_enabled_settings(),
|
||||
).dispatch_due(db_session, now)
|
||||
|
||||
assert summary.sent == MAX_EVENTS_PER_MESSAGE + 1
|
||||
assert len(client.contents) == 2
|
||||
assert future.status == NotificationStatus.pending
|
||||
assert future.attempt_count == 0
|
||||
|
||||
|
||||
def _enabled_settings(**overrides) -> Settings:
|
||||
values = {
|
||||
"wecom_notification_enabled": True,
|
||||
"WECOM_CORP_ID": "test-corp",
|
||||
"WECOM_AGENT_ID": 1000001,
|
||||
"WECOM_APP_SECRET": "test-secret",
|
||||
}
|
||||
values.update(overrides)
|
||||
return Settings(**values)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("to_party", "expected_recipient", "unexpected_key"),
|
||||
[("", ("touser", "@all"), "toparty"), ("2|3", ("toparty", "2|3"), "touser")],
|
||||
)
|
||||
async def test_wecom_uses_application_scope_unless_department_is_explicit(
|
||||
to_party,
|
||||
expected_recipient,
|
||||
unexpected_key,
|
||||
):
|
||||
"""Recipient payloads preserve the app visibility boundary by default."""
|
||||
payloads: list[dict] = []
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
if request.url.path.endswith("/gettoken"):
|
||||
return httpx.Response(
|
||||
200,
|
||||
json={"errcode": 0, "access_token": "token-a", "expires_in": 7200},
|
||||
)
|
||||
payloads.append(json.loads(request.content))
|
||||
return httpx.Response(200, json={"errcode": 0, "errmsg": "ok"})
|
||||
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http:
|
||||
client = WeComClient(
|
||||
_enabled_settings(WECOM_TO_PARTY=to_party),
|
||||
http_client=http,
|
||||
)
|
||||
result = await client.send_text("controlled test message")
|
||||
|
||||
key, value = expected_recipient
|
||||
assert result.success is True
|
||||
assert payloads[0][key] == value
|
||||
assert unexpected_key not in payloads[0]
|
||||
assert payloads[0]["agentid"] == 1000001
|
||||
|
||||
|
||||
async def test_wecom_caches_token_until_expiry_margin():
|
||||
"""Repeated messages do not request a new token before its safe expiry."""
|
||||
calls = {"token": 0, "message": 0}
|
||||
current_time = [datetime(2026, 8, 4, 10, 0)]
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
if request.url.path.endswith("/gettoken"):
|
||||
calls["token"] += 1
|
||||
return httpx.Response(
|
||||
200,
|
||||
json={"errcode": 0, "access_token": "token-a", "expires_in": 120},
|
||||
)
|
||||
calls["message"] += 1
|
||||
return httpx.Response(200, json={"errcode": 0, "errmsg": "ok"})
|
||||
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http:
|
||||
client = WeComClient(
|
||||
_enabled_settings(),
|
||||
http_client=http,
|
||||
now=lambda: current_time[0],
|
||||
)
|
||||
await client.send_text("first")
|
||||
current_time[0] += timedelta(seconds=30)
|
||||
await client.send_text("second")
|
||||
|
||||
assert calls == {"token": 1, "message": 2}
|
||||
|
||||
|
||||
async def test_wecom_refreshes_token_after_safety_margin():
|
||||
"""A token inside the safety margin cannot be reused for a new message."""
|
||||
token_calls = 0
|
||||
current_time = [datetime(2026, 8, 4, 10, 0)]
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
nonlocal token_calls
|
||||
if request.url.path.endswith("/gettoken"):
|
||||
token_calls += 1
|
||||
return httpx.Response(
|
||||
200,
|
||||
json={
|
||||
"errcode": 0,
|
||||
"access_token": f"token-{token_calls}",
|
||||
"expires_in": 120,
|
||||
},
|
||||
)
|
||||
return httpx.Response(200, json={"errcode": 0, "errmsg": "ok"})
|
||||
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http:
|
||||
client = WeComClient(
|
||||
_enabled_settings(),
|
||||
http_client=http,
|
||||
now=lambda: current_time[0],
|
||||
)
|
||||
await client.send_text("first")
|
||||
current_time[0] += timedelta(seconds=61)
|
||||
await client.send_text("second")
|
||||
|
||||
assert token_calls == 2
|
||||
|
||||
|
||||
async def test_wecom_rejects_non_object_json_without_raising():
|
||||
"""Malformed external response shapes return sanitized retry evidence."""
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(200, json=["unexpected", "shape"])
|
||||
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http:
|
||||
result = await WeComClient(
|
||||
_enabled_settings(),
|
||||
http_client=http,
|
||||
).send_text("controlled")
|
||||
|
||||
assert result == DeliveryResult(False, "invalid token response", True)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("status_code", [429, 500, 503])
|
||||
async def test_wecom_marks_rate_limit_and_server_errors_retryable(status_code):
|
||||
"""Transient HTTP classes reach the outbox retry path without response leakage."""
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
if request.url.path.endswith("/gettoken"):
|
||||
return httpx.Response(
|
||||
200,
|
||||
json={"errcode": 0, "access_token": "token-a", "expires_in": 7200},
|
||||
)
|
||||
return httpx.Response(status_code, text="sensitive-response-body")
|
||||
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http:
|
||||
result = await WeComClient(
|
||||
_enabled_settings(),
|
||||
http_client=http,
|
||||
).send_text("sensitive-message-content")
|
||||
|
||||
assert result.success is False
|
||||
assert result.retryable is True
|
||||
assert result.error == f"HTTP {status_code}"
|
||||
|
||||
|
||||
async def test_wecom_transport_failure_is_sanitized_and_retryable(caplog):
|
||||
"""Logs and returned evidence never expose secrets, tokens, bodies, or content."""
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
if request.url.path.endswith("/gettoken"):
|
||||
return httpx.Response(
|
||||
200,
|
||||
json={"errcode": 0, "access_token": "token-a", "expires_in": 7200},
|
||||
)
|
||||
raise httpx.ConnectError("response-sensitive", request=request)
|
||||
|
||||
caplog.set_level(logging.WARNING)
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http:
|
||||
result = await WeComClient(
|
||||
_enabled_settings(WECOM_APP_SECRET="secret-sensitive"),
|
||||
http_client=http,
|
||||
).send_text("message-sensitive")
|
||||
|
||||
assert result == DeliveryResult(False, "transport error", True)
|
||||
assert "secret-sensitive" not in caplog.text
|
||||
assert "token-a" not in caplog.text
|
||||
assert "response-sensitive" not in caplog.text
|
||||
assert "message-sensitive" not in caplog.text
|
||||
Reference in New Issue
Block a user