Refactor delete_leader interface (#2810)

similar to https://github.com/zalando/patroni/pull/2690, but it helps mostly Consul implementation.

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:

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

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

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:

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:

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:

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
 

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

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

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())

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)

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()

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()
 

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('')