diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py index c376ce8b..6f96beb4 100644 --- a/patroni/dcs/etcd.py +++ b/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): diff --git a/tests/test_ctl.py b/tests/test_ctl.py index 71f2ba8f..ccbc0b37 100644 --- a/tests/test_ctl.py +++ b/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) diff --git a/tests/test_etcd.py b/tests/test_etcd.py index 5b8c74fd..f3a32a41 100644 --- a/tests/test_etcd.py +++ b/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}) diff --git a/tests/test_ha.py b/tests/test_ha.py index 1b7f937e..de458b60 100644 --- a/tests/test_ha.py +++ b/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, diff --git a/tests/test_patroni.py b/tests/test_patroni.py index 30c1b585..f9f135cc 100644 --- a/tests/test_patroni.py +++ b/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']