From dcad7a32294e228758f7a339cb13ca6e137aa770 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 10 Jul 2015 10:46:33 +0200 Subject: [PATCH 1/5] Add exhibitor support List of ZooKeeper nodes could be periodically updated from Exhibitor Since we know that each Exhibitor accompanies one ZooKeeper node, list of Exhibitor nodes also maintained. Exhibitor assumes that all ZooKeeper nodes are using the same client port, 2181. The same assumption is valid for Exhibitor, it should always listen on the same port on all nodes. Original list of Exhibitor nodes is cached and used as a fallback when it failed ito query information with using maintained list. --- helpers/zookeeper.py | 82 ++++++++++++++++++++++++++++++++++++++--- postgres0.yml | 11 +++++- postgres1.yml | 11 +++++- tests/test_etcd.py | 2 + tests/test_zookeeper.py | 30 +++++++++++++-- 5 files changed, 126 insertions(+), 10 deletions(-) diff --git a/helpers/zookeeper.py b/helpers/zookeeper.py index 4e7a22c6..cb2918cd 100644 --- a/helpers/zookeeper.py +++ b/helpers/zookeeper.py @@ -1,8 +1,13 @@ import logging +import random +import requests +import time from helpers.dcs import AbstractDCS, Cluster, DCSError, Member, parse_connection_string +from helpers.utils import sleep from kazoo.client import KazooClient, KazooState from kazoo.exceptions import NoNodeError, NodeExistsError +from requests.exceptions import RequestException logger = logging.getLogger(__name__) @@ -11,15 +16,73 @@ class ZooKeeperError(DCSError): pass +class ExhibitorEnsembleProvider: + + 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://{}:{}{}'.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): super(ZooKeeper, self).__init__(name, config) - self.fetch_cluster = True - self.members = [] - self.leader = None - self.last_leader_operation = 0 - self.client = KazooClient(hosts=config['hosts'], + + hosts = config.get('hosts', []) + 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', None) or 30), command_retry={ 'deadline': (config.get('reconnect_timeout', None) or 10), @@ -28,6 +91,12 @@ class ZooKeeper(AbstractDCS): connection_retry={'max_delay': 1, 'max_tries': -1}) self.client.add_listener(self.session_listener) self.cluster_event = self.client.handler.event_object() + + self.fetch_cluster = True + self.members = [] + self.leader = None + self.last_leader_operation = 0 + self.client.start(None) def session_listener(self, state): @@ -87,6 +156,9 @@ class ZooKeeper(AbstractDCS): self.last_leader_operation = int(last_leader_operation[0]) def get_cluster(self): + if self.exhibitor and self.exhibitor.poll(): + self.client.set_hosts(self.exhibitor.zookeeper_hosts) + if self.fetch_cluster: try: self.client.retry(self._inner_load_cluster) diff --git a/postgres0.yml b/postgres0.yml index 0cb694a1..7b8527dc 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -12,7 +12,16 @@ etcd: # scope: batman # session_timeout: *ttl # reconnect_timeout: *loop_wait -# hosts: 127.0.0.1:2181 +# hosts: +# - 127.0.0.1:2181 +# - 127.0.0.2:2181 +# exhibitor: +# poll_interval: 300 +# port: 8181 +# hosts: +# - host1 +# - host2 +# - host3 postgresql: name: postgresql0 listen: 127.0.0.1:5432 diff --git a/postgres1.yml b/postgres1.yml index 680091f1..5be3fc25 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -12,7 +12,16 @@ etcd: # scope: batman # session_timeout: *ttl # reconnect_timeout: *loop_wait -# hosts: 127.0.0.1:2181 +# hosts: +# - 127.0.0.1:2181 +# - 127.0.0.2:2181 +# exhibitor: +# poll_interval: 300 +# port: 8181 +# hosts: +# - host1 +# - host2 +# - host3 postgresql: name: postgresql1 listen: 127.0.0.1:5433 diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 609be245..599b4bb2 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -43,6 +43,8 @@ def requests_get(url, **kwargs): response.content = members elif url.endswith('/bad_response'): response.content = '{' + elif url.startswith('http://exhibitor'): + response.content = '{"servers":["127.0.0.1","127.0.0.2","127.0.0.3"],"port":2181}' elif url.startswith('http://local'): raise requests.exceptions.RequestException() elif url.startswith('http://remote') or url.startswith('http://127.0.0.1') or url.startswith('http://error'): diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index 0799413b..172a8c5f 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -1,11 +1,12 @@ import helpers.zookeeper import unittest +import requests -from helpers.zookeeper import ZooKeeper, ZooKeeperError +from helpers.zookeeper import ExhibitorEnsembleProvider, ZooKeeper, ZooKeeperError from kazoo.client import KazooState from kazoo.exceptions import NoNodeError, NodeExistsError from kazoo.protocol.states import ZnodeStat -from test_etcd import MockPostgresql +from test_etcd import MockPostgresql, requests_get class MockEvent: @@ -88,6 +89,27 @@ class MockKazooClient: self.leader = True raise Exception + def set_hosts(self, hosts, randomize_hosts=None): + pass + + +def exhibitor_sleep(_): + raise Exception + + +class TestExhibitorEnsembleProvider(unittest.TestCase): + + def __init__(self, method_name='runTest'): + self.setUp = self.set_up + super(TestExhibitorEnsembleProvider, self).__init__(method_name) + + def set_up(self): + requests.get = requests_get + helpers.zookeeper.sleep = exhibitor_sleep + + def test_init(self): + self.assertRaises(Exception, ExhibitorEnsembleProvider, ['localhost'], 8181) + class TestZooKeeper(unittest.TestCase): @@ -96,8 +118,9 @@ class TestZooKeeper(unittest.TestCase): super(TestZooKeeper, self).__init__(method_name) def set_up(self): + requests.get = requests_get helpers.zookeeper.KazooClient = MockKazooClient - self.zk = ZooKeeper('foo', {'hosts': 'localhost:2181', 'scope': 'test'}) + self.zk = ZooKeeper('foo', {'exhibitor': {'hosts': ['localhost', 'exhibitor'], 'port': 8181}, 'scope': 'test'}) def test_session_listener(self): self.zk.session_listener(KazooState.SUSPENDED) @@ -112,6 +135,7 @@ class TestZooKeeper(unittest.TestCase): def test_get_cluster(self): self.assertRaises(ZooKeeperError, self.zk.get_cluster) + self.zk.exhibitor.poll = lambda: True self.zk.get_cluster() self.zk.touch_member('foo') self.zk.delete_leader() From 71fce312c25325915d569173505d4587d0414c43 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 10 Jul 2015 10:53:04 +0200 Subject: [PATCH 2/5] Enforce flake8 check --- .travis.yml | 1 + helpers/aws.py | 4 ++-- helpers/postgresql.py | 4 +++- tests/test_postgresql.py | 2 +- 4 files changed, 7 insertions(+), 4 deletions(-) diff --git a/.travis.yml b/.travis.yml index 7c67278e..7debc49e 100644 --- a/.travis.yml +++ b/.travis.yml @@ -9,5 +9,6 @@ install: - pip install coveralls script: - python setup.py test + - python setup.py flake8 after_success: - coveralls diff --git a/helpers/aws.py b/helpers/aws.py index b47e4e0f..352e3617 100644 --- a/helpers/aws.py +++ b/helpers/aws.py @@ -1,5 +1,4 @@ import logging -import re import requests from requests.exceptions import RequestException import yaml @@ -9,6 +8,7 @@ logger = logging.getLogger(__name__) class AWSConnection: + def __init__(self, config): self.available = False self.config = config @@ -43,7 +43,7 @@ class AWSConnection: if not self.available: return False - tags = {'Name': 'spilo_'+self.cluster_name, 'Role': role, 'Instance': self.instance_id} + tags = {'Name': 'spilo_' + self.cluster_name, 'Role': role, 'Instance': self.instance_id} try: conn = boto.ec2.connect_to_region(self.region) volumes = conn.get_all_volumes(filters={'attachment.instance-id': self.instance_id}) diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 560d7a98..bbf5ffe9 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -9,6 +9,7 @@ from helpers.utils import sleep if sys.hexversion >= 0x03000000: from urllib.parse import urlparse + long = int else: from urlparse import urlparse @@ -169,7 +170,8 @@ class Postgresql: try: latest_backup = subprocess.check_output(self.wal_e_path.split() + ['backup-list', '--detail', 'LATEST']) - # name last_modified expanded_size_bytes wal_segment_backup_start wal_segment_offset_backup_start wal_segment_backup_stop wal_segment_offset_backup_stop + # name last_modified expanded_size_bytes wal_segment_backup_start wal_segment_offset_backup_start + # wal_segment_backup_stop wal_segment_offset_backup_stop # base_00000001000000000000007F_00000040 2015-05-18T10:13:25.000Z # 20310671 00000001000000000000007F 00000040 # 00000001000000000000007F 00000240 diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index d6c01f5b..cb2d09a6 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -118,7 +118,7 @@ class TestPostgresql(unittest.TestCase): 'password': 'rep-pass', 'network': '127.0.0.1/32'}, 'parameters': {'foo': 'bar'}, 'recovery_conf': {'foo': 'bar'}}, - on_change_callback=lambda state: True) + on_change_callback=lambda state: True) psycopg2.connect = psycopg2_connect if not os.path.exists(self.p.data_dir): os.makedirs(self.p.data_dir) From 66f4f790d28487e68aa1d7991af77bf472a1bb9f Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 15 Jul 2015 12:49:17 +0200 Subject: [PATCH 3/5] Improve test coverage --- tests/test_postgresql.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index cb2d09a6..30041675 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -153,6 +153,12 @@ class TestPostgresql(unittest.TestCase): self.p.follow_the_leader(self.leader) self.p.follow_the_leader(self.other) + def test_create_connection_users(self): + cfg = self.p.config + cfg['superuser']['username'] = 'test' + p = Postgresql(cfg) + p.create_connection_users() + def test_create_replication_slots(self): self.p.start() cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leader]) From fcead3aed94bc19e7ab2b5246484ae2670849783 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 15 Jul 2015 15:25:20 +0200 Subject: [PATCH 4/5] Update README.md --- README.md | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 3ecbcebb..b894c5d9 100644 --- a/README.md +++ b/README.md @@ -55,8 +55,11 @@ For an example file, see `postgres0.yml`. Below is an explanation of settings: * *scope*: the relative path used on etcd's http api for this deployment, thus you can run multiple HA deployments from a single etcd * *session_timeout*: the TTL to acquire the leader lock. Think of it as the length of time before automatic failover process is initiated. * *reconnects_timeout*: how long we should try to reconnect to ZooKeeper after connection loss. After this timeout we assume that we don't have lock anymore and will restart in read-only mode. - * *hosts*: List of ZooKeeper cluster members in format: 'host1:port1,host2:port2,..etc...' - + * *hosts*: list of ZooKeeper cluster members in format: [ 'host1:port1', 'host2:port2', 'etc...'] + * *exhibitor*: if you are running ZooKeeper cluster under Exhibitor supervisory the following section could be interesting for you + * *poll_interval*: how often list of ZooKeeper and Exhibitor nodes should be updated from Exhibitor + * *port*: Exhibitor port + * *hosts*: initial list of Exhibitor (ZooKeeper) nodes in format: [ 'host1', 'host2', 'etc...' ]. This list would be updated automatically when Exhibitor (ZooKeeper) cluster topology changes. * *postgresql* * *name*: the name of the Postgres host, must be unique for the cluster From d1e47c8ccfbcb298718be3e645dfb2b3644e9aad Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 15 Jul 2015 16:17:38 +0200 Subject: [PATCH 5/5] Update README.md --- README.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/README.md b/README.md index b894c5d9..f90478ff 100644 --- a/README.md +++ b/README.md @@ -67,10 +67,17 @@ For an example file, see `postgres0.yml`. Below is an explanation of settings: * *connect_address*: ip address + port through which Postgres is accessible from other nodes and applications. * *data_dir*: file path to initialize and store Postgres data files * *maximum_lag_on_failover*: the maximum bytes a follower may lag before it is not eligible become leader + * *pg_hba*: list of lines which should be added to pg_hba.conf + * *- host all all 0.0.0.0/0 md5* * *replication* * *username*: replication username, user will be created during initialization * *password*: replication password, user will be created during initialization * *network*: network setting for replication in pg_hba.conf + * *superuser* + * *password*: password for postgres user. It would be set during initialization + * *admin*: + * *username*: admin username, user will be created during initialization. It would have CREATEDB and CREATEROLE privileges + * *password*: admin password, user will be created during initialization. * *recovery_conf*: configuration settings written to recovery.conf when configuring follower * *parameters*: list of configuration settings for Postgres