Repository: zalando/patroni
Pull request #2790 (2023-07-31 08:03:36)

Refactor replica_list
As suggested in
https://github.com/zalando/patroni/pull/2668/files#r1276115738, introduce a couple of classes that represent a single replica and collection of replicas.


Patch:
diff --git a/patroni/config.py b/patroni/config.py
index 66ffd891..facf16fd 100644
--- a/patroni/config.py
+++ b/patroni/config.py
@@ -329,9 +329,19 @@ class Config(object):
 
     @staticmethod
     def _process_postgresql_parameters(parameters: Dict[str, Any], is_local: bool = False) -> Dict[str, Any]:
-        return {name: value for name, value in (parameters or {}).items()
-                if name not in ConfigHandler.CMDLINE_OPTIONS
-                or not is_local and ConfigHandler.CMDLINE_OPTIONS[name][1](value)}
+        pg_params: Dict[str, Any] = {}
+
+        for name, value in (parameters or {}).items():
+            if name not in ConfigHandler.CMDLINE_OPTIONS:
+                pg_params[name] = value
+            elif not is_local:
+                if ConfigHandler.CMDLINE_OPTIONS[name][1](value):
+                    pg_params[name] = value
+                else:
+                    logging.warning("postgresql parameter %s=%s failed validation, defaulting to %s",
+                                    name, value, ConfigHandler.CMDLINE_OPTIONS[name][0])
+
+        return pg_params
 
     def _safe_copy_dynamic_configuration(self, dynamic_configuration: Dict[str, Any]) -> Dict[str, Any]:
         config = deepcopy(self.__DEFAULT_CONFIG)
diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py
index 0bfc4fae..e28da844 100644
--- a/patroni/dcs/etcd3.py
+++ b/patroni/dcs/etcd3.py
@@ -630,6 +630,16 @@ class PatroniEtcd3Client(Etcd3Client):
 
         return ret
 
+    def txn(self, compare: Dict[str, Any], success: Dict[str, Any],
+            failure: Optional[Dict[str, Any]] = None, retry: Optional[Retry] = None) -> Dict[str, Any]:
+        ret = super(PatroniEtcd3Client, self).txn(compare, success, failure, retry)
+        # Here we abuse the fact that the `failure` is only set in the call from update_leader().
+        # In all other cases the txn() call failure may be an indicator of a stale cache,
+        # and therefore we want to restart watcher.
+        if not failure and not ret:
+            self._restart_watcher()
+        return ret
+
 
 class Etcd3(AbstractEtcd):
 
diff --git a/patroni/ha.py b/patroni/ha.py
index 95ca6e05..2445aa02 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -996,9 +996,22 @@ class Ha(object):
                 return ret
 
         if self.state_handler.is_leader():
-            # in pause leader is the healthiest only when no initialize or sysid matches with initialize!
-            return not self.is_paused() or not self.cluster.initialize\
-                or self.state_handler.sysid == self.cluster.initialize
+            if self.is_paused():
+                # in pause leader is the healthiest only when no initialize or sysid matches with initialize!
+                return not self.cluster.initialize or self.state_handler.sysid == self.cluster.initialize
+
+            # We want to protect from the following scenario:
+            # 1. node1 is stressed so much that heart-beat isn't running regularly and the leader lock expires.
+            # 2. node2 promotes, gets heavy load and the situation described in 1 repeats.
+            # 3. Patroni on node1 comes back, notices that Postgres is running as primary but there is
+            #    no leader key and "happily" acquires the leader lock.
+            # That is, node1 discarded promotion of node2. To avoid it we want to detect timeline change.
+            my_timeline = self.state_handler.get_primary_timeline()
+            if my_timeline < self.cluster.timeline:
+                logger.warning('My timeline %s is behind last known cluster timeline %s',
+                               my_timeline, self.cluster.timeline)
+                return False
+            return True
 
         if self.is_paused():
             return False
diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py
index 9461714f..3a61a31d 100644
--- a/patroni/postgresql/config.py
+++ b/patroni/postgresql/config.py
@@ -15,7 +15,7 @@ from ..collections import CaseInsensitiveDict, CaseInsensitiveSet
 from ..dcs import Leader, Member, RemoteMember, slot_name_from_member_name
 from ..exceptions import PatroniFatalException
 from ..utils import compare_values, parse_bool, parse_int, split_host_port, uri, validate_directory, is_subpath
-from ..validator import IntValidator
+from ..validator import IntValidator, EnumValidator
 
 if TYPE_CHECKING:  # pragma: no cover
     from . import Postgresql
@@ -258,14 +258,14 @@ def _false_validator(value: Any) -> bool:
     return False
 
 
