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

Introduce quorum field in the /sync key

Browse Source 
Old keys remain the same for backward compatibility.
This commit is contained in:
Alexander Kukushkin
2023-05-11 11:15:07 +02:00
parent 7941c86775
commit 2223553fe5
10 changed files with 47 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
patroni/dcs/__init__.py
+24 -8
View File
@@ -393,10 +393,13 @@ class SyncState(NamedTuple):
:param index: modification index of a synchronization key in a Configuration Store
:param leader: reference to member that was leader
:param sync_standby: synchronous standby list (comma delimited) which are last synchronized to leader
:param quorum: if the node from sync_standby list is doing a leader race it should
see at least quorum other nodes from the sync_standby + leader list
"""
index: Optional[_Version]
leader: Optional[str]
sync_standby: Optional[str]
quorum: int
@staticmethod
def from_node(index: Optional[_Version], value: Union[str, Dict[str, Any], None]) -> 'SyncState':
@@ -418,13 +421,15 @@ class SyncState(NamedTuple):
if value and isinstance(value, str):
value = json.loads(value)
assert isinstance(value, dict)
return SyncState(index, value.get('leader'), value.get('sync_standby'))
leader = value.get('leader')
quorum = value.get('quorum')
return SyncState(index, leader, value.get('sync_standby'), int(quorum) if leader and quorum else 0)
except (AssertionError, TypeError, ValueError):
return SyncState.empty(index)
@staticmethod
def empty(index: Optional[_Version] = None) -> 'SyncState':
return SyncState(index, None, None)
return SyncState(index, None, None, 0)
@property
def is_empty(self) -> bool:
@@ -441,10 +446,15 @@ class SyncState(NamedTuple):
return list(filter(lambda a: a, [s.strip() for s in value.split(',')]))
@property
def members(self) -> List[str]:
def voters(self) -> List[str]:
""":returns: sync_standby as list."""
return self._str_to_list(self.sync_standby) if not self.is_empty and self.sync_standby else []
@property
def members(self) -> List[str]:
""":returns: leader and all voters as list"""
return [] if not self.leader else [self.leader] + self.voters
def matches(self, name: Optional[str], check_leader: bool = False) -> bool:
"""Checks if node is presented in the /sync state.
@@ -452,7 +462,7 @@ class SyncState(NamedTuple):
: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 = SyncState(1, 'foo', 'bar,zoo', 0)
>>> s.matches('foo')
False
>>> s.matches('fOo', True)
@@ -1077,25 +1087,31 @@ class AbstractDCS(abc.ABC):
"""Delete cluster from DCS"""
@staticmethod
def sync_state(leader: Optional[str], sync_standby: Optional[Collection[str]]) -> Dict[str, Any]:
def sync_state(leader: Optional[str], sync_standby: Optional[Collection[str]],
quorum: Optional[int]) -> Dict[str, Any]:
"""Build sync_state dict.
The sync_standby key being kept for backward compatibility.
:param leader: name of the leader node that manages /sync key
:param sync_standby: collection of currently known synchronous standby node names
:param quorum: if the node from sync_standby list is doing a leader race it should
see at least quorum other nodes from the sync_standby + leader list
:returns: dictionary that later could be serialized to JSON or saved directly to DCS
"""
return {'leader': leader, 'sync_standby': ','.join(sorted(sync_standby)) if sync_standby else None}
return {'leader': leader, 'quorum': quorum,
'sync_standby': ','.join(sorted(sync_standby)) if sync_standby else None}
def write_sync_state(self, leader: Optional[str], sync_standby: Optional[Collection[str]],
index: Optional[Any] = None) -> Optional[SyncState]:
quorum: Optional[int], index: Optional[Any] = None) -> Optional[SyncState]:
"""Write the new synchronous state to DCS.
Calls :func:`sync_state` method to build a dict and than calls DCS specific :func:`set_sync_state_value` method.
:param leader: name of the leader node that manages /sync key
:param sync_standby: collection of currently known synchronous standby node names
:param index: for conditional update of the key/object
:param quorum: if the node from sync_standby list is doing a leader race it should
see at least quorum other nodes from the sync_standby + leader list
:returns: the new :class:`SyncState` object or None
"""
sync_value = self.sync_state(leader, sync_standby)
sync_value = self.sync_state(leader, sync_standby, quorum)
ret = self.set_sync_state_value(json.dumps(sync_value, separators=(',', ':')), index)
if not isinstance(ret, bool):
return SyncState.from_node(ret, sync_value)
patroni/dcs/etcd3.py
+1 -1
View File
@@ -923,6 +923,6 @@ class Etcd3(AbstractEtcd):
return True
try:
return super(Etcd3, self).watch(None, timeout)
return super(Etcd3, self).watch(None, timeout + 0.5)
finally:
self.event.clear()
patroni/dcs/kubernetes.py
+6 -3
View File
@@ -1313,15 +1313,18 @@ class Kubernetes(AbstractDCS):
raise NotImplementedError # pragma: no cover
def write_sync_state(self, leader: Optional[str], sync_standby: Optional[Collection[str]],
index: Optional[str] = None) -> Optional[SyncState]:
quorum: Optional[int], index: Optional[str] = None) -> Optional[SyncState]:
"""Prepare and write annotations to $SCOPE-sync Endpoint or ConfigMap.
:param leader: name of the leader node that manages /sync key
:param sync_standby: collection of currently known synchronous standby node names
:param quorum: if the node from sync_standby list is doing a leader race it should
see at least quorum other nodes from the sync_standby + leader list
:param index: last known `resource_version` for conditional update of the object
:returns: the new :class:`SyncState` object or None
"""
sync_state = self.sync_state(leader, sync_standby)
sync_state = self.sync_state(leader, sync_standby, quorum)
sync_state['quorum'] = str(sync_state['quorum']) if sync_state['quorum'] is not None else None
ret = self.patch_or_create(self.sync_path, sync_state, index, False)
if not isinstance(ret, bool):
return SyncState.from_node(ret.metadata.resource_version, sync_state)
@@ -1333,7 +1336,7 @@ class Kubernetes(AbstractDCS):
:param index: last known `resource_version` for conditional update of the object
:returns: `True` if "delete" was successful
"""
return self.write_sync_state(None, None, index=index) is not None
return self.write_sync_state(None, None, None, index=index) is not None
def watch(self, leader_index: Optional[str], timeout: float) -> bool:
if self.__do_not_watch:
patroni/ha.py
+8 -8
View File
@@ -586,17 +586,17 @@ class Ha(object):
promoting standbys that were guaranteed to be replicating synchronously.
"""
if self.is_synchronous_mode():
current = CaseInsensitiveSet(self.cluster.sync.members)
picked, allow_promote = self.state_handler.sync_handler.current_state(self.cluster)
voters = CaseInsensitiveSet(self.cluster.sync.voters)
if picked != current:
if picked != voters:
sync = self.cluster.sync
# update synchronous standby list in dcs temporarily to point to common nodes in current and picked
sync_common = current & allow_promote
if sync_common != current:
sync_common = voters & allow_promote
if sync_common != voters:
logger.info("Updating synchronous privilege temporarily from %s to %s",
list(current), list(sync_common))
sync = self.dcs.write_sync_state(self.state_handler.name, sync_common, index=sync.index)
list(voters), list(sync_common))
sync = self.dcs.write_sync_state(self.state_handler.name, sync_common, 0, index=sync.index)
if not sync:
return logger.info('Synchronous replication key updated by someone else.')
@@ -614,7 +614,7 @@ class Ha(object):
time.sleep(2)
_, allow_promote = self.state_handler.sync_handler.current_state(self.cluster)
if allow_promote and allow_promote != sync_common:
if not self.dcs.write_sync_state(self.state_handler.name, allow_promote, index=sync.index):
if not self.dcs.write_sync_state(self.state_handler.name, allow_promote, 0, index=sync.index):
return logger.info("Synchronous replication key updated by someone else")
logger.info("Synchronous standby status assigned to %s", list(allow_promote))
else:
@@ -727,7 +727,7 @@ class Ha(object):
if self.is_synchronous_mode():
# Just set ourselves as the authoritative source of truth for now. We don't want to wait for standbys
# to connect. We will try finding a synchronous standby in the next cycle.
if not self.dcs.write_sync_state(self.state_handler.name, None, index=self.cluster.sync.index):
if not self.dcs.write_sync_state(self.state_handler.name, None, 0, index=self.cluster.sync.index):
# Somebody else updated sync state, it may be due to us losing the lock. To be safe, postpone
# promotion until next cycle. TODO: trigger immediate retry of run_cycle
return 'Postponing promotion because synchronous replication state was updated by somebody else'
tests/test_api.py
-2
View File
@@ -187,7 +187,6 @@ class TestRestApiHandler(unittest.TestCase):
def test_do_GET(self):
MockPatroni.dcs.cluster.last_lsn = 20
MockPatroni.dcs.cluster.sync.members = [MockPostgresql.name]
with patch.object(GlobalConfig, 'is_synchronous_mode', PropertyMock(return_value=True)):
MockRestApiServer(RestApiHandler, 'GET /replica')
MockRestApiServer(RestApiHandler, 'GET /replica?lag=1M')
@@ -206,7 +205,6 @@ class TestRestApiHandler(unittest.TestCase):
MockRestApiServer(RestApiHandler, 'GET /synchronous')
MockRestApiServer(RestApiHandler, 'GET /read-only-sync')
with patch.object(RestApiHandler, 'get_postgresql_status', Mock(return_value={'role': 'replica'})):
MockPatroni.dcs.cluster.sync.members = []
MockRestApiServer(RestApiHandler, 'GET /asynchronous')
with patch.object(MockHa, 'is_leader', Mock(return_value=True)):
MockRestApiServer(RestApiHandler, 'GET /replica')
tests/test_etcd.py
+1 -1
View File
@@ -338,7 +338,7 @@ class TestEtcd(unittest.TestCase):
self.assertTrue(self.etcd.watch(None, 1))
def test_sync_state(self):
self.assertIsNone(self.etcd.write_sync_state('leader', None))
self.assertIsNone(self.etcd.write_sync_state('leader', None, 0))
self.assertFalse(self.etcd.delete_sync_state())
def test_set_history_value(self):
tests/test_ha.py
+3 -3
View File
@@ -62,7 +62,7 @@ def get_cluster_initialized_without_leader(leader=False, failover=None, sync=Non
'tags': {'clonefrom': True},
'scheduled_restart': {'schedule': "2100-01-01 10:53:07.560445+00:00",
'postgres_version': '99.0.0'}})
syncstate = SyncState(0 if sync else None, sync and sync[0], sync and sync[1])
syncstate = SyncState(0 if sync else None, sync and sync[0], sync and sync[1], 0)
failsafe = {m.name: m.api_url for m in (m1, m2)} if failsafe else None
return get_cluster(SYSID, leader, [m1, m2], failover, syncstate, cluster_config, failsafe)
@@ -1201,7 +1201,7 @@ class TestHa(PostgresInit):
# When we just became primary nobody is sync
self.assertEqual(self.ha.enforce_primary_role('msg', 'promote msg'), 'promote msg')
mock_set_sync.assert_called_once_with(CaseInsensitiveSet())
mock_write_sync.assert_called_once_with('leader', None, index=0)
mock_write_sync.assert_called_once_with('leader', None, 0, index=0)
mock_set_sync.reset_mock()
@@ -1239,7 +1239,7 @@ class TestHa(PostgresInit):
mock_acquire.assert_called_once()
mock_follow.assert_not_called()
mock_promote.assert_called_once()
mock_write_sync.assert_called_once_with('other', None, index=0)
mock_write_sync.assert_called_once_with('other', None, 0, index=0)
def test_disable_sync_when_restarting(self):
self.ha.is_synchronous_mode = true
tests/test_kubernetes.py
+1 -1
View File
@@ -378,7 +378,7 @@ class TestKubernetesEndpoints(BaseTestKubernetes):
@patch.object(k8s_client.CoreV1Api, 'patch_namespaced_endpoints', mock_namespaced_kind, create=True)
def test_write_sync_state(self):
self.assertIsNotNone(self.k.write_sync_state('a', ['b'], 1))
self.assertIsNotNone(self.k.write_sync_state('a', ['b'], 0, 1))
@patch.object(k8s_client.CoreV1Api, 'patch_namespaced_pod', mock_namespaced_kind, create=True)
@patch.object(k8s_client.CoreV1Api, 'create_namespaced_endpoints', mock_namespaced_kind, create=True)
tests/test_raft.py
+2 -2
View File
@@ -137,8 +137,8 @@ class TestRaft(unittest.TestCase):
self.assertTrue(raft.initialize())
self.assertTrue(raft.cancel_initialization())
self.assertTrue(raft.set_config_value('{}'))
self.assertTrue(raft.write_sync_state('foo', 'bar'))
self.assertFalse(raft.write_sync_state('foo', 'bar', 1))
self.assertTrue(raft.write_sync_state('foo', 'bar', 0))
self.assertFalse(raft.write_sync_state('foo', 'bar', 0, 1))
raft._citus_group = '1'
self.assertTrue(raft.manual_failover('foo', 'bar'))
raft._citus_group = '0'
tests/test_sync.py
+1 -1
View File
@@ -28,7 +28,7 @@ class TestSync(BaseTestPostgresql):
@patch.object(Postgresql, 'last_operation', Mock(return_value=1))
def test_pick_sync_standby(self):
cluster = Cluster(True, None, self.leader, 0, [self.me, self.other, self.leadermem], None,
SyncState(0, self.me.name, self.leadermem.name), None, None, None)
SyncState(0, self.me.name, self.leadermem.name, 0), None, None, None)
pg_stat_replication = [
{'pid': 100, 'application_name': self.leadermem.name, 'sync_state': 'sync', 'flush_lsn': 1},