Merge pull request #156 from zalando/feature/delete-cluster-iface

Implement `delete_cluster` interface in for all available dcs

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,14 @@ 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, 'read', etcd_read)
+@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 +94,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