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