-def _wal_level_validator(value: Any) -> bool:
-    return str(value).lower() in ('hot_standby', 'replica', 'logical')
-
-
 def _bool_validator(value: Any) -> bool:
     return parse_bool(value) is not None
 
 
+def _bool_is_true_validator(value: Any) -> bool:
+    return parse_bool(value) is True
+
+
 class ConfigHandler(object):
 
     # List of parameters which must be always passed to postmaster as command line options
@@ -286,8 +286,8 @@ class ConfigHandler(object):
         'listen_addresses': (None, _false_validator, 90100),
         'port': (None, _false_validator, 90100),
         'cluster_name': (None, _false_validator, 90500),
-        'wal_level': ('hot_standby', _wal_level_validator, 90100),
-        'hot_standby': ('on', _false_validator, 90100),
+        'wal_level': ('hot_standby', EnumValidator(('hot_standby', 'replica', 'logical')), 90100),
+        'hot_standby': ('on', _bool_is_true_validator, 90100),
         'max_connections': (100, IntValidator(min=25), 90100),
         'max_wal_senders': (10, IntValidator(min=3), 90100),
         'wal_keep_segments': (8, IntValidator(min=1), 90100),
@@ -297,7 +297,7 @@ class ConfigHandler(object):
         'track_commit_timestamp': ('off', _bool_validator, 90500),
         'max_replication_slots': (10, IntValidator(min=4), 90400),
         'max_worker_processes': (8, IntValidator(min=2), 90400),
-        'wal_log_hints': ('on', _false_validator, 90400)
+        'wal_log_hints': ('on', _bool_is_true_validator, 90400)
     })
 
     _RECOVERY_PARAMETERS = CaseInsensitiveSet(recovery_parameters.keys())
diff --git a/patroni/postgresql/sync.py b/patroni/postgresql/sync.py
index d0f28586..2280a68c 100644
--- a/patroni/postgresql/sync.py
+++ b/patroni/postgresql/sync.py
@@ -153,6 +153,72 @@ def parse_sync_standby_names(value: str) -> _SSN:
     return _SSN(sync_type, has_star, num, members)
 
 
+class _Replica(NamedTuple):
+    """Class representing a single replica that is eligible to be synchronous.
+
+    Attributes are taken from ``pg_stat_replication`` view and respective ``Cluster.members``.
+
+    :ivar pid: PID of walsender process.
+    :ivar application_name: matches with the ``Member.name``.
+    :ivar sync_state: possible values are: ``async``, ``potential``, ``quorum``, and ``sync``.
+    :ivar lsn: ``write_lsn``, ``flush_lsn``, or ``replay_lsn``, depending on the value of ``synchronous_commit`` GUC.
+    :ivar nofailover: whether the corresponding member has ``nofailover`` tag set to ``True``.
+    """
+    pid: int
+    application_name: str
+    sync_state: str
+    lsn: int
+    nofailover: bool
+
+
+class _ReplicaList(List[_Replica]):
+    """A collection of :class:``_Replica`` objects.
+
+    Values are reverse ordered by ``_Replica.sync_state`` and ``_Replica.lsn``.
+    That is, first there will be replicas that have ``sync_state`` == ``sync``, even if they are not
+    the most up-to-date in term of write/flush/replay LSN. It helps to keep the result of chosing new
+    synchronous nodes consistent in case if a synchronous standby member is slowed down OR async node
+    is receiving changes faster than the sync member. Such cases would trigger sync standby member
+    swapping, but only if lag on this member is exceeding a threshold (``maximum_lag_on_syncnode``).
+
+    :ivar max_lsn: maximum value of ``_Replica.lsn`` among all values. In case if there is just one
+                   element in the list we take value of ``pg_current_wal_lsn()``.
+    """
+
+    def __init__(self, postgresql: 'Postgresql', cluster: Cluster) -> None:
+        """Create :class:``_ReplicaList`` object.
+
+        :param postgresql: reference to :class:``Postgresql`` object.
+        :param cluster: currently known cluster state from DCS.
+        """
+        super().__init__()
+
+        # We want to prioritize candidates based on `write_lsn``, ``flush_lsn``, or ``replay_lsn``.
+        # Which column exactly to pick depends on the values of ``synchronous_commit`` GUC.
+        sort_col = {
+            'remote_apply': 'replay',
+            'remote_write': 'write'
+        }.get(postgresql.synchronous_commit(), 'flush') + '_lsn'
+
+        members = CaseInsensitiveDict({m.name: m for m in cluster.members})
+        for row in postgresql.pg_stat_replication():
+            member = members.get(row['application_name'])
+
+            # We want to consider only rows from ``pg_stat_replication` that:
+            # 1. are known to be streaming (write/flush/replay LSN are not NULL).
+            # 2. can be mapped to a ``Member`` of the ``Cluster``:
+            #   a. ``Member`` doesn't have ``nosync`` tag set;
+            #   b. PostgreSQL on the member is known to be running and accepting client connections.
+            if member and row[sort_col] is not None and member.is_running and not member.tags.get('nosync', False):
+                self.append(_Replica(row['pid'], row['application_name'],
+                                     row['sync_state'], row[sort_col], bool(member.nofailover)))
+
+        # Prefer replicas that are in state ``sync`` and with higher values of ``write``/``flush``/``replay`` LSN.
+        self.sort(key=lambda r: (r.sync_state, r.lsn), reverse=True)
+
+        self.max_lsn = max(self, key=lambda x: x.lsn).lsn if len(self) > 1 else postgresql.last_operation()
+
+
 class SyncHandler(object):
     """Class responsible for working with the `synchronous_standby_names`.
 
@@ -201,6 +267,21 @@ BEGIN
 END;$$""")
         self._postgresql.reset_cluster_info_state(None)  # Reset internal cache to query fresh values
 
