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

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

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:

requirements-py2.txt

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

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