Implement delete_cluster interface in for all available dcs

In addition to that rename confusing `Etcd.client` and
`ZooKeeper.client` into `_client`. This attribute is available from
AbstractDCS and people had wrong impression that it provides the same
interface for different DCS implementations, which is obviously not the
case. For Etcd it has type etcd.Client and for ZooKeeper - KazooClient.

patroni/ctl.py

@@ -363,9 +363,6 @@ def query_member(cluster, cursor, member, role, command, connect_parameters=None
 def remove(config_file, cluster_name, fmt, dcs):
     config, dcs, cluster = ctl_load_config(cluster_name, config_file, dcs)
 
-    if not isinstance(dcs, Etcd):
-        raise PatroniCtlException('We have not implemented this for DCS of type {0}'.format(type(dcs)))
-
     output_members(cluster, fmt=fmt)
 
     confirm = click.prompt('Please confirm the cluster name to remove', type=str)
@@ -384,7 +381,7 @@ def remove(config_file, cluster_name, fmt, dcs):
         if confirm != cluster.leader.name:
             raise PatroniCtlException('You did not specify the current master of the cluster')
 
-    dcs.client.delete(dcs.client_path(''), recursive=True)
+    dcs.delete_cluster()
 
 
 def wait_for_leader(dcs, timeout=30):

patroni/dcs.py

@@ -308,6 +308,10 @@ class AbstractDCS(object):
     def cancel_initialization(self):
         """ Removes the initialize key for a cluster """
 
+    @abc.abstractmethod
+    def delete_cluster(self):
+        """Delete cluster from DCS"""
+
     def watch(self, timeout):
         """If the current node is a master it should just sleep.
         Any other node should watch for changes of leader key with a given timeout

patroni/etcd.py

@@ -199,7 +199,7 @@ class Etcd(AbstractDCS):
                                               etcd.EtcdLeaderElectionInProgress,
                                               etcd.EtcdWatcherCleared,
                                               etcd.EtcdEventIndexCleared))
-        self.client = self.get_etcd_client(config)
+        self._client = self.get_etcd_client(config)
 
     def retry(self, *args, **kwargs):
         return self._retry.copy()(*args, **kwargs)
@@ -221,7 +221,7 @@ class Etcd(AbstractDCS):
 
     def _load_cluster(self):
         try:
-            result = self.retry(self.client.read, self.client_path(''), recursive=True)
+            result = self.retry(self._client.read, self.client_path(''), recursive=True)
             nodes = {os.path.relpath(node.key, result.key): node for node in result.leaves}
 
             # get initialize flag
@@ -256,15 +256,15 @@ class Etcd(AbstractDCS):
 
     @catch_etcd_errors
     def touch_member(self, connection_string, ttl=None):
-        return self.retry(self.client.set, self.member_path, connection_string, ttl or self.ttl)
+        return self.retry(self._client.set, self.member_path, connection_string, ttl or self.ttl)
 
     @catch_etcd_errors
     def take_leader(self):
-        return self.retry(self.client.set, self.leader_path, self._name, self.ttl)
+        return self.retry(self._client.set, self.leader_path, self._name, self.ttl)
 
     def attempt_to_acquire_leader(self):
         try:
-            return bool(self.retry(self.client.write, self.leader_path, self._name, ttl=self.ttl, prevExist=False))
+            return bool(self.retry(self._client.write, self.leader_path, self._name, ttl=self.ttl, prevExist=False))
         except etcd.EtcdAlreadyExist:
             logger.info('Could not take out TTL lock')
         except (RetryFailedError, etcd.EtcdException):
@@ -273,27 +273,31 @@ class Etcd(AbstractDCS):
 
     @catch_etcd_errors
     def set_failover_value(self, value, index=None):
-        return self.client.write(self.failover_path, value, prevIndex=index or 0)
+        return self._client.write(self.failover_path, value, prevIndex=index or 0)
 
     @catch_etcd_errors
     def write_leader_optime(self, last_operation):
-        return self.client.set(self.leader_optime_path, last_operation)
+        return self._client.set(self.leader_optime_path, last_operation)
 
     @catch_etcd_errors
     def update_leader(self):
-        return self.retry(self.client.test_and_set, self.leader_path, self._name, self._name, self.ttl)
+        return self.retry(self._client.test_and_set, self.leader_path, self._name, self._name, self.ttl)
 
     @catch_etcd_errors
     def initialize(self, create_new=True, sysid=""):
-        return self.retry(self.client.write, self.initialize_path, sysid, prevExist=(not create_new))
+        return self.retry(self._client.write, self.initialize_path, sysid, prevExist=(not create_new))
 
     @catch_etcd_errors
     def delete_leader(self):
-        return self.client.delete(self.leader_path, prevValue=self._name)
+        return self._client.delete(self.leader_path, prevValue=self._name)
 
     @catch_etcd_errors
     def cancel_initialization(self):
-        return self.retry(self.client.delete, self.initialize_path)
+        return self.retry(self._client.delete, self.initialize_path)
+
+    @catch_etcd_errors
+    def delete_cluster(self):
+        return self.retry(self._client.delete, self.client_path(''), recursive=True)
 
     def watch(self, timeout):
         cluster = self.cluster
@@ -304,12 +308,12 @@ class Etcd(AbstractDCS):
 
             while index and timeout >= 1:  # when timeout is too small urllib3 doesn't have enough time to connect
                 try:
-                    self.client.watch(self.leader_path, index=index + 1, timeout=timeout + 0.5)
+                    self._client.watch(self.leader_path, index=index + 1, timeout=timeout + 0.5)
                     # Synchronous work of all cluster members with etcd is less expensive
                     # than reestablishing http connection every time from every replica.
                     return True
                 except etcd.EtcdWatchTimedOut:
-                    self.client.http.clear()
+                    self._client.http.clear()
                     return False
                 except etcd.EtcdException:
                     logging.exception('watch')

patroni/zookeeper.py

@@ -83,20 +83,20 @@ class ZooKeeper(AbstractDCS):
             self.exhibitor = ExhibitorEnsembleProvider(exhibitor['hosts'], exhibitor['port'], poll_interval=interval)
             hosts = self.exhibitor.zookeeper_hosts
 
-        self.client = KazooClient(hosts=hosts,
-                                  timeout=(config.get('session_timeout') or 30),
-                                  command_retry={
-                                      'deadline': (config.get('reconnect_timeout') or 10),
-                                      'max_delay': 1,
-                                      'max_tries': -1},
-                                  connection_retry={'max_delay': 1, 'max_tries': -1})
-        self.client.add_listener(self.session_listener)
+        self._client = KazooClient(hosts=hosts,
+                                   timeout=(config.get('session_timeout') or 30),
+                                   command_retry={
+                                        'deadline': (config.get('reconnect_timeout') or 10),
+                                        'max_delay': 1,
+                                        'max_tries': -1},
+                                   connection_retry={'max_delay': 1, 'max_tries': -1})
+        self._client.add_listener(self.session_listener)
 
         self._my_member_data = None
         self.fetch_cluster = True
         self.last_leader_operation = 0
 
-        self.client.start(None)
+        self._client.start(None)
 
     def session_listener(self, state):
         if state in [KazooState.SUSPENDED, KazooState.LOST]:
@@ -108,7 +108,7 @@ class ZooKeeper(AbstractDCS):
 
     def get_node(self, key, watch=None):
         try:
-            ret = self.client.get(key, watch)
+            ret = self._client.get(key, watch)
             return (ret[0].decode('utf-8'), ret[1])
         except NoNodeError:
             return None
@@ -119,7 +119,7 @@ class ZooKeeper(AbstractDCS):
 
     def get_children(self, key, watch=None):
         try:
-            return self.client.get_children(key, watch)
+            return self._client.get_children(key, watch)
         except NoNodeError:
             return []
 
@@ -147,10 +147,10 @@ class ZooKeeper(AbstractDCS):
         # get leader
         leader = self.get_node(self.leader_path) if self._LEADER in nodes else None
         if leader:
-            client_id = self.client.client_id
+            client_id = self._client.client_id
             if leader[0] == self._name and client_id is not None and client_id[0] != leader[1].ephemeralOwner:
                 logger.info('I am leader but not owner of the session. Removing leader node')
-                self.client.delete(self.leader_path)
+                self._client.delete(self.leader_path)
                 leader = None
 
             if leader:
@@ -171,11 +171,11 @@ class ZooKeeper(AbstractDCS):
 
     def _load_cluster(self):
         if self.exhibitor and self.exhibitor.poll():
-            self.client.set_hosts(self.exhibitor.zookeeper_hosts)
+            self._client.set_hosts(self.exhibitor.zookeeper_hosts)
 
         if self.fetch_cluster:
             try:
-                self.client.retry(self._inner_load_cluster)
+                self._client.retry(self._inner_load_cluster)
             except:
                 logger.exception('get_cluster')
                 self.session_listener(KazooState.LOST)
@@ -183,7 +183,7 @@ class ZooKeeper(AbstractDCS):
 
     def _create(self, path, value, **kwargs):
         try:
-            self.client.retry(self.client.create, path, value.encode('utf-8'), **kwargs)
+            self._client.retry(self._client.create, path, value.encode('utf-8'), **kwargs)
             return True
         except:
             return False
@@ -196,7 +196,7 @@ class ZooKeeper(AbstractDCS):
 
     def set_failover_value(self, value, index=None):
         try:
-            self.client.retry(self.client.set, self.failover_path, value.encode('utf-8'), version=index or -1)
+            self._client.retry(self._client.set, self.failover_path, value.encode('utf-8'), version=index or -1)
             return True
         except NoNodeError:
             return value == '' or (not index and self._create(self.failover_path, value))
@@ -206,7 +206,7 @@ class ZooKeeper(AbstractDCS):
 
     def initialize(self, create_new=True, sysid=""):
         return self._create(self.initialize_path, sysid, makepath=True) if create_new \
-            else self.client.retry(self.client.set, self.initialize_path,  sysid.encode("utf-8"))
+            else self._client.retry(self._client.set, self.initialize_path,  sysid.encode("utf-8"))
 
     def touch_member(self, data, ttl=None):
         cluster = self.cluster
@@ -214,9 +214,9 @@ class ZooKeeper(AbstractDCS):
         path = self.member_path
         data = data.encode('utf-8')
         create = not me
-        if me and self.client.client_id is not None and me.session != self.client.client_id[0]:
+        if me and self._client.client_id is not None and me.session != self._client.client_id[0]:
             try:
-                self.client.retry(self.client.delete, path)
+                self._client.retry(self._client.delete, path)
             except NoNodeError:
                 pass
             except:
@@ -228,14 +228,14 @@ class ZooKeeper(AbstractDCS):
 
         try:
             if create:
-                self.client.retry(self.client.create, path, data, makepath=True, ephemeral=True)
+                self._client.retry(self._client.create, path, data, makepath=True, ephemeral=True)
             else:
-                self.client.retry(self.client.set, path, data)
+                self._client.retry(self._client.set, path, data)
             self._my_member_data = data
             return True
         except NodeExistsError:
             try:
-                self.client.retry(self.client.set, path, data)
+                self._client.retry(self._client.set, path, data)
                 self._my_member_data = data
                 return True
             except:
@@ -253,10 +253,10 @@ class ZooKeeper(AbstractDCS):
             self.last_leader_operation = last_operation
             path = self.leader_optime_path
             try:
-                self.client.retry(self.client.set, path, last_operation)
+                self._client.retry(self._client.set, path, last_operation)
             except NoNodeError:
                 try:
-                    self.client.retry(self.client.create, path, last_operation, makepath=True)
+                    self._client.retry(self._client.create, path, last_operation, makepath=True)
                 except:
                     logger.exception('Failed to create %s', path)
             except:
@@ -266,21 +266,27 @@ class ZooKeeper(AbstractDCS):
         return True
 
     def delete_leader(self):
-        self.client.restart()
+        self._client.restart()
         self._my_member_data = None
         return True
 
     def _cancel_initialization(self):
         node = self.get_node(self.initialize_path)
         if node:
-            self.client.delete(self.initialize_path, version=node[1].version)
+            self._client.delete(self.initialize_path, version=node[1].version)
 
     def cancel_initialization(self):
         try:
-            self.client.retry(self._cancel_initialization)
+            self._client.retry(self._cancel_initialization)
         except:
             logger.exception("Unable to delete initialize key")
 
+    def delete_cluster(self):
+        try:
+            return self._client.retry(self._client.delete, self.client_path(''), recursive=True)
+        except NoNodeError:
+            return True
+
     def watch(self, timeout):
         if super(ZooKeeper, self).watch(timeout):
             self.fetch_cluster = True

tests/test_ctl.py

@@ -1,10 +1,10 @@
+import etcd
 import os
 import pytest
 import requests.exceptions
 import unittest
 
 from click.testing import CliRunner
-from etcd import EtcdException
 from mock import patch, Mock, MagicMock
 from patroni.ctl import ctl, members, store_config, load_config, output_members, post_patroni, get_dcs, \
     wait_for_leader, get_all_members, get_any_member, get_cursor, query_member, configure
@@ -44,6 +44,9 @@ def test_rw_config():
 
 
 @patch('patroni.ctl.load_config', Mock(return_value={'dcs': {'scheme': 'etcd', 'hostname': 'localhost', 'port': 4001}}))
+@patch.object(etcd.Client, 'write', etcd_write)
+@patch.object(etcd.Client, 'read', etcd_read)
+@patch.object(etcd.Client, 'delete', Mock(side_effect=etcd.EtcdException))
 class TestCtl(unittest.TestCase):
 
     @patch('socket.getaddrinfo', socket_getaddrinfo)
@@ -52,9 +55,6 @@ class TestCtl(unittest.TestCase):
         with patch.object(Client, 'machines') as mock_machines:
             mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
             self.e = Etcd('foo', {'ttl': 30, 'host': 'ok:2379', 'scope': 'test'})
-            self.e.client.read = etcd_read
-            self.e.client.write = etcd_write
-            self.e.client.delete = Mock(side_effect=EtcdException)
 
     @patch('psycopg2.connect', psycopg2_connect)
     def test_get_cursor(self):
@@ -298,37 +298,28 @@ y''')
             result = self.runner.invoke(ctl, ['restart', 'alpha', '--dcs', '8.8.8.8'], input='y')
 
     @patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_with_leader()))
-    @patch('patroni.etcd.Etcd.get_etcd_client', Mock(return_value=None))
     def test_remove(self):
-        result = self.runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='alpha\nslave')
-        assert 'Please confirm' in result.output
-        assert 'You are about to remove all' in result.output
-        # Not typing an exact confirmation
-        assert result.exit_code == 1
+        with patch('patroni.ctl.get_dcs', Mock(return_value=self.e)):
+            result = self.runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='alpha\nslave')
+            assert 'Please confirm' in result.output
+            assert 'You are about to remove all' in result.output
+            # Not typing an exact confirmation
+            assert result.exit_code == 1
 
-        # master specified does not match master of cluster
-        result = self.runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='''alpha
+            # master specified does not match master of cluster
+            result = self.runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='''alpha
 Yes I am aware
 slave''')
-        assert result.exit_code == 1
+            assert result.exit_code == 1
 
-        # cluster specified on cmdline does not match verification prompt
-        result = self.runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='beta\nleader')
-        assert result.exit_code == 1
+            # cluster specified on cmdline does not match verification prompt
+            result = self.runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='beta\nleader')
+            assert result.exit_code == 1
 
-        with patch('patroni.etcd.Etcd.get_cluster', get_cluster_initialized_with_leader):
-            result = self.runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'],
-                                        input='''alpha
-Yes I am aware
-leader''')
-            assert 'object has no attribute' in str(result.exception)
-
-        with patch('patroni.ctl.get_dcs', Mock(return_value=Mock())):
-            # Not implemented DCS
             result = self.runner.invoke(ctl, ['remove', 'alpha', '--dcs', '8.8.8.8'], input='''alpha
 Yes I am aware
 leader''')
-            assert result.exit_code == 1
+            assert result.exit_code == 0
 
     @patch('patroni.etcd.Etcd.watch', Mock(return_value=None))
     @patch('patroni.etcd.Etcd.get_cluster', Mock(return_value=get_cluster_initialized_with_leader()))

tests/test_etcd.py

@@ -64,7 +64,7 @@ def requests_get(url, **kwargs):
     return response
 
 
-def etcd_watch(key, index=None, timeout=None, recursive=None):
+def etcd_watch(self, key, index=None, timeout=None, recursive=None):
     if timeout == 2.0:
         raise etcd.EtcdWatchTimedOut
     elif timeout == 5.0:
@@ -77,7 +77,7 @@ def etcd_watch(key, index=None, timeout=None, recursive=None):
         return etcd.EtcdResult('set', {'value': 'postgresql2', 'modifiedIndex': index + 1})
 
 
-def etcd_write(key, value, **kwargs):
+def etcd_write(self, key, value, **kwargs):
     if key == '/service/exists/leader':
         raise etcd.EtcdAlreadyExist
     if key in ['/service/test/leader', '/patroni/test/leader'] and \
@@ -86,7 +86,7 @@ def etcd_write(key, value, **kwargs):
     raise etcd.EtcdException
 
 
-def etcd_read(key, **kwargs):
+def etcd_read(self, key, **kwargs):
     if key == '/service/noleader/':
         raise DCSError('noleader')
     elif key == '/service/nocluster/':
@@ -198,15 +198,15 @@ class TestClient(unittest.TestCase):
 
 
 @patch('requests.get', requests_get)
+@patch.object(etcd.Client, 'write', etcd_write)
+@patch.object(etcd.Client, 'read', etcd_read)
+@patch.object(etcd.Client, 'delete', Mock(side_effect=etcd.EtcdException))
 class TestEtcd(unittest.TestCase):
 
     def setUp(self):
         with patch.object(Client, 'machines') as mock_machines:
             mock_machines.__get__ = Mock(return_value=['http://localhost:2379', 'http://localhost:4001'])
             self.etcd = Etcd('foo', {'namespace': '/patroni/', 'ttl': 30, 'host': 'localhost:2379', 'scope': 'test'})
-            self.etcd.client.write = etcd_write
-            self.etcd.client.read = etcd_read
-            self.etcd.client.delete = Mock(side_effect=etcd.EtcdException())
 
     def test_base_path(self):
         self.assertEquals(self.etcd._base_path, '/patroni/test')
@@ -254,8 +254,8 @@ class TestEtcd(unittest.TestCase):
     def test_delete_leader(self):
         self.assertFalse(self.etcd.delete_leader())
 
+    @patch.object(etcd.Client, 'watch', etcd_watch)
     def test_watch(self):
-        self.etcd.client.watch = etcd_watch
         self.etcd.watch(0)
         self.etcd.get_cluster()
         self.etcd.watch(1.5)

tests/test_ha.py

@@ -1,8 +1,8 @@
+import etcd
 import unittest
 import datetime
 import pytz
 
-from etcd import EtcdException
 from mock import Mock, MagicMock, patch
 from patroni.dcs import Cluster, Failover, Leader, Member
 from patroni.etcd import Client, Etcd
@@ -76,10 +76,13 @@ def run_async(func, args=()):
 @patch.object(Postgresql, 'write_recovery_conf', Mock())
 @patch.object(Postgresql, 'query', Mock())
 @patch.object(Postgresql, 'checkpoint', Mock())
+@patch.object(etcd.Client, 'write', etcd_write)
+@patch.object(etcd.Client, 'delete', Mock(side_effect=etcd.EtcdException))
 @patch('subprocess.call', Mock(return_value=0))
 class TestHa(unittest.TestCase):
 
     @patch('socket.getaddrinfo', socket_getaddrinfo)
+    @patch.object(etcd.Client, 'read', etcd_read)
     def setUp(self):
         with patch.object(Client, 'machines') as mock_machines:
             mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
@@ -90,9 +93,6 @@ class TestHa(unittest.TestCase):
             self.p.check_replication_lag = true
             self.p.can_create_replica_without_replication_connection = MagicMock(return_value=False)
             self.e = Etcd('foo', {'ttl': 30, 'host': 'ok:2379', 'scope': 'test'})
-            self.e.client.read = etcd_read
-            self.e.client.write = etcd_write
-            self.e.client.delete = Mock(side_effect=EtcdException())
             self.ha = Ha(MockPatroni(self.p, self.e))
             self.ha._async_executor.run_async = run_async
             self.ha.old_cluster = self.e.get_cluster()

tests/test_patroni.py

@@ -1,3 +1,4 @@
+import etcd
 import sys
 import time
 import unittest
@@ -22,6 +23,8 @@ from test_zookeeper import MockKazooClient
 @patch.object(Postgresql, 'write_recovery_conf', Mock())
 @patch.object(BaseHTTPServer.HTTPServer, '__init__', Mock())
 @patch.object(AsyncExecutor, 'run', Mock())
+@patch.object(etcd.Client, 'write', etcd_write)
+@patch.object(etcd.Client, 'read', etcd_read)
 class TestPatroni(unittest.TestCase):
 
     def setUp(self):
@@ -35,8 +38,6 @@ class TestPatroni(unittest.TestCase):
             with open('postgres0.yml', 'r') as f:
                 config = yaml.load(f)
                 self.p = Patroni(config)
-                self.p.ha.dcs.client.write = etcd_write
-                self.p.ha.dcs.client.read = etcd_read
 
     @patch('patroni.zookeeper.KazooClient', MockKazooClient())
     def test_get_dcs(self):

tests/test_zookeeper.py

@@ -92,7 +92,7 @@ class MockKazooClient(Mock):
             raise Exception
         elif path == '/service/test/members/buzz':
             raise Exception
-        elif path.endswith('/initialize') or path == '/service/test/members/bar':
+        elif path.endswith('/') or path.endswith('/initialize') or path == '/service/test/members/bar':
             raise NoNodeError
 
 
@@ -152,7 +152,7 @@ class TestZooKeeper(unittest.TestCase):
         self.zk._name = 'bar'
         self.zk.touch_member('new')
         self.zk._name = 'na'
-        self.zk.client.exists = 1
+        self.zk._client.exists = 1
         self.zk.touch_member('exists')
         self.zk._name = 'bar'
         self.zk.touch_member('retry')
@@ -172,6 +172,9 @@ class TestZooKeeper(unittest.TestCase):
         self.zk._base_path = self.zk._base_path.replace('test', 'bla')
         self.zk.write_leader_optime('2')
 
+    def test_delete_cluster(self):
+        self.assertTrue(self.zk.delete_cluster())
+
     def test_watch(self):
         self.zk.watch(0)
         self.zk.event.isSet = lambda: True