+    def _process_replica_readiness(self, cluster: Cluster, replica_list: _ReplicaList) -> None:
+        """Flags replicas as truly "synchronous" when they have caught up with ``_primary_flush_lsn``.
+
+        :param cluster: current cluster topology from DCS
+        :param replica_list: collection of replicas that we want to evaluate.
+        """
+        for replica in replica_list:
+            # if standby name is listed in the /sync key we can count it as synchronous, otherwise
+            # it becomes really synchronous when sync_state = 'sync' and it is known that it managed to catch up
+            if replica.application_name not in self._ready_replicas\
+                    and replica.application_name in self._ssn_data.members\
+                    and (cluster.sync.matches(replica.application_name)
+                         or replica.sync_state == 'sync' and replica.lsn >= self._primary_flush_lsn):
+                self._ready_replicas[replica.application_name] = replica.pid
+
     def current_state(self, cluster: Cluster) -> Tuple[CaseInsensitiveSet, CaseInsensitiveSet]:
         """Finds best candidates to be the synchronous standbys.
 
@@ -218,31 +299,8 @@ END;$$""")
         """
         self._handle_synchronous_standby_names_change()
 
-        # Pick candidates based on who has higher replay/remote_write/flush lsn.
-        sort_col = {
-            'remote_apply': 'replay',
-            'remote_write': 'write'
-        }.get(self._postgresql.synchronous_commit(), 'flush') + '_lsn'
-
-        pg_stat_replication = [(r['pid'], r['application_name'], r['sync_state'], r[sort_col])
-                               for r in self._postgresql.pg_stat_replication()
-                               if r[sort_col] is not None]
-
-        members = CaseInsensitiveDict({m.name: m for m in cluster.members})
-        replica_list: List[Tuple[int, str, str, int, bool]] = []
-        # pg_stat_replication.sync_state has 4 possible states - async, potential, quorum, sync.
-        # That is, alphabetically they are in the reversed order of priority.
-        # Since we are doing reversed sort on (sync_state, lsn) tuples, it helps to keep the result
-        # consistent in case if a synchronous standby member is slowed down OR async node receiving
-        # changes faster than the sync member (very rare but possible).
-        # Such cases would trigger sync standby member swapping, but only if lag on a sync node exceeding a threshold.
-        for pid, app_name, sync_state, replica_lsn in sorted(pg_stat_replication, key=lambda r: r[2:4], reverse=True):
-            member = members.get(app_name)
-            if member and member.is_running and not member.tags.get('nosync', False):
-                replica_list.append((pid, member.name, sync_state, replica_lsn, bool(member.nofailover)))
-
-        max_lsn = max(replica_list, key=lambda x: x[3])[3]\
-            if len(replica_list) > 1 else self._postgresql.last_operation()
+        replica_list = _ReplicaList(self._postgresql, cluster)
+        self._process_replica_readiness(cluster, replica_list)
 
         if TYPE_CHECKING:  # pragma: no cover
             assert self._postgresql.global_config is not None
@@ -253,17 +311,11 @@ END;$$""")
         candidates = CaseInsensitiveSet()
         sync_nodes = CaseInsensitiveSet()
         # 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
-            # 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.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:
-                candidates.add(app_name)
-                if sync_state == 'sync' and app_name in self._ready_replicas:
-                    sync_nodes.add(app_name)
+        for replica in sorted(replica_list, key=lambda x: x.nofailover):
+            if sync_node_maxlag <= 0 or replica_list.max_lsn - replica.lsn <= sync_node_maxlag:
+                candidates.add(replica.application_name)
+                if replica.sync_state == 'sync' and replica.application_name in self._ready_replicas:
+                    sync_nodes.add(replica.application_name)
             if len(candidates) >= sync_node_count:
                 break
 
