MySQL MGR集群单主模式的自动搭建和自动化故障修复

共计 12300 个字符，预计需要花费 31 分钟才能阅读完成。

随着 MySQL MGR 的版本的升级以及技术成熟，在把 MHA 拉下神坛之后，MGR 越来越成为 MySQL 高可用的首选方案。
MGR 的搭建并不算很复杂，但是有一系列手工操作步骤，为了简便 MGR 的搭建和故障诊断，这里完成了一个自动化的脚本，来实现 MGR 的自动化搭建，自动化故障诊断以及修复。

 

MGR 自动化搭建
为了简便起见，这里以单机多实例的模式进行测试，
先装好三个 MySQL 实例，端口号分别是 7001,7002,7003，其中 7001 作为写节点，其余两个节点作为读节，8000 节点是笔者的另外一个测试节点，请忽略。
在指明主从节点的情况下，如下为 mgr_tool.py 一键搭建 MGR 集群的测试 demo

 

MGR 故障模拟 1

MGR 节点故障自动监测和自愈实现，如下是搭建完成后的 MGR 集群，目前集群处于完全正常的状态中。

主观造成主从节点间 binlog 的丢失

在主节点上对于对于从节点丢失的数据操作，GTID 无法找到对应的数据，组复制立马熄火

非写入节点出现错误

看下 errorlog

如果是手动解决的话，还是 GTID 跳过错误事物的套路，master 上的 GTID 信息

尝试跳过最新的一个事物 ID，然后重新连接到组，可以正常连接到组，另外一个节点仍旧处于 error 状态

另外一个节点类似，依次解决。

MGR 故障模拟 2

从节点脱离 Group

这种情况倒是比较简单，重新开始组复制即可，start group_replication

 

MGR 故障自动检测和修复

对于如上的两种情况，
1，如果是从节点丢失主节点的事物，尝试在从节点上跳过 GTID，重新开始复制即可
2，如果是从节点非丢失主节点事物，尝试在从节点重新开始组复制即可

实现代码如下

 

对于故障类型 1，GTID 事物不一致的自动化修复

 

 对于故障类型 2 从节点 offline 的自动化修复

 

完整的实现代码

该过程要求 MySQL 实例必须满足 MGR 的基本条件，如果环境本身无法满足 MGR，一切都无从谈起，因此要非常清楚 MGR 环境的最基本要求

完成的实现代码如下，花了一个下午写的，目前来说存在以下不足
1，创建复制用户的时候，没有指定具体的 slave 机器，目前直接指定的 %：create user repl@’%’ identified by repl
2，对于 slave 的修复，目前无法整体修复，只能一台一台修复，其实就是少了一个循环 slave 机器判断的过程
3，目前搭建之前都会 reset master（不管主从，主要是清理可能的残留 GTID），因此只适合新环境的搭建
4，目前只支持 offline 和 gtid 事物冲突的错误类型修复，无法支持其他 MGR 错误类型的修复
5，开发环境是单机多实例模式测试，没有在多机单实例模式下充分测试
以上都会逐步改善 & 加强。

# -*- coding: utf-8 -*-

import pymysql
import logging
import time
import decimal

def execute_query(conn_dict,sql):
    conn = pymysql.connect(host=conn_dict[‘host’],
                          port=conn_dict[‘port’],
                          user=conn_dict[‘user’],
                          passwd=conn_dict[‘password’],
                          db=conn_dict[‘db’])
    cursor = conn.cursor(pymysql.cursors.DictCursor)
    cursor.execute(sql)
    list = cursor.fetchall()
    cursor.close()
    conn.close()
    return list

def execute_noquery(conn_dict,sql):
    conn = pymysql.connect(host=conn_dict[‘host’],
                          port=conn_dict[‘port’],
                          user=conn_dict[‘user’],
                          passwd=conn_dict[‘password’],
                          db=conn_dict[‘db’])
    cursor = conn.cursor()
    cursor.execute(sql)
    conn.commit()
    cursor.close()
    conn.close()
    return list

def get_gtid(conn_dict):
    sql = “show master status;”
    list = execute_query(conn_dict,sql)
    return list

def skip_gtid_on_slave(conn_dict,gtid):
    sql_1 = ‘stop group_replication;’
    sql_2 = ”’set gtid_next='{0}’;”’.format(gtid)
    sql_3 = ‘begin;’
    sql_4 = ‘commit;’
    sql_5 = ”’set gtid_next=’automatic’;”’

    try:
        execute_noquery(conn_dict, sql_1)
        execute_noquery(conn_dict, sql_2)
        execute_noquery(conn_dict, sql_3)
        execute_noquery(conn_dict, sql_4)
        execute_noquery(conn_dict, sql_5)
    except:
        raise

