logan
4c9427e546
- P0: 实例访问控制全线接入权限组 (829测试) - P1: permissions字段激活 + 操作级权限检查 (843测试) - 前端: 删除数据权限页面, 权限组合并为唯一入口 - 新增: CodeMirror编辑器 + 查询模板 + 慢查询分析 - 新增: MongoDB管理页面 + Dashboard增强 - 修复: Pydantic V2迁移 + 前端TS错误修复
315 lines
11 KiB
Python
315 lines
11 KiB
Python
"""Schema 漂移检测 — 检测数据库 Schema 的意外变更。
|
|
|
|
工作原理:
|
|
1. 定期捕获数据库 Schema 快照（表结构、索引、约束）
|
|
2. 将快照与基线比较
|
|
3. 发现差异时生成漂移报告
|
|
4. 发送通知告警
|
|
|
|
支持的检测类型:
|
|
- 新增表/字段
|
|
- 删除表/字段
|
|
- 字段类型变更
|
|
- 索引变更
|
|
"""
|
|
import hashlib
|
|
import json
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
from typing import List, Optional, Dict, Any
|
|
|
|
from app.models.instance import Instance
|
|
from app.services.db_connector import DBConnector
|
|
|
|
|
|
@dataclass
|
|
class ColumnSnapshot:
|
|
"""列定义快照。"""
|
|
table_name: str
|
|
column_name: str
|
|
column_type: str
|
|
is_nullable: bool
|
|
column_default: Optional[str]
|
|
extra: str
|
|
character_set: Optional[str]
|
|
collation: Optional[str]
|
|
checksum: str
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"table_name": self.table_name,
|
|
"column_name": self.column_name,
|
|
"column_type": self.column_type,
|
|
"is_nullable": self.is_nullable,
|
|
"column_default": self.column_default,
|
|
"extra": self.extra,
|
|
"checksum": self.checksum,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class DriftItem:
|
|
"""漂移项。"""
|
|
drift_type: str # added / removed / modified
|
|
table_name: str
|
|
column_name: str
|
|
before: Optional[str]
|
|
after: Optional[str]
|
|
severity: str # low / medium / high
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"drift_type": self.drift_type,
|
|
"table_name": self.table_name,
|
|
"column_name": self.column_name,
|
|
"before": self.before,
|
|
"after": self.after,
|
|
"severity": self.severity,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class DriftReport:
|
|
"""漂移检测报告。"""
|
|
instance_id: int
|
|
database: str
|
|
checked_at: str
|
|
total_columns: int
|
|
drift_count: int
|
|
drifts: List[DriftItem]
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"instance_id": self.instance_id,
|
|
"database": self.database,
|
|
"checked_at": self.checked_at,
|
|
"total_columns": self.total_columns,
|
|
"drift_count": self.drift_count,
|
|
"drifts": [d.to_dict() for d in self.drifts],
|
|
}
|
|
|
|
|
|
class SchemaDriftDetector:
|
|
"""Schema 漂移检测器。"""
|
|
|
|
# 内存中的基线存储: {(instance_id, database): [ColumnSnapshot, ...]}
|
|
_baselines: Dict[tuple, List[ColumnSnapshot]] = {}
|
|
|
|
async def capture_snapshot(self, instance: Instance,
|
|
database: str) -> List[ColumnSnapshot]:
|
|
"""捕获数据库 Schema 快照。"""
|
|
if instance.db_type == "mysql":
|
|
return await self._capture_mysql_schema(instance, database)
|
|
elif instance.db_type in ("postgres", "postgresql"):
|
|
return await self._capture_postgres_schema(instance, database)
|
|
else:
|
|
raise ValueError(f"不支持的 Schema 捕获类型: {instance.db_type}")
|
|
|
|
async def save_baseline(self, instance_id: int,
|
|
database: str,
|
|
snapshot: List[ColumnSnapshot]) -> None:
|
|
"""保存 Schema 基线到内存。"""
|
|
self._baselines[(instance_id, database)] = snapshot
|
|
|
|
async def detect_drift(self, instance: Instance,
|
|
database: str) -> DriftReport:
|
|
"""检测 Schema 漂移，返回漂移报告。"""
|
|
current = await self.capture_snapshot(instance, database)
|
|
baseline = await self._get_baseline(instance.id, database)
|
|
|
|
checked_at = datetime.now().isoformat()
|
|
|
|
if not baseline:
|
|
# 首次检测，保存为基线
|
|
await self.save_baseline(instance.id, database, current)
|
|
return DriftReport(
|
|
instance_id=instance.id,
|
|
database=database,
|
|
checked_at=checked_at,
|
|
total_columns=len(current),
|
|
drift_count=0,
|
|
drifts=[],
|
|
)
|
|
|
|
drifts = self._compare_schemas(current, baseline)
|
|
|
|
# 更新基线为当前状态
|
|
await self.save_baseline(instance.id, database, current)
|
|
|
|
return DriftReport(
|
|
instance_id=instance.id,
|
|
database=database,
|
|
checked_at=checked_at,
|
|
total_columns=len(current),
|
|
drift_count=len(drifts),
|
|
drifts=drifts,
|
|
)
|
|
|
|
async def _capture_mysql_schema(self, instance: Instance,
|
|
database: str) -> List[ColumnSnapshot]:
|
|
"""从 MySQL INFORMATION_SCHEMA 捕获 Schema。"""
|
|
password = instance.password # should be decrypted before calling
|
|
sql = """
|
|
SELECT
|
|
TABLE_NAME, COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE,
|
|
COLUMN_DEFAULT, EXTRA, CHARACTER_SET_NAME, COLLATION_NAME
|
|
FROM INFORMATION_SCHEMA.COLUMNS
|
|
WHERE TABLE_SCHEMA = %s
|
|
AND TABLE_NAME NOT LIKE '\\_%%' -- exclude internal tables
|
|
ORDER BY TABLE_NAME, ORDINAL_POSITION
|
|
"""
|
|
|
|
from app.core.crypto import decrypt_password
|
|
instance.password = decrypt_password(instance.password)
|
|
result = await DBConnector._query_mysql(instance, "information_schema", sql)
|
|
|
|
snapshots = []
|
|
for row in result[0]:
|
|
# 计算列定义的校验和
|
|
content = (
|
|
f"{row.get('COLUMN_TYPE', '')}|"
|
|
f"{row.get('IS_NULLABLE', '')}|"
|
|
f"{row.get('COLUMN_DEFAULT', '') or ''}|"
|
|
f"{row.get('EXTRA', '')}"
|
|
)
|
|
checksum = hashlib.md5(content.encode()).hexdigest()
|
|
|
|
snapshots.append(ColumnSnapshot(
|
|
table_name=row.get("TABLE_NAME", ""),
|
|
column_name=row.get("COLUMN_NAME", ""),
|
|
column_type=row.get("COLUMN_TYPE", ""),
|
|
is_nullable=row.get("IS_NULLABLE", "YES") == "YES",
|
|
column_default=str(row.get("COLUMN_DEFAULT", "")) if row.get("COLUMN_DEFAULT") is not None else None,
|
|
extra=row.get("EXTRA", ""),
|
|
character_set=row.get("CHARACTER_SET_NAME"),
|
|
collation=row.get("COLLATION_NAME"),
|
|
checksum=checksum,
|
|
))
|
|
|
|
return snapshots
|
|
|
|
async def _capture_postgres_schema(self, instance: Instance,
|
|
database: str) -> List[ColumnSnapshot]:
|
|
"""从 PostgreSQL information_schema 捕获 Schema。"""
|
|
sql = """
|
|
SELECT
|
|
c.table_name,
|
|
c.column_name,
|
|
CASE
|
|
WHEN c.data_type = 'character varying' AND c.character_maximum_length IS NOT NULL
|
|
THEN 'varchar(' || c.character_maximum_length || ')'
|
|
WHEN c.data_type = 'character' AND c.character_maximum_length IS NOT NULL
|
|
THEN 'char(' || c.character_maximum_length || ')'
|
|
WHEN c.data_type = 'numeric' AND c.numeric_precision IS NOT NULL
|
|
THEN 'numeric(' || c.numeric_precision || ',' || COALESCE(c.numeric_scale, 0) || ')'
|
|
ELSE c.data_type
|
|
END as column_type,
|
|
c.is_nullable,
|
|
c.column_default,
|
|
'' as extra,
|
|
NULL as character_set_name,
|
|
NULL as collation_name
|
|
FROM information_schema.columns c
|
|
WHERE c.table_schema = 'public'
|
|
AND c.table_catalog = %s
|
|
ORDER BY c.table_name, c.ordinal_position
|
|
"""
|
|
|
|
from app.core.crypto import decrypt_password
|
|
instance.password = decrypt_password(instance.password)
|
|
result = await DBConnector._query_postgres(instance, database, sql)
|
|
|
|
snapshots = []
|
|
for row in result[0]:
|
|
content = (
|
|
f"{row.get('column_type', '')}|"
|
|
f"{row.get('is_nullable', '')}|"
|
|
f"{row.get('column_default', '') or ''}"
|
|
)
|
|
checksum = hashlib.md5(content.encode()).hexdigest()
|
|
|
|
snapshots.append(ColumnSnapshot(
|
|
table_name=row.get("table_name", ""),
|
|
column_name=row.get("column_name", ""),
|
|
column_type=row.get("column_type", ""),
|
|
is_nullable=row.get("is_nullable", "YES") == "YES",
|
|
column_default=str(row.get("column_default", "")) if row.get("column_default") is not None else None,
|
|
extra=row.get("extra", ""),
|
|
character_set=None,
|
|
collation=None,
|
|
checksum=checksum,
|
|
))
|
|
|
|
return snapshots
|
|
|
|
async def _get_baseline(self, instance_id: int,
|
|
database: str) -> Optional[List[ColumnSnapshot]]:
|
|
"""获取保存的基线。"""
|
|
return self._baselines.get((instance_id, database))
|
|
|
|
def _compare_schemas(self, current: List[ColumnSnapshot],
|
|
baseline: List[ColumnSnapshot]) -> List[DriftItem]:
|
|
"""比较当前 Schema 与基线，返回漂移项。"""
|
|
drifts = []
|
|
current_dict = {(s.table_name, s.column_name): s for s in current}
|
|
baseline_dict = {(s.table_name, s.column_name): s for s in baseline}
|
|
|
|
# 检查新增列
|
|
for key, snapshot in current_dict.items():
|
|
if key not in baseline_dict:
|
|
drifts.append(DriftItem(
|
|
drift_type="added",
|
|
table_name=key[0],
|
|
column_name=key[1],
|
|
before=None,
|
|
after=f"{snapshot.column_type} NULL={snapshot.is_nullable}",
|
|
severity="medium",
|
|
))
|
|
|
|
# 检查删除列
|
|
for key, snapshot in baseline_dict.items():
|
|
if key not in current_dict:
|
|
drifts.append(DriftItem(
|
|
drift_type="removed",
|
|
table_name=key[0],
|
|
column_name=key[1],
|
|
before=f"{snapshot.column_type} NULL={snapshot.is_nullable}",
|
|
after=None,
|
|
severity="high", # 删除列是高风险操作
|
|
))
|
|
|
|
# 检查修改列
|
|
for key, current_snap in current_dict.items():
|
|
if key in baseline_dict:
|
|
baseline_snap = baseline_dict[key]
|
|
if current_snap.checksum != baseline_snap.checksum:
|
|
severity = "medium"
|
|
# 类型变更是高风险
|
|
if current_snap.column_type != baseline_snap.column_type:
|
|
severity = "high"
|
|
|
|
drifts.append(DriftItem(
|
|
drift_type="modified",
|
|
table_name=key[0],
|
|
column_name=key[1],
|
|
before=f"{baseline_snap.column_type} (default: {baseline_snap.column_default or 'None'})",
|
|
after=f"{current_snap.column_type} (default: {current_snap.column_default or 'None'})",
|
|
severity=severity,
|
|
))
|
|
|
|
# 按严重性排序
|
|
severity_order = {"high": 0, "medium": 1, "low": 2}
|
|
drifts.sort(key=lambda x: severity_order.get(x.severity, 3))
|
|
|
|
return drifts
|
|
|
|
def clear_baseline(self, instance_id: int, database: str) -> None:
|
|
"""清除指定实例的基线。"""
|
|
key = (instance_id, database)
|
|
if key in self._baselines:
|
|
del self._baselines[key]
|
|
|
|
def clear_all_baselines(self) -> None:
|
|
"""清除所有基线。"""
|
|
self._baselines.clear()
|