From 413a84836bbc40b8a9525920c68aff2650e553ab Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 10 Aug 2016 10:17:37 +0200 Subject: [PATCH] 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 | 90 ++++++++++++++++++++++++++++--------------- tests/test_ctl.py | 5 ++- tests/test_etcd.py | 45 +++++++++++++++------- tests/test_ha.py | 3 +- tests/test_patroni.py | 5 ++- 5 files changed, 98 insertions(+), 50 deletions(-) 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']