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

@@ -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:

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')):

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}

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:

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'

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)