gitops/patroni
1
0
Fork 0
mirror of https://github.com/outbackdingo/patroni.git synced 2026-08-26 15:40:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Make sure Cluster.sync is never empty (#2614)

Browse Source 
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.
This commit is contained in:
Alexander Kukushkin
2023-03-22 16:41:41 +01:00
committed by GitHub
parent 918674e7bb
commit a8b90f0cd6
7 changed files with 47 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
patroni/dcs/__init__.py
+30 -21
View File
@@ -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
patroni/dcs/consul.py
+1 -1
View File
@@ -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')
patroni/dcs/etcd.py
+1 -1
View File
@@ -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
patroni/dcs/raft.py
+1 -1
View File
@@ -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)
patroni/ha.py
+4 -4
View File
@@ -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
tests/test_ha.py
+3 -3
View File
@@ -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):
tests/test_slots.py
+7 -7
View File
@@ -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: