PostgreSQL主从流复制状态监控和自动故障转移的轻量级实现 ...

如何实现PostgreSQL的高可用，之前研究过repmgr以及pg_auto_failover，起作用都是起到主节点故障时，实现“自动故障转移”的目的。
但是repmgr以及pg_auto_failover得缺点是对数据库侵入过多，必要在监控的数据库内部进行一系列的设置操作，同时必要启动第三方服务实现节点的可用性监控，这又引入了额外的不确定性因素。
如果是单纯地为了故障转移，repmgr以及pg_auto_failover都显得过于极重，于是快速尝试了一个轻量级的实现方案
实现思绪

紧张思绪如下：

1. 1，自动故障转移：该方式在正常情况下，作为一个服务持续监控主从复制状态
2. while 1：
3.         轮训PostgreSQL流复制的状态：
4.         if 如果主节点不可达：
5.             if 判断从节点是否可达：
6.             if 从节点状态是是in_recovery：
7.                         promote 从节点，从节点接管读写服务
8.             
11.        
12. 2，手动故障转移：如果是主从节点都正常的情况下，为了演练灾备切换，人工故障转移
13. if 判断从节点是否可达：
14.     if 从节点状态是是in_recovery：
15.             promote 从节点，从节点接管读写服务
16. #把原始主节点作为从节点加入集群中运行
17. if 如果主节点不可达：
18.         if 从节点状态是不是in_recovery：
19.                 1，关闭原始主库
20.                 2，尝试修复时间线
21.                 3，尝试以standby的模式启动主库

复制代码

优点

快速尝试了一个轻量级的实现方案，优点如下：
1，不引入或者依靠于任何第三方中间件，可以实现多组流复制集群的监控和故障转移。
2，不对现有PostgreSQL的复制环境做任何修改和侵入，以“第三方”的视角来监控现有PostgreSQL流复制，可以自行实现PostgreSQL流复制状态监控和故障转移。
3，支持在主节点故障时实现自动故障转移，或者手动故障转移视线灾备切换演练。
4，简单轻量，可以自定义故障转移的模式以及自定义日记等信息，通知信息等 详细实现

详细实现如下，目前在两台EC2上测试环境上，一台腾讯云，一台阿里云上搭建PostgreSQL流复制。经测试：可以一键快速实现主从切换，或者一连故障转移（A节点切换到B节点，B节点切换到A节点，必要修改主从毗连信息）

