From 01afd09ca27f89deac9aab737738cae402377d23 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Sat, 12 Mar 2016 15:49:42 +0100 Subject: [PATCH] Migrate to python-etcd 0.4.3 Despite this release was very buggy it has really nice features: * EtcdWatchTimedOut exception is raised when `watch` call timed out * it supports SRV autodiscovery Since we already implemented our own SRV discovery this feature is not really interesting for us, but it solves the problem of having two requirements files for different python versions, because python-etcd will install dnspython or dnspython3 as a dependency. In order to fix https://github.com/jplana/python-etcd/issues/152 and https://github.com/jplana/python-etcd/pull/154 I had to override `api_execute` method. --- .travis.yml | 3 +- patroni/etcd.py | 62 +++++++++++++++++++----- requirements-py2.txt | 13 ----- requirements-py3.txt => requirements.txt | 0 setup.py | 11 ++--- tests/test_etcd.py | 26 ++++------ 6 files changed, 65 insertions(+), 50 deletions(-) delete mode 100644 requirements-py2.txt rename requirements-py3.txt => requirements.txt (100%) diff --git a/.travis.yml b/.travis.yml index 23c69358..ba4733d3 100644 --- a/.travis.yml +++ b/.travis.yml @@ -4,8 +4,7 @@ python: - "3.4" - "3.5" install: - - 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; fi + - pip install -r requirements.txt - pip install coveralls codacy-coverage script: - python setup.py test diff --git a/patroni/etcd.py b/patroni/etcd.py index 79842ca4..31c0fd37 100644 --- a/patroni/etcd.py +++ b/patroni/etcd.py @@ -14,6 +14,7 @@ from patroni.dcs import AbstractDCS, Cluster, Failover, Leader, Member from patroni.exceptions import DCSError from patroni.utils import Retry, RetryFailedError, sleep from requests.exceptions import RequestException +from six.moves.http_client import HTTPException logger = logging.getLogger(__name__) @@ -49,12 +50,57 @@ class Client(etcd.Client): self._update_machines_cache = True return [self._base_uri] - def api_execute(self, path, method, **kwargs): + def _do_http_request(self, request_executor, method, url, fields=None, **kwargs): + try: + response = request_executor(method, url, fields=fields, **kwargs) + response.data.decode('utf-8') +# self._check_cluster_id(response) + except (urllib3.exceptions.HTTPError, HTTPException, socket.error) as e: + if (isinstance(fields, dict) and fields.get("wait") == "true" and + isinstance(e, urllib3.exceptions.ReadTimeoutError)): + logger.debug("Watch timed out.") + raise etcd.EtcdWatchTimedOut("Watch timed out: {0}".format(e), cause=e) + logger.error("Request to server %s failed: %r", self._base_uri, e) + logger.info("Reconnection allowed, looking for another server.") + self._base_uri = self._next_server(cause=e) + response = False + return response + + def api_execute(self, path, method, params=None, timeout=None): + if not path.startswith('/'): + raise ValueError('Path does not start with /') + + if timeout is None: + timeout = self.read_timeout + + if timeout == 0: + timeout = None + + kwargs = {'timeout': timeout, 'fields': params, 'redirect': self.allow_redirect, + 'headers': self._get_headers(), 'preload_content': False} + + if method in [self._MGET, self._MDELETE]: + request_executor = self.http.request + elif method in [self._MPUT, self._MPOST]: + request_executor = self.http.request_encode_body + kwargs['encode_multipart'] = False + else: + raise etcd.EtcdException('HTTP method {} not supported'.format(method)) + # Update machines_cache if previous attempt of update has failed if self._update_machines_cache: self._load_machines_cache() + + response = False + try: - return super(Client, self).api_execute(path, method, **kwargs) + while not response: + response = self._do_http_request(request_executor, method, self._base_uri + path, **kwargs) + + if response is False and not self._use_proxies: + self._machines_cache = self.machines + self._machines_cache.remove(self._base_uri) + return self._handle_server_response(response) except etcd.EtcdConnectionFailed: self._update_machines_cache = True raise @@ -67,16 +113,6 @@ class Client(etcd.Client): logger.exception('Can not resolve SRV for %s', host) return [] - # try to workarond bug in python-etcd: https://github.com/jplana/python-etcd/issues/81 - def _result_from_response(self, response): - try: - response.data.decode('utf-8') - except urllib3.exceptions.TimeoutError: - raise - except Exception as e: - raise etcd.EtcdException('Unable to decode server response: {0}'.format(e)) - return super(Client, self)._result_from_response(response) - def _get_machines_cache_from_srv(self, discovery_srv): """Fetch list of etcd-cluster member by resolving _etcd-server._tcp. SRV record. This record should contain list of host and peer ports which could be used to run @@ -272,7 +308,7 @@ class Etcd(AbstractDCS): # Synchronous work of all cluster members with etcd is less expensive # than reestablishing http connection every time from every replica. return True - except urllib3.exceptions.TimeoutError: + except etcd.EtcdWatchTimedOut: self.client.http.clear() return False except etcd.EtcdException: diff --git a/requirements-py2.txt b/requirements-py2.txt deleted file mode 100644 index 26c0892d..00000000 --- a/requirements-py2.txt +++ /dev/null @@ -1,13 +0,0 @@ -boto -dnspython -mock -psycopg2>=2.6.1 -PyYAML -requests -six >= 1.7 -kazoo>=2.2.1 -python-etcd==0.4.2 -click>=4.1 -prettytable>=0.7 -tzlocal -python-dateutil diff --git a/requirements-py3.txt b/requirements.txt similarity index 100% rename from requirements-py3.txt rename to requirements.txt diff --git a/setup.py b/setup.py index 437354e2..a8936d7c 100644 --- a/setup.py +++ b/setup.py @@ -51,8 +51,8 @@ CLASSIFIERS = [ 'Operating System :: POSIX :: Linux', 'Programming Language :: Python', 'Programming Language :: Python :: 2.7', - 'Programming Language :: Python :: 3.3', 'Programming Language :: Python :: 3.4', + 'Programming Language :: Python :: 3.5', 'Programming Language :: Python :: Implementation :: CPython', ] @@ -76,7 +76,7 @@ class PyTest(TestCommand): def finalize_options(self): TestCommand.finalize_options(self) if self.cov_xml or self.cov_html: - self.cov = ['--cov', MAIN_PACKAGE, '--cov', MAIN_PACKAGE, '--cov-report', 'term-missing'] + self.cov = ['--cov', MAIN_PACKAGE, '--cov-report', 'term-missing'] if self.cov_xml: self.cov.extend(['--cov-report', 'xml']) if self.cov_html: @@ -116,8 +116,7 @@ def setup_package(): # Some helper variables version = os.getenv('GO_PIPELINE_LABEL', VERSION) - requirements = 'requirements-py2.txt' if sys.version_info[0] == 2 else 'requirements-py3.txt' - install_reqs = get_install_requirements(requirements) + install_reqs = get_install_requirements('requirements.txt') command_options = {'test': {'test_suite': ('setup.py', 'tests')}} if JUNIT_XML: @@ -142,9 +141,9 @@ def setup_package(): packages=setuptools.find_packages(exclude=['tests', 'tests.*']), package_data={MAIN_PACKAGE: ["*.json"]}, install_requires=install_reqs, - setup_requires=['six', 'flake8'], + setup_requires=['flake8'], cmdclass=cmdclass, - tests_require=['pytest-cov', 'pytest'], + tests_require=['mock', 'pytest-cov', 'pytest'], command_options=command_options, entry_points={'console_scripts': CONSOLE_SCRIPTS}, ) diff --git a/tests/test_etcd.py b/tests/test_etcd.py index c14d3cac..fa2538c9 100644 --- a/tests/test_etcd.py +++ b/tests/test_etcd.py @@ -25,11 +25,7 @@ class MockResponse(object): @property def data(self): - if self.content == 'TimeoutError': - raise urllib3.exceptions.TimeoutError - if self.content == 'Exception': - raise Exception - return self.content + return self.content.encode('utf-8') @property def status(self): @@ -70,7 +66,7 @@ def requests_get(url, **kwargs): def etcd_watch(key, index=None, timeout=None, recursive=None): if timeout == 2.0: - raise urllib3.exceptions.TimeoutError + raise etcd.EtcdWatchTimedOut elif timeout == 5.0: return etcd.EtcdResult('delete', {}) elif timeout == 10.0: @@ -147,6 +143,8 @@ def socket_getaddrinfo(*args): def http_request(method, url, **kwargs): + if url == 'http://localhost:2379/timeout': + raise urllib3.exceptions.ReadTimeoutError(None, None, None) if url == 'http://localhost:2379/': return MockResponse() raise socket.error @@ -164,31 +162,27 @@ class TestClient(unittest.TestCase): mock_machines.__get__ = Mock(return_value=['http://localhost:2379', 'http://localhost:4001']) self.client = Client({'discovery_srv': 'test'}) self.client.http.request = http_request + self.client.http.request_encode_body = http_request def test_api_execute(self): self.client._base_uri = 'http://localhost:4001' self.client._machines_cache = ['http://localhost:2379'] - self.client.api_execute('/', 'GET') + self.assertRaises(etcd.EtcdWatchTimedOut, self.client.api_execute, '/timeout', 'POST', params={'wait': 'true'}) + self.client._update_machines_cache = False + self.client.api_execute('/', 'POST', timeout=0) self.client._update_machines_cache = False self.client._base_uri = 'http://localhost:4001' self.client._machines_cache = [] self.assertRaises(etcd.EtcdConnectionFailed, self.client.api_execute, '/', 'GET') self.assertTrue(self.client._update_machines_cache) self.assertRaises(etcd.EtcdException, self.client.api_execute, '/', 'GET') + self.assertRaises(etcd.EtcdException, self.client.api_execute, '/', '') + self.assertRaises(ValueError, self.client.api_execute, '', '') def test_get_srv_record(self): self.assertEquals(self.client.get_srv_record('blabla'), []) self.assertEquals(self.client.get_srv_record('exception'), []) - def test__result_from_response(self): - response = MockResponse() - response.content = 'TimeoutError' - self.assertRaises(urllib3.exceptions.TimeoutError, self.client._result_from_response, response) - response.content = 'Exception' - self.assertRaises(etcd.EtcdException, self.client._result_from_response, response) - response.content = b'{}' - self.assertRaises(etcd.EtcdException, self.client._result_from_response, response) - def test__get_machines_cache_from_srv(self): self.client.get_srv_record = Mock(return_value=[('localhost', 2380)]) self.client._get_machines_cache_from_srv('blabla')