diff --git a/patroni/ctl.py b/patroni/ctl.py index 0cfd6039..5434cebb 100644 --- a/patroni/ctl.py +++ b/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): diff --git a/patroni/dcs.py b/patroni/dcs.py index 2768b13e..aa7e178a 100644 --- a/patroni/dcs.py +++ b/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 diff --git a/patroni/etcd.py b/patroni/etcd.py index 03d6259d..3f798cf8 100644 --- a/patroni/etcd.py +++ b/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') diff --git a/patroni/zookeeper.py b/patroni/zookeeper.py index 79e8a72a..7d458872 100644 --- a/patroni/zookeeper.py +++ b/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 diff --git a/tests/test_ctl.py b/tests/test_ctl.py index ed3aa1b9..806b567d 100644 --- a/tests/test_ctl.py +++ b/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())) diff --git a/tests/test_etcd.py b/tests/test_etcd.py index fa2538c9..9ed5af58 100644 --- a/tests/test_etcd.py +++ b/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) diff --git a/tests/test_ha.py b/tests/test_ha.py index 2f9e5478..9ebf55d3 100644 --- a/tests/test_ha.py +++ b/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() diff --git a/tests/test_patroni.py b/tests/test_patroni.py index 2698de58..d3801abb 100644 --- a/tests/test_patroni.py +++ b/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): diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index f5cbdf13..339aa4bd 100644 --- a/tests/test_zookeeper.py +++ b/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