监控数据容灾切换的脚本如何编写

本文目录导读：

1. 通用脚本设计原则
2. MySQL主从切换脚本（Python + pymysql）
3. Redis哨兵/集群切换脚本（Bash + redis-cli）
4. 通用健康检查 + API调用切换（Python + requests）
5. 关键注意事项

编写监控数据容灾切换脚本需要根据具体的数据库类型、中间件和容灾架构来定制，下面提供几种常见场景的脚本思路和示例，包括 MySQL主从切换、Redis哨兵切换 以及 通用的健康检查+切换逻辑。

通用脚本设计原则

1. 健康检查：定期检查主库是否可用（如端口、查询、复制状态）
2. 故障判定：连续N次检查失败才认为故障，避免误判
3. 切换逻辑：提升备库为主库，修改应用连接指向
4. 通知告警：切换成功后通知运维人员
5. 幂等性：避免重复切换造成脑裂

MySQL主从切换脚本（Python + pymysql）

场景：一主一从，主库故障时自动将从库提升为主库

#!/usr/bin/env python3
import pymysql
import time
import logging
import sys
logging.basicConfig(level=logging.INFO, 
                    format='%(asctime)s - %(levelname)s - %(message)s')
# 配置信息
MASTER_HOST = '192.168.1.10'
MASTER_PORT = 3306
MASTER_USER = 'root'
MASTER_PASS = 'password'
SLAVE_HOST = '192.168.1.20'
SLAVE_PORT = 3306
SLAVE_USER = 'root'
SLAVE_PASS = 'password'
CHECK_INTERVAL = 5  # 秒
MAX_RETRY = 3       # 连续失败次数
def check_mysql_connection(host, port, user, password):
    """检查MySQL是否可达"""
    try:
        conn = pymysql.connect(host=host, port=port, user=user, 
                              password=password, connect_timeout=3)
        conn.ping()
        conn.close()
        return True
    except Exception:
        return False
def check_slave_status(host, port, user, password):
    """检查从库复制状态"""
    try:
        conn = pymysql.connect(host=host, port=port, user=user, 
                              password=password, connect_timeout=3)
        with conn.cursor() as cursor:
            cursor.execute("SHOW SLAVE STATUS")
            result = cursor.fetchone()
            if result:
                slave_io_running = result[10]  # Slave_IO_Running
                slave_sql_running = result[11] # Slave_SQL_Running
                return (slave_io_running == 'Yes' and slave_sql_running == 'Yes')
        conn.close()
        return False
    except Exception:
        return False
def promote_slave_to_master():
    """将从库提升为主库"""
    try:
        conn = pymysql.connect(host=SLAVE_HOST, port=SLAVE_PORT, 
                              user=SLAVE_USER, password=SLAVE_PASS)
        with conn.cursor() as cursor:
            # 停止复制
            cursor.execute("STOP SLAVE")
            cursor.execute("RESET SLAVE ALL")
            # 设置只读关闭（让从库可写）
            cursor.execute("SET GLOBAL read_only = OFF")
        conn.commit()
        conn.close()
        logging.info("从库已提升为主库: %s:%d", SLAVE_HOST, SLAVE_PORT)
        return True
    except Exception as e:
        logging.error("提升从库失败: %s", e)
        return False
def switch_dns_or_config(new_master_ip):
    """修改应用配置或DNS指向新主库（示例为更新配置文件）"""
    config_file = '/etc/app/db.conf'
    try:
        with open(config_file, 'r', encoding='utf-8') as f:
            content = f.read()
        # 替换旧IP为新IP（可根据实际配置调整）
        old_ip = MASTER_HOST
        new_content = content.replace(old_ip, new_master_ip)
        with open(config_file, 'w', encoding='utf-8') as f:
            f.write(new_content)
        logging.info("配置文件已更新，新主库IP: %s", new_master_ip)
        # 重启应用（可选）
        # subprocess.run(['systemctl', 'restart', 'app-service'])
        return True
    except Exception as e:
        logging.error("更新配置文件失败: %s", e)
        return False
def main():
    fail_count = 0
    while True:
        if check_mysql_connection(MASTER_HOST, MASTER_PORT, MASTER_USER, MASTER_PASS):
            # 主库正常，重置计数器
            fail_count = 0
            logging.info("主库正常")
        else:
            fail_count += 1
            logging.warning("主库不可达（第%d次）", fail_count)
            if fail_count >= MAX_RETRY:
                logging.info("主库故障确认，开始容灾切换...")
                # 检查从库状态
                if not check_slave_status(SLAVE_HOST, SLAVE_PORT, SLAVE_USER, SLAVE_PASS):
                    logging.error("从库状态异常，无法切换")
                    fail_count = 0
                    time.sleep(CHECK_INTERVAL)
                    continue
                # 提升从库为主库
                if promote_slave_to_master():
                    # 修改应用配置
                    if switch_dns_or_config(SLAVE_HOST):
                        logging.info("容灾切换完成，新主库: %s:%d", SLAVE_HOST, SLAVE_PORT)
                        # 发送告警（示例为打印日志）
                        # send_alert("MySQL容灾切换完成，新主库: %s" % SLAVE_HOST)
                    else:
                        logging.error("配置更新失败，请手动处理")
                else:
                    logging.error("切换失败，请手动处理")
                # 切换完成后可以退出或继续监控新主库
                # 注意：这里只是演示，实际建议退出脚本并通知人工介入
                break
        time.sleep(CHECK_INTERVAL)
