Compare commits

12 Commits

32 changed files with 3765 additions and 600 deletions
+3
View File
@@ -17,6 +17,8 @@ dist/
# Env
.env
backend/.env
backend/.env.*
!backend/.env.example
# Database
*.db
@@ -26,3 +28,4 @@ backend/.env
.DS_Store
Thumbs.db
.claude/
.worktrees/
+25 -9
View File
@@ -1,20 +1,33 @@
# 运行环境
ENVIRONMENT=development
# 数据库(开发用 SQLite,生产用 PostgreSQL
DATABASE_URL=sqlite+aiosqlite:///./pingwatch.db
# DATABASE_URL=postgresql+asyncpg://user:pass@localhost/pingwatch
# 企业微信
WECOM_CORP_ID=your_corp_id
WECOM_AGENT_ID=1000001
WECOM_APP_SECRET=your_app_secret
WECOM_NOTIFICATION_ENABLED=false
WECOM_CORP_ID=
WECOM_AGENT_ID=0
WECOM_APP_SECRET=
WECOM_TO_PARTY=
WECOM_NOTIFICATION_MAX_ATTEMPTS=5
WECOM_RETRY_BASE_SECONDS=30
# Casdoor
CASDOOR_ENDPOINT=https://casdoor.dhdx.fun
CASDOOR_CLIENT_ID=e46b9e1eb893027bdf2a
CASDOOR_CLIENT_SECRET=b12c7e1688ed51481f3b5c5dae4191b6edbba916
CASDOOR_CLIENT_ID=
CASDOOR_CLIENT_SECRET=
CASDOOR_CERTIFICATE=
CASDOOR_ORGANIZATION=dahua
CASDOOR_APPLICATION=PingWatch
CASDOOR_REDIRECT_URI=http://10.10.10.7:5173/login
CASDOOR_REDIRECT_URI=
# 连通性监测
PROBE_PACKETS_PER_ROUND=3
OFFLINE_CONSECUTIVE_ROUNDS=2
DEGRADED_WINDOW_ROUNDS=5
DEGRADED_LOSS_PERCENT=20.0
RECOVERY_CONSECUTIVE_CLEAN_ROUNDS=3
# Ping 引擎
PING_INTERVAL_SECONDS=30
@@ -31,5 +44,8 @@ OFFLINE_SUPPRESS_RATIO=0.9
PING_RECORD_RETENTION_DAYS=90
ALERT_RETENTION_DAYS=365
# JWT 密钥(请改成随机字符串
SECRET_KEY=change-me-to-a-long-random-string
# JWT 密钥(生产环境必须由受控运行环境注入
SECRET_KEY=
# LogHive(可选)
LOGHIVE_API_KEY=
+34 -4
View File
@@ -1,10 +1,17 @@
"""应用配置,通过环境变量注入,不支持 .env 文件"""
"""Application configuration loaded from environment variables or a local .env file."""
from pydantic import Field
from pydantic_settings import BaseSettings
from typing import Optional
from typing import Literal
LEGACY_DEFAULT_SECRET = "change-me-to-a-long-random-string"
class Settings(BaseSettings):
# ---------- 运行环境 ----------
environment: Literal["development", "test", "production"] = "development"
# ---------- 数据库 ----------
DATABASE_URL: str = "sqlite+aiosqlite:///./pingwatch.db"
# PostgreSQL: "postgresql+asyncpg://user:pass@localhost/pingwatch"
@@ -15,7 +22,7 @@ class Settings(BaseSettings):
WECOM_APP_SECRET: str = ""
# ---------- Casdoor ----------
CASDOOR_ENDPOINT: str = "https://casdoor.dhdx.fun"
casdoor_endpoint: str = "https://casdoor.dhdx.fun"
CASDOOR_CLIENT_ID: str = ""
CASDOOR_CLIENT_SECRET: str = ""
CASDOOR_CERTIFICATE: str = "" # 可选,用于验证 id_token 签名
@@ -23,6 +30,19 @@ class Settings(BaseSettings):
CASDOOR_APPLICATION: str = "PingWatch"
CASDOOR_REDIRECT_URI: str = "http://10.10.10.7:5173/login" # 前端回调地址
# ---------- 连通性监测 ----------
probe_packets_per_round: int = Field(default=3, ge=1, le=10)
offline_consecutive_rounds: int = Field(default=2, ge=1, le=10)
degraded_window_rounds: int = Field(default=5, ge=2, le=60)
degraded_loss_percent: float = Field(default=20.0, ge=1, le=100)
recovery_consecutive_clean_rounds: int = Field(default=3, ge=1, le=20)
# ---------- 企业微信投递 ----------
wecom_notification_enabled: bool = False
wecom_notification_max_attempts: int = Field(default=5, ge=1, le=10)
wecom_retry_base_seconds: int = Field(default=30, ge=1, le=3600)
WECOM_TO_PARTY: str = ""
# ---------- Ping 引擎 ----------
PING_INTERVAL_SECONDS: int = 30
PING_TIMEOUT_SECONDS: float = 5.0
@@ -40,7 +60,7 @@ class Settings(BaseSettings):
ALERT_RETENTION_DAYS: int = 365
# ---------- JWT ----------
SECRET_KEY: str = "change-me-to-a-long-random-string"
secret_key: str = LEGACY_DEFAULT_SECRET
ACCESS_TOKEN_EXPIRE_MINUTES: int = 480
# ---------- LogHive ----------
@@ -53,5 +73,15 @@ class Settings(BaseSettings):
model_config = {"env_file": ".env", "env_file_encoding": "utf-8"}
@property
def CASDOOR_ENDPOINT(self) -> str:
"""Compatibility accessor for existing uppercase configuration consumers."""
return self.casdoor_endpoint
@property
def SECRET_KEY(self) -> str:
"""Compatibility accessor for existing uppercase configuration consumers."""
return self.secret_key
settings = Settings()
+7 -10
View File
@@ -64,11 +64,11 @@ async def exchange_code_for_user(code: str) -> Optional[dict]:
"code": code,
}
async with httpx.AsyncClient(timeout=15, verify=False) as client:
async with httpx.AsyncClient(timeout=15) as client:
try:
resp = await client.post(token_url, data=data)
if resp.status_code != 200:
logger.error(f"Casdoor token 换取失败: {resp.status_code} {resp.text}")
logger.error("Casdoor token 换取失败: status=%s", resp.status_code)
return None
token_data = resp.json()
@@ -77,19 +77,16 @@ async def exchange_code_for_user(code: str) -> Optional[dict]:
logger.error("Casdoor 返回中没有 id_token")
return None
# 解码 id_token (JWT) payload,不验证签名(HTTPS 已保证传输安全)
# 生产环境建议验证 Casdoor 证书
cert = _load_casdoor_certificate()
try:
if not cert:
logger.error("Casdoor token 验证证书未配置")
return None
payload = jwt.decode(
id_token,
key=cert or None,
options={"verify_signature": bool(cert)},
key=cert,
audience=settings.CASDOOR_CLIENT_ID,
)
except JWTError:
# 不验证签名的方式解码
payload = jwt.get_unverified_claims(id_token)
return payload
+10 -3
View File
@@ -6,7 +6,7 @@ from app.config import settings
# 处理 sqlite 协议兼容
db_url = settings.DATABASE_URL
if db_url.startswith("sqlite"):
if db_url.startswith("sqlite://"):
db_url = db_url.replace("sqlite://", "sqlite+aiosqlite://")
engine = create_async_engine(db_url, echo=False, pool_pre_ping=True)
@@ -22,7 +22,14 @@ async def get_db() -> AsyncGenerator[AsyncSession, None]:
async def init_db():
"""创建所有表"""
from app.models.device import Base
"""Apply non-destructive schema upgrades before creating missing tables."""
from importlib import import_module
from app.models import Base
migration = import_module("migrations.versions.20260803_reliability_monitoring")
run_reliability_migration = migration.run_reliability_migration
async with engine.begin() as conn:
await conn.run_sync(run_reliability_migration)
await conn.run_sync(Base.metadata.create_all)
+5 -2
View File
@@ -1,11 +1,14 @@
from .device import Device, DeviceTypeEnum
from .device import Base, Device, DeviceMonitoringPolicy, DeviceTypeEnum
from .ping_record import PingRecord
from .alert_event import AlertEvent, AlertTypeEnum
from .notification_outbox import NotificationOutbox, NotificationStatus
from .user import User, UserRoleEnum
__all__ = [
"Device", "DeviceTypeEnum",
"Base",
"Device", "DeviceMonitoringPolicy", "DeviceTypeEnum",
"PingRecord",
"AlertEvent", "AlertTypeEnum",
"NotificationOutbox", "NotificationStatus",
"User", "UserRoleEnum",
]
+19 -2
View File
@@ -1,11 +1,15 @@
import enum
from datetime import datetime
from sqlalchemy import Column, Integer, String, Boolean, DateTime, BigInteger, Enum
from sqlalchemy import BigInteger, Boolean, Column, DateTime, Enum, Integer, String
from .device import Base
PRIMARY_KEY_TYPE = BigInteger().with_variant(Integer, "sqlite")
class AlertTypeEnum(str, enum.Enum):
offline = "offline" # 设备离线
degraded = "degraded" # 设备丢包故障
recovered = "recovered" # 设备恢复
system = "system" # 系统告警(如上游断网检测)
@@ -14,7 +18,7 @@ class AlertEvent(Base):
"""告警事件表"""
__tablename__ = "alert_events"
id = Column(BigInteger, primary_key=True, autoincrement=True)
id = Column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True)
device_id = Column(Integer, nullable=False, index=True, comment="关联设备 ID")
alert_type = Column(Enum(AlertTypeEnum), nullable=False, comment="告警类型")
message = Column(String(1024), default="", comment="告警消息摘要")
@@ -26,6 +30,19 @@ class AlertEvent(Base):
is_resolved = Column(Boolean, default=False, comment="是否已恢复")
notification_sent = Column(Boolean, default=False, comment="是否已发送通知")
notification_attempts = Column(Integer, default=0, comment="通知尝试次数")
last_notification_error = Column(
String(512),
nullable=True,
comment="最近一次通知失败原因",
)
related_event_id = Column(
BigInteger,
nullable=True,
comment="关联的原始或升级告警事件 ID",
)
previous_status = Column(String(16), nullable=True, comment="状态变更前状态")
current_status = Column(String(16), nullable=True, comment="状态变更后状态")
acknowledged_at = Column(DateTime, nullable=True, comment="用户确认时间")
created_at = Column(DateTime, default=datetime.now, comment="创建时间")
+121 -2
View File
@@ -1,8 +1,14 @@
import enum
from dataclasses import dataclass
from datetime import datetime
from sqlalchemy import Column, Integer, String, Float, Boolean, DateTime, Enum, Text
from typing import TYPE_CHECKING
from sqlalchemy import Boolean, Column, DateTime, Enum, Float, Integer, String
from sqlalchemy.orm import DeclarativeBase
if TYPE_CHECKING:
from app.config import Settings
class Base(DeclarativeBase):
pass
@@ -16,6 +22,88 @@ class DeviceTypeEnum(str, enum.Enum):
other = "other" # 其他
@dataclass(frozen=True)
class DeviceMonitoringPolicy:
"""Resolved monitoring thresholds for one device.
Nullable device fields intentionally inherit the runtime defaults so a
configuration change can apply to devices that do not need exceptions.
"""
probe_packets_per_round: int
offline_consecutive_rounds: int
degraded_window_rounds: int
degraded_loss_percent: float
recovery_consecutive_clean_rounds: int
@staticmethod
def _resolve_override(
override: object,
default: int | float,
minimum: int | float,
maximum: int | float,
) -> int | float:
"""Use a bounded persisted override or retain the validated default."""
if (
isinstance(override, bool)
or not isinstance(override, (int, float))
or override < minimum
or override > maximum
):
return default
return override
@classmethod
def from_device(
cls,
device: "Device",
settings: "Settings",
) -> "DeviceMonitoringPolicy":
"""Resolve device overrides without changing the stored device."""
return cls(
probe_packets_per_round=(
cls._resolve_override(
device.probe_packets_per_round,
settings.probe_packets_per_round,
1,
10,
)
),
offline_consecutive_rounds=(
cls._resolve_override(
device.offline_consecutive_rounds,
settings.offline_consecutive_rounds,
1,
10,
)
),
degraded_window_rounds=(
cls._resolve_override(
device.degraded_window_rounds,
settings.degraded_window_rounds,
2,
60,
)
),
degraded_loss_percent=(
cls._resolve_override(
device.degraded_loss_percent,
settings.degraded_loss_percent,
1,
100,
)
),
recovery_consecutive_clean_rounds=(
cls._resolve_override(
device.recovery_consecutive_clean_rounds,
settings.recovery_consecutive_clean_rounds,
1,
20,
)
),
)
class Device(Base):
__tablename__ = "devices"
@@ -32,8 +120,39 @@ class Device(Base):
alert_threshold = Column(Integer, default=5, comment="连续失败次数判离线")
is_enabled = Column(Boolean, default=True, comment="是否启用监控")
# 设备级监测策略覆盖;NULL 时使用运行环境中的全局默认值。
probe_packets_per_round = Column(
Integer,
nullable=True,
comment="单轮探测发包数覆盖",
)
offline_consecutive_rounds = Column(
Integer,
nullable=True,
comment="连续全丢包离线轮数覆盖",
)
degraded_window_rounds = Column(
Integer,
nullable=True,
comment="故障丢包滑动窗口轮数覆盖",
)
degraded_loss_percent = Column(
Float,
nullable=True,
comment="故障丢包率阈值覆盖",
)
recovery_consecutive_clean_rounds = Column(
Integer,
nullable=True,
comment="连续零丢包恢复轮数覆盖",
)
# 运行状态
current_status = Column(String(16), default="unknown", comment="当前状态: online/offline/unknown")
current_status = Column(
String(16),
default="unknown",
comment="当前状态: online/degraded/offline/unknown",
)
consecutive_failures = Column(Integer, default=0, comment="当前连续失败次数")
last_ping_time = Column(DateTime, nullable=True, comment="最后一次 ping 时间")
last_online_time = Column(DateTime, nullable=True, comment="最后一次在线时间")
+55
View File
@@ -0,0 +1,55 @@
"""Durable delivery queue for alert notifications."""
import enum
from datetime import datetime
from sqlalchemy import BigInteger, Column, DateTime, Enum, Index, Integer, String, Text
from .device import Base
PRIMARY_KEY_TYPE = BigInteger().with_variant(Integer, "sqlite")
class NotificationStatus(str, enum.Enum):
"""Lifecycle states for a notification delivery attempt."""
pending = "pending"
sending = "sending"
sent = "sent"
failed = "failed"
class NotificationOutbox(Base):
"""A notification retained until a dispatcher records its final outcome."""
__tablename__ = "notification_outbox"
id = Column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True)
alert_event_id = Column(BigInteger, nullable=False, index=True, comment="关联告警事件 ID")
message_content = Column(Text, nullable=False, comment="待发送的脱敏消息摘要")
delivery_scope = Column(String(256), nullable=True, comment="受控投递范围摘要")
status = Column(
Enum(NotificationStatus),
nullable=False,
default=NotificationStatus.pending,
server_default=NotificationStatus.pending.value,
comment="投递状态",
)
attempt_count = Column(Integer, nullable=False, default=0, comment="投递尝试次数")
next_attempt_at = Column(DateTime, nullable=True, comment="下次允许投递时间")
locked_at = Column(DateTime, nullable=True, comment="投递器领取时间")
sent_at = Column(DateTime, nullable=True, comment="成功投递时间")
last_error = Column(String(512), nullable=True, comment="最近一次投递错误摘要")
created_at = Column(DateTime, default=datetime.now, nullable=False, comment="创建时间")
updated_at = Column(
DateTime,
default=datetime.now,
onupdate=datetime.now,
nullable=False,
comment="更新时间",
)
__table_args__ = (
Index("idx_notification_outbox_status_next_attempt", "status", "next_attempt_at"),
)
+16 -2
View File
@@ -1,18 +1,32 @@
from datetime import datetime
from sqlalchemy import Column, Integer, Float, Boolean, DateTime, BigInteger
from sqlalchemy import BigInteger, Boolean, Column, DateTime, Float, Index, Integer, String
from .device import Base
PRIMARY_KEY_TYPE = BigInteger().with_variant(Integer, "sqlite")
class PingRecord(Base):
"""单次 ping 结果记录"""
__tablename__ = "ping_records"
id = Column(BigInteger, primary_key=True, autoincrement=True)
id = Column(PRIMARY_KEY_TYPE, primary_key=True, autoincrement=True)
device_id = Column(Integer, nullable=False, index=True)
is_alive = Column(Boolean, nullable=False, comment="是否通")
response_time_ms = Column(Float, nullable=True, comment="响应时间毫秒,不通则为 NULL")
round_num = Column(Integer, nullable=False, comment="轮次编号(从 1 递增)")
created_at = Column(DateTime, default=datetime.now, index=True, comment="记录时间")
sent_count = Column(Integer, nullable=True, comment="本轮发包数")
received_count = Column(Integer, nullable=True, comment="本轮收包数")
packet_loss_percent = Column(Float, nullable=True, comment="本轮丢包率百分比")
average_rtt_ms = Column(Float, nullable=True, comment="本轮平均时延毫秒")
is_valid = Column(Boolean, nullable=False, default=True, comment="探测结果是否有效")
failure_reason = Column(String(512), nullable=True, comment="无效探测的原因")
__table_args__ = (
Index("idx_ping_records_device_created_at", "device_id", "created_at"),
)
def __repr__(self):
return f"<PingRecord(device={self.device_id}, alive={self.is_alive}, rtt={self.response_time_ms})>"
+114 -297
View File
@@ -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)
+178
View File
@@ -0,0 +1,178 @@
"""Execute and parse multi-packet fping probes without shell interpolation."""
import asyncio
from dataclasses import dataclass
import re
from typing import Iterable
FPING_RESULT_PATTERN = re.compile(r"^\s*(\S+)\s+:\s*(.*?)\s*$")
@dataclass(frozen=True)
class ProbeResult:
"""Aggregate result for one IP in one monitoring round."""
ip: str
sent_count: int
received_count: int
average_rtt_ms: float | None
is_valid: bool
failure_reason: str | None = None
@property
def packet_loss_percent(self) -> float:
"""Return packet loss as a percentage of the attempted probes."""
if self.sent_count <= 0:
return 0.0
return (
(self.sent_count - self.received_count)
/ self.sent_count
* 100.0
)
def _invalid_result(
ip: str,
packets_per_round: int,
reason: str,
) -> ProbeResult:
"""Build an invalid aggregate that cannot resemble measured packet loss."""
return ProbeResult(
ip=ip,
sent_count=packets_per_round,
received_count=0,
average_rtt_ms=None,
is_valid=False,
failure_reason=reason,
)
def parse_fping_count_output(
output: str,
expected_ips: set[str],
packets_per_round: int,
) -> dict[str, ProbeResult]:
"""Parse ``fping -C`` reply tokens for every expected IP.
A numeric token is a measured RTT and ``-`` is packet loss. Missing,
duplicated, or malformed lines are represented as invalid aggregates.
"""
if packets_per_round <= 0:
raise ValueError("packets_per_round must be positive")
parsed_lines: dict[str, list[str]] = {}
duplicate_ips: set[str] = set()
for line in output.splitlines():
match = FPING_RESULT_PATTERN.match(line)
if match is None:
continue
ip, replies = match.groups()
if ip not in expected_ips:
continue
if ip in parsed_lines:
duplicate_ips.add(ip)
continue
parsed_lines[ip] = replies.split()
results: dict[str, ProbeResult] = {}
for ip in expected_ips:
tokens = parsed_lines.get(ip)
if tokens is None:
results[ip] = _invalid_result(
ip,
packets_per_round,
"missing fping output",
)
continue
if ip in duplicate_ips or len(tokens) != packets_per_round:
results[ip] = _invalid_result(
ip,
packets_per_round,
"malformed fping replies",
)
continue
reply_times: list[float] = []
malformed = False
for token in tokens:
if token == "-":
continue
try:
reply_time = float(token)
except ValueError:
malformed = True
break
if reply_time < 0:
malformed = True
break
reply_times.append(reply_time)
if malformed:
results[ip] = _invalid_result(
ip,
packets_per_round,
"malformed fping replies",
)
continue
average_rtt_ms = (
sum(reply_times) / len(reply_times)
if reply_times
else None
)
results[ip] = ProbeResult(
ip=ip,
sent_count=packets_per_round,
received_count=len(reply_times),
average_rtt_ms=average_rtt_ms,
is_valid=True,
)
return results
async def run_fping_count(
ips: Iterable[str],
packets_per_round: int,
timeout_ms: int,
executable: str,
) -> dict[str, ProbeResult]:
"""Run one fping count probe and return an aggregate for every IP."""
ip_list = list(ips)
if not ip_list:
return {}
command = [
executable,
"-C",
str(packets_per_round),
"-q",
"-t",
str(timeout_ms),
*ip_list,
]
try:
process = await asyncio.create_subprocess_exec(
*command,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await process.communicate()
except Exception as error:
reason = f"fping execution failed: {type(error).__name__}"
return {
ip: _invalid_result(ip, packets_per_round, reason)
for ip in ip_list
}
output = "\n".join(
stream.decode("utf-8", errors="replace")
for stream in (stdout, stderr)
if stream
)
return parse_fping_count_output(
output,
set(ip_list),
packets_per_round,
)
@@ -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)
+210 -200
View File
@@ -1,242 +1,252 @@
"""
异步 Ping 引擎
"""Persist batched probe summaries and apply device health transitions."""
核心逻辑:
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
import time
from collections import defaultdict
from dataclasses import dataclass
from datetime import datetime
from sqlalchemy import select, update
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.models.device import Device, DeviceTypeEnum
from app.config import Settings, settings
from app.models.device import Device, DeviceMonitoringPolicy
from app.models.ping_record import PingRecord
from app.services.fping_runner import ProbeResult, run_fping_count
from app.services.state_machine import StateDecision, evaluate_health
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
@dataclass(frozen=True)
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
"""A persisted device health transition produced by one probe round."""
device: Device
old_status: str
new_status: str
consecutive_failures: int
event_type: str | None = None
reason: str = ""
class Pinger:
"""
Ping 引擎,使用 fping 批量并发检测。
对所有设备进 ping,返回存活状态和响应时间。
"""
"""Run multi-packet fping probes and update monitored device health."""
def __init__(self):
def __init__(self, runtime_settings: Settings | None = None):
self._settings = runtime_settings or settings
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. 更新设备状态
"""
async def run_one_round(
self,
db: AsyncSession,
) -> list[DeviceStateChange]:
"""Persist one summary per enabled device and commit state changes once."""
self._round_num += 1
round_num = self._round_num
# 1. 加载启用设备
result = await db.execute(
select(Device).where(Device.is_enabled == True)
query_result = await db.execute(
select(Device).where(Device.is_enabled.is_(True))
)
devices = list(result.scalars().all())
devices = list(query_result.scalars().all())
if not devices:
logger.info(f"[Round {round_num}] 没有启用的设备")
logger.info("[Round %s] no enabled devices", 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,
started_at = time.monotonic()
policies = {
device.id: DeviceMonitoringPolicy.from_device(
device,
self._settings,
)
for r in results
]
db.add_all(records)
for device in devices
}
probe_results = await self._probe_by_packet_count(devices, policies)
observed_at = datetime.now()
records_by_device: dict[int, PingRecord] = {}
for device in devices:
policy = policies[device.id]
probe = probe_results.get(
(device.id, policy.probe_packets_per_round)
)
if probe is None:
probe = ProbeResult(
ip=device.ip,
sent_count=policy.probe_packets_per_round,
received_count=0,
average_rtt_ms=None,
is_valid=False,
failure_reason="missing runner result",
)
record = self._to_record(
device=device,
probe=probe,
round_num=round_num,
observed_at=observed_at,
)
records_by_device[device.id] = record
db.add(record)
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:
changes: list[DeviceStateChange] = []
for device in devices:
record = records_by_device[device.id]
device.last_ping_time = observed_at
if not record.is_valid:
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,
policy = policies[device.id]
recent = await self._load_recent_valid(
db,
device.id,
policy,
)
await self._on_state_change(change)
decision = evaluate_health(
device.current_status or "unknown",
recent,
policy,
)
self._update_observation_fields(device, recent, observed_at)
change = self._apply_decision(
device,
decision,
observed_at,
)
if change is not None:
changes.append(change)
await db.commit()
# 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(
f"[Round {round_num}] 完成: {alive_count}/{len(devices)} 在线, "
f"耗时 {elapsed:.2f}s"
"[Round %s] persisted %s device summaries with %s transitions "
"in %.2fs",
round_num,
len(devices),
len(changes),
elapsed,
)
return changes
async def _probe_by_packet_count(
self,
devices: list[Device],
policies: dict[int, DeviceMonitoringPolicy],
) -> dict[tuple[int, int], ProbeResult]:
"""Run one argv-safe batch for each distinct packet-count policy."""
groups: dict[int, list[Device]] = defaultdict(list)
for device in devices:
groups[policies[device.id].probe_packets_per_round].append(device)
results: dict[tuple[int, int], ProbeResult] = {}
timeout_ms = int(self._settings.PING_TIMEOUT_SECONDS * 1000)
for packet_count, grouped_devices in groups.items():
unique_ips = list(dict.fromkeys(
device.ip for device in grouped_devices
))
batch = await run_fping_count(
unique_ips,
packets_per_round=packet_count,
timeout_ms=timeout_ms,
executable=self._settings.FPING_PATH,
)
for device in grouped_devices:
probe = batch.get(device.ip)
if probe is not None:
results[(device.id, packet_count)] = probe
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,
@staticmethod
def _to_record(
device: Device,
probe: ProbeResult,
round_num: int,
observed_at: datetime,
) -> PingRecord:
"""Convert a runner aggregate into the persisted compatibility model."""
return PingRecord(
device_id=device.id,
is_alive=probe.is_valid and probe.received_count > 0,
response_time_ms=probe.average_rtt_ms,
round_num=round_num,
created_at=observed_at,
sent_count=probe.sent_count,
received_count=probe.received_count,
packet_loss_percent=(
probe.packet_loss_percent if probe.is_valid else None
),
average_rtt_ms=probe.average_rtt_ms,
is_valid=probe.is_valid,
failure_reason=probe.failure_reason,
)
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,
@staticmethod
async def _load_recent_valid(
db: AsyncSession,
device_id: int,
policy: DeviceMonitoringPolicy,
) -> list[PingRecord]:
"""Load enough valid summaries for every configured transition rule."""
history_size = max(
policy.offline_consecutive_rounds,
policy.degraded_window_rounds,
policy.recovery_consecutive_clean_rounds,
)
await proc.wait()
elapsed = (time.time() - start) * 1000
return ip, proc.returncode == 0, round(elapsed, 2)
except Exception:
return ip, False, None
query_result = await db.execute(
select(PingRecord)
.where(PingRecord.device_id == device_id)
.where(PingRecord.is_valid.is_(True))
.order_by(PingRecord.created_at.desc(), PingRecord.id.desc())
.limit(history_size)
)
newest_first = list(query_result.scalars().all())
return list(reversed(newest_first))
tasks = [ping_one(ip) for ip in ip_list]
sem = asyncio.Semaphore(settings.PING_CONCURRENCY)
@staticmethod
def _update_observation_fields(
device: Device,
recent: list[PingRecord],
observed_at: datetime,
) -> None:
"""Maintain compatibility timestamps and the full-loss streak counter."""
current = recent[-1]
if current.received_count > 0:
device.last_online_time = observed_at
async def bounded_ping(ip: str):
async with sem:
return await ping_one(ip)
full_loss_count = 0
for record in reversed(recent):
if record.sent_count and record.received_count == 0:
full_loss_count += 1
else:
break
device.consecutive_failures = full_loss_count
results = await asyncio.gather(*[bounded_ping(ip) for ip in ip_list])
return {ip: (alive, rtt) for ip, alive, rtt in results}
@staticmethod
def _apply_decision(
device: Device,
decision: StateDecision,
observed_at: datetime,
) -> DeviceStateChange | None:
"""Apply one pure decision and materialize its transition value."""
old_status = device.current_status or "unknown"
if decision.next_status == old_status:
return None
device.current_status = decision.next_status
if decision.next_status == "offline":
device.last_offline_time = observed_at
if decision.next_status == "online":
device.last_online_time = observed_at
return DeviceStateChange(
device=device,
old_status=old_status,
new_status=decision.next_status,
consecutive_failures=device.consecutive_failures or 0,
event_type=decision.event_type,
reason=decision.reason,
)
+105 -57
View File
@@ -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,21 @@
"""Runtime validation for security-sensitive application settings."""
from app.config import LEGACY_DEFAULT_SECRET, Settings
def validate_runtime_settings(settings: Settings) -> None:
"""Reject unsafe settings before scheduling monitoring work."""
if settings.environment == "production":
if not settings.secret_key.strip() or settings.secret_key == LEGACY_DEFAULT_SECRET:
raise ValueError("SECRET_KEY must be provided by the runtime environment")
if not settings.casdoor_endpoint.startswith("https://"):
raise ValueError("CASDOOR endpoint must use TLS in production")
if settings.wecom_notification_enabled:
credentials = (
settings.WECOM_CORP_ID,
settings.WECOM_AGENT_ID,
settings.WECOM_APP_SECRET,
)
if not all(credentials):
raise ValueError("WECOM delivery requires corp ID, agent ID, and app secret")
+126
View File
@@ -0,0 +1,126 @@
"""Pure state transitions for device connectivity health."""
from dataclasses import dataclass
from typing import Protocol, Sequence
from app.models.device import DeviceMonitoringPolicy
class ProbeSample(Protocol):
"""Packet summary attributes required for health evaluation."""
sent_count: int
received_count: int
is_valid: bool
@dataclass(frozen=True)
class StateDecision:
"""Result of evaluating recent probes against a monitoring policy."""
next_status: str
event_type: str | None
should_notify: bool
reason: str
def _retain(previous_status: str, reason: str) -> StateDecision:
"""Return a non-notifying decision that preserves current health."""
return StateDecision(
next_status=previous_status,
event_type=None,
should_notify=False,
reason=reason,
)
def _transition(
previous_status: str,
next_status: str,
event_type: str,
reason: str,
) -> StateDecision:
"""Return a transition event only when the persisted status changes."""
if previous_status == next_status:
return _retain(previous_status, reason)
return StateDecision(
next_status=next_status,
event_type=event_type,
should_notify=True,
reason=reason,
)
def _is_valid(sample: ProbeSample) -> bool:
"""Reject impossible summaries before they influence device health."""
return (
sample.is_valid
and sample.sent_count > 0
and 0 <= sample.received_count <= sample.sent_count
)
def evaluate_health(
previous_status: str,
recent: Sequence[ProbeSample],
policy: DeviceMonitoringPolicy,
) -> StateDecision:
"""Evaluate ordered oldest-to-newest probe summaries.
Invalid input retains state. Offline has priority over aggregate degraded
loss, and an alerted state recovers only after the configured clean streak.
"""
if not recent:
return _retain(previous_status, "no probe history")
if any(not _is_valid(sample) for sample in recent):
return _retain(previous_status, "invalid probe history")
offline_rounds = policy.offline_consecutive_rounds
if len(recent) >= offline_rounds and all(
sample.received_count == 0
for sample in recent[-offline_rounds:]
):
return _transition(
previous_status,
"offline",
"offline",
f"{offline_rounds} consecutive full-loss rounds",
)
recovery_rounds = policy.recovery_consecutive_clean_rounds
if (
previous_status in {"offline", "degraded"}
and len(recent) >= recovery_rounds
and all(
sample.received_count == sample.sent_count
for sample in recent[-recovery_rounds:]
)
):
return _transition(
previous_status,
"online",
"recovered",
f"{recovery_rounds} consecutive clean rounds",
)
degraded_rounds = policy.degraded_window_rounds
if len(recent) >= degraded_rounds:
window = recent[-degraded_rounds:]
sent_count = sum(sample.sent_count for sample in window)
received_count = sum(sample.received_count for sample in window)
loss_percent = (sent_count - received_count) / sent_count * 100.0
if (
previous_status != "offline"
and loss_percent >= policy.degraded_loss_percent
):
return _transition(
previous_status,
"degraded",
"degraded",
(
f"window packet loss {loss_percent:.2f}% is at or above "
f"{policy.degraded_loss_percent:.2f}%"
),
)
return _retain(previous_status, "no transition threshold met")
@@ -0,0 +1,152 @@
"""Add durable reliability-monitoring persistence without destructive DDL."""
from datetime import datetime
from sqlalchemy import Column, DateTime, MetaData, String, Table, inspect, select, text
from sqlalchemy.engine import Connection
from app.models import AlertEvent, Device, NotificationOutbox, PingRecord
REVISION = "20260803_reliability_monitoring"
_migration_metadata = MetaData()
_schema_migrations = Table(
"schema_migrations",
_migration_metadata,
Column("revision", String(64), primary_key=True),
Column("applied_at", DateTime, nullable=False, default=datetime.now),
)
_DEFAULTS_FOR_EXISTING_ROWS = {
"is_valid": "TRUE",
"notification_attempts": "0",
}
_REVISION_COLUMNS = {
"devices": (
Device.__table__,
{
"probe_packets_per_round",
"offline_consecutive_rounds",
"degraded_window_rounds",
"degraded_loss_percent",
"recovery_consecutive_clean_rounds",
},
),
"ping_records": (
PingRecord.__table__,
{
"sent_count",
"received_count",
"packet_loss_percent",
"average_rtt_ms",
"is_valid",
"failure_reason",
},
),
"alert_events": (
AlertEvent.__table__,
{
"notification_attempts",
"last_notification_error",
"related_event_id",
"previous_status",
"current_status",
},
),
}
def _quote(connection: Connection, identifier: str) -> str:
"""Quote one database identifier through the active dialect."""
return connection.dialect.identifier_preparer.quote(identifier)
def _add_missing_columns(connection: Connection) -> None:
"""Add only newly required nullable/defaulted columns to existing tables."""
inspector = inspect(connection)
existing_tables = set(inspector.get_table_names())
for table_name, (table, column_names) in _REVISION_COLUMNS.items():
if table_name not in existing_tables:
continue
existing_columns = {
column["name"] for column in inspector.get_columns(table_name)
}
for column in table.columns:
if column.name not in column_names or column.name in existing_columns:
continue
column_type = connection.dialect.type_compiler.process(column.type)
definition = f"{_quote(connection, column.name)} {column_type}"
default = _DEFAULTS_FOR_EXISTING_ROWS.get(column.name)
if default is not None:
if connection.dialect.name == "sqlite" and default == "TRUE":
default = "1"
if column.name == "is_valid":
definition = f"{definition} NOT NULL DEFAULT {default}"
else:
definition = f"{definition} DEFAULT {default}"
connection.execute(
text(
f"ALTER TABLE {_quote(connection, table_name)} "
f"ADD COLUMN {definition}"
)
)
def _create_missing_indexes(connection: Connection) -> None:
"""Create new performance indexes for tables that predate this revision."""
inspector = inspect(connection)
table = PingRecord.__table__
if table.name not in inspector.get_table_names():
return
existing_indexes = {
index["name"] for index in inspector.get_indexes(table.name)
}
index = next(
item
for item in table.indexes
if item.name == "idx_ping_records_device_created_at"
)
if index.name not in existing_indexes:
index.create(bind=connection)
def _upgrade_postgresql_alert_type(connection: Connection) -> None:
"""Permit the degraded alert value for databases using PostgreSQL enums."""
if connection.dialect.name == "postgresql":
enum_name = AlertEvent.__table__.c.alert_type.type.name
connection.execute(
text(
f"ALTER TYPE {_quote(connection, enum_name)} "
"ADD VALUE IF NOT EXISTS 'degraded'"
)
)
def run_reliability_migration(connection: Connection) -> None:
"""Apply this revision once while retaining all existing monitoring data.
This runner is deliberately idempotent: it creates only this revision's
outbox table, missing revision columns, and indexes, then writes one
immutable revision marker. It never issues DROP, DELETE, UPDATE, or
data-copy statements. The application owns global metadata creation after
this function returns.
"""
_migration_metadata.create_all(bind=connection)
NotificationOutbox.__table__.create(bind=connection, checkfirst=True)
already_applied = connection.execute(
select(_schema_migrations.c.revision).where(
_schema_migrations.c.revision == REVISION
)
).scalar_one_or_none()
if already_applied is not None:
return
_upgrade_postgresql_alert_type(connection)
_add_missing_columns(connection)
_create_missing_indexes(connection)
connection.execute(_schema_migrations.insert().values(revision=REVISION))
+4
View File
@@ -0,0 +1,4 @@
[pytest]
asyncio_mode = auto
asyncio_default_fixture_loop_scope = function
testpaths = tests
+3 -1
View File
@@ -1,7 +1,7 @@
fastapi>=0.110.0
uvicorn[standard]>=0.29.0
sqlalchemy[asyncio]>=2.0.30
aiosqlite>=0.20.0
aiosqlite==0.20.0
asyncpg>=0.29.0
pydantic>=2.7.0
pydantic-settings>=2.2.0
@@ -9,3 +9,5 @@ python-jose[cryptography]>=3.3.0
httpx>=0.27.0
python-multipart>=0.0.9
websockets>=12.0
pytest==8.3.5
pytest-asyncio==0.25.3
+1
View File
@@ -0,0 +1 @@
"""Shared test configuration for PingWatch backend tests."""
+427
View File
@@ -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()
+18
View File
@@ -0,0 +1,18 @@
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
def test_compose_does_not_embed_password_or_net_admin_capability():
compose = (ROOT / "docker-compose.yml").read_text(encoding="utf-8")
assert "POSTGRES_PASSWORD=pingwatch123" not in compose
assert "NET_ADMIN" not in compose
assert "127.0.0.1:8001:8000" in compose
def test_oauth_client_does_not_disable_tls_verification():
auth_source = (ROOT / "backend" / "app" / "core" / "auth.py").read_text(encoding="utf-8")
assert "verify=False" not in auth_source
+48
View File
@@ -0,0 +1,48 @@
"""Regression coverage for application dependency module imports."""
import importlib
import sys
from pathlib import Path
import pytest
def test_dependency_module_compiles():
"""Database initialization dependencies remain valid Python syntax."""
module_path = Path(__file__).parents[1] / "app" / "core" / "deps.py"
compile(module_path.read_text(encoding="utf-8"), str(module_path), "exec")
@pytest.mark.parametrize(
("database_url", "expected_url"),
[
("sqlite:///./pingwatch.db", "sqlite+aiosqlite:///./pingwatch.db"),
(
"sqlite+aiosqlite:///./pingwatch.db",
"sqlite+aiosqlite:///./pingwatch.db",
),
],
)
def test_dependency_module_normalizes_sqlite_url_once(
monkeypatch, database_url, expected_url
):
"""Plain SQLite URLs gain the async driver without duplicating an existing one."""
monkeypatch.setenv("DATABASE_URL", database_url)
sys.modules.pop("app.core.deps", None)
sys.modules.pop("app.config", None)
deps = importlib.import_module("app.core.deps")
assert str(deps.engine.url) == expected_url
def test_dependency_module_initializes_with_default_sqlite_url(monkeypatch):
"""The default database setting can initialize the asynchronous dependency."""
monkeypatch.delenv("DATABASE_URL", raising=False)
sys.modules.pop("app.core.deps", None)
sys.modules.pop("app.config", None)
deps = importlib.import_module("app.core.deps")
assert str(deps.engine.url) == "sqlite+aiosqlite:///./pingwatch.db"
+322
View File
@@ -0,0 +1,322 @@
"""Behavior tests for multi-packet fping execution and persistence."""
import asyncio
from datetime import datetime, timedelta
import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from app.config import Settings
from app.models.device import Base, Device, DeviceTypeEnum
from app.models.ping_record import PingRecord
from app.services.fping_runner import (
ProbeResult,
parse_fping_count_output,
run_fping_count,
)
from app.services.pinger import Pinger
@pytest.fixture
async def db_session() -> AsyncSession:
"""Provide an isolated database for pinger transaction 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()
def test_parses_loss_and_average_rtt():
"""Numeric replies count as received packets and dashes count as loss."""
output = "10.0.0.8 : 12.4 - 12.6\n10.0.0.9 : - - -"
result = parse_fping_count_output(
output,
{"10.0.0.8", "10.0.0.9"},
3,
)
assert result["10.0.0.8"].received_count == 2
assert result["10.0.0.8"].average_rtt_ms == 12.5
assert result["10.0.0.8"].packet_loss_percent == pytest.approx(
33.3333333333
)
assert result["10.0.0.9"].packet_loss_percent == 100.0
def test_marks_missing_and_malformed_expected_output_invalid():
"""Incomplete fping output cannot be interpreted as device packet loss."""
output = "10.0.0.8 : 10.2 bad-token -"
result = parse_fping_count_output(
output,
{"10.0.0.8", "10.0.0.9"},
3,
)
assert result["10.0.0.8"].is_valid is False
assert result["10.0.0.8"].failure_reason == "malformed fping replies"
assert result["10.0.0.9"].is_valid is False
assert result["10.0.0.9"].failure_reason == "missing fping output"
def test_parses_ipv6_without_confusing_address_colons_for_delimiter():
"""An IPv6 address remains the result key when parsing the output delimiter."""
result = parse_fping_count_output(
"2001:db8::8 : 1.25 - 1.75",
{"2001:db8::8"},
3,
)
assert result["2001:db8::8"].is_valid is True
assert result["2001:db8::8"].received_count == 2
assert result["2001:db8::8"].average_rtt_ms == 1.5
async def test_runner_uses_argument_vector_and_parses_quiet_stderr(monkeypatch):
"""The runner passes IPs as argv entries and reads fping -C quiet output."""
observed = {}
class CompletedProcess:
returncode = 1
async def communicate(self):
return b"", b"10.0.0.8 : 4.2 - 4.4\n"
async def fake_create_subprocess_exec(*args, **kwargs):
observed["args"] = args
observed["kwargs"] = kwargs
return CompletedProcess()
monkeypatch.setattr(
asyncio,
"create_subprocess_exec",
fake_create_subprocess_exec,
)
result = await run_fping_count(
["10.0.0.8"],
packets_per_round=3,
timeout_ms=800,
executable="/test/fping",
)
assert observed["args"] == (
"/test/fping",
"-C",
"3",
"-q",
"-t",
"800",
"10.0.0.8",
)
assert observed["kwargs"] == {
"stdout": asyncio.subprocess.PIPE,
"stderr": asyncio.subprocess.PIPE,
}
assert result["10.0.0.8"].received_count == 2
async def test_runner_failure_returns_invalid_result_instead_of_packet_loss(
monkeypatch,
):
"""A process communication failure remains an invalid probe aggregate."""
class BrokenProcess:
async def communicate(self):
raise RuntimeError("simulated process pipe failure")
async def fake_create_subprocess_exec(*args, **kwargs):
return BrokenProcess()
monkeypatch.setattr(
asyncio,
"create_subprocess_exec",
fake_create_subprocess_exec,
)
result = await run_fping_count(
["10.0.0.8"],
packets_per_round=3,
timeout_ms=800,
executable="/test/fping",
)
assert result["10.0.0.8"].is_valid is False
assert result["10.0.0.8"].failure_reason == (
"fping execution failed: RuntimeError"
)
async def test_pinger_persists_one_summary_for_each_enabled_device(
db_session,
monkeypatch,
):
"""One monitoring round writes exactly one aggregate per enabled device."""
first = Device(
name="edge-a",
ip="10.0.0.8",
device_type=DeviceTypeEnum.switch,
current_status="online",
is_enabled=True,
)
second = Device(
name="edge-b",
ip="10.0.0.9",
device_type=DeviceTypeEnum.switch,
current_status="online",
is_enabled=True,
)
disabled = Device(
name="edge-disabled",
ip="10.0.0.10",
device_type=DeviceTypeEnum.switch,
current_status="online",
is_enabled=False,
)
db_session.add_all([first, second, disabled])
await db_session.commit()
async def fake_run_fping_count(
ips,
packets_per_round,
timeout_ms,
executable,
):
assert set(ips) == {"10.0.0.8", "10.0.0.9"}
return {
"10.0.0.8": ProbeResult("10.0.0.8", 3, 2, 8.5, True),
"10.0.0.9": ProbeResult("10.0.0.9", 3, 3, 9.5, True),
}
monkeypatch.setattr(
"app.services.pinger.run_fping_count",
fake_run_fping_count,
)
changes = await Pinger(Settings()).run_one_round(db_session)
records = list(
(
await db_session.execute(
select(PingRecord).order_by(PingRecord.device_id)
)
)
.scalars()
.all()
)
assert changes == []
assert len(records) == 2
assert [record.device_id for record in records] == [first.id, second.id]
assert records[0].sent_count == 3
assert records[0].received_count == 2
assert records[0].packet_loss_percent == pytest.approx(33.3333333333)
assert records[0].average_rtt_ms == 8.5
assert records[0].is_valid is True
async def test_pinger_persists_invalid_probe_without_changing_device_state(
db_session,
monkeypatch,
):
"""A runner failure remains observable but cannot mark a device offline."""
device = Device(
name="edge-a",
ip="10.0.0.8",
device_type=DeviceTypeEnum.switch,
current_status="online",
is_enabled=True,
)
db_session.add(device)
await db_session.commit()
async def fake_run_fping_count(*args, **kwargs):
return {
device.ip: ProbeResult(
device.ip,
3,
0,
None,
False,
"fping executable not found",
)
}
monkeypatch.setattr(
"app.services.pinger.run_fping_count",
fake_run_fping_count,
)
changes = await Pinger(Settings()).run_one_round(db_session)
record = (
await db_session.execute(select(PingRecord).where(PingRecord.device_id == device.id))
).scalar_one()
assert changes == []
assert device.current_status == "online"
assert record.is_valid is False
assert record.packet_loss_percent is None
assert record.failure_reason == "fping executable not found"
async def test_pinger_evaluates_recent_valid_summaries_and_returns_change(
db_session,
monkeypatch,
):
"""A newly persisted full-loss summary completes the offline streak."""
device = Device(
name="edge-a",
ip="10.0.0.8",
device_type=DeviceTypeEnum.switch,
current_status="online",
is_enabled=True,
offline_consecutive_rounds=2,
)
db_session.add(device)
await db_session.flush()
db_session.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=datetime.now() - timedelta(seconds=30),
)
)
await db_session.commit()
async def fake_run_fping_count(*args, **kwargs):
return {
device.ip: ProbeResult(
device.ip,
3,
0,
None,
True,
)
}
monkeypatch.setattr(
"app.services.pinger.run_fping_count",
fake_run_fping_count,
)
changes = await Pinger(Settings()).run_one_round(db_session)
assert len(changes) == 1
assert changes[0].device is device
assert changes[0].old_status == "online"
assert changes[0].new_status == "offline"
assert changes[0].event_type == "offline"
assert device.current_status == "offline"
assert device.last_offline_time is not None
+292
View File
@@ -0,0 +1,292 @@
"""Persistence contracts for reliability monitoring data."""
import importlib
from pathlib import Path
import pytest
from sqlalchemy import Column, Index, Integer, Table, inspect, text
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
import app.models as models
from app.config import Settings
from app.models.device import Base
@pytest.fixture
async def db_session() -> AsyncSession:
"""Provide an isolated database containing the current model metadata."""
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_probe_record_stores_packet_summary(db_session: AsyncSession):
"""A probe aggregate retains loss and latency needed by the state machine."""
record = models.PingRecord(
device_id=1,
is_alive=True,
response_time_ms=12.5,
round_num=1,
sent_count=3,
received_count=2,
packet_loss_percent=33.33,
average_rtt_ms=12.5,
is_valid=True,
)
db_session.add(record)
await db_session.commit()
assert record.packet_loss_percent == 33.33
assert record.received_count == 2
async def test_outbox_defaults_to_pending(db_session: AsyncSession):
"""New notifications stay eligible for delivery until the dispatcher claims them."""
assert hasattr(models, "NotificationOutbox")
assert hasattr(models, "NotificationStatus")
item = models.NotificationOutbox(
alert_event_id=1,
message_content="masked summary",
)
db_session.add(item)
await db_session.commit()
assert item.status == models.NotificationStatus.pending
def test_device_policy_prefers_explicit_device_overrides():
"""A device-specific threshold supersedes only the configured global default."""
device = models.Device(
probe_packets_per_round=5,
degraded_loss_percent=None,
)
policy = models.DeviceMonitoringPolicy.from_device(device, Settings())
assert policy.probe_packets_per_round == 5
assert policy.degraded_loss_percent == 20.0
def test_device_policy_uses_global_default_for_invalid_persisted_override():
"""An invalid stored override cannot weaken a device's safe probe policy."""
device = models.Device(
probe_packets_per_round=0,
degraded_loss_percent=101.0,
)
policy = models.DeviceMonitoringPolicy.from_device(device, Settings())
assert policy.probe_packets_per_round == 3
assert policy.degraded_loss_percent == 20.0
async def test_migration_is_idempotent_and_records_revision(tmp_path: Path):
"""Rerunning the migration preserves the schema and writes one revision marker."""
migration = importlib.import_module(
"migrations.versions.20260803_reliability_monitoring"
)
assert hasattr(migration, "run_reliability_migration")
database_url = f"sqlite+aiosqlite:///{tmp_path / 'monitoring.db'}"
engine = create_async_engine(database_url)
try:
async with engine.begin() as connection:
await connection.run_sync(migration.run_reliability_migration)
await connection.run_sync(Base.metadata.create_all)
async with engine.begin() as connection:
await connection.run_sync(migration.run_reliability_migration)
revision_count = await connection.scalar(
text(
"SELECT count(*) FROM schema_migrations "
"WHERE revision = '20260803_reliability_monitoring'"
)
)
assert revision_count == 1
finally:
await engine.dispose()
async def test_migration_preserves_legacy_rows_and_owns_only_its_schema(
tmp_path: Path,
):
"""The revision upgrades legacy data twice without creating unrelated models."""
migration = importlib.import_module(
"migrations.versions.20260803_reliability_monitoring"
)
database_url = f"sqlite+aiosqlite:///{tmp_path / 'legacy-monitoring.db'}"
engine = create_async_engine(database_url)
future_table = Table(
"future_unrelated_model",
Base.metadata,
Column("id", Integer, primary_key=True),
)
try:
async with engine.begin() as connection:
await connection.execute(
text(
"CREATE TABLE devices ("
"id INTEGER PRIMARY KEY, name VARCHAR(128) NOT NULL, "
"ip VARCHAR(45) NOT NULL, device_type VARCHAR(16) NOT NULL, "
"location VARCHAR(256), project_name VARCHAR(256), "
"tags VARCHAR(512), ping_interval INTEGER, "
"alert_threshold INTEGER, is_enabled BOOLEAN, "
"current_status VARCHAR(16), consecutive_failures INTEGER, "
"last_ping_time DATETIME, last_online_time DATETIME, "
"last_offline_time DATETIME, created_at DATETIME, "
"updated_at DATETIME)"
)
)
await connection.execute(
text(
"CREATE TABLE ping_records ("
"id INTEGER PRIMARY KEY, device_id INTEGER NOT NULL, "
"is_alive BOOLEAN NOT NULL, response_time_ms FLOAT, "
"round_num INTEGER NOT NULL, created_at DATETIME)"
)
)
await connection.execute(
text(
"CREATE TABLE alert_events ("
"id INTEGER PRIMARY KEY, device_id INTEGER NOT NULL, "
"alert_type VARCHAR(16) NOT NULL, message VARCHAR(1024), "
"start_at DATETIME NOT NULL, end_at DATETIME, "
"duration_minutes INTEGER, is_resolved BOOLEAN, "
"notification_sent BOOLEAN, acknowledged_at DATETIME, "
"created_at DATETIME)"
)
)
await connection.execute(
text(
"INSERT INTO devices (id, name, ip, device_type, current_status) "
"VALUES (1, 'legacy-router', '192.0.2.10', 'router', 'online')"
)
)
await connection.execute(
text(
"INSERT INTO ping_records "
"(id, device_id, is_alive, response_time_ms, round_num) "
"VALUES (2, 1, 1, 8.5, 7)"
)
)
await connection.execute(
text(
"INSERT INTO alert_events "
"(id, device_id, alert_type, message, start_at, notification_sent) "
"VALUES (3, 1, 'offline', 'legacy event', CURRENT_TIMESTAMP, 1)"
)
)
await connection.run_sync(migration.run_reliability_migration)
table_names = await connection.run_sync(
lambda sync_connection: set(
inspect(sync_connection).get_table_names()
)
)
assert "future_unrelated_model" not in table_names
assert "notification_outbox" in table_names
Base.metadata.remove(future_table)
await connection.run_sync(Base.metadata.create_all)
async with engine.begin() as connection:
await connection.run_sync(migration.run_reliability_migration)
legacy_device = await connection.execute(
text("SELECT name, ip, current_status FROM devices WHERE id = 1")
)
legacy_probe = await connection.execute(
text(
"SELECT device_id, is_alive, response_time_ms, round_num "
"FROM ping_records WHERE id = 2"
)
)
legacy_alert = await connection.execute(
text(
"SELECT device_id, alert_type, message, notification_sent "
"FROM alert_events WHERE id = 3"
)
)
probe_columns, alert_columns, probe_indexes = await connection.run_sync(
lambda sync_connection: (
{
column["name"]: column
for column in inspect(sync_connection).get_columns(
"ping_records"
)
},
{
column["name"]
for column in inspect(sync_connection).get_columns(
"alert_events"
)
},
{
index["name"]
for index in inspect(sync_connection).get_indexes(
"ping_records"
)
},
)
)
assert legacy_device.one() == ("legacy-router", "192.0.2.10", "online")
assert legacy_probe.one() == (1, 1, 8.5, 7)
assert legacy_alert.one() == (1, "offline", "legacy event", 1)
assert probe_columns["is_valid"]["nullable"] is False
assert {
"sent_count",
"received_count",
"packet_loss_percent",
"average_rtt_ms",
"failure_reason",
}.issubset(probe_columns)
assert "related_event_id" in alert_columns
assert "idx_ping_records_device_created_at" in probe_indexes
finally:
if future_table in Base.metadata.tables.values():
Base.metadata.remove(future_table)
await engine.dispose()
async def test_migration_does_not_create_a_future_ping_record_index(
tmp_path: Path,
):
"""Historical migration DDL stays limited to its named composite index."""
migration = importlib.import_module(
"migrations.versions.20260803_reliability_monitoring"
)
future_index = Index(
"idx_future_ping_record_round_num",
models.PingRecord.__table__.c.round_num,
)
engine = create_async_engine(
f"sqlite+aiosqlite:///{tmp_path / 'future-index.db'}"
)
try:
async with engine.begin() as connection:
await connection.execute(
text(
"CREATE TABLE ping_records ("
"id INTEGER PRIMARY KEY, device_id INTEGER NOT NULL, "
"is_alive BOOLEAN NOT NULL, response_time_ms FLOAT, "
"round_num INTEGER NOT NULL, created_at DATETIME)"
)
)
await connection.run_sync(migration.run_reliability_migration)
index_names = await connection.run_sync(
lambda sync_connection: {
index["name"]
for index in inspect(sync_connection).get_indexes(
"ping_records"
)
}
)
assert "idx_ping_records_device_created_at" in index_names
assert "idx_future_ping_record_round_num" not in index_names
finally:
models.PingRecord.__table__.indexes.remove(future_index)
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
+68
View File
@@ -0,0 +1,68 @@
import pytest
from pydantic import ValidationError
from app.config import Settings
from app.services.settings_validation import validate_runtime_settings
def test_production_rejects_legacy_jwt_secret():
settings = Settings(
environment="production",
secret_key="change-me-to-a-long-random-string",
)
with pytest.raises(ValueError, match="SECRET_KEY"):
validate_runtime_settings(settings)
@pytest.mark.parametrize("secret_key", ["", " ", "\t\n"])
def test_production_rejects_blank_jwt_secret(secret_key):
settings = Settings(environment="production", secret_key=secret_key)
with pytest.raises(ValueError, match="SECRET_KEY"):
validate_runtime_settings(settings)
@pytest.mark.parametrize("environment", ["prod", "Production", "staging"])
def test_environment_must_be_an_explicit_supported_value(environment):
with pytest.raises(ValidationError, match="environment"):
Settings(environment=environment)
def test_probe_packet_count_must_be_positive():
with pytest.raises(ValidationError):
Settings(probe_packets_per_round=0)
@pytest.mark.parametrize(
("field", "invalid_value"),
[
("offline_consecutive_rounds", 0),
("degraded_window_rounds", 1),
("degraded_loss_percent", 0.0),
("recovery_consecutive_clean_rounds", 0),
("wecom_notification_max_attempts", 0),
("wecom_retry_base_seconds", 0),
],
)
def test_monitoring_and_retry_settings_reject_values_below_their_bounds(field, invalid_value):
with pytest.raises(ValidationError):
Settings(**{field: invalid_value})
def test_production_rejects_tls_disabled_casdoor():
settings = Settings(
environment="production",
secret_key="a-safe-production-secret",
casdoor_endpoint="http://casdoor.internal",
)
with pytest.raises(ValueError, match="CASDOOR"):
validate_runtime_settings(settings)
def test_enabled_wecom_delivery_requires_all_credentials():
settings = Settings(wecom_notification_enabled=True, WECOM_CORP_ID="corp")
with pytest.raises(ValueError, match="WECOM"):
validate_runtime_settings(settings)
+139
View File
@@ -0,0 +1,139 @@
"""Unit tests for the connectivity health state machine."""
from dataclasses import dataclass
import pytest
from app.models.device import DeviceMonitoringPolicy
from app.services.state_machine import evaluate_health
@dataclass(frozen=True)
class RecentProbe:
"""Minimal real-data-shaped record consumed by the pure state machine."""
sent_count: int
received_count: int
is_valid: bool = True
def loss(sent_count: int) -> RecentProbe:
return RecentProbe(sent_count=sent_count, received_count=0)
def partial(sent_count: int, received_count: int) -> RecentProbe:
return RecentProbe(
sent_count=sent_count,
received_count=received_count,
)
def clean(sent_count: int) -> RecentProbe:
return RecentProbe(sent_count=sent_count, received_count=sent_count)
@pytest.fixture
def policy() -> DeviceMonitoringPolicy:
return DeviceMonitoringPolicy(
probe_packets_per_round=3,
offline_consecutive_rounds=2,
degraded_window_rounds=5,
degraded_loss_percent=20.0,
recovery_consecutive_clean_rounds=3,
)
def test_two_full_loss_rounds_take_device_offline(policy):
"""Configured consecutive full-loss rounds transition to offline once."""
decision = evaluate_health("online", [loss(3), loss(3)], policy)
assert (decision.next_status, decision.event_type) == (
"offline",
"offline",
)
assert decision.should_notify is True
def test_offline_priority_wins_over_aggregate_degraded_loss(policy):
"""Full-loss streaks are classified offline even when the window also degrades."""
recent = [partial(3, 2)] * 3 + [loss(3), loss(3)]
decision = evaluate_health("online", recent, policy)
assert decision.next_status == "offline"
assert decision.event_type == "offline"
def test_five_round_window_with_loss_is_degraded(policy):
"""Aggregate loss at the configured window threshold is degraded."""
decision = evaluate_health(
"online",
[partial(3, 2)] * 5,
policy,
)
assert decision.next_status == "degraded"
assert decision.event_type == "degraded"
def test_degraded_requires_a_complete_window(policy):
"""A partial history cannot satisfy an aggregate-window decision."""
decision = evaluate_health(
"online",
[partial(3, 2)] * 4,
policy,
)
assert decision.next_status == "online"
assert decision.event_type is None
def test_three_clean_rounds_recovers(policy):
"""An alerted device recovers only after the configured clean streak."""
decision = evaluate_health(
"offline",
[clean(3)] * 3,
policy,
)
assert (decision.next_status, decision.event_type) == (
"online",
"recovered",
)
assert decision.should_notify is True
def test_clean_online_device_does_not_emit_duplicate_event(policy):
"""Clean probes retain an already-online state without notifying."""
decision = evaluate_health(
"online",
[clean(3)] * 5,
policy,
)
assert decision.next_status == "online"
assert decision.event_type is None
assert decision.should_notify is False
def test_latest_invalid_probe_retains_state(policy):
"""A malformed current result cannot reuse old loss to trigger a fault."""
recent = [loss(3), RecentProbe(3, 0, is_valid=False)]
decision = evaluate_health("online", recent, policy)
assert decision.next_status == "online"
assert decision.event_type is None
assert decision.should_notify is False
def test_offline_does_not_recover_before_clean_streak_is_complete(policy):
"""Recovery debounce retains offline until enough clean rounds exist."""
decision = evaluate_health(
"offline",
[clean(3), clean(3)],
policy,
)
assert decision.next_status == "offline"
assert decision.event_type is None
+5 -7
View File
@@ -6,10 +6,9 @@ services:
env_file:
- ./backend/.env
environment:
- DATABASE_URL=postgresql+asyncpg://pingwatch:pingwatch123@db:5432/pingwatch
- TZ=Asia/Shanghai
ports:
- "8001:8000"
- "127.0.0.1:8001:8000"
depends_on:
db:
condition: service_healthy
@@ -18,8 +17,7 @@ services:
networks:
- pingwatch-net
cap_add:
- NET_RAW # 允许 ICMP ping
- NET_ADMIN # 允许原始套接字
- NET_RAW
frontend:
build: ./frontend
@@ -40,9 +38,9 @@ services:
restart: unless-stopped
environment:
- TZ=Asia/Shanghai
- POSTGRES_DB=pingwatch
- POSTGRES_USER=pingwatch
- POSTGRES_PASSWORD=pingwatch123
- POSTGRES_DB=${POSTGRES_DB:-pingwatch}
- POSTGRES_USER=${POSTGRES_USER:-pingwatch}
- POSTGRES_PASSWORD=${POSTGRES_PASSWORD:?POSTGRES_PASSWORD must be set in backend/.env}
volumes:
- postgres_data:/var/lib/postgresql/data
healthcheck:
@@ -0,0 +1,465 @@
# PingWatch 可靠连通性监测与企业微信告警 Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.
**Goal:** 将 PingWatch 改造成可区分离线与间歇性丢包故障、可靠发送企业微信应用通知,并能在 Ubuntu/OpenResty 环境安全运行的监测服务。
**Architecture:** fping 每轮为每个设备取得多包汇总,独立状态机依据连续轮次和滑动窗口计算 online/degraded/offline。状态转换在同一数据库事务内创建事件与通知出箱;独立投递器重试企业微信消息。OpenResty 在宿主机提供 Vue 构建文件并反向代理仅绑定回环地址的 FastAPI 服务。
**Tech Stack:** Python 3.12、FastAPI、SQLAlchemy asyncio、PostgreSQL 16、fping、httpx、pytest、Vue 3、Element Plus、OpenResty、Docker Compose。
## Global Constraints
- 目标能力等级暂按 C1;所有需求使用 PINGWATCH_(可靠性监测)_001 追溯。
- 默认检测周期为 30 秒,每轮 3 包;离线为连续 2 轮 100% 丢包,故障为最近 5 轮至少 20% 丢包,恢复为连续 3 轮零丢包。
- 只监测显式录入且 is_enabled=true 的 IPv4/IPv6 地址;不得增加网段发现或端口扫描。
- 凭据仅通过运行时环境变量注入;Git、镜像、日志、API 响应和文档不得包含 secret、token 或口令。
- 企业微信默认向应用可见范围发送;不提供系统页面维护接收人。可选 WECOM_TO_PARTY 只允许由受控环境变量设置。
- 后端仅以 127.0.0.1:8001 暴露给 Ubuntu 宿主机 OpenRestyPostgreSQL 不映射宿主机端口。
- 不可信输入必须经 Pydantic/显式校验,fping 只通过参数数组启动,禁止 shell=True。
- 除非命令明确指定其他目录,所有 Python pytest 命令均从 backend/ 目录运行。
---
## Planned File Structure
| 文件 | 责任 |
| --- | --- |
| backend/app/models/{device,ping_record,alert_event,notification_outbox}.py | 设备策略、探测汇总、领域事件及通知出箱持久化。 |
| backend/app/services/{fping_runner,state_machine,pinger,notification_dispatcher,alerter}.py | 命令输出解析、状态计算、轮次编排、可靠投递和事件创建。 |
| backend/app/api/{devices,alerts,health}.py 与 backend/app/schemas/* | 受保护设备/事件接口和不含敏感信息的健康探针。 |
| backend/tests/* | 状态、解析、投递、API、配置和迁移验证。 |
| frontend/src/views/{Devices,Alerts,Dashboard}.vue | 健康度、探测质量、事件和投递结果展示。 |
| deploy/openresty、deploy/systemd、docs/operations | OpenResty、受控启动、部署、回退、备份运行材料。 |
## Task 1: 建立测试与运行配置契约
**Files:**
- Modify: backend/requirements.txt, backend/app/config.py, .gitignore
- Create: backend/.env.example, backend/pytest.ini, backend/app/services/settings_validation.py
- Create: backend/tests/conftest.py, backend/tests/test_settings_validation.py
**Interfaces:**
- Produces: validate_runtime_settings(settings: Settings) -> None.
- Produces: monitoring defaults and bounded retry/WeCom configuration fields.
- [ ] **Step 1: Write the failing configuration tests**
~~~python
import pytest
from pydantic import ValidationError
from app.config import Settings
from app.services.settings_validation import validate_runtime_settings
def test_production_rejects_legacy_jwt_secret():
settings = Settings(environment="production", secret_key="change-me-to-a-long-random-string")
with pytest.raises(ValueError, match="SECRET_KEY"):
validate_runtime_settings(settings)
def test_probe_packet_count_must_be_positive():
with pytest.raises(ValidationError):
Settings(probe_packets_per_round=0)
~~~
- [ ] **Step 2: Run test to verify it fails**
Run: python -m pytest backend/tests/test_settings_validation.py -v
Expected: FAIL because settings fields and validation module are absent.
- [ ] **Step 3: Write minimal configuration implementation**
Add pinned test dependencies pytest, pytest-asyncio and aiosqlite. Add Settings fields: environment, probe_packets_per_round=3 (1..10), offline_consecutive_rounds=2 (1..10), degraded_window_rounds=5 (2..60), degraded_loss_percent=20.0 (1..100), recovery_consecutive_clean_rounds=3 (1..20), wecom_notification_max_attempts, wecom_retry_base_seconds and WECOM_TO_PARTY. validate_runtime_settings rejects production default JWT keys, production TLS-disabled Casdoor, and enabled WeCom delivery without all three credentials.
~~~python
def validate_runtime_settings(settings: Settings) -> None:
if settings.environment == "production" and settings.secret_key == LEGACY_DEFAULT_SECRET:
raise ValueError("SECRET_KEY must be provided by the runtime environment")
~~~
Create backend/.env.example with empty secret values only; keep real .env ignored.
- [ ] **Step 4: Run focused verification**
Run: python -m pytest backend/tests/test_settings_validation.py -v
Expected: PASS.
- [ ] **Step 5: Commit**
~~~bash
git add backend/requirements.txt backend/pytest.ini backend/.env.example .gitignore backend/app/config.py backend/app/services/settings_validation.py backend/tests
git commit -m "test: establish monitoring configuration contract"
~~~
## Task 2: 持久化探测质量、设备策略、事件与通知出箱
**Files:**
- Modify: backend/app/models/device.py, backend/app/models/ping_record.py, backend/app/models/alert_event.py, backend/app/models/__init__.py, backend/app/core/deps.py
- Create: backend/app/models/notification_outbox.py, backend/migrations/versions/20260803_reliability_monitoring.py
- Create: backend/tests/test_models_and_migration.py
**Interfaces:**
- Produces: DeviceMonitoringPolicy.from_device(device, settings) -> DeviceMonitoringPolicy.
- Produces: PingRecord.sent_count, received_count, packet_loss_percent, average_rtt_ms, is_valid, failure_reason.
- Produces: NotificationOutbox with status pending, sending, sent, failed.
- [ ] **Step 1: Write failing persistence tests**
~~~python
async def test_probe_record_stores_packet_summary(db_session):
record = PingRecord(device_id=1, sent_count=3, received_count=2,
packet_loss_percent=33.33, average_rtt_ms=12.5, is_valid=True)
db_session.add(record)
await db_session.commit()
assert record.packet_loss_percent == 33.33
async def test_outbox_defaults_to_pending(db_session):
item = NotificationOutbox(alert_event_id=1, message_content="masked summary")
db_session.add(item)
await db_session.commit()
assert item.status == NotificationStatus.pending
~~~
- [ ] **Step 2: Run test to verify it fails**
Run: python -m pytest backend/tests/test_models_and_migration.py -v
Expected: FAIL because fields and NotificationOutbox do not exist.
- [ ] **Step 3: Implement models and migration**
Extend Device with nullable per-device policy overrides and allow degraded status. Add an index on probe (device_id, created_at), and a status/next_attempt_at index on outbox. Add AlertTypeEnum.degraded and fields previous_status, current_status, last_notification_error, notification_attempts. Implement one idempotent migration runner called before create_all; it records revision 20260803_reliability_monitoring in schema_migrations and never deletes existing data.
- [ ] **Step 4: Run focused verification**
Run: python -m pytest backend/tests/test_models_and_migration.py -v
Expected: PASS, including two sequential migration executions against the same database.
- [ ] **Step 5: Commit**
~~~bash
git add backend/app/models backend/app/core/deps.py backend/migrations backend/tests/test_models_and_migration.py
git commit -m "feat: persist probe quality and notification outbox"
~~~
## Task 3: 实现多包 fping 解析和纯状态机
**Files:**
- Create: backend/app/services/fping_runner.py, backend/app/services/state_machine.py
- Modify: backend/app/services/pinger.py
- Create: backend/tests/test_fping_runner.py, backend/tests/test_state_machine.py
**Interfaces:**
- Produces: ProbeResult(ip, sent_count, received_count, average_rtt_ms, is_valid, failure_reason).
- Produces: parse_fping_count_output(output, expected_ips, packets_per_round) -> dict[str, ProbeResult].
- Produces: evaluate_health(previous_status, recent, policy) -> StateDecision.
- [ ] **Step 1: Write failing parser and transition tests**
~~~python
def test_parses_loss_and_average_rtt():
output = "10.0.0.8 : 12.4 - 12.6\n10.0.0.9 : - - -"
result = parse_fping_count_output(output, {"10.0.0.8", "10.0.0.9"}, 3)
assert result["10.0.0.8"].received_count == 2
assert result["10.0.0.9"].packet_loss_percent == 100.0
def test_two_full_loss_rounds_take_device_offline(policy):
decision = evaluate_health("online", [loss(3), loss(3)], policy)
assert (decision.next_status, decision.event_type) == ("offline", "offline")
def test_five_round_window_with_loss_is_degraded(policy):
decision = evaluate_health("online", [partial(3, 2)] * 5, policy)
assert decision.next_status == "degraded"
def test_three_clean_rounds_recovers(policy):
decision = evaluate_health("offline", [clean(3)] * 3, policy)
assert (decision.next_status, decision.event_type) == ("online", "recovered")
~~~
- [ ] **Step 2: Run test to verify it fails**
Run: python -m pytest backend/tests/test_fping_runner.py backend/tests/test_state_machine.py -v
Expected: FAIL because runner and state machine modules are absent.
- [ ] **Step 3: Implement minimal parser and state machine**
Invoke fping with the argument sequence fping -C <count> -q -t <milliseconds> <ip...>; capture stdout and stderr because quiet count output is emitted to stderr. Numeric reply tokens count as success and dash tokens as loss. Missing/malformed expected IP output creates is_valid=False and cannot trigger a device failure.
State priority: invalid records retain state; configured consecutive full loss gives offline; configured aggregate window loss gives degraded; only offline/degraded plus configured clean records gives online; otherwise retain state. StateDecision contains next_status, event_type, should_notify, reason.
- [ ] **Step 4: Refactor Pinger to persist exactly one summary per enabled device**
Pinger.run_one_round saves every ProbeResult, reads the required recent valid records, computes StateDecision, updates Device, and returns DeviceStateChange values within one transaction. It must use asyncio.create_subprocess_exec(*command) and never a command string.
- [ ] **Step 5: Run focused verification**
Run: python -m pytest backend/tests/test_fping_runner.py backend/tests/test_state_machine.py -v
Expected: PASS.
- [ ] **Step 6: Commit**
~~~bash
git add backend/app/services/fping_runner.py backend/app/services/state_machine.py backend/app/services/pinger.py backend/tests/test_fping_runner.py backend/tests/test_state_machine.py
git commit -m "feat: classify offline and degraded connectivity"
~~~
## Task 4: 用事务出箱可靠投递企业微信事件
**Files:**
- Create: backend/app/services/notification_dispatcher.py
- Modify: backend/app/services/alerter.py, backend/app/services/scheduler.py, backend/app/services/pinger.py
- Create: backend/tests/test_notification_dispatcher.py, backend/tests/test_alert_workflow.py
**Interfaces:**
- Consumes: list[DeviceStateChange] from Pinger.run_one_round().
- Produces: Alerter.record_transition(change, db) -> AlertEvent.
- Produces: NotificationDispatcher.dispatch_due(db, now) -> DispatchSummary.
- Produces: WeComClient.send_text(content) -> DeliveryResult.
- [ ] **Step 1: Write failing outbox tests**
~~~python
async def test_transition_creates_event_and_pending_outbox_in_one_commit(db_session, change):
event = await Alerter().record_transition(change, db_session)
outbox = await db_session.scalar(select(NotificationOutbox).where(
NotificationOutbox.alert_event_id == event.id))
assert outbox.status == NotificationStatus.pending
async def test_retryable_failure_reschedules(db_session, pending_message, frozen_time):
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
~~~
- [ ] **Step 2: Run test to verify it fails**
Run: python -m pytest backend/tests/test_notification_dispatcher.py backend/tests/test_alert_workflow.py -v
Expected: FAIL because dispatcher interfaces are absent.
- [ ] **Step 3: Implement atomic event/outbox recording**
Replace in-memory pending alerts. record_transition inserts AlertEvent and NotificationOutbox before caller commit. Offline escalation annotates/resolves the prior open degraded event. Recovery closes one unresolved event, stores duration, and creates a recovered notification. Render notifications with event type, device, IP, location, project, loss summary, time and duration; split batches at a fixed safe count.
- [ ] **Step 4: Implement WeCom client and retry dispatcher**
Use httpx.AsyncClient with explicit connect/read/write timeout. Cache access_token until server expiry minus safety margin. Default payload uses touser="@all"; only use toparty when WECOM_TO_PARTY is non-empty. Never log token-bearing URLs, request JSON, response body, secret, or message content. Retry transport/timeouts, 429 and 5xx with next_attempt_at = now + base_seconds * 2 ** (attempt_count - 1); mark failed at configured maximum.
- [ ] **Step 5: Connect dispatcher to scheduler**
Run dispatch_due after a successful probe transaction and at a small bounded interval. The scheduler has one task and a lock; cancellation waits for the active probe/dispatch cycle to close its session.
- [ ] **Step 6: Run verification and commit**
Run: python -m pytest backend/tests/test_notification_dispatcher.py backend/tests/test_alert_workflow.py -v
Expected: PASS.
~~~bash
git add backend/app/services/notification_dispatcher.py backend/app/services/alerter.py backend/app/services/scheduler.py backend/app/services/pinger.py backend/tests/test_notification_dispatcher.py backend/tests/test_alert_workflow.py
git commit -m "feat: deliver alert events through persistent outbox"
~~~
## Task 5: 校验设备输入并提供健康/审计 API
**Files:**
- Modify: backend/app/schemas/device.py, backend/app/schemas/alert.py, backend/app/api/devices.py, backend/app/api/alerts.py, backend/app/main.py
- Create: backend/app/api/health.py
- Create: backend/tests/test_devices_api.py, backend/tests/test_alerts_api.py, backend/tests/test_health_api.py
**Interfaces:**
- Produces: DeviceCreate and DeviceUpdate using IPvAnyAddress and bounded policy fields.
- Produces: GET /api/health -> {"status": "ok", "database": "ready"} after SELECT 1.
- Produces: alert DTO transition and notification-status fields.
- [ ] **Step 1: Write failing API tests**
~~~python
async def test_create_rejects_command_like_ip(authenticated_client):
response = await authenticated_client.post("/api/devices",
json={"name": "bad", "ip": "127.0.0.1;id"})
assert response.status_code == 422
async def test_health_requires_database_readiness(client):
response = await client.get("/api/health")
assert response.json() == {"status": "ok", "database": "ready"}
~~~
- [ ] **Step 2: Run test to verify it fails**
Run: python -m pytest backend/tests/test_devices_api.py backend/tests/test_alerts_api.py backend/tests/test_health_api.py -v
Expected: FAIL on missing validation/health behavior.
- [ ] **Step 3: Implement schema and import validation**
Use IPvAnyAddress, Field bounds and field_validator to map blank overrides to None. Before decoding CSV enforce CSV_MAX_BYTES; reject invalid UTF-8 as 400; limit returned line errors to 100; de-duplicate both database IPs and earlier accepted CSV rows. Device and alert responses expose only aggregate notification status, never message content or external error body.
- [ ] **Step 4: Implement health and API projections**
Health router runs SELECT 1. The lifespan calls validate_runtime_settings before scheduling. CORS accepts only exact configured origins. Device lists return latest valid summary and policy fields; alerts return degraded/recovered filters, transition and retry fields.
- [ ] **Step 5: Run verification and commit**
Run: python -m pytest backend/tests/test_devices_api.py backend/tests/test_alerts_api.py backend/tests/test_health_api.py -v
Expected: PASS.
~~~bash
git add backend/app/schemas backend/app/api backend/app/main.py backend/tests/test_devices_api.py backend/tests/test_alerts_api.py backend/tests/test_health_api.py
git commit -m "feat: expose validated monitoring health and event APIs"
~~~
## Task 6: 更新前端运维视图
**Files:**
- Modify: frontend/package.json, frontend/package-lock.json, frontend/src/api/index.js
- Modify: frontend/src/views/Devices.vue, frontend/src/views/Alerts.vue, frontend/src/views/Dashboard.vue
- Create: frontend/src/utils/status.js, frontend/src/utils/status.test.js
**Interfaces:**
- Produces: statusLabel(status) and statusTagType(status) supporting online, degraded, offline, unknown.
- Consumes: enhanced Task 5 device and alert DTOs.
- [ ] **Step 1: Write a failing presentation test**
~~~javascript
import { describe, expect, it } from "vitest"
import { statusLabel, statusTagType } from "./status"
describe("status presentation", () => {
it("marks degraded connectivity as a warning", () => {
expect(statusLabel("degraded")).toBe("故障")
expect(statusTagType("degraded")).toBe("warning")
})
})
~~~
- [ ] **Step 2: Run test to verify it fails**
Run: npm --prefix frontend run test -- --run
Expected: FAIL because the test script is absent.
- [ ] **Step 3: Implement minimal UI contract**
Add vitest development dependency and test script. Add status utility with explicit mappings. Devices view adds state, packet loss, RTT and last probe columns plus bounded policy form fields. Alerts view adds degraded filter, transition, delivery status and retry count. Dashboard adds degraded count. Use Vue interpolation exclusively; never use v-html for device data.
- [ ] **Step 4: Run verification and commit**
Run: npm --prefix frontend run test -- --run
Expected: PASS.
Run: npm --prefix frontend run build
Expected: PASS.
~~~bash
git add frontend/package.json frontend/package-lock.json frontend/src/api/index.js frontend/src/utils frontend/src/views/Devices.vue frontend/src/views/Alerts.vue frontend/src/views/Dashboard.vue
git commit -m "feat: display degraded connectivity and delivery state"
~~~
## Task 7: 加固认证、容器暴露与 Ubuntu/OpenResty 交付
**Files:**
- Modify: backend/app/core/auth.py, backend/Dockerfile, docker-compose.yml
- Create: deploy/openresty/pingwatch.conf, deploy/systemd/pingwatch.service
- Create: docs/operations/ubuntu-openresty-deployment.md, docs/operations/rollback-and-backup.md
- Create: backend/tests/test_auth_security.py, deploy/tests/test_compose_exposure.ps1
**Interfaces:**
- Produces: exchange_code_for_user(code) that never accepts unverified JWT claims in production.
- Produces: OpenResty routes for /, /api/, /ws and /api/health.
- [ ] **Step 1: Write failing security/deployment assertions**
~~~python
async def test_production_rejects_unverified_casdoor_token(monkeypatch):
monkeypatch.setattr(settings, "environment", "production")
monkeypatch.setattr(settings, "casdoor_certificate", "")
assert await exchange_code_for_user("code") is None
~~~
~~~powershell
$compose = Get-Content -Raw .\docker-compose.yml
if ($compose -match 'POSTGRES_PASSWORD=') { throw 'tracked compose contains password' }
if ($compose -match 'NET_ADMIN') { throw 'container has excess capability' }
if ($compose -notmatch '127.0.0.1:8001:8000') { throw 'backend is not loopback-bound' }
~~~
- [ ] **Step 2: Run test to verify it fails**
Run: python -m pytest backend/tests/test_auth_security.py -v
Expected: FAIL because OAuth disables TLS verification or accepts unverified claims.
Run: pwsh -File deploy/tests/test_compose_exposure.ps1
Expected: FAIL because Compose contains an embedded password and NET_ADMIN.
- [ ] **Step 3: Implement minimal hardening**
Require certificate/TLS verification before accepting production Casdoor id_token; remove unverified-claims fallback. Compose reads database credentials only from server-only environment file, exposes backend as 127.0.0.1:8001:8000, does not run frontend container in production, does not expose database, uses cap_drop ALL plus cap_add NET_RAW and no-new-privileges. Backend image has non-secret health check.
- [ ] **Step 4: Add OpenResty and operating material**
OpenResty listens on the agreed IP port, serves /opt/pingwatch/frontend, uses SPA try_files, proxies /api/ to 127.0.0.1:8001 and upgrades /ws, limits request body to 15m, and has separate access/error logs. Deployment guide gives preflight, server-only env permissions, build/copy, Compose lifecycle, OpenResty test/reload, health/API/WebSocket smoke tests and approved rollback. Backup guide defines PostgreSQL volume backup before upgrades and restore of prior frontend build/image.
- [ ] **Step 5: Run verification and commit**
Run: python -m pytest backend/tests/test_auth_security.py -v
Expected: PASS.
Run: pwsh -File deploy/tests/test_compose_exposure.ps1
Expected: PASS.
Run: docker compose config
Expected: PASS with no database port and no plaintext secret in tracked Compose file.
~~~bash
git add backend/app/core/auth.py backend/Dockerfile docker-compose.yml deploy docs/operations backend/tests/test_auth_security.py
git commit -m "feat: secure OpenResty deployment package"
~~~
## Task 8: 全量验证、追溯证据与发布准备
**Files:**
- Create: docs/traceability/2026-08-03-reliability-monitoring.md
- Create: docs/test-reports/2026-08-03-reliability-monitoring.md
- Create: README.md
**Interfaces:**
- Produces: each of five requirement IDs mapped to code, test and Ubuntu/OpenResty validation evidence.
- [ ] **Step 1: Execute full automated verification**
Run: python -m pytest backend/tests -v from backend/
Expected: PASS.
Run: npm --prefix frontend run test -- --run
Expected: PASS.
Run: npm --prefix frontend run build
Expected: PASS.
Run: docker compose config
Expected: PASS.
- [ ] **Step 2: Execute local Compose smoke test**
Run: docker compose up --build -d, then curl -fsS http://127.0.0.1:8001/api/health, then docker compose down.
Expected: health contains status=ok and database=ready; no test data or secret file is committed.
- [ ] **Step 3: Write traceability/test reports**
Traceability maps requirements _01 through _05 to implementation file, exact automated test name and Ubuntu/OpenResty validation command. Test report records command, timestamp, result, limitations, and states that real network validation needs authorized test IPs and configured enterprise WeChat credentials. It excludes credentials, private IP inventories and full notification contents.
- [ ] **Step 4: Scan final changes and commit evidence**
Run: git diff main...HEAD --check
Expected: PASS.
Run: rg -n '(SECRET|PASSWORD|TOKEN|PRIVATE KEY)\s*=\s*[^" ]+' -g '!*.example' -g '!docs/superpowers/**' .
Expected: no hardcoded secret assignment.
~~~bash
git add README.md docs/traceability docs/test-reports
git commit -m "docs: add reliability monitoring verification evidence"
~~~
## Self-Review
- Spec coverage: Tasks 1-4 implement multi-packet probing, state transitions, durable notification, recovery and batch/node safety; Tasks 5-6 implement API/UI; Task 7 implements OpenResty, secrets, authentication and container controls; Task 8 provides C1 deployment evidence and traceability.
- Placeholder scan: the plan contains executable commands, named files, test behavior and data contracts for each task.
- Type consistency: ProbeResult, DeviceMonitoringPolicy, StateDecision, DeviceStateChange, NotificationOutbox and DeliveryResult are defined before their consuming tasks.
@@ -0,0 +1,119 @@
# PingWatch 连通性监测与企业微信告警改造设计
**日期:** 2026-08-03
**状态:** 已确认,待实现
**需求编号:** `PINGWATCH_(可靠性监测)_001`
**目标能力等级:** 暂按 C1 基线,待项目负责人确认
## 1. 目标、范围与验收
### 1.1 目标
将现有的批量 ICMP 连通性检测改造成可区分离线与业务故障的监测系统,并向企业微信应用可见部门发送及时、可追溯的通知。
| 编号 | 可验证需求 | 验收条件 |
| --- | --- | --- |
| `PINGWATCH_(可靠性监测)_001_01` | 及时发现离线设备 | 每 30 秒检测一轮,每轮 3 个 ICMP 包;连续 2 轮 100% 丢包进入 `offline` 并创建/通知事件。 |
| `PINGWATCH_(可靠性监测)_001_02` | 识别影响业务的间歇性丢包 | 最近 5 轮(15 包)丢包率不低于 20%、但未离线时进入 `degraded` 并通知。 |
| `PINGWATCH_(可靠性监测)_001_03` | 发现设备恢复 | 已告警设备连续 3 轮零丢包后进入 `online`,关闭关联事件并通知恢复。 |
| `PINGWATCH_(可靠性监测)_001_04` | 通知可靠可追溯 | 企业微信调用失败时事件保留,按有限退避重试;每次投递结果可查询。 |
| `PINGWATCH_(可靠性监测)_001_05` | Ubuntu/OpenResty 运行 | 以 `http://10.10.10.14:<port>` 提供访问;OpenResty 承载静态站点并反代 API/WebSocket。 |
### 1.2 非目标
- 不增加未授权网段扫描、端口扫描或资产发现。
- 不在首期实现按历史基线自动学习阈值。
- 不自建企业微信成员/部门管理,应用可见范围及接收部门由企业微信管理端控制。
### 1.3 数据分类与风险
- 设备名称、IP、位置、项目和告警记录按内部运维数据处理;企业微信密钥、JWT 密钥、数据库口令和 OAuth 凭据为敏感配置。
- 主要风险:网络抖动误告警、监控节点自身断网、企业微信不可用、存储膨胀、特权过大、凭据泄露。
## 2. 方案比较与决策
| 方案 | 优点 | 缺点 | 决策 |
| --- | --- | --- | --- |
| 单包连续失败阈值 | 简单 | 无法量化间歇性丢包 | 不采用 |
| 多包探测 + 滑动窗口状态机 | 规则可解释、及时且抗抖动 | 需要保存汇总数据 | **采用** |
| 自适应历史基线 | 对不同链路更精细 | 学习期、复杂度和解释成本高 | 后续评估 |
## 3. 架构与数据流
```text
受管 IP → fping 批量探测 → 轮次汇总/状态机 → 事件出箱 → 企业微信应用(应用可见范围)
│ │
├→ PostgreSQL ←──────┤
└→ API/WebSocket → OpenResty → 浏览器(IP:端口)
```
1. 调度器不允许检测轮次重叠;每 30 秒加载启用设备并批量执行 `fping`
2. 一轮记录每台设备的发包数、收包数、丢包率、平均 RTT 与结果有效性。命令或解析异常产生系统事件,不能被当作设备丢包。
3. 状态机基于最新有效检测记录计算状态;状态改变时原子地写入领域事件和待投递记录。
4. 投递 worker 从数据库读取待发送记录,向企业微信应用发送应用消息;由企业微信应用的可见范围限定接收部门。成功、失败、下次重试时间均持久化。
5. 前端通过 API/WebSocket 展示设备当前健康度、最近丢包率、时延与事件投递结果。
## 4. 状态机与告警规则
| 现状态 | 条件 | 新状态 | 动作 |
| --- | --- | --- | --- |
| `unknown` / `online` | 连续 2 轮均 100% 丢包 | `offline` | 创建离线事件,进入待通知队列。 |
| `unknown` / `online` | 最近 5 轮丢包率 ≥20%,且不满足离线 | `degraded` | 创建业务故障事件,进入待通知队列。 |
| `degraded` | 满足离线规则 | `offline` | 升级现有故障事件,通知离线。 |
| `offline` / `degraded` | 连续 3 轮零丢包 | `online` | 关闭未恢复事件,发送恢复通知。 |
| 任意 | 无有效检测结果 | 保持原状态 | 创建受限频率的系统事件;不发送设备故障通知。 |
- 同一事件只在首次状态变更时通知;未恢复事件默认每 4 小时提醒一次,间隔可配置。
- 每个设备可覆盖全局规则:检测间隔、单轮发包数、离线轮数、故障窗口、故障丢包率、恢复轮数和提醒间隔。
- 上游心跳失败或单轮大面积离线时,创建“监控节点异常/批量故障”系统事件并通知;不将设备事件静默丢弃。
## 5. 数据、接口与页面
### 5.1 数据模型
- 扩展 `ping_records``sent_count``received_count``packet_loss_percent``average_rtt_ms``is_valid``failure_reason`
- 扩展 `devices`:监测策略覆盖字段与 `current_status`(加入 `degraded`)。
- 扩展 `alert_events`:加入 `degraded`、状态变更前后值、关联恢复事件、通知尝试数、最后通知错误和下一次通知时间。
- 新增 `notification_outbox`:为每个待发送通知保存内容摘要、投递范围摘要、状态、尝试次数、锁定时间和投递结果。
数据库变更应使用可重复执行、可回退的迁移;保留现有数据。
### 5.2 API 与 UI
- 设备 API 返回当前状态、最新有效结果和设备级策略;创建、编辑、批量导入均校验 IPv4/IPv6、数值范围、单文件大小、字符编码及重复 IP。
- 告警 API 支持 `offline``degraded``recovered``system` 筛选,并返回持续时长、恢复关联和通知状态。
- 设备列表展示“在线/故障/离线/未知”、最近丢包率、时延和最近检测时间;告警页展示事件、持续时间和投递状态。
## 6. 企业微信与错误处理
- 应用使用环境变量提供的 Corp ID、Agent ID、Secret;默认向应用可见范围内的成员发送,接收部门仅在企业微信应用管理端配置。若后续需要缩小范围,可增加受控的 `WECOM_TO_PARTY` 环境变量,不在系统页面维护接收人。
- access token 仅在进程内带有效期缓存;请求使用明确连接/读取超时。失败事件按有限次数的指数退避重试,达到上限后标记失败,保留人工处理证据。
- 日志只记录错误码、事件 ID、设备 ID、轮次和重试次数;不得记录 token、secret、口令或完整敏感响应。
- 认证/OAuth 必须校验证书与 JWT 签名;不得以 `verify=False` 或未验签方式绕过校验。
## 7. Ubuntu + OpenResty 部署
1. 宿主机 OpenResty 监听指定 IP 端口,直接提供前端构建目录,反向代理 `/api/``/ws/``127.0.0.1` 后端端口。
2. Docker Compose 运行后端和 PostgreSQLPostgreSQL 不映射宿主机端口,后端只绑定回环地址。
3. OpenResty 配置 WebSocket Upgrade、超时、请求体大小限制、访问日志与健康检查路由。没有域名时首期在受控内网使用 HTTP;需要 TLS 时由企业内部 CA 或含 IP SAN 的证书在 OpenResty 终止。
4. 后端容器按最小权限运行,移除 `NET_ADMIN`,仅保留探测所需的 `NET_RAW`;所有运行密钥由受控环境文件注入,不进入 Git 或镜像。
5. 交付运行手册,包括配置清单、备份/恢复、启动/停止、OpenResty 重载、健康检查、升级和回退。发布前执行备份与冒烟检查。
## 8. 测试和首批验收
- 单元测试:状态转移、阈值、窗口计算、防抖、事件去重、过期提醒、企业微信重试、IP/CSV/配置校验。
- 集成测试:异步数据库迁移、API 权限、出箱持久化、推送失败恢复、调度器不重叠。
- 部署测试:Compose 启动、OpenResty 反向代理、WebSocket、健康检查、PostgreSQL 不可从宿主机访问、容器能力最小化。
- 运行验证:使用明确授权的测试 IP 验证离线、间歇丢包、恢复、监控节点异常和部门通知。
## 9. 合规控制与可追溯性
| 分类 | 控制 | 依据/说明 |
| --- | --- | --- |
| 规范要求(C1) | 部署前完成测试并提交报告;提供版本、制品/代码关联、部署计划、操作步骤、回退及备份方案;部署后按用例验证。 | `13-deployment.md` §2.2.1、§2.2.2、§2.6。 |
| 规范要求 | 防止将不可信输入用于拼接 SQL、命令执行或不安全文件上传。 | `12-security.md` §3.6.1.1、§3.6.1.3、§3.6.2.2。 |
| 规范要求 | 生产变更前开启主机安全管控与防火墙,并纳入运行状态监控。 | `12-security.md` §5.1(强制)。 |
| 工程建议 | 使用事务性事件出箱、状态机防抖、最小容器能力、依赖/密钥扫描与版本固定。 | 可靠性和安全工程实践,不宣称为上述规范的特定强制目录或工具。 |
| 待确认 | 项目最终 C 级别、企业微信部门 ID、对外端口、内网 TLS 要求、备份责任人和部署审批人。 | 由项目负责人/运行方确认。 |
需求、设计、实现、测试、制品和部署记录应通过上述需求编号建立双向关联。实现完成后补充追踪矩阵和实际证据。