Repository: zalando/patroni
Pull request #2614 (2023-03-22 10:40:52)

Make sure Cluster.sync is never empty
It was possible to have it empty if the all cluster keys are missing in DCS. In this case the `Cluster` object was manually created with all values set to `None` or `[]` (including sync).
It already resulted in #2217, which is in fact wasn't a correct fix.

In order to solve it and reduce code duplication we introduce `Cluster.empty()` and `SyncState.empty()` methods, which will create corresponding empty objects and start using `Cluster.empty()` from all places where the empty `Cluster` object was manually created.


Patch:
diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py
index 09d4be66..4ae26150 100644
--- a/patroni/dcs/__init__.py
+++ b/patroni/dcs/__init__.py
@@ -14,6 +14,7 @@ from collections import defaultdict, namedtuple
 from copy import deepcopy
 from random import randint
 from threading import Event, Lock
+from typing import Any, Dict, List, Optional, Union
 from urllib.parse import urlparse, urlunparse, parse_qsl
 
 from ..exceptions import PatroniFatalException
@@ -374,7 +375,7 @@ class SyncState(namedtuple('SyncState', 'index,leader,sync_standby')):
     """
 
     @staticmethod
-    def from_node(index, value):
+    def from_node(index: Union[str, int], value: Union[str, Dict[str, Any]]) -> 'SyncState':
         """
         >>> SyncState.from_node(1, None).leader is None
         True
@@ -389,27 +390,31 @@ class SyncState(namedtuple('SyncState', 'index,leader,sync_standby')):
         >>> SyncState.from_node(1, {"leader": "leader"}).leader == "leader"
         True
         """
-        if isinstance(value, dict):
-            data = value
-        elif value:
-            try:
-                data = json.loads(value)
-                if not isinstance(data, dict):
-                    data = {}
-            except (TypeError, ValueError):
-                data = {}
-        else:
-            data = {}
-        return SyncState(index, data.get('leader'), data.get('sync_standby'))
+        try:
+            if value and isinstance(value, str):
+                value = json.loads(value)
+            if not isinstance(value, dict):
+                return SyncState.empty(index)
+            return SyncState(index, value.get('leader'), value.get('sync_standby'))
+        except (TypeError, ValueError):
+            return SyncState.empty(index)
+
+    @staticmethod
+    def empty(index: Optional[Union[str, int]] = '') -> 'SyncState':
+        return SyncState(index, None, '')
 
     @property
-    def members(self):
-        """ Returns sync_standby in list """
-        return self.sync_standby and self.sync_standby.split(',') or []
+    def is_empty(self) -> bool:
+        """:returns: True if /sync key doesn't have a leader"""
+        return self.leader is None
 
-    def matches(self, name):
-        """
-        Returns if a node name matches one of the nodes in the sync state
+    @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 []
+
+    def matches(self, name: str) -> bool:
+        """:returns: True if a node name matches one of the nodes in the sync state (including leader)
 
         >>> s = SyncState(1, 'foo', 'bar,zoo')
         >>> s.matches('foo')
