灾难恢复演练实战：从预案设计到真实切换的完整执行手册

灾难恢复演练实战从预案设计到真实切换的完整执行手册一、体系化起点灾难恢复演练不是跑个脚本就行了我在团队里推行过多次灾难恢复演练最深刻的体会是第一次真刀真枪的演练几乎没有一次能在预定时间内完成。不是因为预案写得不好而是因为真实环境的复杂性远超纸面推演能覆盖的范围。典型的翻车场景包括切换脚本中硬编码的 IP 已经过期实际主库 IP 几个月前就变了。DNS/VIP 切换后应用连接池没刷新大量请求照样打到已宕机的节点。数据校验脚本对 NULL 值的处理不一致导致误报数据不一致。运维人员在压力下误操作把主库当备库执行了RESET MASTER。这些问题的根因在于灾难恢复是一个跨团队、跨系统、跨流程的复杂协同动作而不仅仅是一组数据库切换命令。flowchart TB subgraph 演练前 A[预案编写br/切换SOP] -- B[预检清单br/环境一致性检查] B -- C[沙箱预演br/在独立环境验证] end subgraph 演练中 D[故障注入br/模拟主库宕机] -- E[故障检测br/监控告警触发] E -- F[决策阶段br/确认是否切换] F -- G[执行切换br/VIP漂移/GTID同步] G -- H[验证阶段br/业务连通性/数据一致性] end subgraph 演练后 I[复盘记录br/时间线还原] -- J[问题清单br/改进跟踪] J -- K[预案更新br/SOP迭代] end C -- D H -- I style D fill:#ffcdd2 style G fill:#fff3e0 style H fill:#c8e6c9 style I fill:#e3f2fd整张图展示了一个完整的演练生命周期——从预案到复盘。每个阶段的投入产出比是不同的前期预案的精细化程度直接决定了演练时的手忙脚乱程度。二、预案核心设计一份能打硬仗的SOP长什么样一份合格的灾难恢复 SOPStandard Operating Procedure必须满足三个标准可执行性任何一位值班 DBA 拿到 SOP 后即使从来没有操作过也能在 15 分钟内完成切换。幂等性切换步骤的设计必须支持重入——如果某一步失败了重跑整个脚本不会造成二次破坏。可回滚性切换后发现数据不一致时必须有办法回滚到切换前的状态。以下是一份 MySQL 主从切换的完整 SOP# 灾难恢复 SOP - MySQL 主从故障切换 # 版本: v3.2 | 最后更新: 2026-06-30 | 维护者: DBA Team metadata: sop_id: DR-SOP-MYSQL-001 version: 3.2 applicable_scenarios: - 主库宕机无法恢复 - 主库网络隔离 - 主库数据文件损坏 rto_target: 300 # 目标恢复时间 5分钟 rpo_target: 30 # 目标数据丢失 30秒 pre_check: # Phase 1: 故障确认5分钟内完成 - step: 1.1 action: 确认主库不可用 command: | mysqladmin ping -h $MASTER_HOST -u monitor -p$MONITOR_PASS --connect-timeout3 expected: 连接失败或超时 on_failure: 检查网络连通性排除网络问题后再进入下一步 - step: 1.2 action: 确认主库进程状态 command: | ssh $MASTER_HOST ps aux | grep mysqld | grep -v grep expected: 进程不存在或状态异常 - step: 1.3 action: 获取所有备库的复制状态 command: | for SLAVE in $SLAVE_LIST; do echo $SLAVE mysql -h $SLAVE -u monitor -p$MONITOR_PASS -e SHOW SLAVE STATUS\G 2/dev/null | grep -E Slave_IO_Running|Slave_SQL_Running|Seconds_Behind_Master|Exec_Master_Log_Pos|Retrieved_Gtid_Set|Executed_Gtid_Set done note: 选择延迟最小的备库作为新主库候选 failover_execution: # Phase 2: 执行切换 - step: 2.1 action: 确认新主库候选 params: new_master: $BEST_SLAVE command: | echo 选定新主库: $NEW_MASTER if [ -z $NEW_MASTER ]; then echo 错误: 未找到可用的备库 exit 1 fi - step: 2.2 action: 新主库停止复制并提升为主 command: | mysql -h $NEW_MASTER -u admin -p$ADMIN_PASS EOF -- 1. 停止 IO 线程不再接收 Binlog STOP SLAVE IO_THREAD; -- 2. 等待 SQL 线程应用完所有已接收的 Relay Log -- 最多等待 30 秒 SELECT SLEEP(1); -- 3. 完全停止复制 STOP SLAVE; -- 4. 重置复制配置清除从库身份 RESET SLAVE ALL; -- 5. 开启写权限 SET GLOBAL read_only OFF; SET GLOBAL super_read_only OFF; -- 6. 记录切换后的 GTID用于回滚 SELECT GLOBAL.GTID_EXECUTED AS checkpoint_gtid; EOF on_failure: 回滚在新主库上执行 SET GLOBAL read_onlyON - step: 2.3 action: VIP 漂移到新主库 command: | # 使用 Keepalived 或手动操作 VIP # 先确认旧 VIP 已下线 ssh $OLD_MASTER ip addr del $VIP_ADDR/24 dev eth0 2/dev/null || true sleep 2 # VIP 绑定到新主库 ssh $NEW_MASTER ip addr add $VIP_ADDR/24 dev eth0 ssh $NEW_MASTER arping -c 3 -A -I eth0 $VIP_ADDR echo VIP $VIP_ADDR 已切换至 $NEW_MASTER verify: | # 验证 VIP 可达 mysqladmin ping -h $VIP_ADDR -u monitor -p$MONITOR_PASS --connect-timeout3 - step: 2.4 action: 其他备库切换到新主库 command: | for SLAVE in $OTHER_SLAVES; do echo 处理备库: $SLAVE mysql -h $SLAVE -u admin -p$ADMIN_PASS EOF STOP SLAVE; -- 使用 GTID_AUTO_POSITION 自动找到同步点 CHANGE MASTER TO MASTER_HOST $NEW_MASTER, MASTER_PORT 3306, MASTER_USER repl, MASTER_PASSWORD $REPL_PASS, MASTER_AUTO_POSITION 1; START SLAVE; -- 等待复制追上最多 60 秒 SELECT SLEEP(3); SHOW SLAVE STATUS\G EOF # 检查复制延迟 delay$(mysql -h $SLAVE -u monitor -p$MONITOR_PASS -N \ -e SELECT COALESCE(Seconds_Behind_Master, 99999)) if [ $delay -gt 60 ]; then echo 警告: $SLAVE 复制延迟 $delay 秒 fi done verification: # Phase 3: 业务验证 - step: 3.1 action: 业务连通性测试 command: | # 模拟业务连接 mysql -h $VIP_ADDR -u app_user -p$APP_PASS -e SELECT 1 AS connectivity_test; SELECT COUNT(*) FROM information_schema.PROCESSLIST WHERE USER app_user; - step: 3.2 action: 数据一致性快速校验 command: | # 对核心业务表做采样校验 TABLESusers orders payments products for TABLE in $TABLES; do echo 校验表: $TABLE # 在复制恢复后做一次校验 mysql -h $NEW_MASTER -u admin -p$ADMIN_PASS -N \ -e SELECT $TABLE, COUNT(*), MAX(id), MAX(updated_at) FROM $DB.$TABLE done - step: 3.3 action: 通知上下游 checklist: - 通知应用团队数据库已完成切换VIP 保持不变 - 通知监控团队更新监控告警目标 - 通知备份团队确认备份任务在新主库上正常运行 - 更新 CMDB记录新主库信息 rollback: # 回滚方案仅在数据不一致时使用 condition: 新主库数据完整性校验失败 或 业务出现大面积异常 steps: - step: R1 action: 新主库设置只读 command: | mysql -h $NEW_MASTER -u admin -p$ADMIN_PASS \ -e SET GLOBAL read_onlyON; SET GLOBAL super_read_onlyON; - step: R2 action: 恢复旧主库如可恢复 command: | # 从备份恢复旧主库到切换前的状态 # 利用步骤 2.2 记录的 checkpoint_gtid echo 回滚操作请联系 DBA 负责人三、GTID 冲突处理切换中真正卡住人的技术点在 MySQL 主从切换中最让 DBA 头疼的问题通常是 GTID 一致性。常见的故障场景场景一主库有 GTID备库没有。主库宕机前已经写入了 BinlogGTIDuuid:100但 IO/SQL 线程还没来得及同步到备库。此时备库提升为主库后这个 GTID 对应的数据就永久丢失了。场景二备库多执行了 GTID。如果备库曾经因为手动修数据而执行了额外的写操作会产生独立的 GTID。这些 GTID 切换到新主库后不会造成问题因为是备库自己写的但会影响后续的数据校验。处理方案如下#!/usr/bin/env python3 MySQL 主从切换中的 GTID 一致性校验工具 在切换前和切换后验证 GTID 状态 import pymysql import json from typing import Dict, List, Optional, Set, Tuple from dataclasses import dataclass from datetime import datetime dataclass class GTIDStatus: 单个节点的 GTID 状态 host: str executed: Set[str] purged: Set[str] gtid_mode: str classmethod def from_mysql(cls, host: str, conn_params: Dict) - GTIDStatus: 从 MySQL 实例获取 GTID 状态 conn pymysql.connect(hosthost, **conn_params) try: with conn.cursor() as cursor: cursor.execute(SELECT GLOBAL.GTID_MODE) gtid_mode cursor.fetchone()[0] cursor.execute(SELECT GLOBAL.GTID_EXECUTED) executed_raw cursor.fetchone()[0] or cursor.execute(SELECT GLOBAL.GTID_PURGED) purged_raw cursor.fetchone()[0] or executed set(parse_gtid_set(executed_raw)) purged set(parse_gtid_set(purged_raw)) return cls( hosthost, executedexecuted, purgedpurged, gtid_modegtid_mode ) finally: conn.close() def parse_gtid_set(gtid_str: str) - List[str]: 解析 GTID 集合为列表 输入: uuid1:1-100:200-300, uuid2:1-50 输出: [uuid1:1, uuid1:2, ..., uuid1:100, uuid1:200, ...] if not gtid_str or gtid_str.strip() : return [] result [] for segment in gtid_str.split(,): segment segment.strip() if not segment: continue # uuid:intervals if : not in segment: continue uuid_part, intervals_part segment.split(:, 1) for interval in intervals_part.split(:): interval interval.strip() if - in interval: start, end interval.split(-) for i in range(int(start), int(end) 1): result.append(f{uuid_part}:{i}) else: try: result.append(f{uuid_part}:{int(interval)}) except ValueError: pass return result class GTIDConsistencyChecker: GTID 一致性校验器 staticmethod def find_missing_gtids( master: GTIDStatus, slaves: List[GTIDStatus] ) - Dict[str, List[str]]: 找出主库有但备库缺失的 GTID即将丢失的数据 result {} for slave in slaves: missing master.executed - slave.executed if missing: result[slave.host] sorted(missing) return result staticmethod def find_extra_gtids( master: GTIDStatus, slaves: List[GTIDStatus] ) - Dict[str, List[str]]: 找出备库有但主库没有的 GTID备库可能被手动写过 result {} for slave in slaves: extra slave.executed - master.executed if extra: result[slave.host] sorted(extra) return result staticmethod def select_best_slave( master: GTIDStatus, slaves: List[GTIDStatus] ) - Tuple[str, int]: 选择 GTID 完备性最好的备库 best_host None best_overlap -1 for slave in slaves: overlap len(master.executed slave.executed) if overlap best_overlap: best_overlap overlap best_host slave.host return best_host, best_overlap staticmethod def can_safely_failover( master: GTIDStatus, new_master_candidate: GTIDStatus, other_slaves: List[GTIDStatus] ) - Tuple[bool, str]: 判断是否可以安全切换 # 检查1: 新主库的 GTID 必须是主库的子集不能有多余数据 extra new_master_candidate.executed - master.executed if extra: return False, f新主库 {new_master_candidate.host} 有多余 GTID: {list(extra)[:5]}... # 检查2: 主库有哪些数据将丢失 missing master.executed - new_master_candidate.executed if missing: return False, f主库有 {len(missing)} 个 GTID 未同步到新主库数据可能丢失 # 检查3: 其他备库是否能跟上新主库 for slave in other_slaves: if slave.host new_master_candidate.host: continue slave_missing new_master_candidate.executed - slave.executed slave_extra slave.executed - new_master_candidate.executed if slave_extra: return False, f备库 {slave.host} 有多余 GTID无法自动切换 return True, GTID 一致性检查通过可以安全切换 class FailoverOrchestrator: 切换编排器 def __init__(self, conn_params: Dict): self.conn_params conn_params def pre_failover_check(self, master_host: str, slave_hosts: List[str]) - Dict: 切换前完整检查 report { check_time: datetime.now().isoformat(), master: master_host, slaves: slave_hosts, checks: {} } try: # 获取所有节点的 GTID 状态 master_gtid GTIDStatus.from_mysql(master_host, self.conn_params) slave_gtids [] for host in slave_hosts: try: gtid GTIDStatus.from_mysql(host, self.conn_params) slave_gtids.append(gtid) except Exception as e: report[checks][fconnect_{host}] { status: fail, message: str(e) } checker GTIDConsistencyChecker() # 查找数据丢失风险 missing checker.find_missing_gtids(master_gtid, slave_gtids) report[checks][missing_gtids] { status: pass if not missing else fail, details: {host: len(g) for host, g in missing.items()}, sample: {host: g[:3] for host, g in missing.items()} if missing else {} } # 查找备库额外数据 extra checker.find_extra_gtids(master_gtid, slave_gtids) report[checks][extra_gtids] { status: pass if not extra else warn, details: {host: len(g) for host, g in extra.items()} } # 选择最佳备库 best_host, overlap checker.select_best_slave(master_gtid, slave_gtids) report[recommended_new_master] best_host report[gtid_overlap_count] overlap except Exception as e: report[checks][fatal_error] { status: fail, message: str(e) } return report def execute_failover(self, new_master_host: str, other_slaves: List[str]) - Dict: 执行切换 log {steps: [], start_time: datetime.now().isoformat()} def log_step(step_name: str, success: bool, details: str ): log[steps].append({ step: step_name, success: success, details: details, timestamp: datetime.now().isoformat() }) # Step 1: 新主库停止复制 try: conn pymysql.connect(hostnew_master_host, **self.conn_params) with conn.cursor() as cursor: cursor.execute(STOP SLAVE IO_THREAD) import time; time.sleep(3) cursor.execute(STOP SLAVE) cursor.execute(RESET SLAVE ALL) cursor.execute(SET GLOBAL read_only OFF) cursor.execute(SET GLOBAL super_read_only OFF) cursor.execute(SELECT GLOBAL.GTID_EXECUTED) checkpoint_gtid cursor.fetchone()[0] conn.commit() conn.close() log_step(promote_new_master, True, fcheckpoint_gtid{checkpoint_gtid}) except Exception as e: log_step(promote_new_master, False, str(e)) return log # Step 2: 其他备库切换到新主库 for slave in other_slaves: if slave new_master_host: continue try: conn pymysql.connect(hostslave, **self.conn_params) with conn.cursor() as cursor: cursor.execute(STOP SLAVE) cursor.execute(f CHANGE MASTER TO MASTER_HOST {new_master_host}, MASTER_PORT 3306, MASTER_USER repl, MASTER_PASSWORD {self.conn_params[password]}, MASTER_AUTO_POSITION 1 ) cursor.execute(START SLAVE) conn.close() log_step(freslave_{slave}, True) except Exception as e: log_step(freslave_{slave}, False, str(e)) log[end_time] datetime.now().isoformat() return log if __name__ __main__: conn_params { user: admin, password: ***, port: 3306, connect_timeout: 5 } orchestrator FailoverOrchestrator(conn_params) # 切换前检查 report orchestrator.pre_failover_check( master_host192.168.1.10, slave_hosts[192.168.1.11, 192.168.1.12, 192.168.1.13] ) print(json.dumps(report, ensure_asciiFalse, indent2))四、复盘即进化如何从每次演练中提取真正的价值演练不是目的复盘才是。我在每次演练后都会要求团队产出一份《演练复盘报告》核心关注四个维度的数据时间线还原。从告警报出到业务恢复正常的每个时间节点精确到秒。比较实际耗时与 SOP 中预估耗时的差距。通常第一次演练的实际耗时是预估的 2-3 倍。人为错误分类。将演练中出现的所有操作失误分为三类流程缺陷SOP 写错了、知识缺陷DBA 不知道某条命令的含义、压力失误在紧张下点错按钮/敲错命令。工具链改进点。演练中暴露的工具问题脚本报错、监控漏报、自动化未生效是最有价值的产出因为它们可以直接转化为工程改进。业务影响量化。RTO 是否达标RPO 是否达标哪些业务因切换中断中断时长是否可接受只有将复盘结论落实到 SOP 的版本迭代中灾难恢复能力才会真正越练越强。五、总结灾难恢复演练的终极目标是当真正的故障发生时全团队不需要思考就能做出正确反应。这个目标只有通过反复演练、持续改进才能达到。演练不要追求一次完美。相反第一次演练暴露出越多问题越好——每一个暴露的问题都意味着在真实灾难中不会再犯同样的错误。从这个角度看失败的演练比成功的演练更有价值。最后提醒一个常被忽略的点演练通知必须提前告知所有相关方业务方、运维、安全、管理层并设置明确的终止条件如演练开始后 30 分钟内未完成切换立即回滚。灾难恢复演练的底线是练可以失败但绝不能把演练搞成真正的灾难。