153 lines
5.1 KiB
Python
153 lines
5.1 KiB
Python
"""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))
|