diff --git a/.travis.yml b/.travis.yml index 38dddc77..73d49104 100644 --- a/.travis.yml +++ b/.travis.yml @@ -21,8 +21,7 @@ install: - sudo apt-get update - sudo apt-get -y install postgresql-9.5 - sudo /etc/init.d/postgresql stop - - 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 - curl -L https://github.com/coreos/etcd/releases/download/v${ETCDVERSION}/etcd-v${ETCDVERSION}-linux-amd64.tar.gz | tar xz -C . --strip=1 --wildcards --no-anchored etcd - pip install behave codacy-coverage coverage coveralls script: diff --git a/patroni/etcd.py b/patroni/etcd.py index 79842ca4..03d6259d 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 {0} 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-py3.txt b/requirements-py3.txt deleted file mode 100644 index a827efb3..00000000 --- a/requirements-py3.txt +++ /dev/null @@ -1,13 +0,0 @@ -boto -mock -dnspython3 -psycopg2>=2.6.1 -PyYAML -requests -six -kazoo>=2.2.1 -python-etcd==0.4.2 -click>=4.1 -prettytable>=0.7 -tzlocal -python-dateutil diff --git a/requirements-py2.txt b/requirements.txt similarity index 76% rename from requirements-py2.txt rename to requirements.txt index 26c0892d..66b059f7 100644 --- a/requirements-py2.txt +++ b/requirements.txt @@ -1,12 +1,10 @@ boto -dnspython -mock psycopg2>=2.6.1 PyYAML requests six >= 1.7 kazoo>=2.2.1 -python-etcd==0.4.2 +python-etcd==0.4.3 click>=4.1 prettytable>=0.7 tzlocal 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')