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)):