数据库自治系统的可观测性挑战：如何在AI做出决策后追溯和审计

一、自治系统上线一个月后，没人知道它改了什么参数

数据库自治系统上线后的蜜月期通常在两周左右。AI自动优化器表现出色——自动调整了Buffer Pool大小、优化了索引、关闭了不必要的查询缓存，P99延迟下降了25%，团队一片欢呼。但一个月后，当一条核心查询突然变慢时，问题来了：不知道是哪个自动优化操作导致的。有索引被AI重建了吗？参数被AI调整过吗？查询计划被强制HINT过吗？没有审计记录，答案是一连串的"不确定"。

这种不可追溯性造成的后果远不止排查困难。在金融、医疗等强合规行业，监管要求所有对数据库结构的变更必须有完整的审计轨迹——谁、何时、做了什么、为什么。AI代为执行的变更在合规层面必须等同于人工操作。更进一步，当AI的决策导致了业务问题（比如自动建的索引反而拖慢了写入），复盘时无法回溯AI的推理过程，就无法修正模型中的缺陷，同样的错误会再次发生。可观测性因此是数据库自治系统的核心非功能需求，其重要性不亚于AI模型本身的精度。

二、AI决策的四层审计模型：数据→推理→决策→影响

完整的审计追索需要在四个层次上记录信息。

数据层记录AI做决策时所依据的输入数据。对于一个索引推荐决策，这包括近期的查询日志、当前表结构和统计信息、服务器资源使用情况等。这一层确保决策的输入是可复原的——即使几个月后，也能重新加载当时的数据快照并重现AI的推理过程。

推理层记录AI内部的推理路径。对于规则引擎，就是触发规则的序列和每条规则的条件匹配结果。对于机器学习模型，就是特征重要性排名、SHAP值、以及关键决策边界信息。这一层回答"为什么AI给出了这个建议"。

决策层记录AI最终做出的决策和执行状态。包括建议内容（如"建立idx_foo_bar索引"）、置信度、是否被自动执行、执行时间、执行结果（成功/失败/部分成功）。

影响层记录决策执行后的效果验证。包括关键性能指标的变化（QPS、延迟、CPU使用率）、是否有任何指标劣化、是否触发了回滚。

这四层信息共同构成一个完整的决策事件。查询时可以从任意一层切入——按时间段查询"过去一周AI做了哪些变更"，按影响查询"哪些AI决策导致了延迟恶化"，按推理查询"哪些决策的特征重要性排名异常"。

三、决策追溯日志的存储与查询实现

