From dcad7a32294e228758f7a339cb13ca6e137aa770 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 10 Jul 2015 10:46:33 +0200 Subject: [PATCH 1/2] 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/2] 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)