fix: preserve legacy monitoring migration data

2026-08-04 11:22:21 +08:00
parent 80a31b8dd0
commit fee707f42d
4 changed files with 270 additions and 35 deletions
@@ -36,6 +36,11 @@ class AlertEvent(Base):
        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="用户确认时间")
@@ -36,6 +36,23 @@ class DeviceMonitoringPolicy:
    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,
@@ -45,29 +62,44 @@ class 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
+                cls._resolve_override(
+                    device.probe_packets_per_round,
+                    settings.probe_packets_per_round,
+                    1,
+                    10,
+                )
            ),
            offline_consecutive_rounds=(
-                device.offline_consecutive_rounds
-                if device.offline_consecutive_rounds is not None
-                else settings.offline_consecutive_rounds
+                cls._resolve_override(
+                    device.offline_consecutive_rounds,
+                    settings.offline_consecutive_rounds,
+                    1,
+                    10,
+                )
            ),
            degraded_window_rounds=(
-                device.degraded_window_rounds
-                if device.degraded_window_rounds is not None
-                else settings.degraded_window_rounds
+                cls._resolve_override(
+                    device.degraded_window_rounds,
+                    settings.degraded_window_rounds,
+                    2,
+                    60,
+                )
            ),
            degraded_loss_percent=(
-                device.degraded_loss_percent
-                if device.degraded_loss_percent is not None
-                else settings.degraded_loss_percent
+                cls._resolve_override(
+                    device.degraded_loss_percent,
+                    settings.degraded_loss_percent,
+                    1,
+                    100,
+                )
            ),
            recovery_consecutive_clean_rounds=(
-                device.recovery_consecutive_clean_rounds
-                if device.recovery_consecutive_clean_rounds is not None
-                else settings.recovery_consecutive_clean_rounds
+                cls._resolve_override(
+                    device.recovery_consecutive_clean_rounds,
+                    settings.recovery_consecutive_clean_rounds,
+                    1,
+                    20,
+                )
            ),
        )

@@ -5,7 +5,7 @@ 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
+from app.models import AlertEvent, Device, NotificationOutbox, PingRecord


 REVISION = "20260803_reliability_monitoring"
@@ -23,6 +23,40 @@ _DEFAULTS_FOR_EXISTING_ROWS = {
    "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."""
@@ -33,15 +67,15 @@ 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:
+    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)
+            column["name"] for column in inspector.get_columns(table_name)
        }
        for column in table.columns:
-            if column.name in existing_columns:
+            if column.name not in column_names or column.name in existing_columns:
                continue

            column_type = connection.dialect.type_compiler.process(column.type)
@@ -50,10 +84,13 @@ def _add_missing_columns(connection: Connection) -> None:
            if default is not None:
                if connection.dialect.name == "sqlite" and default == "TRUE":
                    default = "1"
-                definition = f"{definition} DEFAULT {default}"
+                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"ALTER TABLE {_quote(connection, table_name)} "
                    f"ADD COLUMN {definition}"
                )
            )
@@ -62,32 +99,40 @@ def _add_missing_columns(connection: Connection) -> None:
 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)
+    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)
+    }
+    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":
+        enum_name = AlertEvent.__table__.c.alert_type.type.name
        connection.execute(
-            text("ALTER TYPE alerttypeenum ADD VALUE IF NOT EXISTS 'degraded'")
+            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 absent tables,
-    columns, and indexes, then writes one immutable revision marker. It never
-    issues DROP, DELETE, UPDATE, or data-copy statements.
+    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)
-    Base.metadata.create_all(bind=connection)
+    NotificationOutbox.__table__.create(bind=connection, checkfirst=True)

    already_applied = connection.execute(
        select(_schema_migrations.c.revision).where(
@@ -4,7 +4,7 @@ import importlib
 from pathlib import Path

 import pytest
-from sqlalchemy import text
+from sqlalchemy import Column, Integer, Table, inspect, text
 from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

 import app.models as models
@@ -72,6 +72,19 @@ def test_device_policy_prefers_explicit_device_overrides():
    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(
@@ -96,3 +109,143 @@ async def test_migration_is_idempotent_and_records_revision(tmp_path: Path):
        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()