diff --git a/patroni/etcd.py b/patroni/etcd.py index 3f798cf8..47f0b636 100644 --- a/patroni/etcd.py +++ b/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 diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 742c57c5..ef61f413 100644 --- a/tests/test_etcd.py +++ b/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):