Use config file as a fallback when all current etcd nodes failed (#2599)

If communication with etcd nodes failed it is logical to start from scratch, from nodes that are listed in the config. But, it could happen that config is in fact outdated and all nodes in the real cluster were replaced.

Previously we used to track whether config file was changed, which turned out not to work in all possible cases.
The new strategy is a bit more different - if communication with all nodes failed we will continue keeping the last know topology and at the same time will try to figure out the new one by merging two lists together, the cached list and the list from the config file.

patroni/dcs/etcd.py

@@ -16,6 +16,7 @@ from dns import resolver
 from http.client import HTTPException
 from queue import Queue
 from threading import Thread
+from typing import List, Optional
 from urllib.parse import urlparse
 from urllib3 import Timeout
 from urllib3.exceptions import HTTPError, ReadTimeoutError, ProtocolError
@@ -98,7 +99,6 @@ class AbstractEtcdClientWithFailover(abc.ABC, etcd.Client):
         # Workaround for the case when https://github.com/jplana/python-etcd/pull/196 is not applied
         self.http.connection_pool_kw.pop('ssl_version', None)
         self._config = config
-        self._initial_machines_cache = []
         self._load_machines_cache()
         self._allow_reconnect = True
         # allow passing retry argument to api_execute in params
@@ -168,19 +168,12 @@ class AbstractEtcdClientWithFailover(abc.ABC, etcd.Client):
         base_uri, cache = self._base_uri, self._machines_cache
         return ([base_uri] if base_uri in cache else []) + [machine for machine in cache if machine != base_uri]
 
-    @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 empty list of machines.
-        Later, during next `api_execute` call we will forcefully update machines_cache.
+    def _get_machines_list(self, machines_cache: List[str]) -> List[str]:
+        """Gets list of members from Etcd cluster using API
 
-        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."""
-
-        machines_cache = self.machines_cache
+        :param machines_cache: initial list of Etcd members
+        :returns: list of clientURLs retrieved from Etcd cluster
+        :raises EtcdConnectionFailed: if failed"""
         kwargs = self._prepare_get_members(len(machines_cache))
 
         for base_uri in machines_cache:
@@ -198,6 +191,22 @@ class AbstractEtcdClientWithFailover(abc.ABC, etcd.Client):
 
         raise etcd.EtcdConnectionFailed('No more machines in the cluster')
 
+    @property
+    def machines(self) -> List[str]:
+        """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 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.
+
+        After the next refactoring the whole logic was moved to the _get_machines_list() method."""
+
+        return self._get_machines_list(self.machines_cache)
+
     def set_read_timeout(self, timeout):
         self._read_timeout = timeout
 
@@ -367,36 +376,46 @@ class AbstractEtcdClientWithFailover(abc.ABC, etcd.Client):
         # enforce resolving dns name,they might get new ips
         self._update_dns_cache(self._dns_resolver.remove, machines_cache)
 
-        # The etcd cluster could change its topology over time and depending on how we resolve the initial
-        # topology (list of hosts in the Patroni config or DNS records, A or SRV) we might get into the situation
-        # the the real topology doesn't match anymore with the topology resolved from the configuration file.
-        # In case if the "initial" topology is the same as before we will not override the `_machines_cache`.
-        ret = set(machines_cache) != set(self._initial_machines_cache)
-        if ret:
-            self._initial_machines_cache = self._machines_cache = machines_cache
-
-            # After filling up the initial list of machines_cache we should ask etcd-cluster about actual list
-            self._refresh_machines_cache(True)
+        # after filling up the initial list of machines_cache we should ask etcd-cluster about actual list
+        ret = self._refresh_machines_cache(machines_cache)
 
         self._update_machines_cache = False
         return ret
 
-    def _refresh_machines_cache(self, updating_cache=False):
+    def _refresh_machines_cache(self, machines_cache: Optional[List[str]] = None) -> bool:
+        """Get etcd cluster topology using Etcd API and put it to self._machines_cache
+
+        :param machines_cache: the list of nodes we want to run through executing API request
+                               in addition to values stored in the self._machines_cache
+        :returns: `True` if self._machines_cache was updated with new values
+        :raises EtcdException: if failed to get topology and `machines_cache` was specified.
+
+        The self._machines_cache will not be updated if nodes from the list are
+        not accessible or if they are not returning correct results."""
+
         if self._use_proxies:
-            self._machines_cache = self._get_machines_cache_from_config()
+            value = self._get_machines_cache_from_config()
         else:
             try:
-                self._machines_cache = self.machines
+                # we want to go through the list obtained from the config file + last known health topology
+                value = self._get_machines_list(list(set((machines_cache or []) + self.machines_cache)))
             except etcd.EtcdConnectionFailed:
-                if updating_cache:
-                    raise etcd.EtcdException("Could not get the list of servers, "
-                                             "maybe you provided the wrong "
-                                             "host(s) to connect to?")
-                return
+                value = []
+
+        if value:
+            ret = set(self._machines_cache) != set(value)
+            self._machines_cache = value
+        elif machines_cache:  # we are just starting or all nodes were not available at some point
+            raise etcd.EtcdException("Could not get the list of servers, "
+                                     "maybe you provided the wrong "
+                                     "host(s) to connect to?")
+        else:
+            return False
 
         if self._base_uri not in self._machines_cache:
             self.set_base_uri(self._machines_cache[0])
         self._machines_cache_updated = time.time()
+        return ret
 
     def set_base_uri(self, value):
         if self._base_uri != value:

tests/test_ctl.py

@@ -28,12 +28,11 @@ class TestCtl(unittest.TestCase):
     TEST_ROLES = ('master', 'primary', 'leader')
 
     @patch('socket.getaddrinfo', socket_getaddrinfo)
+    @patch.object(AbstractEtcdClientWithFailover, '_get_machines_list', Mock(return_value=['http://remotehost:2379']))
     def setUp(self):
-        with patch.object(AbstractEtcdClientWithFailover, '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},
-                              'citus': {'group': 0}}, 'foo', None)
+        self.runner = CliRunner()
+        self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'retry_timeout': 10},
+                          'citus': {'group': 0}}, 'foo', None)
 
     @patch('patroni.ctl.logging.debug')
     def test_load_config(self, mock_logger_debug):

tests/test_etcd.py

@@ -135,12 +135,12 @@ class TestClient(unittest.TestCase):
     @patch('dns.resolver.query', dns_query)
     @patch('socket.getaddrinfo', socket_getaddrinfo)
     @patch('patroni.dcs.etcd.requests_get', requests_get)
+    @patch.object(EtcdClient, '_get_machines_list',
+                  Mock(return_value=['http://localhost:2379', 'http://localhost:4001']))
     def setUp(self):
-        with patch.object(EtcdClient, 'machines') as mock_machines:
-            mock_machines.__get__ = Mock(return_value=['http://localhost:2379', 'http://localhost:4001'])
-            self.client = EtcdClient({'srv': 'test', 'retry_timeout': 3}, DnsCachingResolver())
-            self.client.http.request = http_request
-            self.client.http.request_encode_body = http_request
+        self.client = EtcdClient({'srv': 'test', 'retry_timeout': 3}, DnsCachingResolver())
+        self.client.http.request = http_request
+        self.client.http.request_encode_body = http_request
 
     def test_machines(self):
         self.client._base_uri = 'http://localhost:4002'
@@ -156,9 +156,9 @@ class TestClient(unittest.TestCase):
         except Exception:
             self.assertIsNone(machines)
 
-    @patch.object(EtcdClient, 'machines')
-    def test_api_execute(self, mock_machines):
-        mock_machines.__get__ = Mock(return_value=['http://localhost:4001', 'http://localhost:2379'])
+    @patch.object(EtcdClient, '_get_machines_list',
+                  Mock(return_value=['http://localhost:4001', 'http://localhost:2379']))
+    def test_api_execute(self):
         self.client._base_uri = 'http://localhost:4001'
         self.assertRaises(etcd.EtcdException, self.client.api_execute, '/', 'POST', timeout=0)
         self.client._base_uri = 'http://localhost:4001'
@@ -196,11 +196,10 @@ class TestClient(unittest.TestCase):
     def test__get_machines_cache_from_dns(self):
         self.client._get_machines_cache_from_dns('error', 2379)
 
-    @patch.object(EtcdClient, 'machines')
-    def test__refresh_machines_cache(self, mock_machines):
-        mock_machines.__get__ = Mock(side_effect=etcd.EtcdConnectionFailed)
-        self.assertIsNone(self.client._refresh_machines_cache())
-        self.assertRaises(etcd.EtcdException, self.client._refresh_machines_cache, True)
+    @patch.object(EtcdClient, '_get_machines_list', Mock(side_effect=etcd.EtcdConnectionFailed))
+    def test__refresh_machines_cache(self):
+        self.assertFalse(self.client._refresh_machines_cache())
+        self.assertRaises(etcd.EtcdException, self.client._refresh_machines_cache, ['http://localhost:2379'])
 
     def test__load_machines_cache(self):
         self.client._config = {}
@@ -230,28 +229,27 @@ class TestClient(unittest.TestCase):
 class TestEtcd(unittest.TestCase):
 
     @patch('socket.getaddrinfo', socket_getaddrinfo)
+    @patch.object(EtcdClient, '_get_machines_list',
+                  Mock(return_value=['http://localhost:2379', 'http://localhost:4001']))
     def setUp(self):
-        with patch.object(EtcdClient, 'machines') as mock_machines:
-            mock_machines.__get__ = Mock(return_value=['http://localhost:2379', 'http://localhost:4001'])
-            self.etcd = Etcd({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 10,
-                              'host': 'localhost:2379', 'scope': 'test', 'name': 'foo'})
+        self.etcd = Etcd({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 10,
+                          'host': 'localhost:2379', 'scope': 'test', 'name': 'foo'})
 
     def test_base_path(self):
         self.assertEqual(self.etcd._base_path, '/patroni/test')
 
     @patch('dns.resolver.query', dns_query)
+    @patch('time.sleep', Mock(side_effect=SleepException))
+    @patch.object(EtcdClient, '_get_machines_list', Mock(side_effect=etcd.EtcdConnectionFailed))
     def test_get_etcd_client(self):
-        with patch('time.sleep', Mock(side_effect=SleepException)),\
-                patch.object(EtcdClient, 'machines') as mock_machines:
-            mock_machines.__get__ = Mock(side_effect=etcd.EtcdException)
-            self.assertRaises(SleepException, self.etcd.get_etcd_client,
-                              {'discovery_srv': 'test', 'retry_timeout': 10, 'cacert': '1', 'key': '1', 'cert': 1},
-                              EtcdClient)
-            self.assertRaises(SleepException, self.etcd.get_etcd_client,
-                              {'url': 'https://test:2379', 'retry_timeout': 10}, EtcdClient)
-            self.assertRaises(SleepException, self.etcd.get_etcd_client,
-                              {'hosts': 'foo:4001,bar', 'retry_timeout': 10}, EtcdClient)
-            mock_machines.__get__ = Mock(return_value=[])
+        self.assertRaises(SleepException, self.etcd.get_etcd_client,
+                          {'discovery_srv': 'test', 'retry_timeout': 10, 'cacert': '1', 'key': '1', 'cert': 1},
+                          EtcdClient)
+        self.assertRaises(SleepException, self.etcd.get_etcd_client,
+                          {'url': 'https://test:2379', 'retry_timeout': 10}, EtcdClient)
+        self.assertRaises(SleepException, self.etcd.get_etcd_client,
+                          {'hosts': 'foo:4001,bar', 'retry_timeout': 10}, EtcdClient)
+        with patch.object(EtcdClient, '_get_machines_list', Mock(return_value=[])):
             self.assertRaises(SleepException, self.etcd.get_etcd_client,
                               {'proxy': 'https://user:password@test:2379', 'retry_timeout': 10}, EtcdClient)
 

tests/test_ha.py

@@ -199,21 +199,20 @@ class TestHa(PostgresInit):
     @patch('socket.getaddrinfo', socket_getaddrinfo)
     @patch('patroni.dcs.dcs_modules', Mock(return_value=['patroni.dcs.etcd']))
     @patch.object(etcd.Client, 'read', etcd_read)
+    @patch.object(AbstractEtcdClientWithFailover, '_get_machines_list', Mock(return_value=['http://remotehost:2379']))
     def setUp(self):
         super(TestHa, self).setUp()
-        with patch.object(AbstractEtcdClientWithFailover, 'machines') as mock_machines:
-            mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
-            self.p.set_state('running')
-            self.p.set_role('replica')
-            self.p.postmaster_start_time = MagicMock(return_value=str(postmaster_start_time))
-            self.p.can_create_replica_without_replication_connection = MagicMock(return_value=False)
-            self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'scope': 'test',
-                                       'name': 'foo', 'retry_timeout': 10},
-                              'citus': {'database': 'citus', 'group': None}})
-            self.ha = Ha(MockPatroni(self.p, self.e))
-            self.ha.old_cluster = self.e.get_cluster()
-            self.ha.cluster = get_cluster_initialized_without_leader()
-            self.ha.load_cluster_from_dcs = Mock()
+        self.p.set_state('running')
+        self.p.set_role('replica')
+        self.p.postmaster_start_time = MagicMock(return_value=str(postmaster_start_time))
+        self.p.can_create_replica_without_replication_connection = MagicMock(return_value=False)
+        self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'scope': 'test',
+                                   'name': 'foo', 'retry_timeout': 10},
+                          'citus': {'database': 'citus', 'group': None}})
+        self.ha = Ha(MockPatroni(self.p, self.e))
+        self.ha.old_cluster = self.e.get_cluster()
+        self.ha.cluster = get_cluster_initialized_without_leader()
+        self.ha.load_cluster_from_dcs = Mock()
 
     def test_update_lock(self):
         self.ha.is_failsafe_mode = true

tests/test_patroni.py

@@ -66,7 +66,7 @@ class TestPatroni(unittest.TestCase):
     @patch.object(HTTPServer, '__init__', Mock())
     @patch.object(etcd.Client, 'read', etcd_read)
     @patch.object(Thread, 'start', Mock())
-    @patch.object(AbstractEtcdClientWithFailover, 'machines', PropertyMock(return_value=['http://remotehost:2379']))
+    @patch.object(AbstractEtcdClientWithFailover, '_get_machines_list', Mock(return_value=['http://remotehost:2379']))
     def setUp(self):
         self._handlers = logging.getLogger().handlers[:]
         RestApiServer._BaseServer__is_shut_down = Mock()
@@ -88,7 +88,7 @@ class TestPatroni(unittest.TestCase):
     @patch('sys.argv', ['patroni.py', 'postgres0.yml'])
     @patch('time.sleep', Mock(side_effect=SleepException))
     @patch.object(etcd.Client, 'delete', Mock())
-    @patch.object(AbstractEtcdClientWithFailover, 'machines', PropertyMock(return_value=['http://remotehost:2379']))
+    @patch.object(AbstractEtcdClientWithFailover, '_get_machines_list', Mock(return_value=['http://remotehost:2379']))
     @patch.object(Thread, 'join', Mock())
     def test_patroni_patroni_main(self):
         with patch('subprocess.call', Mock(return_value=1)):