1
2
3

SELECT SUBSTRING_INDEX(host, ':', 1) AS slave_hostname
FROM information_schema.processlist
WHERE command IN ('Binlog Dump', 'Binlog Dump GTID');

1
2
3

{
  "ReplicationLagQuery": "select absolute_lag from meta.heartbeat_view"
}

1
2
3
4

-- 创建元数据库和账号
CREATE DATABASE orchestrator;
CREATE USER 'orchestrator'@'127.0.0.1' IDENTIFIED BY '123456';
GRANT ALL PRIVILEGES ON orchestrator.* TO 'orchestrator'@'127.0.0.1';

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# 下载二进制包
wget https://github.com/openark/orchestrator/releases/download/v3.2.4/orchestrator-3.2.4-linux-amd64.tar.gz

# 创建安装目录(自行选择)
mkdir /home/orchestrator

# 解压包到安装目录
tar xf orchestrator-3.2.4-linux-amd64.tar.gz -C /home/orchestrator

# 拷贝 orchestrator 命令及静态资源到安装目录
cp /home/orchestrator/usr/local/orchestrator/orchestrator /home/orchestrator
cp -r /home/orchestrator/usr/local/orchestrator/resources /home/orchestrator

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

{
  "MySQLTopologyUser": "orch_client",
  "MySQLTopologyPassword": "123456",
  "MySQLOrchestratorHost": "127.0.0.1",
  "MySQLOrchestratorPort": 3306,
  "MySQLOrchestratorDatabase": "orchestrator",
  "MySQLOrchestratorUser": "orchestrator",
  "MySQLOrchestratorPassword": "123456",
  "MySQLHostnameResolveMethod": "@@report_host",
  "RaftEnabled": true,
  "RaftDataDir": "/home/orchestrator",
  "RaftBind": "10.xxxx",
  "RaftNodes": [
    "10.xxxx",
    "10.xxxx",
    "10.xxxx"
  ]
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# cat /etc/systemd/system/orchestrator.service
[Unit]
Description=orchestrator: MySQL replication management and visualization
Documentation=https://github.com/openark/orchestrator
After=syslog.target network.target mysqld.service mysql.service

[Service]
User=root
Group=root
Type=simple
WorkingDirectory=/home/orchestrator
ExecStart=/bin/sh -c '/home/orchestrator/orchestrator --config=/home/orchestrator/orchestrator.conf.json http >> /home/var/log/orchestrator.log 2>&1'
EnvironmentFile=-/etc/sysconfig/orchestrator
ExecReload=/bin/kill -HUP $MAINPID

[Install]
WantedBy=multi-user.target

1
2

systemctl daemon-reload
systemctl start orchestrator.service

1
2

CREATE USER 'orch_client'@'%' IDENTIFIED BY 'XXXXXXXXXXXXXXXXXXXXXX';
GRANT RELOAD, PROCESS, SUPER, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'orch_client'@'%';

1
2
3
4
5

{
  "Debug": true,                    // debug模式，输出详细信息
  "EnableSyslog": false,            // 是否输出到系统日志里
  "ListenAddress": ":3000"          // orchestrator的监听端口，web端口
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

{
  "MySQLTopologyUser": "orch_client",         // 后端被管理的mysql实例中的账号，所有实例都要有
  "MySQLTopologyPassword": "123456",          // 密码
  "MySQLTopologyCredentialsConfigFile": "",   // 验证的配置文件，账号密码可以直接写入文件，读取
  "MySQLTopologySSLPrivateKeyFile": "",       // ssl验证文件
  "MySQLTopologySSLCertFile": "",
  "MySQLTopologySSLCAFile": "",
  "MySQLTopologySSLSkipVerify": true,         // 跳过验证
  "MySQLTopologyUseMutualTLS": false          // 使用TLS验证
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

{
  "MySQLOrchestratorHost": "127.0.0.1",              // orchestrator的IP，也可以是本机IP
  "MySQLOrchestratorPort": 3306,                     // orchestrator所在的端口
  "MySQLOrchestratorDatabase": "orchestrator",       // orchestrator元数据的数据库名称
  "MySQLOrchestratorUser": "root",                   // 管理orchestrator数据库的账户
  "MySQLOrchestratorPassword": "123456",             // 密码
  "MySQLOrchestratorCredentialsConfigFile": "",
  "MySQLOrchestratorSSLPrivateKeyFile": "",
  "MySQLOrchestratorSSLCertFile": "",
  "MySQLOrchestratorSSLCAFile": "",
  "MySQLOrchestratorSSLSkipVerify": true,
  "MySQLOrchestratorUseMutualTLS": false
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

{
  "MySQLConnectTimeoutSeconds": 1,               // orchestrator连接mysql超时秒数
  "DefaultInstancePort": 3307,                   // mysql实例的端口，对外提供服务的实例
  "DiscoverByShowSlaveHosts": true,              // 是否启用审查和自动发现
  "InstancePollSeconds": 5,                      // orchestrator探测mysql间隔秒数
  "SkipMaxScaleCheck": true,                     // 没有MaxScale binlogserver设置为true
  "UnseenInstanceForgetHours": 240,              // 忘记看不见的实例的小时数
  "SnapshotTopologiesIntervalHours": 0,          // 快照拓扑调用之间的小时间隔。默认值:0(禁用)
  "InstanceBulkOperationsWaitTimeoutSeconds": 10 // 执行批量(多个实例)操作时在单个实例上等待的时间
}

1
2
3
4
5
6
7

{
  "HostnameResolveMethod": "none",                   // 解析主机名，默认default 不解析为none
  "MySQLHostnameResolveMethod": "@@hostname",        // MySQL主机名解析
  "SkipBinlogServerUnresolveCheck": true,            // 跳过二进制服务器检测
  "ExpiryHostnameResolvesMinutes": 60,               // 域名检测过期分钟数
  "RejectHostnameResolvePattern": ""                 // 禁止的域名正则表达式
}

1
2
3
4
5
6
7
8
9

{
  "ReasonableReplicationLagSeconds": 10,                  // 复制延迟高于10秒表示异常
  "ProblemIgnoreHostnameFilters": [],                     // 主机正则匹配筛选最小化
  "VerifyReplicationFilters": false,                      // 重构前检查复制筛选器
  "ReasonableMaintenanceReplicationLagSeconds": 20,       // 上移和下移的阈值
  "CandidateInstanceExpireMinutes": 60,                   // 实例过期分钟数
  "AuditLogFile": "",                                     // 审计日志
  "AuditToSyslog": false                                  // 审计日志输出到系统日志
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

{
  "RemoveTextFromHostnameDisplay": ":3306",    // 去除集群的文本
  "ReadOnly": true,                            // 全局只读
  "AuthenticationMethod": "",                  // 身份验证模式
  "HTTPAuthUser": "",                          // http验证用户名
  "HTTPAuthPassword": "",                      // http验证密码
  "AuthUserHeader": "",                        // 指示身份验证用户的HTTP标头，当AuthenticationMethod为"proxy"时
  "PowerAuthUsers": [                          // 在AuthenticationMethod=="proxy"上，可以进行更改的用户列表。所有其他都是只读的
    "*"
  ],
  "ClusterNameToAlias": {                      // 正则表达式匹配群集名称与人类友好别名之间的映射
    "127.0.0.1": "test suite"
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

{
  "SlaveLagQuery": "",                          // 使用SHOW SLAVE STATUS进行延迟判断
  "DetectClusterAliasQuery": "SELECT SUBSTRING_INDEX(@@hostname, '.', 1)",  // 查询集群别名
  "DetectClusterDomainQuery": "",               // 可选查询（在拓扑实例上执行），返回此集群主服务器的VIP/CNAME/别名/任何域名。查询将仅在集群主机上执行。如果提供，必须返回一行，一列
  "DetectInstanceAliasQuery": "",               // 可选查询（在拓扑实例上执行），返回实例的别名。如果提供，必须返回一行，一列
  "DetectPromotionRuleQuery": "",               // 可选查询（在拓扑实例上执行），返回实例的提升规则。如果提供，必须返回一行，一列
  "DataCenterPattern": "[.]([^.]+)[.][^.]+[.]mydomain[.]com",                // 从正则表达式中筛选数据中心名称
  "PhysicalEnvironmentPattern": "[.]([^.]+[.][^.]+)[.]mydomain[.]com",       // 返回实例的物理环境
  "PromotionIgnoreHostnameFilters": [],         // Orchestrator不会使用主机名匹配模式来提升副本(通过-c recovery;例如，避免使用dev专用计算机)
  "DetectSemiSyncEnforcedQuery": ""             // 查询以确定是否强制完全半同步写入
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

{
  "ServeAgentsHttp": false,           // 产生一个agent的http接口
  "AgentsServerPort": ":3001",        // 可选，对于raft设置，此节点将向其对等方通告的HTTP地址
  "AgentsUseSSL": false,              // 当"true" orchestrator将使用SSL侦听代理端口以及通过SSL连接到代理时
  "AgentsUseMutualTLS": false,
  "AgentSSLSkipVerify": false,
  "AgentSSLPrivateKeyFile": "",
  "AgentSSLCertFile": "",
  "AgentSSLCAFile": "",
  "AgentSSLValidOUs": [],
  "UseSSL": false,                    // 在服务器Web端口上使用SSL
  "UseMutualTLS": false,
  "SSLSkipVerify": false,
  "SSLPrivateKeyFile": "",
  "SSLCertFile": "",
  "SSLCAFile": "",
  "SSLValidOUs": [],
  "URLPrefix": "",                    // 在非根Web路径上运行orchestrator的URL前缀，例如/orchestrator将其置于nginx之后
  "StatusEndpoint": "/api/status",    // 使用相互TLS时的有效组织单位
  "StatusSimpleHealth": true,
  "StatusOUVerify": false,
  "AgentPollMinutes": 60,             // 代理轮询之间的分钟数
  "UnseenAgentForgetHours": 6,        // 忘记看不见代理的小时数
  "StaleSeedFailMinutes": 60,         // 认为陈旧(无进展)种子失败的分钟数
  "SeedAcceptableBytesDiff": 8192     // 仍被视为成功复制的种子源和目标数据大小之间的字节数差异
}

1
2
3
4
5
6
7
8
9

{
  "PseudoGTIDPattern": "",                          // 为空禁用伪GTID
  "PseudoGTIDPatternIsFixedSubstring": false,       // 如果为true，则PseudoGTIDPattern不被视为正则表达式而是固定子字符串，并且可以提高搜索时间
  "PseudoGTIDMonotonicHint": "asc:",                // Pseudo-GTID条目中的子字符串，表示Pseudo-GTID条目预计会单调递增
  "DetectPseudoGTIDQuery": "",                      // 可选查询，用于权威地决定是否在实例上启用伪gtid
  "BinlogEventsChunkSize": 10000,                   // SHOW BINLOG|RELAYLOG EVENTS LIMIT ?,X语句。较小意味着更少的锁定和工作要做
  "SkipBinlogEventsContaining": [],                 // 扫描/比较Pseudo-GTID的binlog时，跳过包含给定文本的条目。这些不是正则表达式(扫描binlog时会消耗太多的CPU)，只需查找子字符串
  "ReduceReplicationAnalysisCount": true            // 当为true时，复制分析将仅报告可能首先处理问题的实例(例如，不报告大多数叶子节点)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

{
  "RaftEnabled": true,                  // raft模式
  "BackendDB": "mysql",                // 后台数据库类型
  "RaftBind": "192.168.89.103",        // 绑定地址，本机IP
  "RaftDataDir": "/var/lib/orchestrator",  // 数据目录，如果不存在，则自动创建
  "DefaultRaftPort": 10008,            // raft通信端口，所有机器必须保持一致
  "RaftNodes": [                       // raft节点，必须包含所有节点
    "192.168.89.100",
    "192.168.89.102",
    "192.168.89.103"
  ],
  "ConsulAddress": "",                 // 找到Consul HTTP api的地址。示例：127.0.0.1:8500
  "ConsulAclToken": ""                 // 用于写入Consul KV的ACL令牌
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

{
  "FailureDetectionPeriodBlockMinutes": 60,       // 该时间内发现故障，不被多次发现
  "RecoveryPeriodBlockSeconds": 3600,             // 该时间内发现故障，不会多次转移
  "RecoveryIgnoreHostnameFilters": [],            // 恢复会忽略的主机
  "RecoverMasterClusterFilters": [                // 仅在匹配这些正则表达式模式的集群上进行主恢复(当然".*"模式匹配所有内容)
    "*"
  ],
  "RecoverIntermediateMasterClusterFilters": [    // 仅在与这些正则表达式模式匹配的集群上进行IM恢复
    "*"
  ]
}

1
2
3

"OnFailureDetectionProcesses": [
  "echo 'Detected {failureType} on {failureCluster}. Affected replicas: {countSlaves} autoMasterRecovery: {autoMasterRecovery} losthost: {lostSlaves} slavehost: {slaveHosts} orchestratorHost: {orchestratorHost}' >> /tmp/recovery.log"
]

1
2
3

"PreGracefulTakeoverProcesses": [
  "echo 'Planned takeover about to take place on {failureCluster}. Master will switch to read_only autoMasterRecovery: {autoMasterRecovery} losthost: {lostSlaves} slavehost: {slaveHosts} orchestratorHost: {orchestratorHost}' >> /tmp/recovery.log"
]

1
2
3

"PreFailoverProcesses": [
  "echo 'Will recover from {failureType} on {failureCluster}' >> /tmp/recovery.log"
]

1
2
3

"PostFailoverProcesses": [
  "echo '(for all types) Recovered from {failureType} on {failureCluster}. Failed: {failedHost}:{failedPort}; Successor: {successorHost}:{successorPort}' >> /tmp/recovery.log"
]

1

"PostUnsuccessfulFailoverProcesses": []

1
2
3

"PostMasterFailoverProcesses": [
  "echo 'Recovered from {failureType} on {failureCluster}. Failed: {failedHost}:{failedPort}; Promoted: {successorHost}:{successorPort}' >> /tmp/recovery.log"
]

1
2
3

"PostIntermediateMasterFailoverProcesses": [
  "echo 'Recovered from {failureType} on {failureCluster}. Failed: {failedHost}:{failedPort}; Successor: {successorHost}:{successorPort}' >> /tmp/recovery.log"
]

1
2
3

"PostGracefulTakeoverProcesses": [
  "echo 'Planned takeover complete' >> /tmp/recovery.log"
]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

{
  "CoMasterRecoveryMustPromoteOtherCoMaster": true,     // 当'false'时，任何东西都可以升级(候选人优先于其他人)。当'true'时，orchestrator将促进其他共同主人或否则失败
  "DetachLostSlavesAfterMasterFailover": true,          // 恢复可能丢失一些副本。DetachLostReplicasAfterMasterFailover的同义词
  "ApplyMySQLPromotionAfterMasterFailover": true,       // orchestrator应该自己应用MySQL主促销：设置read_only=0，分离复制等
  "PreventCrossDataCenterMasterFailover": false,        // 当为true(默认值：false)时，不允许跨DC主故障转移，orchestrator将尽其所能只在同一DC内进行故障转移，否则根本不进行故障转移
  "MasterFailoverDetachSlaveMasterHost": false,         // 确保新主不复制旧主的数据。MasterFailoverDetachReplicaMasterHost的同义词
  "MasterFailoverLostInstancesDowntimeMinutes": 0,      // 在主故障转移(包括失败的主副本和丢失的副本)之后丢失的任何服务器停机的分钟数。0表示禁用
  "PostponeSlaveRecoveryOnLagMinutes": 0,               // PostponeReplicaRecoveryOnLagMinutes的同义词
  "OSCIgnoreHostnameFilters": []                        // OSC副本推荐将忽略与给定模式匹配的副本主机名
}

1
2
3
4
5

{
  "GraphiteAddr": "",
  "GraphitePath": "",
  "GraphiteConvertHostnameDotsToUnderscores": true
}

1

orchestrator-client -c ack-cluster-recoveries -alias somealias

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
orchestrator 自动切换后的 hook 脚本
"""
import sys

def _prompt(cluster_alias: str, failure_type: str, failed_host: str):
    log.error(f'{cluster_alias} 发生故障，故障类型：{failure_type} ,故障节点: {failed_host}')

if __name__ == '__main__':
    log.debug(f'参数个数为：{len(sys.argv)}')
    log.debug(f'参数详情：{sys.argv}')
    _prompt(cluster_alias=sys.argv[1],
            failure_type=sys.argv[2], failed_host=sys.argv[3])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
FileName: PreFailoverProcesses.py
Description:
"""
import sys
import json
from datetime import datetime, timedelta

def _run_command(command: str, ip: str, sudo: bool = False, command_timeout: int = None):
    try:
        return ssh.run(
            host=ip,
            command=command,
            sudo=sudo,
            command_timeout=command_timeout,
        )
    except Exception as err:
        log.error(f'主库 {ip} 执行命令 {command} 出错，错误信息：{err}')

def _process(old_ip: str, cluster_alias: str, failure_type: str):
    """
    Args:
        old_ip:
    Returns:
    """
    # 判断是否需要进行切换
    log.info(f'检查 failure_type 类型 {failure_type}')
    if failure_type not in ('DeadCoMaster', 'DeadMaster', 'DeadCoMasterAndSomeReplicas'):
        log.error(f'集群有节点发生异常，但不是 DeadCoMaster或DeadMaster或DeadCoMasterAndSomeReplicas类型，所以不做切换动作')
        sys.exit(1)

    log.info(f' kill旧主：{old_ip} 的mysqld进程')
    mysql_kill = 'kill -9 $(pgrep -x mysqld)'
    ssh.run_command(mysql_kill, old_ip, command_timeout = 5)

if __name__ == '__main__':
    log.debug(f' pre_graceful_takeover 参数个数为：{len(sys.argv)}')
    log.debug(f'pre_graceful_takeover 参数详情：{sys.argv}')
    _process(old_ip=sys.argv[1], cluster_alias=sys.argv[2], failure_type=sys.argv[3])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

"""
FileName: PostFailoverProcesses.py
Description:
orchestrator 自动切换后的 hook 脚本
"""
import sys
import uuid
from datetime import datetime, timedelta

def _change_vip_dns(domain: str, new_ip: str):
    #### 更换 dns或者vip

def _send_success_msg(new_ip: str, old_ip: str, cluster_alias: str, orc_command: str = ''):
    command_tag = f'[{orc_command}]' if orc_command else ''
    action_detail = {
        '': f'实例 {cluster_alias} 发生自动容灾切换，{old_ip} => {new_ip}'
    }

def _process(new_ip: str, old_ip: str, cluster_alias: str, orc_command: str = ''):
    _change_vip_dns(domain, new_ip)
    _send_success_msg(new_ip, old_ip, orc_command=orc_command)

if __name__ == '__main__':
    log.debug(f'auto_recover 参数个数为：{len(sys.argv)}')
    log.debug(f'auto_recover 参数详情：{sys.argv}')
    _process(new_ip=sys.argv[1], old_ip=sys.argv[2],
             cluster_alias=sys.argv[3], orc_command=sys.argv[4])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
FileName: PostUnsuccessfulFailoverProcesses.py
Description:
"""
import sys

def rollback_readonly(old_ip: str):
    try:
        with mysql_connect(ip=old_ip, user=user, password=password) as mysql:
            mysql.execute('set global read_only=0;')
    except Exception as err:
        log.error(f'集群切换时失败，并且尝试将原主read_only 回退失败')

def _process(old_ip: str):
    """
    Args:
        old_ip:
    Returns:
    """
    rollback_readonly(old_ip=old_ip)
    log.error(f'集群主节点发生异常，但进行切换时失败，请检查集群状态是否正常')

if __name__ == '__main__':
    log.debug(f' post_unsuccessful_failover 参数个数为：{len(sys.argv)} , 参数详情：{sys.argv}')
    _process(old_ip=sys.argv[1])

1

SELECT * FROM database_instance WHERE oracle_gtid=0 \G