diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index e536e9dc..279ce265 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -1754,26 +1754,29 @@ class AbstractDCS(abc.ABC): """ @abc.abstractmethod - def _delete_leader(self) -> bool: + def _delete_leader(self, leader: Leader) -> bool: """Remove leader key from DCS. This method should remove leader key if current instance is the leader. + :param leader: :class:`Leader` object with information about the leader. + :returns: ``True`` if successfully committed to DCS. """ - def delete_leader(self, last_lsn: Optional[int] = None) -> bool: + def delete_leader(self, leader: Optional[Leader], last_lsn: Optional[int] = None) -> bool: """Update ``optime/leader`` and voluntarily remove leader key from DCS. This method should remove leader key if current instance is the leader. + :param leader: :class:`Leader` object with information about the leader. :param last_lsn: latest checkpoint location in bytes. :returns: boolean result of called abstract :meth:`~AbstractDCS._delete_leader`. """ if last_lsn: self.write_status({self._OPTIME: last_lsn}) - return self._delete_leader() + return bool(leader) and self._delete_leader(leader) @abc.abstractmethod def cancel_initialization(self) -> bool: diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index 7b310d15..5dd69c86 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -643,12 +643,8 @@ class Consul(AbstractDCS): return self._client.kv.put(self.history_path, value) @catch_consul_errors - def _delete_leader(self) -> bool: - cluster = self.cluster - if cluster and isinstance(cluster.leader, Leader) and\ - cluster.leader.name == self._name and isinstance(cluster.leader.version, int): - return self._client.kv.delete(self.leader_path, cas=cluster.leader.version) - return True + def _delete_leader(self, leader: Leader) -> bool: + return self._client.kv.delete(self.leader_path, cas=int(leader.version)) @catch_consul_errors def set_sync_state_value(self, value: str, version: Optional[int] = None) -> Union[int, bool]: diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index 447c7dab..335cf7a7 100644 --- a/patroni/dcs/etcd.py +++ b/patroni/dcs/etcd.py @@ -809,7 +809,7 @@ class Etcd(AbstractEtcd): return bool(self.retry(self._client.write, self.initialize_path, sysid, prevExist=(not create_new))) @catch_etcd_errors - def _delete_leader(self) -> bool: + def _delete_leader(self, leader: Leader) -> bool: return bool(self._client.delete(self.leader_path, prevValue=self._name)) @catch_etcd_errors diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py index e5e069c5..5b1acac3 100644 --- a/patroni/dcs/etcd3.py +++ b/patroni/dcs/etcd3.py @@ -912,11 +912,10 @@ class Etcd3(AbstractEtcd): return self.retry(self._client.put, self.initialize_path, sysid, create_revision='0' if create_new else None) @catch_etcd_errors - def _delete_leader(self) -> bool: - cluster = self.cluster - if cluster and isinstance(cluster.leader, Leader) and cluster.leader.name == self._name: - return self._client.deleterange(self.leader_path, mod_revision=cluster.leader.version) - return True + def _delete_leader(self, leader: Leader) -> bool: + fields = build_range_request(self.leader_path) + compare = {'key': fields['key'], 'target': 'VALUE', 'value': base64_encode(self._name)} + return bool(self._client.txn(compare, {'request_delete_range': fields})) @catch_etcd_errors def cancel_initialization(self) -> bool: diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py index 4a66df71..fec544b0 100644 --- a/patroni/dcs/kubernetes.py +++ b/patroni/dcs/kubernetes.py @@ -1308,11 +1308,11 @@ class Kubernetes(AbstractDCS): if cluster and cluster.config and cluster.config.version else None return self.patch_or_create_config({self._INITIALIZE: sysid}, resource_version) - def _delete_leader(self) -> bool: + def _delete_leader(self, leader: Leader) -> bool: """Unused""" raise NotImplementedError # pragma: no cover - def delete_leader(self, last_lsn: Optional[int] = None) -> bool: + def delete_leader(self, leader: Optional[Leader], last_lsn: Optional[int] = None) -> bool: ret = False kind = self._kinds.get(self.leader_path) if kind and (kind.metadata.annotations or {}).get(self._LEADER) == self._name: diff --git a/patroni/dcs/raft.py b/patroni/dcs/raft.py index c4286f1e..3f9337cb 100644 --- a/patroni/dcs/raft.py +++ b/patroni/dcs/raft.py @@ -446,7 +446,7 @@ class Raft(AbstractDCS): def initialize(self, create_new: bool = True, sysid: str = '') -> bool: return self._sync_obj.set(self.initialize_path, sysid, prevExist=(not create_new)) is not False - def _delete_leader(self) -> bool: + def _delete_leader(self, leader: Leader) -> bool: return self._sync_obj.delete(self.leader_path, prevValue=self._name, timeout=1) def cancel_initialization(self) -> bool: diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index ed2b71ed..29d159e6 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -466,7 +466,7 @@ class ZooKeeper(AbstractDCS): return False return True - def _delete_leader(self) -> bool: + def _delete_leader(self, leader: Leader) -> bool: self._client.restart() return True diff --git a/patroni/ha.py b/patroni/ha.py index 886dcc8e..0aa9b9d0 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -1078,7 +1078,7 @@ class Ha(object): def _delete_leader(self, last_lsn: Optional[int] = None) -> None: self.set_is_leader(False) - self.dcs.delete_leader(last_lsn) + self.dcs.delete_leader(self.cluster.leader, last_lsn) self.dcs.reset_cluster() def release_leader_key_voluntarily(self, last_lsn: Optional[int] = None) -> None: @@ -1874,7 +1874,7 @@ class Ha(object): # location, we can remove the leader key and allow them to start leader race. if self.is_failover_possible(cluster_lsn=checkpoint_location): - self.dcs.delete_leader(checkpoint_location) + self.dcs.delete_leader(self.cluster.leader, checkpoint_location) status['deleted'] = True else: self.dcs.write_leader_optime(checkpoint_location) @@ -1891,7 +1891,7 @@ class Ha(object): if not self.state_handler.is_running(): if self.is_leader() and not status['deleted']: checkpoint_location = self.state_handler.latest_checkpoint_location() - self.dcs.delete_leader(checkpoint_location) + self.dcs.delete_leader(self.cluster.leader, checkpoint_location) self.touch_member() else: # XXX: what about when Patroni is started as the wrong user that has access to the watchdog device diff --git a/tests/test_consul.py b/tests/test_consul.py index ebbbb90e..83ee67d8 100644 --- a/tests/test_consul.py +++ b/tests/test_consul.py @@ -197,9 +197,10 @@ class TestConsul(unittest.TestCase): @patch.object(consul.Consul.KV, 'delete', Mock(return_value=True)) def test_delete_leader(self): - self.c.delete_leader() + leader = self.c.get_cluster().leader + self.c.delete_leader(leader) self.c._name = 'other' - self.c.delete_leader() + self.c.delete_leader(leader) @patch.object(consul.Consul.KV, 'put', Mock(return_value=True)) def test_initialize(self): diff --git a/tests/test_etcd.py b/tests/test_etcd.py index f9e07144..90402b5f 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -313,7 +313,7 @@ class TestEtcd(unittest.TestCase): self.assertFalse(self.etcd.cancel_initialization()) def test_delete_leader(self): - self.assertFalse(self.etcd.delete_leader()) + self.assertFalse(self.etcd.delete_leader(self.etcd.get_cluster().leader)) def test_delete_cluster(self): self.assertFalse(self.etcd.delete_cluster()) diff --git a/tests/test_etcd3.py b/tests/test_etcd3.py index 2e3ed59d..9aed7eb1 100644 --- a/tests/test_etcd3.py +++ b/tests/test_etcd3.py @@ -298,9 +298,10 @@ class TestEtcd3(BaseTestEtcd3): self.etcd3.cancel_initialization() def test_delete_leader(self): - self.etcd3.delete_leader() + leader = self.etcd3.get_cluster().leader + self.etcd3.delete_leader(leader) self.etcd3._name = 'other' - self.etcd3.delete_leader() + self.etcd3.delete_leader(leader) def test_delete_cluster(self): self.etcd3.delete_cluster() @@ -312,7 +313,7 @@ class TestEtcd3(BaseTestEtcd3): self.etcd3.set_sync_state_value('', 1) def test_delete_sync_state(self): - self.etcd3.delete_sync_state() + self.etcd3.delete_sync_state('1') def test_watch(self): self.etcd3.set_ttl(10) diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py index 9eaa173a..694cd505 100644 --- a/tests/test_kubernetes.py +++ b/tests/test_kubernetes.py @@ -324,7 +324,7 @@ class TestKubernetesConfigMaps(BaseTestKubernetes): self.k.initialize() def test_delete_leader(self): - self.k.delete_leader(1) + self.k.delete_leader(self.k.get_cluster().leader, 1) def test_cancel_initialization(self): self.k.cancel_initialization() diff --git a/tests/test_raft.py b/tests/test_raft.py index 1fe8733c..9bb109e9 100644 --- a/tests/test_raft.py +++ b/tests/test_raft.py @@ -142,25 +142,25 @@ class TestRaft(unittest.TestCase): raft._citus_group = '1' self.assertTrue(raft.manual_failover('foo', 'bar')) raft._citus_group = '0' + self.assertTrue(raft.take_leader()) cluster = raft.get_cluster() self.assertIsInstance(cluster, Cluster) self.assertIsInstance(cluster.workers[1], Cluster) + leader = cluster.leader + self.assertTrue(raft.delete_leader(leader)) self.assertTrue(raft._sync_obj.set(raft.status_path, '{"optime":1234567,"slots":{"ls":12345}}')) - leader = raft.get_cluster().leader + raft.get_cluster() self.assertTrue(raft.update_leader(leader, '1', failsafe={'foo': 'bat'})) self.assertTrue(raft._sync_obj.set(raft.failsafe_path, '{"foo"}')) self.assertTrue(raft._sync_obj.set(raft.status_path, '{')) raft.get_citus_coordinator() self.assertTrue(raft.delete_sync_state()) - self.assertTrue(raft.delete_leader()) self.assertTrue(raft.set_history_value('')) self.assertTrue(raft.delete_cluster()) raft._citus_group = '1' self.assertTrue(raft.delete_cluster()) raft._citus_group = None raft.get_cluster() - self.assertTrue(raft.take_leader()) - raft.get_cluster() raft.watch(None, 0.001) raft._sync_obj.destroy() diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index c72fefe9..45aeac10 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -202,7 +202,7 @@ class TestZooKeeper(unittest.TestCase): mock_logger.assert_called_once() def test_delete_leader(self): - self.assertTrue(self.zk.delete_leader()) + self.assertTrue(self.zk.delete_leader(self.zk.get_cluster().leader)) def test_set_failover_value(self): self.zk.set_failover_value('')