if __name__ == "__main__":
    main()

Redis哨兵/集群切换脚本（Bash + redis-cli）

场景：Redis哨兵架构，主库故障时自动更新本地应用配置

#!/bin/bash
# Redis 容灾切换检测脚本
REDIS_MASTER="192.168.1.30"
REDIS_PORT=6379
SENTINEL_HOST="192.168.1.31"
SENTINEL_PORT=26379
APP_CONFIG="/etc/app/redis.conf"
CHECK_RETRY=3
FAIL_COUNT=0
check_master() {
    redis-cli -h $REDIS_MASTER -p $REDIS_PORT PING > /dev/null 2>&1
    return $?
}
get_new_master_from_sentinel() {
    # 通过哨兵获取当前主库
    redis-cli -h $SENTINEL_HOST -p $SENTINEL_PORT sentinel get-master-addr-by-name mymaster 2>/dev/null | head -1
}
switch_redis_config() {
    local new_ip=$1
    sed -i "s/^redis_host=.*/redis_host=$new_ip/" $APP_CONFIG
    echo "更新Redis配置为新主库: $new_ip"
    # 重启应用
    systemctl restart app-service
    echo "应用已重启"
}
# 主监控循环
while true; do
    if check_master; then
        FAIL_COUNT=0
        echo "$(date) 主库正常"
    else
        ((FAIL_COUNT++))
        echo "$(date) 主库不可达 ($FAIL_COUNT/$CHECK_RETRY)"
        if [ $FAIL_COUNT -ge $CHECK_RETRY ]; then
            echo "主库故障确认，启动切换流程..."
            # 获取新主库
            NEW_MASTER=$(get_new_master_from_sentinel)
            if [ -z "$NEW_MASTER" ]; then
                echo "无法从哨兵获取新主库"
                FAIL_COUNT=0
                sleep 5
                continue
            fi
            # 检查新主库是否可用
            redis-cli -h $NEW_MASTER -p $REDIS_PORT PING > /dev/null 2>&1
            if [ $? -eq 0 ]; then
                echo "新主库 $NEW_MASTER 可用，更新配置..."
                switch_redis_config $NEW_MASTER
                echo "切换完成"
            else
                echo "新主库 $NEW_MASTER 也不可用，请人工介入"
            fi
            break
        fi
    fi
    sleep 5
done

通用健康检查 + API调用切换（Python + requests）

如果需要对接云平台的API或自定义的切换服务,可以使用通用模式：

import requests
import time
import json
import logging
logging.basicConfig(level=logging.INFO)
class HealthCheck:
    def __init__(self, endpoint, api_url, api_token):
        self.endpoint = endpoint
        self.api_url = api_url
        self.api_token = api_token
        self.fail_count = 0
        self.max_retry = 3
    def check_health(self):
        """通用健康检查，支持HTTP/TCP/ICMP"""
        try:
            r = requests.get(self.endpoint, timeout=5)
            return r.status_code == 200
        except:
            return False
    def trigger_switch(self):
        """调用容灾切换API"""
        headers = {'Authorization': f'Bearer {self.api_token}'}
        payload = {'action': 'failover', 'resource': 'db-primary'}
        try:
            r = requests.post(self.api_url, json=payload, headers=headers, timeout=10)
            if r.status_code == 200:
                logging.info("切换API调用成功: %s", r.json())
                return True
            else:
                logging.error("切换API返回错误: %s", r.text)
                return False
        except Exception as e:
            logging.error("调用切换API失败: %s", e)
            return False
    def run(self):
        while True:
            if self.check_health():
                self.fail_count = 0
                logging.info("服务正常")
            else:
                self.fail_count += 1
                logging.warning("服务异常 (第%d次)", self.fail_count)
                if self.fail_count >= self.max_retry:
                    logging.info("触发容灾切换...")
                    if self.trigger_switch():
                        logging.info("切换完成")
                    else:
                        logging.error("切换失败，请人工处理")
                    break
            time.sleep(10)
if __name__ == "__main__":
    checker = HealthCheck(
        endpoint="http://10.0.0.1:8080/health",
        api_url="http://switch-api.internal/failover",
        api_token="your_token_here"
    )
    checker.run()

关键注意事项

编写容灾切换脚本时,建议：

1. 先实现监控检测，再增加切换逻辑
2. 使用稳定成熟的客户端库（如pymysql、redis-py）
3. 配置外部化，避免硬编码
4. 测试环境充分演练再上线
5. 监控脚本本身（比如脚本进程挂了要有告警）

根据你的实际架构（云原生、虚拟机、物理机）和数据库类型，选择对应的最佳实践即可。