Merge branch 'master' of github.com:zalando/patroni into feature/dynamic-configuration

SETTINGS.rst

@@ -37,10 +37,16 @@ Consul
 ------
 -  **host**: the host:port for the Consul endpoint.
 
-etcd
+Etcd
 ----
 -  **host**: the host:port for the etcd endpoint.
 
+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
 ----------
 -  **authentication**:
@@ -76,7 +82,3 @@ REST API
 ZooKeeper
 ----------
 -  **hosts**: list of ZooKeeper cluster members in format: ['host1:port1', 'host2:port2', 'etc...'].
--  **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

@@ -143,14 +143,11 @@ class PatroniController(AbstractController):
             dcs_config = config.pop('etcd')
             dcs_config.pop('host')
 
-            if dcs == 'consul':
-                config[dcs] = dcs_config
-            else:
             if dcs == 'exhibitor':
-                    dcs_config['exhibitor'] = {'hosts': ['127.0.0.1'], 'port': 8181}
+                dcs_config.update({'hosts': ['127.0.0.1'], 'port': 8181})
-                else:
+            elif dcs == 'zookeeper':
                 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.safe_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, 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__(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, ClusterConfig, 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, config):
@@ -76,16 +21,8 @@ class ZooKeeper(AbstractDCS):
         if isinstance(hosts, list):
             hosts = ','.join(hosts)
 
-        self.exhibitor = None
+        self._client = KazooClient(hosts, timeout=config['ttl'], connection_retry={'max_delay': 1, 'max_tries': -1},
-        if 'exhibitor' in config:
+                                   command_retry={'deadline': config['retry_timeout'], 'max_delay': 1, 'max_tries': -1})
-            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['ttl'],
-                                   command_retry={'deadline': config['retry_timeout'], '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
@@ -104,8 +41,7 @@ class ZooKeeper(AbstractDCS):
 
     def set_ttl(self, ttl):
         ttl = int(ttl * 1000)
-        # I know, it's weird to access private attributes, but there is
+        # I know, it's weird to access private attributes
-        # no other way to change session_timeout without losing session
         if self._client._session_timeout != ttl:
             self._client._session_timeout = ttl
             self._client.restart()
@@ -180,9 +116,6 @@ class ZooKeeper(AbstractDCS):
         self._cluster = Cluster(initialize, config, 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)

patroni/postgresql.py

@@ -63,6 +63,7 @@ class Postgresql(object):
         self.config = config
         self.name = config['name']
         self.scope = config['scope']
+        self._database = config.get('database', 'postgres')
         self._data_dir = config['data_dir']
         self._pending_restart = False
         self._server_parameters = self.get_server_parameters(config)
@@ -79,8 +80,9 @@ class Postgresql(object):
 
         self._pgpass = config.get('pgpass') or os.path.join(os.path.expanduser('~'), 'pgpass')
         self.callback = config.get('callbacks') or {}
-        self._postgresql_conf = os.path.join(self._data_dir, 'postgresql.conf')
+        config_base_name = config.get('config_base_name', 'postgresql')
-        self._postgresql_base_conf_name = 'postgresql.base.conf'
+        self._postgresql_conf = os.path.join(self._data_dir, config_base_name + '.conf')
+        self._postgresql_base_conf_name = config_base_name + '.base.conf'
         self._postgresql_base_conf = os.path.join(self._data_dir, self._postgresql_base_conf_name)
         self._recovery_conf = os.path.join(self._data_dir, 'recovery.conf')
         self._configuration_to_save = (self._postgresql_conf, self._postgresql_base_conf,
@@ -130,8 +132,8 @@ class Postgresql(object):
 
     def resolve_connection_addresses(self):
         self._local_address = self.get_local_address()
-        self.connection_string = 'postgres://{username}:{password}@{connect_address}/postgres'.format(
+        self.connection_string = 'postgres://{username}:{password}@{connect_address}/{database}'.format(
-            connect_address=self._connect_address or self._local_address, **self._replication)
+            connect_address=self._connect_address or self._local_address, database=self._database, **self._replication)
 
     def reload_config(self, config):
         server_parameters = self.get_server_parameters(config)
@@ -244,7 +246,7 @@ class Postgresql(object):
 
     @property
     def _connect_kwargs(self):
-        r = parseurl('postgres://{0}/postgres'.format(self._local_address))
+        r = parseurl('postgres://{0}/{1}'.format(self._local_address, self._database))
         if 'username' in self._superuser:
             r['user'] = self._superuser['username']
         if 'password' in self._superuser:
@@ -613,8 +615,9 @@ class Postgresql(object):
         r = parseurl(leader.conn_url)
         r.update(self._superuser)
         r['user'] = r.pop('username')
+        r['database'] = self._database
         env = self.write_pgpass(r)
-        pc = "user={user} host={host} port={port} dbname=postgres sslmode=prefer sslcompression=1".format(**r)
+        pc = "user={user} host={host} port={port} dbname={database} sslmode=prefer sslcompression=1".format(**r)
         # first run a checkpoint on a promoted master in order
         # to make it store the new timeline ([email protected])
         self.checkpoint(r)
@@ -660,7 +663,7 @@ class Postgresql(object):
         for opt, val in sorted((options or {}).items()):
             cmd.extend(['-c', '{0}={1}'.format(opt, val)])
         # need a database name to connect
-        cmd.append('postgres')
+        cmd.append(self._database)
         p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=open(os.devnull, 'w'), stderr=subprocess.STDOUT)
         if p:
             if command:

postgres0.yml

@@ -50,6 +50,7 @@ bootstrap:
       options:
         - createrole
         - createdb
+
 postgresql:
   listen: 127.0.0.1:5432
   connect_address: 127.0.0.1:5432

postgres1.yml

@@ -23,7 +23,7 @@ etcd:
 #  hosts:
 #    - 127.0.0.1:2181
 #    - 127.0.0.2:2181
-#  exhibitor:
+#exhibitor:
 #  poll_interval: 300
 #  port: 8181
 #  hosts:

postgres2.yml

@@ -23,7 +23,7 @@ etcd:
 #  hosts:
 #    - 127.0.0.1:2181
 #    - 127.0.0.2:2181
-#  exhibitor:
+#exhibitor:
 #  poll_interval: 300
 #  port: 8181
 #  hosts:

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,31 @@
+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({'hosts': ['localhost', 'exhibitor'], 'port': 8181, 'scope': 'test',
+                            'name': 'foo', 'ttl': 30, 'retry_timeout': 10})
+
+    @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 patroni.dcs.zookeeper import Leader, ZooKeeper, ZooKeeperError
-from test_etcd import SleepException, requests_get
 
 
 class MockKazooClient(Mock):
@@ -14,6 +13,9 @@ class MockKazooClient(Mock):
     leader = False
     exists = True
 
+    def __init__(self, *args, **kwargs):
+        super(MockKazooClient, self).__init__()
+
     @property
     def client_id(self):
         return (-1, '')
@@ -90,21 +92,12 @@ 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({'exhibitor': {'hosts': ['localhost', 'exhibitor'], 'port': 8181},
+        self.zk = ZooKeeper({'hosts': ['localhost:2181'], 'scope': 'test',
-                             'scope': 'test', 'name': 'foo', 'ttl': 30, 'retry_timeout': 10})
+                             'name': 'foo', 'ttl': 30, 'retry_timeout': 10})
 
     def test_session_listener(self):
         self.zk.session_listener(KazooState.SUSPENDED)
@@ -129,11 +122,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('')