1. import psycopg2
2. import time
3. import paramiko
6. # 连接测试
7. def conn_to_postgres(db_config):
8.     try:
9.         conn = psycopg2.connect(**db_config)
10.         conn.close()
11.         return True
12.     except Exception as e:
13.         print(f"Error connecting to master: {e}")
14.         return False
17. # 判断节点身份
18. def is_postgresql_recovery(db_config):
19.     try:
20.         with psycopg2.connect(**db_config) as conn:
21.             with conn.cursor() as cur:
22.                 cur.execute("SELECT pg_is_in_recovery();")
23.                 in_recovery = cur.fetchone()[0]
24.         return in_recovery
25.     except Exception as e:
26.         print(f"Error connecting to master: {e}")
27.         return False
30. # 探测节点是否可达
31. def is_postgresql_reachable(db_config,retry_times):
32.     # 这里仅仅判断主节点是否可连通
33.     # 判断节点是否可达的时候，作为仲裁方，可以增加其他判断，比如当前节点是否能ping的通DNS服务器等，标明当前节点的判断是没有问题的
34.     while retry_times > 0:
35.         if not conn_to_postgres(db_config):
36.             print('the postgres cannot reachable,retrying......')
37.             time.sleep(10)
38.             retry_times = retry_times -1
39.         else:
40.             return True
41.     else:
42.         return False
45. def ssh_conn(host, username, password, port):
46.     ssh = paramiko.SSHClient()
47.     ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
48.     try:
49.         ssh.connect(hostname=host, port=port, username=username, password=password)
50.         return ssh
51.     except Exception as e:
52.         return None, str(e)
55. def create_replication_slot_if_not_exists(db_config, slot_name):
56.     try:
57.         with psycopg2.connect(**db_config) as conn:
58.             with conn.cursor() as cur:
59.                 cur.execute("SELECT slot_name FROM pg_replication_slots WHERE slot_name = '{0}';".format(slot_name))
60.                 var_slot_name = cur.fetchall()
61.                 if not var_slot_name:
62.                     cur.execute("SELECT * FROM pg_create_physical_replication_slot('{0}');".format(slot_name))
63.                     print(f"Replication slot '{slot_name}' created.")
64.                 else:
65.                     print(f"Replication slot '{slot_name}' already exists.")
66.     except Exception as e:
67.         print(f"Error connecting to master: {e}")
68.         return False
71. def promote_standby_to_primary(db_config):
72.     try:
73.         with psycopg2.connect(**db_config) as conn:
74.             conn.autocommit = True
75.             with conn.cursor() as cur:
76.                 cur.execute("SELECT pg_promote(true);")
77.         print("{0} Standby promoted to primary successfully.".format(db_config['host']))
78.     except Exception as e:
79.         print("{0} Standby promoted to primary failed : {1}".format(db_config[0],e))
82. # 时间线修复
83. def run_pg_rewind(primary_node, standby_node):
84.     # 新的主服务器的连接信息（pg_rewind需要这些信息）
85.     primary_db_config = "host={0} port={1} user={2} password={3}".format(primary_node['db_config']['host'],
86.                                                                          primary_node['db_config']['port'],
87.                                                                          primary_node['db_config']['user'],
88.                                                                          primary_node['db_config']['password'])
89.     # 构建pg_rewind命令
90.     pg_rewind_cmd = ("sudo -u postgres {0}/pg_rewind --target-pgdata={1} --source-server='host={2} port={3} user={4} dbname='postgres' password={5}'"
91.                      .format(standby_node['pg_base_dir'],
92.                              standby_node['pg_data_dir'],
93.                              primary_node['db_config']['host'],
94.                              primary_node['db_config']['port'],
95.                              primary_node['db_config']['user'],
96.                              primary_node['db_config']['password'] ))
97.     ssh = ssh_conn(host=standby_node['host'], username=standby_node['user_name'], port=standby_node['port'], password=standby_node['password'])
99.     try:
100.         stdin, stdout, stderr = ssh.exec_command('sudo systemctl stop postgresql9000')
101.         output = stdout.read().decode('utf-8')
102.         error = stderr.read().decode('utf-8')
103.         if stderr.channel.recv_exit_status() != 0:
104.             raise Exception(f"Failed to stop postgresql9000")
105.             print(error)
107.         print('######begin to rewind standby node')
108.         print(pg_rewind_cmd)
109.         stdin, stdout, stderr = ssh.exec_command(pg_rewind_cmd)
110.         output = stdout.read().decode('utf-8')
111.         error = stderr.read().decode('utf-8')
112.         if stderr.channel.recv_exit_status() != 0:
113.             print('standby node pg_rewind failed')
114.             raise Exception(f"{0} Failed to rewind commend".format(standby_node['db_config']['host']))
115.         print('######standby node pg_rewind successfully')
116.         return True
117.     except Exception as err:
118.         print('pg_rewind failed ' + str(err))
119.         return False
120.     finally:
121.         ssh.close()
124. def startup_as_replica(primary_node,standby_node):
125.     standby_auto_conf_path = '{0}/postgresql.auto.conf'.format(standby_node['pg_data_dir'])
126.     standby_signal_path = '{0}/standby.signal'.format(standby_node['pg_data_dir'])
128.     ssh = ssh_conn(host=standby_node['host'], username=standby_node['user_name'], port=standby_node['port'], password=standby_node['password'])
130.     # 要写入postgresql.auto.conf的内容（模版）
131.     auto_conf_content = """
132.         primary_conninfo = 'user={2} password=''{3}'' channel_binding=prefer host={0} port={1} sslmode=prefer sslcompression=0 sslcertmode=allow sslsni=1 ssl_min_protocol_version=TLSv1.2 gssencmode=disable krbsrvname=postgres gssdelegation=0 target_session_attrs=any load_balance_hosts=disable'
133.         primary_slot_name = '{4}'
134.         """.format(primary_node['db_config']['host'],
135.                    primary_node['db_config']['port'],
136.                    primary_node['repl_user'],
137.                    primary_node['repl_password'],
138.                    primary_node['slot_name']).lstrip()
140.     # 要创建的standby.signal文件（空文件即可，表示该节点为备用节点）
141.     try:
143.         stdin, stdout, stderr = ssh.exec_command('sudo systemctl stop postgresql9000')
144.         output = stdout.read().decode('utf-8')
145.         error = stderr.read().decode('utf-8')
146.         if stderr.channel.recv_exit_status() != 0:
147.             raise Exception(f"Failed to stop postgresql9000")
148.             print(error)
150.         try:
151.             # 创建standby.signal文件（如果尚不存在）
152.             print('###### {0} touch {1}'.format(standby_node['host'],standby_signal_path))
153.             stdin, stdout, stderr = ssh.exec_command(f'sudo -u postgres touch {standby_signal_path}')
154.             output = stdout.read().decode('utf-8')
155.             error = stderr.read().decode('utf-8')
156.             # 检查命令是否成功执行
157.             if stderr.channel.recv_exit_status() != 0:
158.                 raise Exception(f"Failed to create {standby_signal_path}: {stderr.read().decode()}")
160.             print('###### {0} touch {1}'.format(standby_node['host'], standby_auto_conf_path))
161.             stdin, stdout, stderr = ssh.exec_command(f'sudo -u postgres touch {standby_auto_conf_path}')
162.             stdin, stdout, stderr = ssh.exec_command('''sudo  echo "{0}"  > {1}'''.format(auto_conf_content,standby_auto_conf_path))
163.             output = stdout.read().decode('utf-8')
164.             error = stderr.read().decode('utf-8')
165.             # 检查命令是否成功执行
166.             if stderr.channel.recv_exit_status() != 0:
167.                 print(error)
168.                 raise Exception(f"Failed to create {standby_signal_path}: {stderr.read().decode()}")
169.         except Exception as err:
170.             print(err)
172.         stdin, stdout, stderr = ssh.exec_command('sudo systemctl restart postgresql9000')
173.         output = stdout.read().decode('utf-8')
174.         error = stderr.read().decode('utf-8')
175.         # 检查命令是否成功执行
176.         if stderr.channel.recv_exit_status() != 0:
177.             print(error)
178.             raise Exception(f"{0} restart postgresql failed".format(standby_node['host']))
179.     finally:
180.         ssh.close()
183. # 人工故障转移：在主节点可达的情况，把主节点作为从节点运行，可以是灾备切换的预演
184. def execute_failover(primary_node,standby_node):
185.     print('promote 从节点为主节点')
186.     # 判断standby节点是否可达，如果不可达，重试5次
187.     if is_postgresql_reachable(standby_node['db_config'],retry_times=5):
188.         # 判断standby节点是否处于recovery模式
189.         if is_postgresql_recovery(standby_node['db_config']):
190.             # 创建复制槽，为从节点复制做准备
191.             create_replication_slot_if_not_exists(standby_node['db_config'], standby_node['slot_name'])
192.             # promote standby节点
193.             promote_standby_to_primary(standby_node['db_config'])
194.         else:
195.             print('当前节点非recovery模式，不可执行promote')
197.     # 将旧的主节点作为从节点加入新的主节点复制
198.     # 判断原主节点是否可达
199.     print('将原始主节点，重新以从节点身份加入复制集群')
200.     if is_postgresql_reachable(primary_node['db_config'],retry_times=5):
201.         # 执行pg_rewind修复原始主节点的时间线，请注意：此时从节点提升为主节点，主节点已经变为从节点，所以需要注意节点身份的变化和参数的变化
202.         pg_rewind_result = run_pg_rewind(primary_node=standby_node,standby_node=primary_node)
203.         if pg_rewind_result:
204.             # 注意此时主从节点身份已经变换，所以这里参数也变化,原来的从节点已经成为主节点，所以注意参数的交换
205.             startup_as_replica(primary_node=standby_node, standby_node=primary_node)
208. # 自动检测和故障转移
209. def auto_failover(primary_node,standby_node):
210.     while True:
211.         if not is_postgresql_reachable(primary_node['db_config'], retry_times=5):
212.             # 告警通知，说明当前配置的主节点故障，即将开始主从切换
213.             execute_failover(primary_node,standby_node)
214.             # 执行完切换之后，进行写日志，邮件/短信告警通知等
215.             # 因为主从身份发生了变化，不再对当前的主从配置进行监控，应该第一时间人工介入：1，确认原始主节点是否正常，2，原始主节点是否以从节点的身份加入集群
216.             exit(0)
217.         # 定期检查间隔
218.         time.sleep(15)
221. if __name__ == "__main__":
222.    
223.     # 数据库连接配置
224.     primary_node  = {
225.         'host':'***.***.***.***',                                #主机名
226.         'user_name':"root",                                      #系统用户名
227.         'password': "******",                                    #系统密码
228.         'port':22,                                               #ssh端口号
229.         'pg_base_dir':'/usr/local/pgsql16/server/bin',           #PostgreSQL Server路径
230.         'pg_data_dir':'/usr/local/pgsql16/pg9000/data',          #PostgreSQL 数据路径
231.         'repl_user':'replica_user',                              #流复制用户名
232.         'repl_password':'******',                                #流复制用户密码
233.         'slot_name': 'pgstandby_slave01',                        #流复制slot名字
234.         'db_config': {                                           #数据库配置
235.             'host': '***.***.***.***',
236.             'user': 'postgres',
237.             'password': '******',
238.             'dbname': 'postgres',
239.             'port': '******'
240.         },
241.     }
243.     standby_node1 = {
244.         'host':'***.***.***.***',
245.         'user_name':"root",
246.         'password': "******",
247.         'port':22,
248.         'pg_base_dir':'/usr/local/pgsql16/server/bin',
249.         'pg_data_dir':'/usr/local/pgsql16/pg9000/data',
250.         'repl_user':'replica_user',
251.         'repl_password':'******',
252.         'slot_name': 'pgstandby_slave01',
253.         'db_config': {
254.             'host': '***.***.***.***',
255.             'user': 'postgres',
256.             'password': '******',
257.             'dbname': 'postgres',
258.             'port': '******'
259.         },
260.     }
262.     # 手动故障转移，主从节点以及复制均正常的情况下，实现参数里的主从节点交换身份
263.     execute_failover(primary_node=primary_node, standby_node=standby_node1)
265.     # 单独的时间线修复操作
266.     #pg_rewind_result = run_pg_rewind(primary_node=standby_node1, standby_node=primary_node)
268.     # 单独的讲一个节点作为从节点运行
269.     # 注意此时主从节点身份已经变换，所以这里参数也变化,原来的从节点已经成为主节点，所以注意参数的交换
270.     # startup_as_replica(primary_node=standby_node1, standby_node=primary_node)

复制代码

 
待改进

1，用户名密码均已明文方式写在主机信息中
2，pg_rewind的时候必要重启服务器，这里把PostgreSQL的服务写死了，笔者的测试环境PostgreSQL服务名是postgresql9000
3，如果是自动切换模式，自动故障转移后，尚未修改主从设置信息，此时节点的身份信息还是切换前的
4，判定节点是否正常的逻辑中，仅仅通过是否可以毗连来实现，如果监控服务自己无法与被监控节点通信，可能会出现误判的环境

 

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。