Split ZooKeeper and Exhibitor

Originally Exhibitor was supported in the ZooKeeper class and
configuration for Exhibitor was taken also from `zookeeper` section in
the yaml config file. In fact, Exhibitor just extends ZooKeeper and now
it is reflected in the code and also Exhibitor got it's own section in
the config.yaml file. It will make it easier to configure Exhibitor
hosts and port via environment variables when PR#211 will be merged.

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.

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)

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}'"}}
 

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

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)

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

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)

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