Merge pull request #212 from zalando/feature/split-exhibitor

Split ZooKeeper and Exhibitor

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)

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

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

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

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