def get_group_replication_status(conn_dict):
    sql = ”’select MEMBER_STATE from performance_schema.replication_group_members
            where (MEMBER_HOST = ‘{0}’ or ifnull(MEMBER_HOST,”) = ”) 
            AND (MEMBER_PORT={1} or ifnull(MEMBER_PORT,”) =”) ; ”’.format(conn_dict[“host”], conn_dict[“port”])
    result = execute_query(conn_dict,sql)
    if result:
        return result
    else:
        return None

def check_replication_group_members(conn_dict):
    print(‘——————————————————-‘)
    result = execute_query(conn_dict, ” select * from performance_schema.replication_group_members; “)
    if result:
        column = result[0].keys()
        current_row = ”
        for key in column:
            current_row += str(key) + ”    “
        print(current_row)

        for row in result:
            current_row = ”
            for key in row.values():
                current_row += str(key) + ”    “
            print(current_row)
    print(‘——————————————————-‘)

def auto_fix_mgr_error(conn_master_dict,conn_slave_dict):
    group_replication_status = get_group_replication_status(conn_slave_dict)
    if(group_replication_status[0][“MEMBER_STATE”]==”ERROR” or group_replication_status[0][“MEMBER_STATE”] == “OFFLINE”):
        print(conn_slave_dict[“host”]+str(conn_slave_dict[“port”])+’——>’+group_replication_status[0][“MEMBER_STATE”])
        print(“auto fixing……”)
        while 1 > 0:
            master_gtid_list = get_gtid(conn_master_dict)
            slave_gtid_list = get_gtid(conn_slave_dict)
            master_executed_gtid_value = int((master_gtid_list[-1][“Executed_Gtid_Set”]).split(“-“)[-1])
            slave_executed_gtid_value = int(slave_gtid_list[-1][“Executed_Gtid_Set”].split(“-“)[-1])
            slave_executed_gtid_prefix = slave_gtid_list[-1][“Executed_Gtid_Set”].split(“:”)[0]
            slave_executed_skiped_gtid = slave_executed_gtid_value + 1
            if (master_executed_gtid_value > slave_executed_gtid_value):
                print(“skip gtid and restart group replication,skiped gtid is “
                      + slave_gtid_list[-1][“Executed_Gtid_Set”].split(“:”)[-1].split(“-“)[0]
                      + “:”+str(slave_executed_skiped_gtid))
                slave_executed_skiped_gtid = slave_executed_gtid_prefix+”:”+str(slave_executed_skiped_gtid)
                skip_gtid_on_slave(conn_slave_dict,slave_executed_skiped_gtid)
                time.sleep(10)
                start_group_replication(conn_slave_dict)
                if(get_group_replication_status(conn_slave_dict)[0][“MEMBER_STATE”]==”ONLINE”):
                    print(“mgr cluster fixed,back to normal”)
                    break
            else:
                start_group_replication(conn_slave_dict)
                if(get_group_replication_status(conn_slave_dict)[0][“MEMBER_STATE”]==”ONLINE”):
                    print(“mgr cluster fixed,back to normal”)
                break
    elif (group_replication_status[0][‘MEMBER_STATE’] == ‘ONLINE’):
        print(“mgr cluster is normal,nothing to do”)
        check_replication_group_members(conn_slave_dict)

”’
reset master
”’
def reset_master(conn_dict):
    try:
        execute_noquery(conn_dict, “reset master;”)
    except:
        raise

def install_group_replication_plugin(conn_dict):
    get_plugin_sql = “SELECT name,dl FROM mysql.plugin WHERE name = ‘group_replication’;”
    install_plugin_sql = ”’install plugin group_replication soname ‘group_replication.so’; ”’
    try:
        result = execute_query(conn_dict, get_plugin_sql)
        if not result:
            execute_noquery(conn_dict, install_plugin_sql)
    except:
        raise

def create_mgr_repl_user(conn_master_dict,user,password):
    try:
        reset_master(conn_master_dict)
        sql_exists_user = ”’select user from mysql.user where user = ‘{0}’; ”’.format(user)
        user_list = execute_query(conn_master_dict,sql_exists_user)
        if not user_list:
            create_user_sql = ”’create user {0}@’%’ identified by ‘{1}’; ”’.format(user,password)
            grant_privilege_sql = ”’grant replication slave on *.* to {0}@’%’;”’.format(user)
            execute_noquery(conn_master_dict,create_user_sql)
            execute_noquery(conn_master_dict, grant_privilege_sql)
            execute_noquery(conn_master_dict, “flush privileges;”)
    except:
        raise

def set_super_read_only_off(conn_dict):
    super_read_only_off = ”’set global super_read_only = 0;”’
    execute_noquery(conn_dict, super_read_only_off)

def open_group_replication_bootstrap_group(conn_dict):
    sql = ”’select variable_name,variable_value from performance_schema.global_variables where variable_name = ‘group_replication_bootstrap_group’;”’
    result = execute_query(conn_dict, sql)
    open_bootstrap_group_sql = ”’set @@global.group_replication_bootstrap_group=on;”’
    if result and result[0][‘variable_value’]==”OFF”:
        execute_noquery(conn_dict, open_bootstrap_group_sql)

