From 2223553fe5fe4106f2b64197a50d27e3dc40f8e2 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Thu, 11 May 2023 11:15:07 +0200 Subject: [PATCH] Introduce quorum field in the /sync key Old keys remain the same for backward compatibility. --- patroni/dcs/__init__.py | 32 ++++++++++++++++++++++++-------- patroni/dcs/etcd3.py | 2 +- patroni/dcs/kubernetes.py | 9 ++++++--- patroni/ha.py | 16 ++++++++-------- tests/test_api.py | 2 -- tests/test_etcd.py | 2 +- tests/test_ha.py | 6 +++--- tests/test_kubernetes.py | 2 +- tests/test_raft.py | 4 ++-- tests/test_sync.py | 2 +- 10 files changed, 47 insertions(+), 30 deletions(-) diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index f1429c6f..3936beaf 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -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) diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py index bdf5cc33..c7a5973a 100644 --- a/patroni/dcs/etcd3.py +++ b/patroni/dcs/etcd3.py @@ -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() diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index e33a8509..0bb37c03 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -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: diff --git a/patroni/ha.py b/patroni/ha.py index e1559a27..611d945f 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -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' diff --git a/tests/test_api.py b/tests/test_api.py index 85c01b49..2e947bb3 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -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') diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 6699127a..1b0c4c50 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -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): diff --git a/tests/test_ha.py b/tests/test_ha.py index c9734517..db11706d 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -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 diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index bbe55157..61f858d6 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -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) diff --git a/tests/test_raft.py b/tests/test_raft.py index 2029a61e..bf033e37 100644 --- a/tests/test_raft.py +++ b/tests/test_raft.py @@ -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' diff --git a/tests/test_sync.py b/tests/test_sync.py index 0cfba7f8..a93b98f3 100644 --- a/tests/test_sync.py +++ b/tests/test_sync.py @@ -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},