diff --git a/backend/app/core/deps.py b/backend/app/core/deps.py index 6e57789..af29349 100644 --- a/backend/app/core/deps.py +++ b/backend/app/core/deps.py @@ -22,7 +22,12 @@ 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 app.models import Base + from migrations.versions.20260803_reliability_monitoring import ( + run_reliability_migration, + ) + async with engine.begin() as conn: + await conn.run_sync(run_reliability_migration) await conn.run_sync(Base.metadata.create_all) diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index cc4340f..81bfaf4 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -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", ] diff --git a/backend/app/models/alert_event.py b/backend/app/models/alert_event.py index c78c4da..e5587c2 100644 --- a/backend/app/models/alert_event.py +++ b/backend/app/models/alert_event.py @@ -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,14 @@ 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="最近一次通知失败原因", + ) + 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="创建时间") diff --git a/backend/app/models/device.py b/backend/app/models/device.py index 19f4a8a..19e1c0c 100644 --- a/backend/app/models/device.py +++ b/backend/app/models/device.py @@ -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,56 @@ 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 + + @classmethod + def from_device( + cls, + device: "Device", + settings: "Settings", + ) -> "DeviceMonitoringPolicy": + """Resolve device overrides without changing the stored device.""" + return cls( + probe_packets_per_round=( + device.probe_packets_per_round + if device.probe_packets_per_round is not None + else settings.probe_packets_per_round + ), + offline_consecutive_rounds=( + device.offline_consecutive_rounds + if device.offline_consecutive_rounds is not None + else settings.offline_consecutive_rounds + ), + degraded_window_rounds=( + device.degraded_window_rounds + if device.degraded_window_rounds is not None + else settings.degraded_window_rounds + ), + degraded_loss_percent=( + device.degraded_loss_percent + if device.degraded_loss_percent is not None + else settings.degraded_loss_percent + ), + recovery_consecutive_clean_rounds=( + device.recovery_consecutive_clean_rounds + if device.recovery_consecutive_clean_rounds is not None + else settings.recovery_consecutive_clean_rounds + ), + ) + + class Device(Base): __tablename__ = "devices" @@ -32,8 +88,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="最后一次在线时间") diff --git a/backend/app/models/notification_outbox.py b/backend/app/models/notification_outbox.py new file mode 100644 index 0000000..5adfb78 --- /dev/null +++ b/backend/app/models/notification_outbox.py @@ -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"), + ) diff --git a/backend/app/models/ping_record.py b/backend/app/models/ping_record.py index 2757c9c..9422ba2 100644 --- a/backend/app/models/ping_record.py +++ b/backend/app/models/ping_record.py @@ -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"" diff --git a/backend/migrations/versions/20260803_reliability_monitoring.py b/backend/migrations/versions/20260803_reliability_monitoring.py new file mode 100644 index 0000000..09c78d3 --- /dev/null +++ b/backend/migrations/versions/20260803_reliability_monitoring.py @@ -0,0 +1,103 @@ +"""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 Base + + +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", +} + + +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 in Base.metadata.sorted_tables: + 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 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" + 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) + for table in Base.metadata.sorted_tables: + existing_indexes = { + index["name"] for index in inspector.get_indexes(table.name) + } + for index in table.indexes: + 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": + connection.execute( + text("ALTER TYPE alerttypeenum 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 absent tables, + columns, and indexes, then writes one immutable revision marker. It never + issues DROP, DELETE, UPDATE, or data-copy statements. + """ + _migration_metadata.create_all(bind=connection) + Base.metadata.create_all(bind=connection) + + 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)) diff --git a/backend/tests/test_models_and_migration.py b/backend/tests/test_models_and_migration.py new file mode 100644 index 0000000..03934a9 --- /dev/null +++ b/backend/tests/test_models_and_migration.py @@ -0,0 +1,98 @@ +"""Persistence contracts for reliability monitoring data.""" + +import importlib +from pathlib import Path + +import pytest +from sqlalchemy import 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 + + +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()