diff --git a/SETTINGS.rst b/SETTINGS.rst index 6b0d4153..a4004bcf 100644 --- a/SETTINGS.rst +++ b/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**: @@ -68,7 +74,7 @@ REST API -------- - **connect\_address**: IP address and port through which restapi is accessible. - **listen**: IP address and port that Patroni will listen to, to provide health-check information for HAProxy. -- **Optional**: +- **Optional**: - **auth**: 'username:password' to protect dangerous REST API endpoints. - **certfile**: Specifies a file with the certificate in the PEM format. If the certfile is not specified or is left empty, the API server will work without SSL. - **keyfile**: Specifies a file with the secret key in the PEM format. @@ -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. diff --git a/features/environment.py b/features/environment.py index 40b2f377..3fc52a2d 100644 --- a/features/environment.py +++ b/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} - else: - dcs_config['hosts'] = ['127.0.0.1:2181'] - config['zookeeper'] = dcs_config + if dcs == 'exhibitor': + dcs_config.update({'hosts': ['127.0.0.1'], 'port': 8181}) + elif dcs == 'zookeeper': + dcs_config['hosts'] = ['127.0.0.1:2181'] + config[dcs] = dcs_config with open(patroni_config_path, 'w') as f: yaml.safe_dump(config, f, default_flow_style=False) diff --git a/patroni/ctl.py b/patroni/ctl.py index 3d93d2fd..499f6563 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..56105af1 --- /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, 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() diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index e077f726..dae08ffb 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, 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 - 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['ttl'], - command_retry={'deadline': config['retry_timeout'], 'max_delay': 1, 'max_tries': -1}, - connection_retry={'max_delay': 1, 'max_tries': -1}) + self._client = KazooClient(hosts, timeout=config['ttl'], connection_retry={'max_delay': 1, 'max_tries': -1}, + command_retry={'deadline': config['retry_timeout'], '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 - # no other way to change session_timeout without losing session + # I know, it's weird to access private attributes 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) diff --git a/patroni/postgresql.py b/patroni/postgresql.py index 5672c616..3622b865 100644 --- a/patroni/postgresql.py +++ b/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') - self._postgresql_base_conf_name = 'postgresql.base.conf' + config_base_name = config.get('config_base_name', 'postgresql') + 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( - connect_address=self._connect_address or self._local_address, **self._replication) + self.connection_string = 'postgres://{username}:{password}@{connect_address}/{database}'.format( + 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 (5540277D.8020309@iki.fi) 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: diff --git a/postgres0.yml b/postgres0.yml index e09cb4f0..e1210c88 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -50,6 +50,7 @@ bootstrap: options: - createrole - createdb + postgresql: listen: 127.0.0.1:5432 connect_address: 127.0.0.1:5432 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 ff491ba6..eb63e153 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..d9c0d6d3 --- /dev/null +++ b/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) diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 71c9093f..01555d8e 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): @@ -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}, - 'scope': 'test', 'name': 'foo', 'ttl': 30, 'retry_timeout': 10}) + self.zk = ZooKeeper({'hosts': ['localhost:2181'], 'scope': 'test', + '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('')