Update etcd topology only after original request succeed (#254)

There is no point to try to update topology until original request is
not performed. Also for us it is more important to execute original
request rather then keep topology of etcd cluster in sync.

In addition to that implement the same retry-timeout logic in the
`machines` property which already is used in `api_execute` method.

patroni/dcs/etcd.py

@@ -31,24 +31,57 @@ class Client(etcd.Client):
         self._load_machines_cache()
         self._allow_reconnect = True
 
+    def _build_request_parameters(self):
+        kwargs = {'headers': self._get_headers(), 'redirect': self.allow_redirect}
+
+        # calculate the number of retries and timeout *per node*
+        # actual number of retries depends on the number of nodes
+        etcd_nodes = len(self._machines_cache) + 1
+        kwargs['retries'] = 0 if etcd_nodes > 3 else (1 if etcd_nodes > 1 else 2)
+
+        # if etcd_nodes > 3:
+        #     kwargs.update({'retries': 0, 'timeout': float(self.read_timeout)/etcd_nodes})
+        # elif etcd_nodes > 1:
+        #     kwargs.update({'retries': 1, 'timeout': self.read_timeout/2.0/etcd_nodes})
+        # else:
+        #     kwargs.update({'retries': 2, 'timeout': self.read_timeout/3.0})
+        kwargs['timeout'] = self.read_timeout/float(kwargs['retries'] + 1)/etcd_nodes
+        return kwargs
+
     @property
     def machines(self):
         """Original `machines` method(property) of `etcd.Client` class raise exception
         when it failed to get list of etcd cluster members. This method is being called
         only when request failed on one of the etcd members during `api_execute` call.
-        For us it's more important to execute original request rather then get new
-        topology of etcd cluster. So we will catch this exception and return valid list
-        of machines with setting flag `self._update_machines_cache` to `!True`.
-        Later, during next `api_execute` call we will forcefully update machines_cache"""
-        try:
-            ret = super(Client, self).machines
-            random.shuffle(ret)
-            return ret
-        except etcd.EtcdException:
-            if self._update_machines_cache:  # We are updating machines_cache
-                raise  # This exception is fatal, we should re-raise it.
-            self._update_machines_cache = True
-            return [self._base_uri]
+        For us it's more important to execute original request rather then get new topology
+        of etcd cluster. So we will catch this exception and return empty list of machines.
+        Later, during next `api_execute` call we will forcefully update machines_cache.
+
+        Also this method implements the same timeout-retry logic as `api_execute`, because
+        the original method was retrying 2 times with the `read_timeout` on each node."""
+
+        kwargs = self._build_request_parameters()
+
+        while True:
+            try:
+                response = self.http.request(self._MGET, self._base_uri + self.version_prefix + '/machines', **kwargs)
+                machines = [n.strip() for n in self._handle_server_response(response).data.decode('utf-8').split(',')]
+                logger.debug("Retrieved list of machines: %s", machines)
+                random.shuffle(machines)
+                return machines
+            except Exception as e:
+                # We can't get the list of machines, if one server is in the
+                # machines cache, try on it
+                logger.error("Failed to get list of machines from %s%s: %r", self._base_uri, self.version_prefix, e)
+                if self._machines_cache:
+                    self._base_uri = self._machines_cache.pop(0)
+                    logger.info("Retrying on %s", self._base_uri)
+                elif self._update_machines_cache:
+                    raise etcd.EtcdException("Could not get the list of servers, "
+                                             "maybe you provided the wrong "
+                                             "host(s) to connect to?")
+                else:
+                    return []
 
     def set_read_timeout(self, timeout):
         self._read_timeout = timeout
@@ -73,8 +106,7 @@ class Client(etcd.Client):
         if not path.startswith('/'):
             raise ValueError('Path does not start with /')
 
-        kwargs = {'fields': params, 'redirect': self.allow_redirect,
-                  'headers': self._get_headers(), 'preload_content': False}
+        kwargs = {'fields': params, 'preload_content': False}
 
         if method in [self._MGET, self._MDELETE]:
             request_executor = self.http.request
@@ -88,35 +120,29 @@ class Client(etcd.Client):
         if self._update_machines_cache:
             self._load_machines_cache()
 
-        if timeout is None:
-            # calculate the number of retries and timeout *per node*
-            # actual number of retries depends on the number of nodes
-            etcd_nodes = len(self._machines_cache) + 1
-            kwargs['retries'] = 0 if etcd_nodes > 3 else (1 if etcd_nodes > 1 else 2)
+        kwargs.update(self._build_request_parameters())
 
-            # if etcd_nodes > 3:
-            #     kwargs.update({'retries': 0, 'timeout': float(self.read_timeout)/etcd_nodes})
-            # elif etcd_nodes > 1:
-            #     kwargs.update({'retries': 1, 'timeout': self.read_timeout/2.0/etcd_nodes})
-            # else:
-            #     kwargs.update({'retries': 2, 'timeout': self.read_timeout/3.0})
-            kwargs['timeout'] = self.read_timeout/float(kwargs['retries'] + 1)/etcd_nodes
-        else:
+        if timeout is not None:
             kwargs.update({'retries': 0, 'timeout': timeout})
 
         response = False
 
         try:
+            some_request_failed = False
             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
+                if response is False:
+                    some_request_failed = True
+            if some_request_failed and not self._use_proxies:
+                self._machines_cache = self.machines
+                if self._base_uri in self._machines_cache:
                     self._machines_cache.remove(self._base_uri)
-            return self._handle_server_response(response)
         except etcd.EtcdConnectionFailed:
             self._update_machines_cache = True
-            raise
+            if not response:
+                raise
+        return self._handle_server_response(response)
 
     @staticmethod
     def get_srv_record(host):

tests/test_ctl.py

@@ -8,6 +8,7 @@ from click.testing import CliRunner
 from mock import patch, Mock
 from patroni.ctl import ctl, members, store_config, load_config, output_members, request_patroni, get_dcs, parse_dcs, \
     wait_for_leader, get_all_members, get_any_member, get_cursor, query_member, configure, PatroniCtlException
+from patroni.dcs.etcd import Client
 from psycopg2 import OperationalError
 from test_etcd import etcd_read, requests_get, socket_getaddrinfo, MockResponse
 from test_ha import get_cluster_initialized_without_leader, get_cluster_initialized_with_leader, \
@@ -33,9 +34,9 @@ class TestCtl(unittest.TestCase):
 
     @patch('socket.getaddrinfo', socket_getaddrinfo)
     def setUp(self):
-        self.runner = CliRunner()
-        with patch.object(etcd.Client, 'machines') as mock_machines:
+        with patch.object(Client, 'machines') as mock_machines:
             mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
+            self.runner = CliRunner()
             self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'retry_timeout': 10}}, 'foo')
 
     @patch('psycopg2.connect', psycopg2_connect)

tests/test_etcd.py

@@ -132,6 +132,10 @@ def socket_getaddrinfo(*args):
 def http_request(method, url, **kwargs):
     if url == 'http://localhost:2379/timeout':
         raise ReadTimeoutError(None, None, None)
+    if url == 'http://localhost:2379/v2/machines':
+        ret = MockResponse()
+        ret.content = 'http://localhost:2379,http://localhost:4001'
+        return ret
     if url == 'http://localhost:2379/':
         return MockResponse()
     raise socket.error
@@ -145,26 +149,39 @@ class TestClient(unittest.TestCase):
     @patch('dns.resolver.query', dns_query)
     @patch('requests.get', requests_get)
     def setUp(self):
-        with patch.object(etcd.Client, 'machines') as mock_machines:
+        with patch.object(Client, 'machines') as mock_machines:
             mock_machines.__get__ = Mock(return_value=['http://localhost:2379', 'http://localhost:4001'])
             self.client = Client({'discovery_srv': 'test', 'retry_timeout': 3})
             self.client.http.request = http_request
             self.client.http.request_encode_body = http_request
 
-    def test_api_execute(self):
+    def test_machines(self):
         self.client._base_uri = 'http://localhost:4001'
         self.client._machines_cache = ['http://localhost:2379']
-        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.assertIsNotNone(self.client.machines)
         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.assertIsNotNone(self.client.machines)
+        self.client._update_machines_cache = True
+        machines = None
+        try:
+            machines = self.client.machines
+            self.assertFail()
+        except Exception:
+            self.assertIsNone(machines)
+
+    @patch.object(Client, 'machines')
+    def test_api_execute(self, mock_machines):
+        mock_machines.__get__ = Mock(return_value=['http://localhost:2379'])
         self.assertRaises(ValueError, self.client.api_execute, '', '')
+        self.client._base_uri = 'http://localhost:4001'
+        self.client._machines_cache = ['http://localhost:2379']
+        self.client.api_execute('/', 'POST', timeout=0)
+        self.assertRaises(etcd.EtcdWatchTimedOut, self.client.api_execute, '/timeout', 'POST', params={'wait': 'true'})
+        self.assertRaises(etcd.EtcdException, self.client.api_execute, '/', '')
+        self.client._update_machines_cache = True
+        with patch.object(Client, '_load_machines_cache', Mock(side_effect=etcd.EtcdException)):
+            self.assertRaises(etcd.EtcdException, self.client.api_execute, '/', 'GET')
 
     def test_get_srv_record(self):
         self.assertEquals(self.client.get_srv_record('blabla'), [])
@@ -177,7 +194,9 @@ class TestClient(unittest.TestCase):
     def test__get_machines_cache_from_dns(self):
         self.client._get_machines_cache_from_dns('error:2379')
 
-    def test__load_machines_cache(self):
+    @patch.object(Client, 'machines')
+    def test__load_machines_cache(self, mock_machines):
+        mock_machines.__get__ = Mock(return_value=['http://localhost:2379'])
         self.client._config = {}
         self.assertRaises(Exception, self.client._load_machines_cache)
         self.client._config = {'discovery_srv': 'blabla'}
@@ -201,9 +220,9 @@ class TestEtcd(unittest.TestCase):
 
     @patch('dns.resolver.query', dns_query)
     def test_get_etcd_client(self):
-        with patch.object(etcd.Client, 'machines') as mock_machines:
+        with patch.object(Client, 'machines') as mock_machines:
             mock_machines.__get__ = Mock(side_effect=etcd.EtcdException)
-            with patch('time.sleep', Mock(side_effect=SleepException())):
+            with patch('time.sleep', Mock(side_effect=SleepException)):
                 self.assertRaises(SleepException, self.etcd.get_etcd_client,
                                   {'discovery_srv': 'test', 'retry_timeout': 10})
 

tests/test_ha.py

@@ -7,6 +7,7 @@ import unittest
 from mock import Mock, MagicMock, patch
 from patroni.config import Config
 from patroni.dcs import Cluster, Failover, Leader, Member, get_dcs
+from patroni.dcs.etcd import Client
 from patroni.exceptions import DCSError, PostgresException
 from patroni.ha import Ha
 from patroni.postgresql import Postgresql
@@ -114,7 +115,7 @@ class TestHa(unittest.TestCase):
     @patch('socket.getaddrinfo', socket_getaddrinfo)
     @patch.object(etcd.Client, 'read', etcd_read)
     def setUp(self):
-        with patch.object(etcd.Client, 'machines') as mock_machines:
+        with patch.object(Client, 'machines') as mock_machines:
             mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
             self.p = Postgresql({'name': 'postgresql0', 'scope': 'dummy', 'listen': '127.0.0.1:5432',
                                  'data_dir': 'data/postgresql0', 'retry_timeout': 10,

tests/test_patroni.py

@@ -6,6 +6,7 @@ import unittest
 from mock import Mock, patch
 from patroni.api import RestApiServer
 from patroni.async_executor import AsyncExecutor
+from patroni.dcs.etcd import Client
 from patroni.exceptions import DCSError
 from patroni import Patroni, main as _main
 from six.moves import BaseHTTPServer
@@ -31,7 +32,7 @@ class TestPatroni(unittest.TestCase):
         RestApiServer._BaseServer__is_shut_down = Mock()
         RestApiServer._BaseServer__shutdown_request = True
         RestApiServer.socket = 0
-        with patch.object(etcd.Client, 'machines') as mock_machines:
+        with patch.object(Client, 'machines') as mock_machines:
             mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
             sys.argv = ['patroni.py', 'postgres0.yml']
             self.p = Patroni()
@@ -44,7 +45,7 @@ class TestPatroni(unittest.TestCase):
 
     @patch('time.sleep', Mock(side_effect=SleepException))
     @patch.object(etcd.Client, 'delete', Mock())
-    @patch.object(etcd.Client, 'machines')
+    @patch.object(Client, 'machines')
     def test_patroni_main(self, mock_machines):
         with patch('subprocess.call', Mock(return_value=1)):
             sys.argv = ['patroni.py', 'postgres0.yml']