diff --git a/patroni/validator.py b/patroni/validator.py
index b68f9654..a8ada1f5 100644
--- a/patroni/validator.py
+++ b/patroni/validator.py
@@ -785,7 +785,7 @@ class IntValidator(object):
         self.base_unit = base_unit
         self.raise_assert = raise_assert
 
-    def __call__(self, value: Union[int, str]) -> bool:
+    def __call__(self, value: Any) -> bool:
         """Check if *value* is a valid integer and within the expected range.
 
         .. note::
@@ -821,7 +821,7 @@ class EnumValidator(object):
         self.allowed_values = set(allowed_values) if case_sensitive else CaseInsensitiveSet(allowed_values)
         self.raise_assert = raise_assert
 
-    def __call__(self, value: str) -> bool:
+    def __call__(self, value: Any) -> bool:
         """Check if provided *value* could be found within *allowed_values*.
 
         .. note::
@@ -829,7 +829,7 @@ class EnumValidator(object):
         :param value: value to be checked.
         :returns: ``True`` if *value* could be found within *allowed_values*.
         """
-        ret = value in self.allowed_values
+        ret = isinstance(value, str) and value in self.allowed_values
 
         if self.raise_assert:
             assert_(ret)
diff --git a/postgres2.yml b/postgres2.yml
index c77e734a..581fa719 100644
--- a/postgres2.yml
+++ b/postgres2.yml
@@ -121,4 +121,4 @@ tags:
     nofailover: false
     noloadbalance: false
     clonefrom: false
-    replicatefrom: postgres1
+#    replicatefrom: postgresql1


Test patch:
diff --git a/tests/test_config.py b/tests/test_config.py
index 57a717f5..3f7e4049 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -21,7 +21,8 @@ class TestConfig(unittest.TestCase):
         with patch.object(Config, '_build_effective_configuration', Mock(side_effect=Exception)):
             self.assertFalse(self.config.set_dynamic_configuration({'foo': 'bar'}))
         self.assertTrue(self.config.set_dynamic_configuration({'standby_cluster': {}, 'postgresql': {
-            'parameters': {'cluster_name': 1, 'wal_keep_size': 1, 'track_commit_timestamp': 1, 'wal_level': 1}}}))
+            'parameters': {'cluster_name': 1, 'hot_standby': 1, 'wal_keep_size': 1,
+                           'track_commit_timestamp': 1, 'wal_level': 1}}}))
 
     def test_reload_local_configuration(self):
         os.environ.update({
diff --git a/tests/test_etcd3.py b/tests/test_etcd3.py
index ace59a62..2e3ed59d 100644
--- a/tests/test_etcd3.py
+++ b/tests/test_etcd3.py
@@ -127,10 +127,16 @@ class TestPatroniEtcd3Client(BaseTestEtcd3):
         request = {'key': base64_encode('/patroni/test/leader')}
         mock_urlopen.return_value = MockResponse()
         mock_urlopen.return_value.content = '{"succeeded":true,"header":{"revision":"1"}}'
-        self.client.call_rpc('/kv/txn', {'success': [{'request_delete_range': request}]})
         self.client.call_rpc('/kv/put', request)
         self.client.call_rpc('/kv/deleterange', request)
 
+    @patch.object(urllib3.PoolManager, 'urlopen')
+    def test_txn(self, mock_urlopen):
+        mock_urlopen.return_value = MockResponse()
+        mock_urlopen.return_value.content = '{"header":{"revision":"1"}}'
+        self.client.txn({'target': 'MOD', 'mod_revision': '1'},
+                        {'request_delete_range': {'key': base64_encode('/patroni/test/leader')}})
+
     @patch('time.time', Mock(side_effect=[1, 10.9, 100]))
     def test__wait_cache(self):
         with self.kv_cache.condition:
diff --git a/tests/test_ha.py b/tests/test_ha.py
index 1f5f903d..ae6be3e2 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -375,6 +375,12 @@ class TestHa(PostgresInit):
     def test_acquire_lock_as_primary(self):
         self.assertEqual(self.ha.run_cycle(), 'acquired session lock as a leader')
 
+    def test_leader_race_stale_primary(self):
+        with patch.object(Postgresql, 'get_primary_timeline', Mock(return_value=1)), \
+                patch('patroni.ha.logger.warning') as mock_logger:
+            self.assertEqual(self.ha.run_cycle(), 'demoting self because i am not the healthiest node')
+            self.assertEqual(mock_logger.call_args[0][0], 'My timeline %s is behind last known cluster timeline %s')
+
     def test_promoted_by_acquiring_lock(self):
         self.ha.is_healthiest_node = true
         self.p.is_leader = false