Merge branch 'master' of github.com:zalando/patroni into feature/acceptance_tests_behave

.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:

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:

requirements-py3.txt

@@ -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

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

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},
     )

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')