Merge pull request #188 from zalando/bugfix/python351

urllib3.exceptions.HTTPError fixes for python 3.5.1

patroni/etcd.py

@@ -6,13 +6,13 @@ import random
 import requests
 import socket
 import time
-import urllib3
 
 from dns.exception import DNSException
 from dns import resolver
 from patroni.dcs import AbstractDCS, Cluster, Failover, Leader, Member
 from patroni.exceptions import DCSError
 from patroni.utils import Retry, RetryFailedError, sleep
+from urllib3.exceptions import HTTPError, ReadTimeoutError
 from requests.exceptions import RequestException
 from six.moves.http_client import HTTPException
 
@@ -55,9 +55,9 @@ class Client(etcd.Client):
             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:
+        except (HTTPError, HTTPException, socket.error, socket.timeout) as e:
             if (isinstance(fields, dict) and fields.get("wait") == "true" and
-                    isinstance(e, urllib3.exceptions.ReadTimeoutError)):
+                    isinstance(e, 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)
@@ -302,13 +302,12 @@ class Etcd(AbstractDCS):
     def watch(self, timeout):
         cluster = self.cluster
         # watch on leader key changes if it is defined and current node is not lock owner
-        if cluster and cluster.leader and cluster.leader.name != self._name:
+        if cluster and cluster.leader and cluster.leader.name != self._name and cluster.leader.index:
             end_time = time.time() + timeout
-            index = cluster.leader.index
 
-            while index and timeout >= 1:  # when timeout is too small urllib3 doesn't have enough time to connect
+            while timeout >= 1:  # when timeout is too small urllib3 doesn't have enough time to connect
                 try:
-                    self._client.watch(self.leader_path, index=index + 1, timeout=timeout + 0.5)
+                    self._client.watch(self.leader_path, index=cluster.leader.index + 1, timeout=timeout + 0.5)
                     # Synchronous work of all cluster members with etcd is less expensive
                     # than reestablishing http connection every time from every replica.
                     return True

tests/test_etcd.py

@@ -1,15 +1,15 @@
 import etcd
 import json
 import requests
-import urllib3
 import socket
 import unittest
 
 from dns.exception import DNSException
 from mock import Mock, patch
-from patroni.dcs import Cluster
+from patroni.dcs import Cluster, AbstractDCS
 from patroni.etcd import Client, Etcd, EtcdError
 from patroni.exceptions import DCSError
+from urllib3.exceptions import ReadTimeoutError
 
 
 class MockResponse(object):
@@ -61,8 +61,6 @@ def etcd_watch(self, key, index=None, timeout=None, recursive=None):
         return etcd.EtcdResult('delete', {})
     elif timeout == 10.0:
         raise etcd.EtcdException
-    elif index == 20729:
-        return etcd.EtcdResult('set', {'value': 'postgresql1', 'modifiedIndex': index + 1})
 
 
 def etcd_write(self, key, value, **kwargs):
@@ -132,7 +130,7 @@ def socket_getaddrinfo(*args):
 
 def http_request(method, url, **kwargs):
     if url == 'http://localhost:2379/timeout':
-        raise urllib3.exceptions.ReadTimeoutError(None, None, None)
+        raise ReadTimeoutError(None, None, None)
     if url == 'http://localhost:2379/':
         return MockResponse()
     raise socket.error
@@ -248,8 +246,8 @@ class TestEtcd(unittest.TestCase):
         self.etcd.get_cluster()
         self.etcd.watch(1.5)
         self.etcd.watch(4.5)
-        self.etcd.watch(9.5)
-        self.etcd.watch(100)
+        with patch.object(AbstractDCS, 'watch', Mock()):
+            self.etcd.watch(9.5)
 
     @patch('patroni.etcd.Etcd.retry', Mock(side_effect=AttributeError("foo")))
     def test_other_exceptions(self):