From f42bab50814844070d3d0094f5b2a176221fc334 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 28 Mar 2023 07:36:45 +0200 Subject: [PATCH] Improve behaviour of SyncState.matches() (#2619) Previously it used to compare between the leader and sync_standbys, while in some cases (actually most of them) the leader should be excluded. This commit makes `matches()` method flexible: 1. The leader will be included to comparison only if requested 2. checks will be performed as case insensitive (like PG does) Besides that, everywhere in code start using `cluster.sync.matches()` instead of `name in cluster.sync.members`. --- patroni/api.py | 6 ++--- patroni/dcs/__init__.py | 43 ++++++++++++++++++++++++------- patroni/ha.py | 53 ++++++++++++++++++++++++++------------ patroni/postgresql/sync.py | 5 ++-- patroni/utils.py | 2 +- tests/test_api.py | 1 + 6 files changed, 77 insertions(+), 33 deletions(-) diff --git a/patroni/api.py b/patroni/api.py index 5b74cac5..2c84ea8f 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -546,13 +546,13 @@ class RestApiHandler(BaseHTTPRequestHandler): if leader and (not cluster.leader or cluster.leader.name != leader): return 'leader name does not match' if candidate: - if action == 'switchover' and cluster.is_synchronous_mode() and candidate not in cluster.sync.members: + if action == 'switchover' and cluster.is_synchronous_mode() and not cluster.sync.matches(candidate): return 'candidate name does not match with sync_standby' members = [m for m in cluster.members if m.name == candidate] if not members: return 'candidate does not exists' elif cluster.is_synchronous_mode(): - members = [m for m in cluster.members if m.name in cluster.sync.members] + members = [m for m in cluster.members if cluster.sync.matches(m.name)] if not members: return action + ' is not possible: can not find sync_standby' else: @@ -690,7 +690,7 @@ class RestApiHandler(BaseHTTPRequestHandler): result['role'] = postgresql.role if result['role'] == 'replica' and cluster and cluster.is_synchronous_mode()\ - and cluster.sync and postgresql.name in cluster.sync.members: + and cluster.sync.matches(postgresql.name): result['sync_standby'] = True if row[1] > 0: diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 4ae26150..b63eb4c9 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -405,32 +405,55 @@ class SyncState(namedtuple('SyncState', 'index,leader,sync_standby')): @property def is_empty(self) -> bool: - """:returns: True if /sync key doesn't have a leader""" - return self.leader is None + """:returns: True if /sync key is not valid (doesn't have a leader).""" + return not self.leader + + @staticmethod + def _str_to_list(value: str) -> List[str]: + """Splits a string by comma and returns list of strings. + + :param value: a comma separated string + :returns: list of non-empty strings after splitting an input value by comma + """ + return list(filter(lambda a: a, [s.strip() for s in value.split(',')])) @property def members(self) -> List[str]: - """:returns: sync_standby as list""" - return list(filter(lambda a: a, [s.strip() for s in self.sync_standby.split(',')])) if self.sync_standby else [] + """:returns: sync_standby as list.""" + return self._str_to_list(self.sync_standby) if not self.is_empty and self.sync_standby else [] - def matches(self, name: str) -> bool: - """:returns: True if a node name matches one of the nodes in the sync state (including leader) + def matches(self, name: Union[str, None], check_leader: Optional[bool] = False) -> bool: + """Checks if node is presented in the /sync state. + Since PostgreSQL does case-insensitive checks for synchronous_standby_name we do it also. + :param name: name of the node + :param check_leader: by default the name is searched in members, check_leader=True will include leader to list + :returns: `True` if the /sync key not :func:`is_empty` and a given name is among presented in the sync state >>> s = SyncState(1, 'foo', 'bar,zoo') >>> s.matches('foo') + False + >>> s.matches('fOo', True) True - >>> s.matches('bar') + >>> s.matches('Bar') True - >>> s.matches('zoo') + >>> s.matches('zoO') True >>> s.matches('baz') False >>> s.matches(None) False - >>> SyncState(1, None, None).matches('foo') + >>> SyncState.empty(1).matches('foo') False """ - return name is not None and name in [self.leader] + self.members + ret = False + if name and not self.is_empty: + search_str = (self.sync_standby or '') + (',' + self.leader if check_leader else '') + ret = name.lower() in self._str_to_list(search_str.lower()) + return ret + + def leader_matches(self, name: Union[str, None]) -> bool: + """:returns: `True` if name is matching the `SyncState.leader` value.""" + return name and not self.is_empty and name.lower() == self.leader.lower() class TimelineHistory(namedtuple('TimelineHistory', 'index,value,lines')): diff --git a/patroni/ha.py b/patroni/ha.py index 53f6d419..919e6f6e 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -9,6 +9,7 @@ import uuid from collections import namedtuple from multiprocessing.pool import ThreadPool from threading import RLock +from typing import List, Optional, Union from . import psycopg from .async_executor import AsyncExecutor, CriticalTask @@ -17,7 +18,7 @@ from .postgresql.callback_executor import CallbackAction from .postgresql.misc import postgres_version_to_int from .postgresql.rewind import Rewind from .utils import polling_loop, tzutc, is_standby_cluster as _is_standby_cluster, parse_int -from .dcs import Cluster, Leader, RemoteMember +from .dcs import Cluster, Leader, Member, RemoteMember logger = logging.getLogger(__name__) @@ -625,7 +626,7 @@ class Ha(object): cluster = self.dcs.get_cluster() except DCSError: return logger.warning("Could not get cluster state from DCS during process_sync_replication()") - if not cluster.sync.is_empty and cluster.sync.leader != self.state_handler.name: + if not cluster.sync.is_empty and not cluster.sync.leader_matches(self.state_handler.name): logger.info("Synchronous replication key updated by someone else") return if not self.dcs.write_sync_state(self.state_handler.name, allow_promote, index=cluster.sync.index): @@ -637,9 +638,10 @@ class Ha(object): logger.info("Disabled synchronous replication") self.state_handler.sync_handler.set_synchronous_standby_names([]) - def is_sync_standby(self, cluster): - return cluster.leader and cluster.sync.leader == cluster.leader.name \ - and self.state_handler.name in cluster.sync.members + def is_sync_standby(self, cluster: Cluster) -> bool: + """:returns: `True` if the current node is a synchronous standby.""" + return cluster.leader and cluster.sync.leader_matches(cluster.leader.name) \ + and cluster.sync.matches(self.state_handler.name) def while_not_sync_standby(self, func): """Runs specified action while trying to make sure that the node is not assigned synchronous standby status. @@ -863,16 +865,25 @@ class Ha(object): logger.info('Wal position of %s is ahead of my wal position', st.member.name) # In synchronous mode the former leader might be still accessible and even be ahead of us. # We should not disqualify himself from the leader race in such a situation. - if not self.is_synchronous_mode() or st.member.name != self.cluster.sync.leader: + if not self.is_synchronous_mode() or self.cluster.sync.is_empty\ + or not self.cluster.sync.leader_matches(st.member.name): return False logger.info('Ignoring the former leader being ahead of us') return True - def is_failover_possible(self, members, check_synchronous=True, cluster_lsn=None): + def is_failover_possible(self, members: List[Member], check_synchronous: Optional[bool] = True, + cluster_lsn: Optional[int] = 0) -> bool: + """Checks whether one of the members from the list can possibly win the leader race. + + :param members: list of members to check + :param check_synchronous: consider only members that are known to be listed in /sync key when sync replication. + :param cluster_lsn: to calculate replication lag and exclude member if it is laggin + :returns: `True` if there are members eligible to be the new leader + """ ret = False cluster_timeline = self.cluster.timeline members = [m for m in members if m.name != self.state_handler.name and not m.nofailover and m.api_url] - if check_synchronous and self.is_synchronous_mode(): + if check_synchronous and self.is_synchronous_mode() and not self.cluster.sync.is_empty: members = [m for m in members if self.cluster.sync.matches(m.name)] if members: for st in self.fetch_nodes_statuses(members): @@ -893,7 +904,12 @@ class Ha(object): logger.warning('manual failover: members list is empty') return ret - def manual_failover_process_no_leader(self): + def manual_failover_process_no_leader(self) -> Union[bool, None]: + """Handles manual failover/switchover when the old leader already stepped down. + + :returns: - `True` if the current node is the best candidate to become the new leader + - `None` if the current node is running as a primary and requested candidate doesn't exist + """ failover = self.cluster.failover if failover.candidate: # manual failover to specific member if failover.candidate == self.state_handler.name: # manual failover to me @@ -901,8 +917,8 @@ class Ha(object): elif self.is_paused(): # Remove failover key if the node to failover has terminated to avoid waiting for it indefinitely # In order to avoid attempts to delete this key from all nodes only the primary is allowed to do it. - if (not self.cluster.get_member(failover.candidate, fallback_to_leader=False) and - self.state_handler.is_leader()): + if not self.cluster.get_member(failover.candidate, fallback_to_leader=False)\ + and self.state_handler.is_leader(): logger.warning("manual failover: removing failover key because failover candidate is not running") self.dcs.manual_failover('', '', index=self.cluster.failover.index) return None @@ -910,7 +926,7 @@ class Ha(object): # in synchronous mode when our name is not in the /sync key # we shouldn't take any action even if the candidate is unhealthy - if self.is_synchronous_mode() and not self.cluster.sync.matches(self.state_handler.name): + if self.is_synchronous_mode() and not self.cluster.sync.matches(self.state_handler.name, True): return False # find specific node and check that it is healthy @@ -945,7 +961,12 @@ class Ha(object): members = [m for m in self.cluster.members if m.name != failover.leader] return self._is_healthiest_node(members, check_replication_lag=False) - def is_healthiest_node(self): + def is_healthiest_node(self) -> bool: + """Performs a series of checks to determine that the current node is the best candidate. + + In case if manual failover/switchover is requested it calls :func:`manual_failover_process_no_leader` method. + :returns: `True` if the current node is among the best candidates to become the new leader. + """ if time.time() - self._released_leader_key_timestamp < self.dcs.ttl: logger.info('backoff: skip leader race after pre_promote script failure and releasing the lock voluntarily') return False @@ -973,7 +994,7 @@ class Ha(object): if self.cluster.failover: # When doing a switchover in synchronous mode only synchronous nodes and former leader are allowed to race if self.is_synchronous_mode() and self.cluster.failover.leader and \ - not self.cluster.sync.matches(self.state_handler.name): + not self.cluster.sync.is_empty and not self.cluster.sync.matches(self.state_handler.name, True): return False return self.manual_failover_process_no_leader() @@ -995,10 +1016,10 @@ class Ha(object): # When in sync mode, only last known primary and sync standby are allowed to promote automatically. if self.is_synchronous_mode() and not self.cluster.sync.is_empty: - if not self.cluster.sync.matches(self.state_handler.name): + if not self.cluster.sync.matches(self.state_handler.name, True): return False # pick between synchronous candidates so we minimize unnecessary failovers/demotions - members = {m.name: m for m in all_known_members if self.cluster.sync.matches(m.name)} + members = {m.name: m for m in all_known_members if self.cluster.sync.matches(m.name, True)} else: # run usual health check members = {m.name: m for m in all_known_members} diff --git a/patroni/postgresql/sync.py b/patroni/postgresql/sync.py index 578224c6..550248f7 100644 --- a/patroni/postgresql/sync.py +++ b/patroni/postgresql/sync.py @@ -217,10 +217,9 @@ class SyncHandler(object): # Prefer members without nofailover tag. We are relying on the fact that sorts are guaranteed to be stable. for pid, app_name, sync_state, replica_lsn, _ in sorted(replica_list, key=lambda x: x[4]): # if standby name is listed in the /sync key we can count it as synchronous, otherwice - # ig becomes really synchronous when sync_state = 'sync' and it is known that it managed to catch up + # it becomes really synchronous when sync_state = 'sync' and it is known that it managed to catch up if app_name not in self._ready_replicas and app_name in self._ssn_data['members'] and\ - (cluster.sync and app_name in cluster.sync.members or - sync_state == 'sync' and replica_lsn >= self._primary_flush_lsn): + (cluster.sync.matches(app_name) or sync_state == 'sync' and replica_lsn >= self._primary_flush_lsn): self._ready_replicas[app_name] = pid if sync_node_maxlag <= 0 or max_lsn - replica_lsn <= sync_node_maxlag: diff --git a/patroni/utils.py b/patroni/utils.py index 086617d5..a52a69e2 100644 --- a/patroni/utils.py +++ b/patroni/utils.py @@ -412,7 +412,7 @@ def cluster_as_json(cluster): if m.name == leader_name: config = cluster.config.data if cluster.config and cluster.config.modify_index else {} role = 'standby_leader' if is_standby_cluster(config.get('standby_cluster')) else 'leader' - elif m.name in cluster.sync.members: + elif cluster.sync.matches(m.name): role = 'sync_standby' else: role = 'replica' diff --git a/tests/test_api.py b/tests/test_api.py index 3ca1eece..42b41d03 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -519,6 +519,7 @@ class TestRestApiHandler(unittest.TestCase): MockRestApiServer(RestApiHandler, request) cluster.leader.name = 'postgresql1' + cluster.sync.matches.return_value = False for cluster.is_synchronous_mode.return_value in (True, False): MockRestApiServer(RestApiHandler, request)