From ddac8683e6265667388bb548518a850ae45ea9eb Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Tue, 14 Mar 2023 15:54:17 +0100 Subject: [PATCH] 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 | 81 ++++++++++++++++++++++++++----------------- tests/test_ctl.py | 9 +++-- tests/test_etcd.py | 54 ++++++++++++++--------------- tests/test_ha.py | 25 +++++++------ tests/test_patroni.py | 4 +-- 5 files changed, 94 insertions(+), 79 deletions(-) diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index d0541c38..ca862df9 100644 --- a/patroni/dcs/etcd.py +++ b/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: diff --git a/tests/test_ctl.py b/tests/test_ctl.py index b8294956..39ee2520 100644 --- a/tests/test_ctl.py +++ b/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): diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 7e91b9e9..0bb597f8 100644 --- a/tests/test_etcd.py +++ b/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) diff --git a/tests/test_ha.py b/tests/test_ha.py index 78872623..d30689bf 100644 --- a/tests/test_ha.py +++ b/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 diff --git a/tests/test_patroni.py b/tests/test_patroni.py index a3706e78..ad8e775c 100644 --- a/tests/test_patroni.py +++ b/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)):