From c79fba7656431f043bcab4c5434122c8273a836d Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 9 Jun 2015 09:51:53 +0200 Subject: [PATCH 01/12] Support work with etcd as a cluster In case if one member of a cluster is not available it will retry with another one and fetch the new cluster configuration. Default timeout for all requests to etcd is 5 seconds. Initial cluster configuration can be resolved through: 1) /v2/members call on one of the cluster members on a client port 2) when it is possible to resolve hostname into multiple ip's it will iterate through list and try to perform action from 1) 3) If there is discovery_srv defined in etcd section of config file it will resolve peer addresses of all cluster members and will fetch cluster configuration with using peer protocol by doing /members call on a peer port --- .travis.yml | 3 +- helpers/errors.py | 4 + helpers/etcd.py | 206 ++++++++++++++++++----- postgres0.yml | 1 + postgres1.yml | 1 + requirements.txt => requirements-py2.txt | 3 +- requirements-py3.txt | 4 + setup.py | 3 +- tests/test_etcd.py | 122 ++++++++++++-- tests/test_governor.py | 3 +- tests/test_ha.py | 2 +- tox.ini | 6 + 12 files changed, 299 insertions(+), 59 deletions(-) rename requirements.txt => requirements-py2.txt (71%) create mode 100644 requirements-py3.txt diff --git a/.travis.yml b/.travis.yml index a0659429..221ceca1 100644 --- a/.travis.yml +++ b/.travis.yml @@ -4,7 +4,8 @@ python: - "3.3" - "3.4" install: - - pip install -r requirements.txt + - [[ $TRAVIS_PYTHON_VERSION == 2* ]] && pip install -r requirements-py2.txt --use-mirrors + - [[ $TRAVIS_PYTHON_VERSION == 3* ]] && pip install -r requirements-py3.txt --use-mirrors - pip install coveralls script: - python setup.py test diff --git a/helpers/errors.py b/helpers/errors.py index b86bd0f0..3a56a1e0 100644 --- a/helpers/errors.py +++ b/helpers/errors.py @@ -9,3 +9,7 @@ class EtcdError(Exception): class CurrentLeaderError(EtcdError): pass + + +class EtcdConnectionFailed(EtcdError): + pass diff --git a/helpers/etcd.py b/helpers/etcd.py index 32345106..cc9ae95c 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -1,11 +1,14 @@ import logging import requests +import socket import sys -from requests.exceptions import RequestException from collections import namedtuple -from helpers.errors import CurrentLeaderError, EtcdError +from dns.exception import DNSException +from dns import resolver +from helpers.errors import CurrentLeaderError, EtcdError, EtcdConnectionFailed from helpers.utils import sleep +from requests.exceptions import RequestException if sys.hexversion >= 0x03000000: from urllib.parse import urlparse, urlunparse, parse_qsl @@ -35,58 +38,185 @@ class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,mem return not (self.leader and self.leader.hostname) +class Client: + + API_VERSION = 'v2' + + def __init__(self, config): + self._config = config + self.timeout = 5 + self._base_uri = None + self._members_cache = [] + self.load_members() + + def client_url(self, path): + return self._base_uri + path + + def _next_server(self): + self._base_uri = None + try: + self._base_uri = self._members_cache.pop() + except IndexError: + logger.error('Members cache is empty, can not retry.') + raise EtcdConnectionFailed('No more members in the cluster') + else: + logger.info('Selected new etcd server %s', self._base_uri) + + def _get(self, path): + response = None + while response is None: + uri = self.client_url(path) + try: + logger.info(uri) + response = requests.get(uri, timeout=self.timeout) + except RequestException: + self._next_server() + + logger.debug([response.status_code, response.content]) + try: + return response.json(), response.status_code + except (TypeError, ValueError): + raise EtcdError('Bad response from %s: %s' % (uri, response.content)) + + @staticmethod + def get_srv_record(host): + try: + return [(r.target.rstrip('.'), r.port) for r in resolver.query('_etcd-server._tcp.' + host, 'SRV')] + except DNSException: + logger.exception('Can not resolve SRV for %s', host) + return [] + + @staticmethod + def get_peers_urls_from_dns(host): + return ['http://{}:{}'.format(h, p) for h, p in Client.get_srv_record(host)] + + @staticmethod + def get_client_urls_from_dns(addr): + host, port = addr.split(':') + ret = [] + try: + for r in set(socket.getaddrinfo(host, port, socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP)): + ret.append('http://{}:{}/{}'.format(r[4][0], r[4][1], Client.API_VERSION)) + except socket.error: + logger.exception('Can not resolve %s', host) + return list(set(ret)) if ret else ['http://{}:{}/{}'.format(host, port, Client.API_VERSION)] + + def load_members(self): + load_from_srv = False + if not self._base_uri: + if 'discovery_srv' in self._config: + load_from_srv = True + self._members_cache = self.get_peers_urls_from_dns(self._config['discovery_srv']) + elif 'host' in self._config: + self._members_cache = self.get_client_urls_from_dns(self._config['host']) + else: + raise Exception('Neither discovery_srv nor host are defined in etcd section of config') + self._next_server() + + response, status_code = self._get('/members') + if status_code != 200: + raise EtcdError('Got response with code=%s from %s' % (status_code, self._base_uri)) + + members_cache = [] + for member in response if load_from_srv else response['members']: + members_cache.extend([m + '/' + self.API_VERSION for m in member['clientURLs']]) + self._members_cache = list(set(members_cache)) # TODO: randomize + if load_from_srv: + self._next_server() + else: + try: + self._members_cache.remove(self._base_uri) + except ValueError: + pass + + def get(self, path): + if not self._base_uri: + self.load_members() + old_base_uri = self._base_uri + try: + return self._get(path) + finally: + if self._base_uri != old_base_uri: + try: + self.load_members() + except EtcdError: + logger.exception('load_members') + + def put(self, path, **data): + if not self._base_uri: + self.load_members() + old_base_uri = self._base_uri + response = None + while response is None: + uri = self.client_url(path) + try: + logger.info(uri) + response = requests.put(uri, timeout=self.timeout, data=data) + except RequestException: + logger.exception('PUT %s data=%s', uri, data) + self._next_server() + + if self._base_uri != old_base_uri: + try: + self.load_members() + except EtcdError: + logger.exception('load_members') + return response.status_code in [200, 201, 202, 204] + + def delete(self, path): + if not self._base_uri: + self.load_members() + old_base_uri = self._base_uri + response = None + while response is None: + uri = self.client_url(path) + try: + response = requests.delete(uri, timeout=self.timeout) + except RequestException: + logger.exception('DELETE %s', uri) + self._next_server() + + if self._base_uri != old_base_uri: + try: + self.load_members() + except EtcdError: + logger.exception('load_members') + return response.status_code in [200, 202, 204] + + class Etcd: def __init__(self, config): self.ttl = config['ttl'] self.member_ttl = config.get('member_ttl', 3600) - self.base_client_url = 'http://{host}/v2/keys/service/{scope}'.format(**config) - self.postgres_cluster = None + self._base_path = '/keys/service/' + config['scope'] + self.client = self.get_etcd_client(config) - def get_client_path(self, path, max_attempts=1): - attempts = 0 - response = None - - while True: - ex = None + def get_etcd_client(self, config): + client = None + while not client: try: - response = requests.get(self.client_url(path)) - if response.status_code == 200: - break - except RequestException as e: - logger.exception('get_client_path') - ex = e + client = Client(config) + except EtcdError: + logger.info('waiting on etcd') + sleep(5) + return client - attempts += 1 - if attempts < max_attempts: - logger.info('Failed to return %s, trying again. (%s of %s)', path, attempts, max_attempts) - sleep(3) - elif ex: - raise ex - else: - break + def client_path(self, path): + return self._base_path + path - return response.json(), response.status_code + def get_client_path(self, path): + return self.client.get(self.client_path(path)) def put_client_path(self, path, **data): - try: - response = requests.put(self.client_url(path), data=data) - return response.status_code in [200, 201, 202, 204] - except RequestException: - logger.exception('PUT %s data=%s', path, data) - raise EtcdError('Etcd is not responding properly') + return self.client.put(self.client_path(path), **data) def delete_client_path(self, path): try: - response = requests.delete(self.client_url(path)) - return response.status_code in [200, 202, 204] - except RequestException: - logger.exception('DELETE %s', path) + return self.client.delete(self.client_path(path)) + except EtcdConnectionFailed: return False - def client_url(self, path): - return self.base_client_url + path - @staticmethod def find_node(node, key): """ diff --git a/postgres0.yml b/postgres0.yml index 9a8e6212..0b5910f9 100644 --- a/postgres0.yml +++ b/postgres0.yml @@ -6,6 +6,7 @@ etcd: scope: batman ttl: 30 host: 127.0.0.1:4001 + #discovery_srv: my-etcd.domain postgresql: name: postgresql0 listen: 127.0.0.1:5432 diff --git a/postgres1.yml b/postgres1.yml index 2f5f2d0a..5d5687d3 100644 --- a/postgres1.yml +++ b/postgres1.yml @@ -6,6 +6,7 @@ etcd: scope: batman ttl: 30 host: 127.0.0.1:4001 + #discovery_srv: my-etcd.domain postgresql: name: postgresql1 listen: 127.0.0.1:5433 diff --git a/requirements.txt b/requirements-py2.txt similarity index 71% rename from requirements.txt rename to requirements-py2.txt index 2185e5c6..436f88ab 100644 --- a/requirements.txt +++ b/requirements-py2.txt @@ -1,3 +1,4 @@ -PyYAML +dnspython psycopg2 +PyYAML requests diff --git a/requirements-py3.txt b/requirements-py3.txt new file mode 100644 index 00000000..13f53d0e --- /dev/null +++ b/requirements-py3.txt @@ -0,0 +1,4 @@ +dnspython3 +psycopg2 +PyYAML +requests diff --git a/setup.py b/setup.py index cafa4c23..fa1d4ae5 100644 --- a/setup.py +++ b/setup.py @@ -102,7 +102,8 @@ def setup_package(): # Some helper variables version = os.getenv('GO_PIPELINE_LABEL', VERSION) - install_reqs = get_install_requirements('requirements.txt') + requirements = 'requirements-py2.txt' if sys.version_info[0] == 2 else 'requirements-py3.txt' + install_reqs = get_install_requirements(requirements) command_options = {'test': {'test_suite': ('setup.py', 'tests')}} if JUNIT_XML: diff --git a/tests/test_etcd.py b/tests/test_etcd.py index a7ab93ee..cdb5cc0a 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -1,10 +1,13 @@ -import unittest -import requests -import time +import dns.resolver import json +import requests +import socket +import time +import unittest -from helpers.etcd import Cluster, Etcd -from helpers.errors import EtcdError, CurrentLeaderError +from dns.exception import DNSException +from helpers.errors import EtcdError, CurrentLeaderError, EtcdConnectionFailed +from helpers.etcd import Client, Cluster, Etcd class MockResponse: @@ -25,10 +28,19 @@ class MockPostgresql: def requests_get(url, **kwargs): - if url.startswith('http://local'): - raise requests.exceptions.RequestException() + members = '[{"id":14855829450254237642,"peerURLs":["http://localhost:2380","http://localhost:7001"],"name":"default","clientURLs":["http://localhost:2379","http://localhost:4001"]}]' response = MockResponse() - if url.startswith('http://remote') or url.startswith('http://127.0.0.1'): + if url.endswith('/v2/members'): + response.content = '{"members": ' + members + '}' + if url.startswith('http://error'): + response.status_code = 404 + elif url.endswith('/members'): + response.content = members + elif url.endswith('/bad_response'): + response.content = '{' + 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'): response.content = '{"action":"get","node":{"key":"/service/batman5","dir":true,"nodes":[{"key":"/service/batman5/initialize","value":"postgresql0","modifiedIndex":1582,"createdIndex":1582},{"key":"/service/batman5/leader","value":"postgresql1","expiration":"2015-05-15T09:11:00.037397538Z","ttl":21,"modifiedIndex":20728,"createdIndex":20434},{"key":"/service/batman5/optime","dir":true,"nodes":[{"key":"/service/batman5/optime/leader","value":"2164261704","modifiedIndex":20729,"createdIndex":20729}],"modifiedIndex":20437,"createdIndex":20437},{"key":"/service/batman5/members","dir":true,"nodes":[{"key":"/service/batman5/members/postgresql1","value":"postgres://replicator:rep-pass@127.0.0.1:5434/postgres?application_name=http://127.0.0.1:8009/governor","expiration":"2015-05-15T09:10:59.949384522Z","ttl":21,"modifiedIndex":20727,"createdIndex":20727},{"key":"/service/batman5/members/postgresql0","value":"postgres://replicator:rep-pass@127.0.0.1:5433/postgres?application_name=http://127.0.0.1:8008/governor","expiration":"2015-05-15T09:11:09.611860899Z","ttl":30,"modifiedIndex":20730,"createdIndex":20730}],"modifiedIndex":1581,"createdIndex":1581}],"modifiedIndex":1581,"createdIndex":1581}}' elif url.startswith('http://other'): response.status_code = 404 @@ -47,7 +59,7 @@ def requests_put(url, **kwargs): return response -def requests_delete(url): +def requests_delete(url, **kwargs): if url.startswith('http://local'): raise requests.exceptions.RequestException() response = MockResponse() @@ -59,6 +71,81 @@ def time_sleep(_): pass +def time_sleep_exception(_): + raise Exception() + + +class MockSRV: + port = 2380 + target = '127.0.0.1' + + +def dns_query(name, type): + if name == '_etcd-server._tcp.blabla': + return [] + elif name == '_etcd-server._tcp.exception': + raise DNSException() + return [MockSRV()] + + +def socket_getaddrinfo(*args): + if args[0] == 'ok': + return [(2, 1, 6, '', ('127.0.0.1', 2379)), (2, 1, 6, '', ('127.0.0.1', 2379))] + raise socket.error() + + +class TestClient(unittest.TestCase): + + def __init__(self, method_name='runTest'): + self.setUp = self.set_up + super(TestClient, self).__init__(method_name) + + def set_up(self): + socket.getaddrinfo = socket_getaddrinfo + requests.get = requests_get + requests.put = requests_put + requests.delete = requests_delete + dns.resolver.query = dns_query + self.client = Client({'discovery_srv': 'test'}) + + def test__get(self): + self.assertRaises(EtcdError, self.client._get, '/bad_response') + + def test_get_srv_record(self): + self.assertEquals(Client.get_srv_record('blabla'), []) + self.assertEquals(Client.get_srv_record('exception'), []) + + def test_get_client_urls_from_dns(self): + self.assertEquals(Client.get_client_urls_from_dns('ok:2379'), ['http://127.0.0.1:2379/v2']) + + def test_load_members(self): + self.client._base_uri = self.client._base_uri.replace('localhost', 'error_code') + self.assertRaises(EtcdError, self.client.load_members) + self.client._base_uri = None + self.client._config = {} + self.assertRaises(Exception, self.client.load_members) + + def test_get(self): + self.client._base_uri = None + self.assertRaises(EtcdConnectionFailed, self.client.get, '') + self.client._members_cache = ['http://error_code:4001/v2'] + self.client.get('') + + def test_put(self): + self.client._base_uri = None + self.assertRaises(EtcdConnectionFailed, self.client.put, '') + self.client._base_uri = 'http://localhost:4001/v2' + self.client._members_cache = ['http://error_code:4001/v2'] + self.client.put('') + + def test_delete(self): + self.client._base_uri = None + self.assertRaises(EtcdConnectionFailed, self.client.delete, '') + self.client._base_uri = 'http://localhost:4001/v2' + self.client._members_cache = ['http://error_code:4001/v2'] + self.client.delete('') + + class TestEtcd(unittest.TestCase): def __init__(self, method_name='runTest'): @@ -66,11 +153,16 @@ class TestEtcd(unittest.TestCase): super(TestEtcd, self).__init__(method_name) def set_up(self): + socket.getaddrinfo = socket_getaddrinfo requests.get = requests_get requests.put = requests_put requests.delete = requests_delete time.sleep = time_sleep - self.etcd = Etcd({'ttl': 30, 'host': 'localhost', 'scope': 'test'}) + self.etcd = Etcd({'ttl': 30, 'host': 'localhost:2379', 'scope': 'test'}) + + def test_get_etcd_client(self): + time.sleep = time_sleep_exception + self.assertRaises(Exception, self.etcd.get_etcd_client, {'host': 'error:2379'}) def test_get_client_path(self): self.assertRaises(Exception, self.etcd.get_client_path, '', 2) @@ -83,12 +175,12 @@ class TestEtcd(unittest.TestCase): def test_get_cluster(self): self.assertRaises(EtcdError, self.etcd.get_cluster) - self.etcd.base_client_url = self.etcd.base_client_url.replace('local', 'remote') + self.etcd.client._base_uri = self.etcd.client._base_uri.replace('local', 'remote') cluster = self.etcd.get_cluster() self.assertIsInstance(cluster, Cluster) - self.etcd.base_client_url = self.etcd.base_client_url.replace('remote', 'other') + self.etcd.client._base_uri = self.etcd.client._base_uri.replace('remote', 'other') self.etcd.get_cluster() - self.etcd.base_client_url = self.etcd.base_client_url.replace('other', 'noleader') + self.etcd.client._base_uri = self.etcd.client._base_uri.replace('other', 'noleader') self.etcd.get_cluster() def test_current_leader(self): @@ -104,9 +196,9 @@ class TestEtcd(unittest.TestCase): self.assertFalse(self.etcd.attempt_to_acquire_leader('')) def test_update_leader(self): - self.etcd.base_client_url = self.etcd.base_client_url.replace('local', 'remote') + url = self.etcd.client._base_uri = self.etcd.client._base_uri.replace('local', 'remote') self.assertTrue(self.etcd.update_leader(MockPostgresql())) - self.etcd.base_client_url = self.etcd.base_client_url.replace('remote', 'other') + self.etcd.client._base_uri = url.replace('remote', 'other') self.assertFalse(self.etcd.update_leader(MockPostgresql())) def test_race(self): diff --git a/tests/test_governor.py b/tests/test_governor.py index 17db3421..9ecb88f1 100644 --- a/tests/test_governor.py +++ b/tests/test_governor.py @@ -68,8 +68,7 @@ class TestGovernor(unittest.TestCase): return True def test_governor_initialize(self): - self.g.etcd.base_client_url = 'http://remote' - self.g.etcd.client_url + self.g.etcd.client._base_uri = 'http://remote' self.g.postgresql.data_directory_empty = true self.g.etcd.race = true self.g.initialize() diff --git a/tests/test_ha.py b/tests/test_ha.py index 59495b95..574f949f 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -71,7 +71,7 @@ class TestHa(unittest.TestCase): requests.put = requests_put requests.delete = requests_delete self.p = MockPostgresql() - self.e = Etcd({'ttl': 30, 'host': 'remotehost', 'scope': 'test'}) + self.e = Etcd({'ttl': 30, 'host': 'remotehost:2379', 'scope': 'test'}) self.ha = Ha(self.p, self.e) self.ha.load_cluster_from_etcd() self.ha.cluster = Cluster(False, None, None, []) diff --git a/tox.ini b/tox.ini index aa079ec5..bf407bae 100644 --- a/tox.ini +++ b/tox.ini @@ -1,2 +1,8 @@ [flake8] max-line-length=120 + +[testenv:py27] +deps = -rrequirements-py2.txt + +[testenv:py33] +deps = -rrequirements-py3.txt From 5b7a60657f9505b2d7480ea905b32af204065def Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 9 Jun 2015 10:04:07 +0200 Subject: [PATCH 02/12] Fix syntax in .travis.yml --- .travis.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.travis.yml b/.travis.yml index 221ceca1..7c67278e 100644 --- a/.travis.yml +++ b/.travis.yml @@ -4,8 +4,8 @@ python: - "3.3" - "3.4" install: - - [[ $TRAVIS_PYTHON_VERSION == 2* ]] && pip install -r requirements-py2.txt --use-mirrors - - [[ $TRAVIS_PYTHON_VERSION == 3* ]] && pip install -r requirements-py3.txt --use-mirrors + - if [[ $TRAVIS_PYTHON_VERSION == 2* ]]; then pip install -r requirements-py2.txt --use-mirrors; fi + - if [[ $TRAVIS_PYTHON_VERSION == 3* ]]; then pip install -r requirements-py3.txt --use-mirrors; fi - pip install coveralls script: - python setup.py test From da4f9e8445322f161def57fb4b7deaddbe1988ca Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 9 Jun 2015 10:25:33 +0200 Subject: [PATCH 03/12] Make logging of http requests more verbose --- governor.py | 1 + helpers/etcd.py | 5 +++-- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/governor.py b/governor.py index 5267d07c..fe2ecd8a 100755 --- a/governor.py +++ b/governor.py @@ -74,6 +74,7 @@ class Governor: def main(): logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO) + logging.getLogger('requests').setLevel(logging.WARNING) setup_signal_handlers() if len(sys.argv) < 2 or not os.path.isfile(sys.argv[1]): diff --git a/helpers/etcd.py b/helpers/etcd.py index cc9ae95c..72c0995e 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -67,7 +67,7 @@ class Client: while response is None: uri = self.client_url(path) try: - logger.info(uri) + logger.info('GET %s', uri) response = requests.get(uri, timeout=self.timeout) except RequestException: self._next_server() @@ -150,7 +150,7 @@ class Client: while response is None: uri = self.client_url(path) try: - logger.info(uri) + logger.info('PUT %s', uri) response = requests.put(uri, timeout=self.timeout, data=data) except RequestException: logger.exception('PUT %s data=%s', uri, data) @@ -171,6 +171,7 @@ class Client: while response is None: uri = self.client_url(path) try: + logger.info('DELETE %s', uri) response = requests.delete(uri, timeout=self.timeout) except RequestException: logger.exception('DELETE %s', uri) From 261fbf272131081fd18fe2e87bae76e348261eb4 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 9 Jun 2015 10:27:09 +0200 Subject: [PATCH 04/12] Replace logging with logger in ha.py --- helpers/ha.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/helpers/ha.py b/helpers/ha.py index 771e0bfd..9617228f 100644 --- a/helpers/ha.py +++ b/helpers/ha.py @@ -42,7 +42,7 @@ class Ha: self.state_handler.start() if not has_lock: return 'started as a secondary' - logging.info('started as readonly because i had the session lock') + logger.info('started as readonly because i had the session lock') self.load_cluster_from_etcd() if self.cluster.is_unlocked(): From a2988361c5699edfc22ca8567fcfb7350db9441f Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 9 Jun 2015 12:30:16 +0200 Subject: [PATCH 05/12] Retry with host from config when it failed to resolve SRV --- helpers/etcd.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/helpers/etcd.py b/helpers/etcd.py index 72c0995e..c579d481 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -104,13 +104,17 @@ class Client: def load_members(self): load_from_srv = False if not self._base_uri: + if not 'discovery_srv' in self._config and not 'host' in self._config: + raise Exception('Neither discovery_srv nor host are defined in etcd section of config') + if 'discovery_srv' in self._config: load_from_srv = True self._members_cache = self.get_peers_urls_from_dns(self._config['discovery_srv']) - elif 'host' in self._config: + + if not self._members_cache and 'host' in self._config: + load_from_srv = False self._members_cache = self.get_client_urls_from_dns(self._config['host']) - else: - raise Exception('Neither discovery_srv nor host are defined in etcd section of config') + self._next_server() response, status_code = self._get('/members') From 57914821d0193fe402234bee28cb6a6bbc6d822e Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 9 Jun 2015 12:30:58 +0200 Subject: [PATCH 06/12] Randomize list of etcd cluster members --- helpers/etcd.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/helpers/etcd.py b/helpers/etcd.py index c579d481..4fb39f47 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -1,4 +1,5 @@ import logging +import random import requests import socket import sys @@ -124,7 +125,8 @@ class Client: members_cache = [] for member in response if load_from_srv else response['members']: members_cache.extend([m + '/' + self.API_VERSION for m in member['clientURLs']]) - self._members_cache = list(set(members_cache)) # TODO: randomize + self._members_cache = list(set(members_cache)) + random.shuffle(self._members_cache) if load_from_srv: self._next_server() else: From a64dad267e64ee7cccc3b19243a57a11f4c24a5b Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 9 Jun 2015 15:46:34 +0200 Subject: [PATCH 07/12] Bugfix: target field of SRV object should be casted to string --- helpers/etcd.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/helpers/etcd.py b/helpers/etcd.py index 4fb39f47..2479822d 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -82,7 +82,7 @@ class Client: @staticmethod def get_srv_record(host): try: - return [(r.target.rstrip('.'), r.port) for r in resolver.query('_etcd-server._tcp.' + host, 'SRV')] + return [(str(r.target).rstrip('.'), r.port) for r in resolver.query('_etcd-server._tcp.' + host, 'SRV')] except DNSException: logger.exception('Can not resolve SRV for %s', host) return [] From 971bc3b6e8297f54c0c1fbac0c1d61be836f1abe Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 10 Jun 2015 11:56:16 +0200 Subject: [PATCH 08/12] Increase verbosity of failed PUT and DELETE requests. Hope will help to debug some etcd related issues --- helpers/etcd.py | 12 +++++++++--- tests/test_etcd.py | 2 +- 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/helpers/etcd.py b/helpers/etcd.py index 2479822d..143a8ff1 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -105,7 +105,7 @@ class Client: def load_members(self): load_from_srv = False if not self._base_uri: - if not 'discovery_srv' in self._config and not 'host' in self._config: + if 'discovery_srv' not in self._config and 'host' not in self._config: raise Exception('Neither discovery_srv nor host are defined in etcd section of config') if 'discovery_srv' in self._config: @@ -167,7 +167,10 @@ class Client: self.load_members() except EtcdError: logger.exception('load_members') - return response.status_code in [200, 201, 202, 204] + if response.status_code in [200, 201, 202, 204]: + return True + logger.error('Unexpected response: %s %s', response.status_code, response.content) + return False def delete(self, path): if not self._base_uri: @@ -188,7 +191,10 @@ class Client: self.load_members() except EtcdError: logger.exception('load_members') - return response.status_code in [200, 202, 204] + if response.status_code in [200, 202, 204]: + return True + logger.error('Unexpected response: %s %s', response.status_code, response.content) + return False class Etcd: diff --git a/tests/test_etcd.py b/tests/test_etcd.py index cdb5cc0a..c2e07c55 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -63,7 +63,7 @@ def requests_delete(url, **kwargs): if url.startswith('http://local'): raise requests.exceptions.RequestException() response = MockResponse() - response.status_code = 204 + response.status_code = 503 if url.startswith('http://error') else 204 return response From 275e18e58a3b523b5276f48a34bca2e485d73781 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 10 Jun 2015 14:21:23 +0200 Subject: [PATCH 09/12] Refactor tests --- tests/test_postgresql.py | 15 ++++++--------- 1 file changed, 6 insertions(+), 9 deletions(-) diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py index 9eba294b..7fae838a 100644 --- a/tests/test_postgresql.py +++ b/tests/test_postgresql.py @@ -101,7 +101,9 @@ class TestPostgresql(unittest.TestCase): psycopg2.connect = psycopg2_connect if not os.path.exists(self.p.data_dir): os.makedirs(self.p.data_dir) - self.leader = Member('leader', 'postgres://replicator:rep-pass@127.0.0.1:5434/postgres', None, 28) + self.leader = Member('leader', 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres', None, None, 28) + self.other = Member('test1', 'postgres://replicator:rep-pass@127.0.0.1:5433/postgres', None, None, 28) + self.me = Member('test0', 'postgres://replicator:rep-pass@127.0.0.1:5434/postgres', None, None, 28) def tear_down(self): shutil.rmtree('data') @@ -128,13 +130,11 @@ class TestPostgresql(unittest.TestCase): self.p.follow_the_leader(None) self.p.demote(self.leader) self.p.follow_the_leader(self.leader) - self.p.follow_the_leader(Member('leader', 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres', None, 28)) + self.p.follow_the_leader(self.other) def test_create_replication_slots(self): self.p.start() - me = Member('test0', 'postgres://replicator:rep-pass@127.0.0.1:5434/postgres', None, 28) - other = Member('test1', 'postgres://replicator:rep-pass@127.0.0.1:5433/postgres', None, 28) - cluster = Cluster(True, self.leader, 0, [me, other, self.leader]) + cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leader]) self.p.create_replication_slots(cluster) def test_query(self): @@ -148,10 +148,7 @@ class TestPostgresql(unittest.TestCase): self.assertRaises(psycopg2.OperationalError, self.p.query, 'blabla') def test_is_healthiest_node(self): - leader = Member('leader', 'postgres://replicator:rep-pass@127.0.0.1:5435/postgres', None, 28) - me = Member('test0', 'postgres://replicator:rep-pass@127.0.0.1:5434/postgres', None, 28) - other = Member('test1', 'postgres://replicator:rep-pass@127.0.0.1:5433/postgres', None, 28) - cluster = Cluster(True, leader, 0, [me, other, leader]) + cluster = Cluster(True, self.leader, 0, [self.me, self.other, self.leader]) self.assertTrue(self.p.is_healthiest_node(cluster)) self.p.is_leader = false self.assertFalse(self.p.is_healthiest_node(cluster)) From 44e06b4cffa61826461e46fb017f6ccaa07f9c37 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 10 Jun 2015 14:22:11 +0200 Subject: [PATCH 10/12] Add expiration field into Member --- helpers/etcd.py | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) diff --git a/helpers/etcd.py b/helpers/etcd.py index 143a8ff1..3a222a3b 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -19,18 +19,16 @@ else: logger = logging.getLogger(__name__) -class Member(namedtuple('Member', 'hostname,conn_url,api_url,ttl')): +class Member(namedtuple('Member', 'hostname,conn_url,api_url,expiration,ttl')): @staticmethod def fromNode(node): scheme, netloc, path, params, query, fragment = urlparse(node['value']) conn_url = urlunparse((scheme, netloc, path, params, '', fragment)) - api_url = None - for name, value in parse_qsl(query): - if name == 'application_name' and value: - api_url = value - break - return Member(node['key'].split('/')[-1], conn_url, api_url, node.get('ttl', None)) + api_url = ([v for n, v in parse_qsl(query) if n == 'application_name'] or [None])[0] + expiration = node.get('expiration', None) + ttl = node.get('ttl', None) + return Member(node['key'].split('/')[-1], conn_url, api_url, expiration, ttl) class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,members')): @@ -271,7 +269,7 @@ class Etcd: leader = m break if not leader: - leader = Member(node['value'], None, None, None) + leader = Member(node['value'], None, None, None, None) return Cluster(initialize, leader, last_leader_operation, members) elif status_code == 404: From 744026b4bb44eaeb93f7ba950c42ef9aff8f5b0b Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 23 Jun 2015 14:02:32 +0200 Subject: [PATCH 11/12] Do not update member TTL when it is far from being expired --- governor.py | 8 +++++++- helpers/etcd.py | 5 ++++- helpers/utils.py | 29 +++++++++++++++++++++++++++++ tests/test_etcd.py | 15 ++++++++++++++- tests/test_governor.py | 9 +++++++++ 5 files changed, 63 insertions(+), 3 deletions(-) diff --git a/governor.py b/governor.py index fe2ecd8a..ba48fefd 100755 --- a/governor.py +++ b/governor.py @@ -22,9 +22,15 @@ class Governor: host, port = config['restapi']['listen'].split(':') self.api = RestApiServer(self, config['restapi']) self.next_run = time.time() + self.shutdown_member_ttl = 300 def touch_member(self, ttl=None): connection_string = self.postgresql.connection_string + '?application_name=' + self.api.connection_string + if self.ha.cluster: + for m in self.ha.cluster.members: + # Do not update member TTL when it is far from being expired + if m.hostname == self.postgresql.name and m.real_ttl() > self.shutdown_member_ttl: + return True return self.etcd.touch_member(self.postgresql.name, connection_string, ttl) def initialize(self): @@ -91,7 +97,7 @@ def main(): except KeyboardInterrupt: pass finally: - governor.touch_member(300) # schedule member removal + governor.touch_member(governor.shutdown_member_ttl) # schedule member removal governor.postgresql.stop() governor.etcd.delete_leader(governor.postgresql.name) diff --git a/helpers/etcd.py b/helpers/etcd.py index 3a222a3b..f63cc364 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -8,7 +8,7 @@ from collections import namedtuple from dns.exception import DNSException from dns import resolver from helpers.errors import CurrentLeaderError, EtcdError, EtcdConnectionFailed -from helpers.utils import sleep +from helpers.utils import calculate_ttl, sleep from requests.exceptions import RequestException if sys.hexversion >= 0x03000000: @@ -30,6 +30,9 @@ class Member(namedtuple('Member', 'hostname,conn_url,api_url,expiration,ttl')): ttl = node.get('ttl', None) return Member(node['key'].split('/')[-1], conn_url, api_url, expiration, ttl) + def real_ttl(self): + return calculate_ttl(self.expiration) + class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,members')): diff --git a/helpers/utils.py b/helpers/utils.py index e08dd657..c12f389a 100644 --- a/helpers/utils.py +++ b/helpers/utils.py @@ -1,10 +1,39 @@ +import datetime import os +import re import signal import sys import time received_sigchld = False +_DATE_TIME_RE = re.compile(r'''^ +(?P\d{4})\-(?P\d{2})\-(?P\d{2}) # date +T +(?P\d{2}):(?P\d{2}):(?P\d{2})\.(?P\d{6}) # time +\d*Z$''', re.X) + + +def parse_datetime(time_str): + """ + >>> parse_datetime('2015-06-10T12:56:30.552539016Z') + datetime.datetime(2015, 6, 10, 12, 56, 30, 552539) + >>> parse_datetime('2015-06-10 12:56:30.552539016Z') + """ + m = _DATE_TIME_RE.match(time_str) + if not m: + return None + p = dict((n, int(m.group(n))) for n in 'year month day hour minute second microsecond'.split(' ')) + return datetime.datetime(**p) + + +def calculate_ttl(expiration): + expiration = parse_datetime(expiration) + if not expiration: + return None + now = datetime.datetime.utcnow() + return int((expiration - now).total_seconds()) + def lsn_to_bytes(value): """ diff --git a/tests/test_etcd.py b/tests/test_etcd.py index c2e07c55..b9e88df7 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -1,3 +1,4 @@ +import datetime import dns.resolver import json import requests @@ -7,7 +8,7 @@ import unittest from dns.exception import DNSException from helpers.errors import EtcdError, CurrentLeaderError, EtcdConnectionFailed -from helpers.etcd import Client, Cluster, Etcd +from helpers.etcd import Client, Cluster, Etcd, Member class MockResponse: @@ -94,6 +95,18 @@ def socket_getaddrinfo(*args): raise socket.error() +class TestMember(unittest.TestCase): + + def __init__(self, method_name='runTest'): + super(TestMember, self).__init__(method_name) + + def test_real_ttl(self): + now = datetime.datetime.utcnow() + member = Member('a', 'b', 'c', (now + datetime.timedelta(seconds=2)).strftime('%Y-%m-%dT%H:%M:%S.%fZ'), None) + self.assertLess(member.real_ttl(), 2) + self.assertIsNone(Member('a', 'b', 'c', '', None).real_ttl()) + + class TestClient(unittest.TestCase): def __init__(self, method_name='runTest'): diff --git a/tests/test_governor.py b/tests/test_governor.py index 9ecb88f1..b84637a7 100644 --- a/tests/test_governor.py +++ b/tests/test_governor.py @@ -1,3 +1,4 @@ +import datetime import psycopg2 import requests import subprocess @@ -7,6 +8,7 @@ import unittest import yaml from governor import Governor, main +from helpers.etcd import Cluster, Member from test_ha import true, false from test_postgresql import Postgresql, subprocess_call, psycopg2_connect from test_etcd import requests_get, requests_put, requests_delete @@ -67,6 +69,13 @@ class TestGovernor(unittest.TestCase): return False return True + def test_touch_member(self): + now = datetime.datetime.utcnow() + member = Member(self.g.postgresql.name, 'b', 'c', (now + datetime.timedelta( + seconds=self.g.shutdown_member_ttl + 10)).strftime('%Y-%m-%dT%H:%M:%S.%fZ'), None) + self.g.ha.cluster = Cluster(True, member, 0, [member]) + self.g.touch_member() + def test_governor_initialize(self): self.g.etcd.client._base_uri = 'http://remote' self.g.postgresql.data_directory_empty = true From b7c73fdba8d2de0113c3a5aa8021d18ea925782a Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 23 Jun 2015 14:07:00 +0200 Subject: [PATCH 12/12] Rename hostname to name in a Member --- governor.py | 2 +- helpers/etcd.py | 6 +++--- helpers/ha.py | 2 +- helpers/postgresql.py | 6 +++--- 4 files changed, 8 insertions(+), 8 deletions(-) diff --git a/governor.py b/governor.py index ba48fefd..d8a1e620 100755 --- a/governor.py +++ b/governor.py @@ -29,7 +29,7 @@ class Governor: if self.ha.cluster: for m in self.ha.cluster.members: # Do not update member TTL when it is far from being expired - if m.hostname == self.postgresql.name and m.real_ttl() > self.shutdown_member_ttl: + if m.name == self.postgresql.name and m.real_ttl() > self.shutdown_member_ttl: return True return self.etcd.touch_member(self.postgresql.name, connection_string, ttl) diff --git a/helpers/etcd.py b/helpers/etcd.py index f63cc364..0a1f7cb7 100644 --- a/helpers/etcd.py +++ b/helpers/etcd.py @@ -19,7 +19,7 @@ else: logger = logging.getLogger(__name__) -class Member(namedtuple('Member', 'hostname,conn_url,api_url,expiration,ttl')): +class Member(namedtuple('Member', 'name,conn_url,api_url,expiration,ttl')): @staticmethod def fromNode(node): @@ -37,7 +37,7 @@ class Member(namedtuple('Member', 'hostname,conn_url,api_url,expiration,ttl')): class Cluster(namedtuple('Cluster', 'initialize,leader,last_leader_operation,members')): def is_unlocked(self): - return not (self.leader and self.leader.hostname) + return not (self.leader and self.leader.name) class Client: @@ -268,7 +268,7 @@ class Etcd: node = self.find_node(response['node'], '/leader') if node: for m in members: - if m.hostname == node['value']: + if m.name == node['value']: leader = m break if not leader: diff --git a/helpers/ha.py b/helpers/ha.py index 9617228f..0453686e 100644 --- a/helpers/ha.py +++ b/helpers/ha.py @@ -23,7 +23,7 @@ class Ha: return self.etcd.update_leader(self.state_handler) def has_lock(self): - lock_owner = self.cluster.leader and self.cluster.leader.hostname + lock_owner = self.cluster.leader and self.cluster.leader.name logger.info('Lock owner: %s; I am %s', lock_owner, self.state_handler.name) return lock_owner == self.state_handler.name diff --git a/helpers/postgresql.py b/helpers/postgresql.py index 3d2cd183..861074e8 100644 --- a/helpers/postgresql.py +++ b/helpers/postgresql.py @@ -182,7 +182,7 @@ class Postgresql: return False for member in cluster.members: - if member.hostname == self.name: + if member.name == self.name: continue try: r = parseurl(member.conn_url) @@ -195,7 +195,7 @@ class Postgresql: row = member_cursor.fetchone() member_cursor.close() member_conn.close() - logger.error([self.name, member.hostname, row]) + logger.error([self.name, member.name, row]) if not row[0] or row[1] < 0: return False except psycopg2.Error: @@ -269,7 +269,7 @@ primary_conninfo = '{}' self.members = [r[0] for r in cursor] def create_replication_slots(self, cluster): - members = [m.hostname for m in cluster.members if m.hostname != self.name] + members = [m.name for m in cluster.members if m.name != self.name] # drop unused slots for slot in set(self.members) - set(members): self.query("""SELECT pg_drop_replication_slot(%s)