def close_group_replication_bootstrap_group(conn_dict):
    sql = ”’select variable_name,variable_value from performance_schema.global_variables where variable_name = ‘group_replication_bootstrap_group’;”’
    result = execute_query(conn_dict, sql)
    close_bootstrap_group_sql = ”’set @@global.group_replication_bootstrap_group=off;”’
    if result and result[0][‘variable_value’] == “ON”:
        execute_noquery(conn_dict, close_bootstrap_group_sql)

def start_group_replication(conn_dict):
    start_group_replication = ”’start group_replication;”’
    group_replication_status = get_group_replication_status(conn_dict)
    if not (group_replication_status[0][‘MEMBER_STATE’] == ‘ONLINE’):
        execute_noquery(conn_dict, start_group_replication)

def connect_to_group(conn_dict,repl_user,repl_password):
    connect_to_group_sql = ”’change master to
                                    master_user='{0}’,
                                    master_password='{1}’
                                    for channel ‘group_replication_recovery’; ”’.format(repl_user,repl_password)
    try:
        execute_noquery(conn_dict, connect_to_group_sql)
    except:
        raise

def start_mgr_on_master(conn_master_dict,repl_user,repl_password):
    try:
        set_super_read_only_off(conn_master_dict)
        reset_master(conn_master_dict)
        create_mgr_repl_user(conn_master_dict,repl_user,repl_password)
        connect_to_group(conn_master_dict,repl_user,repl_password)

        open_group_replication_bootstrap_group(conn_master_dict)
        start_group_replication(conn_master_dict)
        close_group_replication_bootstrap_group(conn_master_dict)

        group_replication_status = get_group_replication_status(conn_master_dict)
        if (group_replication_status[0][‘MEMBER_STATE’] == ‘ONLINE’):
            print(“master added in mgr and run successfully”)
            return True
    except:
        raise
        print(“############start master mgr error################”)
        exit(1)

def start_mgr_on_slave(conn_slave_dict,repl_user,repl_password):
    try:
        set_super_read_only_off(conn_slave_dict)
        reset_master(conn_slave_dict)
        connect_to_group(conn_slave_dict,repl_user,repl_password)
        start_group_replication(conn_slave_dict)
        # wait for 10
        time.sleep(10)
        # then check mgr status
        group_replication_status = get_group_replication_status(conn_slave_dict)
        if (group_replication_status[0][‘MEMBER_STATE’] == ‘ONLINE’):
            print(“slave added in mgr and run successfully”)
        if (group_replication_status[0][‘MEMBER_STATE’] == ‘RECOVERING’):
            print(“slave is recovering”)
    except:
        print(“############start slave mgr error################”)
        exit(1)

def auto_mgr(conn_master,conn_slave_1,conn_slave_2,repl_user,repl_password):
    install_group_replication_plugin(conn_master)
    master_replication_status = get_group_replication_status(conn_master)

    if not (master_replication_status[0][‘MEMBER_STATE’] == ‘ONLINE’):
        start_mgr_on_master(conn_master,repl_user,repl_password)

    slave1_replication_status = get_group_replication_status(conn_slave_1)
    if not (slave1_replication_status[0][‘MEMBER_STATE’] == ‘ONLINE’):
        install_group_replication_plugin(conn_slave_1)
        start_mgr_on_slave(conn_slave_1, repl_user, repl_user)

    slave2_replication_status = get_group_replication_status(conn_slave_2)
    if not (slave2_replication_status[0][‘MEMBER_STATE’] == ‘ONLINE’):
        install_group_replication_plugin(conn_slave_2)
        start_mgr_on_slave(conn_slave_2, repl_user, repl_user)

    check_replication_group_members(conn_master)

if __name__ == ‘__main__’:
    conn_master  = {‘host’: ‘127.0.0.1’, ‘port’: 7001, ‘user’: ‘root’, ‘password’: ‘root’, ‘db’: ‘mysql’, ‘charset’: ‘utf8mb4’}
    conn_slave_1 = {‘host’: ‘127.0.0.1’, ‘port’: 7002, ‘user’: ‘root’, ‘password’: ‘root’, ‘db’: ‘mysql’, ‘charset’: ‘utf8mb4’}
    conn_slave_2 = {‘host’: ‘127.0.0.1’, ‘port’: 7003, ‘user’: ‘root’, ‘password’: ‘root’, ‘db’: ‘mysql’, ‘charset’: ‘utf8mb4’}
    repl_user = “repl”
    repl_password = “repl”
    #auto_mgr(conn_master,conn_slave_1,conn_slave_2,repl_user,repl_password)

    auto_fix_mgr_error(conn_master,conn_slave_1)
    check_replication_group_members(conn_master)