diff --git a/SETTINGS.rst b/SETTINGS.rst index a5ba11b5..8e240989 100644 --- a/SETTINGS.rst +++ b/SETTINGS.rst @@ -13,14 +13,20 @@ Consul - **scope**: the relative path used on Consul's HTTP API for this deployment; makes it possible to run multiple HA deployments from a single Consul cluster. - **ttl**: the TTL to acquire the leader lock. Think of it as the length of time before initiation of the automatic failover process. -etcd +Etcd ---- - **host**: the host:port for the etcd endpoint. - **scope**: the relative path used on etcd's HTTP API for this deployment. Makes it possible to run multiple HA deployments from a single etcd cluster. - **ttl**: the TTL to acquire the leader lock. Think of it as the length of time before initiation of the automatic failover process. +Exhibitor +--------- +- **hosts**: initial list of Exhibitor (ZooKeeper) nodes in format: ['host1', 'host2', 'etc...' ]. This list updates automatically whenever the Exhibitor (ZooKeeper) cluster topology changes. +- **poll\_interval**: how often the list of ZooKeeper and Exhibitor nodes should be updated from Exhibitor +- **port**: Exhibitor port. + PostgreSQL ---------------- +---------- - **admin**: - **password**: admin password; user is created during initialization. - **username**: admin username; user is created during initialization. It will have CREATEDB and CREATEROLE privileges. @@ -69,7 +75,3 @@ ZooKeeper - **scope**: the relative path used on ZooKeeper for this deployment. Makes it possible to run multiple HA deployments from a single ZooKeeper cluster. - **session\_timeout**: the TTL to acquire the leader lock. Think of it as the length of time before initiation of the automatic failover process. -- **exhibitor**: If you are running a ZooKeeper cluster under the Exhibitor supervisory, this section might interest you: - - **hosts**: initial list of Exhibitor (ZooKeeper) nodes in format: ['host1', 'host2', 'etc...' ]. This list updates automatically whenever the Exhibitor (ZooKeeper) cluster topology changes. - - **poll\_interval**: how often the list of ZooKeeper and Exhibitor nodes should be updated from Exhibitor - - **port**: Exhibitor port. diff --git a/features/environment.py b/features/environment.py index bc1106f4..a8d2f408 100644 --- a/features/environment.py +++ b/features/environment.py @@ -132,15 +132,13 @@ class PatroniController(AbstractController): dcs_config = config.pop('etcd') dcs_config.pop('host') - if dcs == 'consul': - config[dcs] = dcs_config - else: + if dcs != 'consul': dcs_config.update({'session_timeout': dcs_config.pop('ttl'), 'reconnect_timeout': config['loop_wait']}) if dcs == 'exhibitor': - dcs_config['exhibitor'] = {'hosts': ['127.0.0.1'], 'port': 8181} + dcs_config.update({'hosts': ['127.0.0.1'], 'port': 8181}) else: dcs_config['hosts'] = ['127.0.0.1:2181'] - config['zookeeper'] = dcs_config + config[dcs] = dcs_config with open(patroni_config_path, 'w') as f: yaml.dump(config, f, default_flow_style=False) diff --git a/patroni/ctl.py b/patroni/ctl.py index 19b48feb..412fa354 100644 --- a/patroni/ctl.py +++ b/patroni/ctl.py @@ -26,7 +26,7 @@ CONFIG_DIR_PATH = click.get_app_dir('patroni') CONFIG_FILE_PATH = os.path.join(CONFIG_DIR_PATH, 'patronictl.yaml') LOGLEVEL = 'WARNING' DCS_DEFAULTS = {'zookeeper': {'port': 2181, 'template': "zookeeper:\n hosts: ['{host}:{port}']"}, - 'exhibitor': {'port': 8181, 'template': "zookeeper:\n exhibitor:\n hosts: [{host}]\n port: {port}"}, + 'exhibitor': {'port': 8181, 'template': "exhibitor:\n hosts: [{host}]\n port: {port}"}, 'consul': {'port': 8500, 'template': "consul:\n host: '{host}:{port}'"}, 'etcd': {'port': 4001, 'template': "etcd:\n host: '{host}:{port}'"}} diff --git a/patroni/dcs/exhibitor.py b/patroni/dcs/exhibitor.py new file mode 100644 index 00000000..e5053809 --- /dev/null +++ b/patroni/dcs/exhibitor.py @@ -0,0 +1,75 @@ +import logging +import random +import requests +import time + +from patroni.dcs.zookeeper import ZooKeeper +from patroni.utils import sleep +from requests.exceptions import RequestException + +logger = logging.getLogger(__name__) + + +class ExhibitorEnsembleProvider(object): + + TIMEOUT = 3.1 + + def __init__(self, hosts, port, uri_path='/exhibitor/v1/cluster/list', poll_interval=300): + self._exhibitor_port = port + self._uri_path = uri_path + self._poll_interval = poll_interval + self._exhibitors = hosts + self._master_exhibitors = hosts + self._zookeeper_hosts = '' + self._next_poll = None + while not self.poll(): + logger.info('waiting on exhibitor') + sleep(5) + + def poll(self): + if self._next_poll and self._next_poll > time.time(): + return False + + json = self._query_exhibitors(self._exhibitors) + if not json: + json = self._query_exhibitors(self._master_exhibitors) + + if isinstance(json, dict) and 'servers' in json and 'port' in json: + self._next_poll = time.time() + self._poll_interval + zookeeper_hosts = ','.join([h + ':' + str(json['port']) for h in sorted(json['servers'])]) + if self._zookeeper_hosts != zookeeper_hosts: + logger.info('ZooKeeper connection string has changed: %s => %s', self._zookeeper_hosts, zookeeper_hosts) + self._zookeeper_hosts = zookeeper_hosts + self._exhibitors = json['servers'] + return True + return False + + def _query_exhibitors(self, exhibitors): + random.shuffle(exhibitors) + for host in exhibitors: + uri = 'http://{0}:{1}{2}'.format(host, self._exhibitor_port, self._uri_path) + try: + response = requests.get(uri, timeout=self.TIMEOUT) + return response.json() + except RequestException: + pass + return None + + @property + def zookeeper_hosts(self): + return self._zookeeper_hosts + + +class Exhibitor(ZooKeeper): + + def __init__(self, name, config): + interval = config.get('poll_interval', 300) + self._ensemble_provider = ExhibitorEnsembleProvider(config['hosts'], config['port'], poll_interval=interval) + config = config.copy() + config['hosts'] = self._ensemble_provider.zookeeper_hosts + super(Exhibitor, self).__init__(name, config) + + def _load_cluster(self): + if self._ensemble_provider.poll(): + self._client.set_hosts(self._ensemble_provider.zookeeper_hosts) + return super(Exhibitor, self)._load_cluster() diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index e245e5e6..9ec12af3 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -1,14 +1,9 @@ import logging -import random -import requests -import time from kazoo.client import KazooClient, KazooState from kazoo.exceptions import NoNodeError, NodeExistsError from patroni.dcs import AbstractDCS, Cluster, Failover, Leader, Member from patroni.exceptions import DCSError -from patroni.utils import sleep -from requests.exceptions import RequestException logger = logging.getLogger(__name__) @@ -17,56 +12,6 @@ class ZooKeeperError(DCSError): pass -class ExhibitorEnsembleProvider(object): - - TIMEOUT = 3.1 - - def __init__(self, hosts, port, uri_path='/exhibitor/v1/cluster/list', poll_interval=300): - self._exhibitor_port = port - self._uri_path = uri_path - self._poll_interval = poll_interval - self._exhibitors = hosts - self._master_exhibitors = hosts - self._zookeeper_hosts = '' - self._next_poll = None - while not self.poll(): - logger.info('waiting on exhibitor') - sleep(5) - - def poll(self): - if self._next_poll and self._next_poll > time.time(): - return False - - json = self._query_exhibitors(self._exhibitors) - if not json: - json = self._query_exhibitors(self._master_exhibitors) - - if isinstance(json, dict) and 'servers' in json and 'port' in json: - self._next_poll = time.time() + self._poll_interval - zookeeper_hosts = ','.join([h + ':' + str(json['port']) for h in sorted(json['servers'])]) - if self._zookeeper_hosts != zookeeper_hosts: - logger.info('ZooKeeper connection string has changed: %s => %s', self._zookeeper_hosts, zookeeper_hosts) - self._zookeeper_hosts = zookeeper_hosts - self._exhibitors = json['servers'] - return True - return False - - def _query_exhibitors(self, exhibitors): - random.shuffle(exhibitors) - for host in exhibitors: - uri = 'http://{0}:{1}{2}'.format(host, self._exhibitor_port, self._uri_path) - try: - response = requests.get(uri, timeout=self.TIMEOUT) - return response.json() - except RequestException: - pass - return None - - @property - def zookeeper_hosts(self): - return self._zookeeper_hosts - - class ZooKeeper(AbstractDCS): def __init__(self, name, config): @@ -76,13 +21,6 @@ class ZooKeeper(AbstractDCS): if isinstance(hosts, list): hosts = ','.join(hosts) - self.exhibitor = None - if 'exhibitor' in config: - exhibitor = config['exhibitor'] - interval = exhibitor.get('poll_interval', 300) - 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}, @@ -167,9 +105,6 @@ class ZooKeeper(AbstractDCS): self._cluster = Cluster(initialize, leader, self._last_leader_operation, members, failover) def _load_cluster(self): - if self.exhibitor and self.exhibitor.poll(): - self._client.set_hosts(self.exhibitor.zookeeper_hosts) - if self._fetch_cluster or self._cluster is None: try: self._client.retry(self._inner_load_cluster) diff --git a/postgres0.yml b/postgres0.yml index 22ab5a73..067ab743 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -23,13 +23,13 @@ etcd: # hosts: # - 127.0.0.1:2181 # - 127.0.0.2:2181 -# exhibitor: -# poll_interval: 300 -# port: 8181 -# hosts: -# - host1 -# - host2 -# - host3 +#exhibitor: +# poll_interval: 300 +# port: 8181 +# hosts: +# - host1 +# - host2 +# - host3 postgresql: name: postgresql0 scope: *scope diff --git a/postgres1.yml b/postgres1.yml index 02e731e3..5d2b58ee 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -23,13 +23,13 @@ etcd: # hosts: # - 127.0.0.1:2181 # - 127.0.0.2:2181 -# exhibitor: -# poll_interval: 300 -# port: 8181 -# hosts: -# - host1 -# - host2 -# - host3 +#exhibitor: +# poll_interval: 300 +# port: 8181 +# hosts: +# - host1 +# - host2 +# - host3 postgresql: name: postgresql1 scope: *scope diff --git a/postgres2.yml b/postgres2.yml index c5efa813..48c3b55d 100644 --- a/postgres2.yml +++ b/postgres2.yml @@ -23,13 +23,13 @@ etcd: # hosts: # - 127.0.0.1:2181 # - 127.0.0.2:2181 -# exhibitor: -# poll_interval: 300 -# port: 8181 -# hosts: -# - host1 -# - host2 -# - host3 +#exhibitor: +# poll_interval: 300 +# port: 8181 +# hosts: +# - host1 +# - host2 +# - host3 postgresql: name: postgresql2 scope: *scope diff --git a/tests/test_ctl.py b/tests/test_ctl.py index d7394db3..c9bd1258 100644 --- a/tests/test_ctl.py +++ b/tests/test_ctl.py @@ -70,7 +70,7 @@ class TestCtl(unittest.TestCase): assert parse_dcs('') == {'etcd': {'host': 'localhost:4001'}} assert parse_dcs('localhost:8500') == {'consul': {'host': 'localhost:8500'}} assert parse_dcs('zookeeper://localhost') == {'zookeeper': {'hosts': ['localhost:2181']}} - assert parse_dcs('exhibitor://dummy') == {'zookeeper': {'exhibitor': {'hosts': ['dummy'], 'port': 8181}}} + assert parse_dcs('exhibitor://dummy') == {'exhibitor': {'hosts': ['dummy'], 'port': 8181}} assert parse_dcs('consul://localhost') == {'consul': {'host': 'localhost:8500'}} self.assertRaises(PatroniCtlException, parse_dcs, 'invalid://test') diff --git a/tests/test_exhibitor.py b/tests/test_exhibitor.py new file mode 100644 index 00000000..be60e562 --- /dev/null +++ b/tests/test_exhibitor.py @@ -0,0 +1,30 @@ +import unittest + +from mock import Mock, patch +from patroni.dcs.exhibitor import ExhibitorEnsembleProvider, Exhibitor +from patroni.dcs.zookeeper import ZooKeeperError +from test_etcd import SleepException, requests_get +from test_zookeeper import MockKazooClient + + +@patch('requests.get', requests_get) +@patch('time.sleep', Mock(side_effect=SleepException)) +class TestExhibitorEnsembleProvider(unittest.TestCase): + + def test_init(self): + self.assertRaises(SleepException, ExhibitorEnsembleProvider, ['localhost'], 8181) + + def test_poll(self): + self.assertFalse(ExhibitorEnsembleProvider(['exhibitor'], 8181).poll()) + + +class TestExhibitor(unittest.TestCase): + + @patch('requests.get', requests_get) + @patch('patroni.dcs.zookeeper.KazooClient', MockKazooClient) + def setUp(self): + self.e = Exhibitor('foo', {'hosts': ['localhost', 'exhibitor'], 'port': 8181, 'scope': 'test'}) + + @patch.object(ExhibitorEnsembleProvider, 'poll', Mock(return_value=True)) + def test_get_cluster(self): + self.assertRaises(ZooKeeperError, self.e.get_cluster) diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index d2a288f0..163d999c 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -5,8 +5,7 @@ from kazoo.client import KazooState from kazoo.exceptions import NoNodeError, NodeExistsError from kazoo.protocol.states import ZnodeStat from mock import Mock, patch -from patroni.dcs.zookeeper import Leader, ExhibitorEnsembleProvider, ZooKeeper, ZooKeeperError -from test_etcd import SleepException, requests_get +from patroni.dcs.zookeeper import Leader, ZooKeeper, ZooKeeperError class MockKazooClient(Mock): @@ -90,20 +89,11 @@ class MockKazooClient(Mock): raise NoNodeError -@patch('requests.get', requests_get) -@patch('time.sleep', Mock(side_effect=SleepException)) -class TestExhibitorEnsembleProvider(unittest.TestCase): - - def test_init(self): - self.assertRaises(SleepException, ExhibitorEnsembleProvider, ['localhost'], 8181) - - class TestZooKeeper(unittest.TestCase): - @patch('requests.get', requests_get) @patch('patroni.dcs.zookeeper.KazooClient', MockKazooClient) def setUp(self): - self.zk = ZooKeeper('foo', {'exhibitor': {'hosts': ['localhost', 'exhibitor'], 'port': 8181}, 'scope': 'test'}) + self.zk = ZooKeeper('foo', {'hosts': ['localhost:2181'], 'scope': 'test'}) def test_session_listener(self): self.zk.session_listener(KazooState.SUSPENDED) @@ -122,11 +112,12 @@ class TestZooKeeper(unittest.TestCase): def test_get_cluster(self): self.assertRaises(ZooKeeperError, self.zk.get_cluster) - self.zk.exhibitor.poll = lambda: True cluster = self.zk.get_cluster() self.assertIsInstance(cluster.leader, Leader) self.zk.touch_member('foo') - self.zk.delete_leader() + + def test_delete_leader(self): + self.assertTrue(self.zk.delete_leader()) def test_set_failover_value(self): self.zk.set_failover_value('')