import json import time import hashlib import logging from dataclasses import dataclass, field, asdict from typing import List, Dict, Optional, Any from datetime import datetime from enum import Enum logger = logging.getLogger(__name__) class DecisionType(Enum): INDEX_RECOMMEND = "index_recommend" PARAM_TUNE = "param_tune" QUERY_REWRITE = "query_rewrite" SCALING = "scaling" class DecisionStatus(Enum): PROPOSED = "proposed" # 仅建议，未执行 APPROVED = "approved" # 人工审批通过 EXECUTING = "executing" # 执行中 SUCCEEDED = "succeeded" # 执行成功 FAILED = "failed" # 执行失败 ROLLED_BACK = "rolled_back" # 已回滚 @dataclass class AuditEvent: """一个完整的AI决策审计事件""" event_id: str timestamp: float decision_type: DecisionType status: DecisionStatus # 数据层 input_snapshot: Dict[str, Any] input_snapshot_hash: str = "" # 推理层 reasoning_path: Dict[str, Any] # 例如: {"features": {"query_freq": 0.35, "selectivity": 0.28, ...}, # "shap_values": {"query_freq": 0.12, "selectivity": 0.08, ...}, # "rules_triggered": ["high_freq_query", "low_selectivity"]} # 决策层 decision_content: Dict[str, Any] confidence: float = 0.0 # 例如: {"action": "create_index", # "target": "orders", # "columns": ["user_id", "create_time"]} # 影响层 metrics_before: Dict[str, float] = field(default_factory=dict) metrics_after: Dict[str, float] = field(default_factory=dict) validation_result: Optional[str] = None # 'improved', 'degraded', 'neutral' def __post_init__(self): if not self.input_snapshot_hash: # 对输入快照做哈希，便于后续验证 snapshot_bytes = json.dumps(self.input_snapshot, sort_keys=True).encode() self.input_snapshot_hash = hashlib.sha256(snapshot_bytes).hexdigest()[:16] class AuditLogger: """自治决策审计日志系统""" def __init__(self, storage_backend: str = "elasticsearch"): self.storage_backend = storage_backend self.events: List[AuditEvent] = [] self.retention_days = 90 # 审计日志保留90天 def log_decision( self, decision_type: DecisionType, input_snapshot: Dict[str, Any], reasoning_path: Dict[str, Any], decision_content: Dict[str, Any], confidence: float, ) -> AuditEvent: """记录一个新的AI决策""" event = AuditEvent( event_id=f"{decision_type.value}_{int(time.time() * 1000)}_{hashlib.md5(str(decision_content).encode()).hexdigest()[:8]}", timestamp=time.time(), decision_type=decision_type, status=DecisionStatus.PROPOSED, input_snapshot=input_snapshot, reasoning_path=reasoning_path, decision_content=decision_content, confidence=confidence, ) self.events.append(event) logger.info( f"Audit: {event.event_id} type={decision_type.value} " f"confidence={confidence:.2f}" ) return event def update_status(self, event_id: str, status: DecisionStatus, metrics_after: Dict[str, float] = None): """更新决策状态和影响指标""" for event in self.events: if event.event_id == event_id: event.status = status if metrics_after: event.metrics_after = metrics_after self._evaluate_impact(event) logger.info(f"Audit update: {event_id} -> {status.value}") return True logger.error(f"Event not found: {event_id}") return False def _evaluate_impact(self, event: AuditEvent): """评估决策影响""" before = event.metrics_before after = event.metrics_after if not before or not after: return # 简单的阈值判定 if 'p99_latency_ms' in before and 'p99_latency_ms' in after: p99_delta = (after['p99_latency_ms'] - before['p99_latency_ms']) / max(before['p99_latency_ms'], 1) if p99_delta < -0.1: event.validation_result = 'improved' elif p99_delta > 0.2: event.validation_result = 'degraded' else: event.validation_result = 'neutral' def query_by_time_range(self, start_ts: float, end_ts: float) -> List[AuditEvent]: """按时间范围查询审计事件""" return [ e for e in self.events if start_ts <= e.timestamp <= end_ts ] def query_by_impact(self, impact_filter: str = 'degraded') -> List[AuditEvent]: """查询有劣化影响的决策""" return [ e for e in self.events if e.validation_result == impact_filter ] def export_audit_report(self, start_ts: float, end_ts: float) -> Dict: """导出审计报告""" events = self.query_by_time_range(start_ts, end_ts) summary = { 'period': f"{datetime.fromtimestamp(start_ts)} - {datetime.fromtimestamp(end_ts)}", 'total_decisions': len(events), 'by_type': {}, 'by_status': {}, 'degraded_count': 0, 'improved_count': 0, 'avg_confidence': 0.0, } for e in events: dtype = e.decision_type.value summary['by_type'][dtype] = summary['by_type'].get(dtype, 0) + 1 s = e.status.value summary['by_status'][s] = summary['by_status'].get(s, 0) + 1 if e.validation_result == 'degraded': summary['degraded_count'] += 1 elif e.validation_result == 'improved': summary['improved_count'] += 1 if events: summary['avg_confidence'] = sum(e.confidence for e in events) / len(events) return summary

四、审计日志膨胀与存储成本：全量记录vs采样记录的抉择

每天几百次AI决策的审计日志看似数据量不大，但如果包含完整的查询日志快照和表统计信息，单条事件可能达到几MB。一个日均1000次决策的集群，每天的审计日志量高达几GB，90天保留期累计几百GB。对于只有几十个节点的集群，这个存储成本尚可接受；但对于上千个节点的超大规模集群，审计日志本身就是一笔不小的开销。

分级存储策略是解决之道。热数据（近7天的审计日志）存储在Elasticsearch中，支持实时检索和聚合分析。温数据（8-30天）下沉到ClickHouse或其他列存，保留聚合查询能力但牺牲搜索延迟。冷数据（31-90天）压缩归档到对象存储（S3/OSS），仅保留"不可删除"的合规要求底线。超过90天的数据，按照数据治理策略做合规删除。

采样记录适用于数据层的输入快照。全量保存每次决策的完整查询日志既昂贵也不必要。可以采用"基线+增量"的方式——保存一份全量的每日查询日志基线，每次决策只记录相对基线的增量变化（新增的查询模板和频率变化）。这样可以将单条事件的数据量从MB级压缩到KB级。

不可篡改性在审计场景中不可忽视。对于合规要求严格的场景，审计日志需要附加Merkle树或简单链式哈希，确保日志一旦写入就无法被篡改。简单的实现方式是每条审计事件的哈希值包含前一条事件的哈希，形成链式验证。

五、总结

数据库自治系统的可观测性建设应当与AI模型开发同步进行，而非事后补课。四层审计模型覆盖了从数据输入到决策执行再到效果验证的完整链路，为事后追溯和合规审计提供了结构化框架。分级存储策略在成本和质量之间取得平衡——近实时检索+列存分析+归档留存。从工程实践来看，审计日志本身不宜过于详尽也不宜过于简略——过于详尽带来存储膨胀，过于简略失去追溯价值。最实用的折中是"结构化元数据全量+原始数据采样"的策略，确保在大多数排查场景下无需回查原始数据即可定位问题。