@@ -471,6 +476,10 @@ class Cluster(namedtuple('Cluster', 'initialize,config,leader,last_lsn,members,'
             args = args + ({},)
         return super(Cluster, cls).__new__(cls, *args)
 
+    @staticmethod
+    def empty():
+        return Cluster(None, None, None, 0, [], None, SyncState.empty(), None, None, None)
+
     @property
     def leader_name(self):
         return self.leader and self.leader.name
@@ -812,8 +821,8 @@ class AbstractDCS(abc.ABC):
         if isinstance(groups, Cluster):  # Zookeeper could return a cached version
             cluster = groups
         else:
-            cluster = groups.pop(CITUS_COORDINATOR_GROUP_ID,
-                                 Cluster(None, None, None, None, [], None, None, None, None, None))
+            assert isinstance(groups, dict)
+            cluster = groups.pop(CITUS_COORDINATOR_GROUP_ID, Cluster.empty())
             cluster.workers.update(groups)
         return cluster
 
diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py
index 5404c961..5c988833 100644
--- a/patroni/dcs/consul.py
+++ b/patroni/dcs/consul.py
@@ -409,7 +409,7 @@ class Consul(AbstractDCS):
         try:
             return loader(path)
         except NotFound:
-            return Cluster(None, None, None, None, [], None, None, None, None, None)
+            return Cluster.empty()
         except Exception:
             logger.exception('get_cluster')
             raise ConsulError('Consul is not responding properly')
diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py
index ca862df9..154a7de0 100644
--- a/patroni/dcs/etcd.py
+++ b/patroni/dcs/etcd.py
@@ -704,7 +704,7 @@ class Etcd(AbstractEtcd):
         try:
             cluster = loader(path)
         except etcd.EtcdKeyNotFound:
-            cluster = Cluster(None, None, None, None, [], None, None, None, None, None)
+            cluster = Cluster.empty()
         except Exception as e:
             self._handle_exception(e, 'get_cluster', raise_ex=EtcdError('Etcd is not responding properly'))
         self._has_failed = False
diff --git a/patroni/dcs/raft.py b/patroni/dcs/raft.py
index 0d0e4568..e09b3290 100644
--- a/patroni/dcs/raft.py
+++ b/patroni/dcs/raft.py
@@ -383,7 +383,7 @@ class Raft(AbstractDCS):
     def _cluster_loader(self, path):
         response = self._sync_obj.get(path, recursive=True)
         if not response:
-            return Cluster(None, None, None, None, [], None, None, None, None, None)
+            return Cluster.empty()
         nodes = {key[len(path):]: value for key, value in response.items()}
         return self._cluster_from_nodes(nodes)
 
diff --git a/patroni/ha.py b/patroni/ha.py
index 6eda0a3e..53f6d419 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -590,7 +590,7 @@ class Ha(object):
         """
         if self.is_synchronous_mode():
             sync_node_count = self.patroni.config['synchronous_node_count']
-            current = self.cluster.sync.leader and self.cluster.sync.members or []
+            current = [] if self.cluster.sync.is_empty else self.cluster.sync.members
             picked, allow_promote = self.state_handler.sync_handler.current_state(self.cluster, sync_node_count,
                                                                                   self.patroni.config[
                                                                                       'maximum_lag_on_syncnode'])
@@ -625,7 +625,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 cluster.sync.leader and cluster.sync.leader != self.state_handler.name:
+                    if not cluster.sync.is_empty and cluster.sync.leader != 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):
@@ -633,7 +633,7 @@ class Ha(object):
                         return
                     logger.info("Synchronous standby status assigned to %s", allow_promote)
         else:
-            if self.cluster.sync.leader and self.dcs.delete_sync_state(index=self.cluster.sync.index):
+            if not self.cluster.sync.is_empty and self.dcs.delete_sync_state(index=self.cluster.sync.index):
                 logger.info("Disabled synchronous replication")
             self.state_handler.sync_handler.set_synchronous_standby_names([])
 
@@ -994,7 +994,7 @@ class Ha(object):
         all_known_members += self.cluster.members
 
         # When in sync mode, only last known primary and sync standby are allowed to promote automatically.
-        if self.is_synchronous_mode() and self.cluster.sync and self.cluster.sync.leader:
+        if self.is_synchronous_mode() and not self.cluster.sync.is_empty:
             if not self.cluster.sync.matches(self.state_handler.name):
                 return False
             # pick between synchronous candidates so we minimize unnecessary failovers/demotions
diff --git a/patroni/version.py b/patroni/version.py
index b7a55314..7962f44c 100644
--- a/patroni/version.py
+++ b/patroni/version.py
@@ -1,1 +1,5 @@
+"""This module specifies the current Patroni version.
+
+:var __version__: the current Patroni version.
+"""
 __version__ = '3.0.1'


Test patch:
diff --git a/tests/test_ha.py b/tests/test_ha.py
index d30689bf..70e465c6 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -42,11 +42,11 @@ def get_cluster(initialize, leader, members, failover, sync, cluster_config=None
 
 
 def get_cluster_not_initialized_without_leader(cluster_config=None):
-    return get_cluster(None, None, [], None, SyncState(None, None, None), cluster_config)
+    return get_cluster(None, None, [], None, SyncState.empty(), cluster_config)
 
 
 def get_cluster_bootstrapping_without_leader(cluster_config=None):
-    return get_cluster("", None, [], None, SyncState(None, None, None), cluster_config)
+    return get_cluster("", None, [], None, SyncState.empty(), cluster_config)
 
 
 def get_cluster_initialized_without_leader(leader=False, failover=None, sync=None, cluster_config=None, failsafe=False):
@@ -72,7 +72,7 @@ def get_cluster_initialized_with_leader(failover=None, sync=None):
 
 def get_cluster_initialized_with_only_leader(failover=None, cluster_config=None):
     leader = get_cluster_initialized_without_leader(leader=True, failover=failover).leader
-    return get_cluster(True, leader, [leader.member], failover, None, cluster_config)
+    return get_cluster(True, leader, [leader.member], failover, SyncState.empty(), cluster_config)
 
 
 def get_standby_cluster_initialized_with_only_leader(failover=None, sync=None):
diff --git a/tests/test_slots.py b/tests/test_slots.py
index 95e55851..f88cd881 100644
--- a/tests/test_slots.py
+++ b/tests/test_slots.py
@@ -7,7 +7,7 @@ from mock import Mock, PropertyMock, patch
 from threading import Thread
 
 from patroni import psycopg
-from patroni.dcs import Cluster, ClusterConfig, Member
+from patroni.dcs import Cluster, ClusterConfig, Member, SyncState
 from patroni.postgresql import Postgresql
 from patroni.postgresql.misc import fsync_dir
 from patroni.postgresql.slots import SlotsAdvanceThread, SlotsHandler
@@ -31,15 +31,15 @@ class TestSlotsHandler(BaseTestPostgresql):
         self.s = self.p.slots_handler
         self.p.start()
         config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1)
-        self.cluster = Cluster(True, config, self.leader, 0,
-                               [self.me, self.other, self.leadermem], None, None, None, {'ls': 12345}, None)
+        self.cluster = Cluster(True, config, self.leader, 0, [self.me, self.other, self.leadermem],
+                               None, SyncState.empty(), None, {'ls': 12345}, None)
 
     def test_sync_replication_slots(self):
         config = ClusterConfig(1, {'slots': {'test_3': {'database': 'a', 'plugin': 'b'},
                                              'A': 0, 'ls': 0, 'b': {'type': 'logical', 'plugin': '1'}},
                                    'ignore_slots': [{'name': 'blabla'}]}, 1)
-        cluster = Cluster(True, config, self.leader, 0,
-                          [self.me, self.other, self.leadermem], None, None, None, {'test_3': 10}, None)
+        cluster = Cluster(True, config, self.leader, 0, [self.me, self.other, self.leadermem],
+                          None, SyncState.empty(), None, {'test_3': 10}, None)
         with mock.patch('patroni.postgresql.Postgresql._query', Mock(side_effect=psycopg.OperationalError)):
             self.s.sync_replication_slots(cluster, False)
         self.p.set_role('standby_leader')
@@ -70,8 +70,8 @@ class TestSlotsHandler(BaseTestPostgresql):
     def test_process_permanent_slots(self):
         config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}},
                                    'ignore_slots': [{'name': 'blabla'}]}, 1)
-        cluster = Cluster(True, config, self.leader, 0,
-                          [self.me, self.other, self.leadermem], None, None, None, None, None)
+        cluster = Cluster(True, config, self.leader, 0, [self.me, self.other, self.leadermem],
+                          None, SyncState.empty(), None, None, None)
 
         self.s.sync_replication_slots(cluster, False)
         with patch.object(Postgresql, '_query') as mock_query: