From 5fe74bec3b2edfbd9bf57f78c67fdf4b8809fd68 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Wed, 10 Aug 2016 10:15:09 +0200 Subject: [PATCH] Make different kazoo timeouts depend on loop_wait (#243) * Make different kazoo timeouts dependant on loop_wait ping timeout ~ 1/2 * loop_wait connect_timeout ~ 1/2 * loop_wait Originally these values were calculated from negotiated session timeout and didn't worked very well, because it was taking significant time to figure out that connection is dead and reconnect (up to session timeout) and not giving us time to retry. * Address the code review --- patroni/__init__.py | 7 +--- patroni/api.py | 22 +++++----- patroni/dcs/__init__.py | 17 +++++++- patroni/dcs/zookeeper.py | 91 ++++++++++++++++++++++++++++------------ patroni/ha.py | 4 +- tests/test_api.py | 54 ++++++++++++++---------- tests/test_ha.py | 1 - tests/test_patroni.py | 2 +- tests/test_zookeeper.py | 34 +++++++++++---- 9 files changed, 153 insertions(+), 79 deletions(-) diff --git a/patroni/__init__.py b/patroni/__init__.py index 1403dacf..37a8e848 100644 --- a/patroni/__init__.py +++ b/patroni/__init__.py @@ -30,7 +30,6 @@ class Patroni(object): self.ha = Ha(self) self.tags = self.get_tags() - self.nap_time = self.config['loop_wait'] self.next_run = time.time() self.scheduled_restart = {} @@ -57,9 +56,7 @@ class Patroni(object): def reload_config(self): try: self.tags = self.get_tags() - self.nap_time = self.config['loop_wait'] - self.dcs.set_ttl(self.config.get('ttl') or 30) - self.dcs.set_retry_timeout(self.config.get('retry_timeout') or self.nap_time) + self.dcs.reload_config(self.config) self.api.reload_config(self.config['restapi']) self.postgresql.reload_config(self.config['postgresql']) except Exception: @@ -82,7 +79,7 @@ class Patroni(object): return self.tags.get('noloadbalance', False) def schedule_next_run(self): - self.next_run += self.nap_time + self.next_run += self.dcs.loop_wait current_time = time.time() nap_time = self.next_run - current_time if nap_time <= 0: diff --git a/patroni/api.py b/patroni/api.py index 0029d8de..6f3a6ae9 100644 --- a/patroni/api.py +++ b/patroni/api.py @@ -110,7 +110,7 @@ class RestApiHandler(BaseHTTPRequestHandler): self._write_status_response(200, response) def do_GET_config(self): - cluster = self.server.patroni.ha.dcs.cluster or self.server.patroni.ha.dcs.get_cluster() + cluster = self.server.patroni.dcs.cluster or self.server.patroni.dcs.get_cluster() if cluster.config: self._write_json_response(200, cluster.config.data) else: @@ -134,11 +134,11 @@ class RestApiHandler(BaseHTTPRequestHandler): def do_PATCH_config(self): request = self._read_json_content() if request: - cluster = self.server.patroni.ha.dcs.get_cluster() + cluster = self.server.patroni.dcs.get_cluster() data = cluster.config.data.copy() if patch_config(data, request): value = json.dumps(data, separators=(',', ':')) - if not self.server.patroni.ha.dcs.set_config_value(value, cluster.config.index): + if not self.server.patroni.dcs.set_config_value(value, cluster.config.index): return self.send_error(409) self._write_json_response(200, data) @@ -146,10 +146,10 @@ class RestApiHandler(BaseHTTPRequestHandler): def do_PUT_config(self): request = self._read_json_content() if request: - cluster = self.server.patroni.ha.dcs.get_cluster() + cluster = self.server.patroni.dcs.get_cluster() if not deep_compare(request, cluster.config.data): value = json.dumps(request, separators=(',', ':')) - if not self.server.patroni.ha.dcs.set_config_value(value): + if not self.server.patroni.dcs.set_config_value(value): return self.send_error(502) self._write_json_response(200, request) @@ -249,16 +249,16 @@ class RestApiHandler(BaseHTTPRequestHandler): @check_auth def do_POST_reinitialize(self): - ha = self.server.patroni.ha - cluster = ha.dcs.get_cluster() + patroni = self.server.patroni + cluster = patroni.dcs.get_cluster() if cluster.is_unlocked(): status_code = 503 data = 'Cluster has no leader, can not reinitialize' - elif cluster.leader.name == ha.state_handler.name: + elif cluster.leader.name == patroni.ha.state_handler.name: status_code = 503 data = 'I am the leader, can not reinitialize' else: - action = ha.schedule_reinitialize() + action = patroni.ha.schedule_reinitialize() if action is not None: status_code = 503 data = action + ' already in progress' @@ -268,7 +268,7 @@ class RestApiHandler(BaseHTTPRequestHandler): self._write_response(status_code, data) def poll_failover_result(self, leader, candidate): - timeout = 10 if self.server.patroni.nap_time < 10 else self.server.patroni.nap_time + timeout = max(10, self.server.patroni.dcs.loop_wait) for _ in range(0, timeout*2): time.sleep(1) try: @@ -309,7 +309,7 @@ class RestApiHandler(BaseHTTPRequestHandler): leader = request.get('leader') candidate = request.get('candidate') or request.get('member') scheduled_at = request.get('scheduled_at') - cluster = self.server.patroni.ha.dcs.get_cluster() + cluster = self.server.patroni.dcs.get_cluster() status_code = 500 logger.info("received failover request with leader=%s candidate=%s scheduled_at=%s", diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py index 39049792..e7f850b3 100644 --- a/patroni/dcs/__init__.py +++ b/patroni/dcs/__init__.py @@ -44,8 +44,8 @@ def get_dcs(config): available_implementations.add(name) if name in config: # which has configuration section in the config file # propagate some parameters - config[name].update({p: config[p] for p in ('namespace', 'name', - 'scope', 'ttl', 'retry_timeout') if p in config}) + config[name].update({p: config[p] for p in ('namespace', 'name', 'scope', + 'loop_wait', 'ttl', 'retry_timeout') if p in config}) return value(config[name]) raise PatroniException("""Can not find suitable configuration of distributed configuration store Available implementations: """ + ', '.join(available_implementations)) @@ -225,6 +225,7 @@ class AbstractDCS(object): self._name = config['name'] self._namespace = '/{0}'.format(config.get('namespace', '/service/').strip('/')) self._base_path = '/'.join([self._namespace, config['scope']]) + self._set_loop_wait(config.get('loop_wait', 10)) self._cluster = None self._cluster_thread_lock = Lock() @@ -269,6 +270,18 @@ class AbstractDCS(object): def set_retry_timeout(self, retry_timeout): """Set the new value for retry_timeout""" + def _set_loop_wait(self, loop_wait): + self._loop_wait = loop_wait + + def reload_config(self, config): + self._set_loop_wait(config['loop_wait']) + self.set_ttl(config['ttl']) + self.set_retry_timeout(config['retry_timeout']) + + @property + def loop_wait(self): + return self._loop_wait + @abc.abstractmethod def _load_cluster(self): """Internally this method should build `Cluster` object which diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py index be6fe45a..9f8c9e12 100644 --- a/patroni/dcs/zookeeper.py +++ b/patroni/dcs/zookeeper.py @@ -20,7 +20,7 @@ class PatroniSequentialThreadingHandler(SequentialThreadingHandler): self.set_connect_timeout(connect_timeout) def set_connect_timeout(self, connect_timeout): - self._connect_timeout = max(1.0, connect_timeout/4.0) + self._connect_timeout = max(1.0, connect_timeout/2.0) # try to connect to zookeeper node during loop_wait/2 def create_connection(self, *args, **kwargs): """This method is trying to establish connection with one of the zookeeper nodes. @@ -59,8 +59,27 @@ class ZooKeeper(AbstractDCS): self._fetch_cluster = True self._last_leader_operation = 0 + self._orig_kazoo_connect = self._client._connection._connect + self._client._connection._connect = self._kazoo_connect + self._client.start() + def _kazoo_connect(self, host, port): + + """Kazoo is using Ping's to determine health of connection to zookeeper. If there is no + response on Ping after Ping interval (1/2 from read_timeout) it will consider current + connection dead and try to connect to another node. Without this "magic" it was taking + up to 2/3 from session timeout (ttl) to figure out that connection was dead and we had + only small time for reconnect and retry. + + This method is needed to return different value of read_timeout, which is not calculated + from negotiated session timeout but from value of `loop_wait`. And it is 2 sec smaller + than loop_wait, because we can spend up to 2 seconds when calling `touch_member()` and + `write_leader_optime()` methods, which also may hang...""" + + ret = self._orig_kazoo_connect(host, port) + return max(self.loop_wait - 2, 2)*1000, ret[1] + def session_listener(self, state): if state in [KazooState.SUSPENDED, KazooState.LOST]: self.cluster_watcher(None) @@ -69,15 +88,34 @@ class ZooKeeper(AbstractDCS): self._fetch_cluster = True self.event.set() + def reload_config(self, config): + self.set_retry_timeout(config['retry_timeout']) + + loop_wait = config['loop_wait'] + + loop_wait_changed = self._loop_wait != loop_wait + self._loop_wait = loop_wait + self._client.handler.set_connect_timeout(loop_wait) + + # We need to reestablish connection to zookeeper if we want to change + # read_timeout (and Ping interval respectively), because read_timeout + # is calculated in `_kazoo_connect` method. If we are changing ttl at + # the same time, set_ttl method will reestablish connection and return + # `!True`, otherwise we will close existing connection and let kazoo + # open the new one. + if not self.set_ttl(int(config['ttl'] * 1000)) and loop_wait_changed: + self._client._connection._socket.close() + def set_ttl(self, ttl): - ttl = int(ttl * 1000) - # I know, it's weird to access private attributes + """It is not possible to change ttl (session_timeout) in zookeeper without + destroying old session and creating the new one. This method returns `!True` + if session_timeout has been changed (`restart()` has been called).""" if self._client._session_timeout != ttl: self._client._session_timeout = ttl self._client.restart() + return True def set_retry_timeout(self, retry_timeout): - self._client.handler.set_connect_timeout(retry_timeout) self._client._retry.deadline = retry_timeout def get_node(self, key, watch=None): @@ -150,7 +188,7 @@ class ZooKeeper(AbstractDCS): if self._fetch_cluster or self._cluster is None: try: self._client.retry(self._inner_load_cluster) - except: + except Exception: logger.exception('get_cluster') self.cluster_watcher(None) raise ZooKeeperError('ZooKeeper in not responding properly') @@ -195,36 +233,35 @@ class ZooKeeper(AbstractDCS): def touch_member(self, data, ttl=None): cluster = self.cluster member = cluster and ([m for m in cluster.members if m.name == self._name] or [None])[0] - path = self.member_path data = data.encode('utf-8') if member and self._client.client_id is not None and member.session != self._client.client_id[0]: try: - self._client.retry(self._client.delete, path) + self._client.delete_async(self.member_path).get(timeout=1) except NoNodeError: pass except: return False member = None - if member and data == self._my_member_data: - return True - - try: - if member: - self._client.retry(self._client.set, path, data) - else: - self._client.retry(self._client.create, path, data, makepath=True, ephemeral=True) - self._my_member_data = data - return True - except NodeExistsError: + if member: + if data == self._my_member_data: + return True + else: try: - self._client.retry(self._client.set, path, data) + self._client.create_async(self.member_path, data, makepath=True, ephemeral=True).get(timeout=1) self._my_member_data = data return True - except: - logger.exception('touch_member') + except Exception as e: + if not isinstance(e, NodeExistsError): + logger.exception('touch_member') + return False + try: + self._client.set_async(self.member_path, data).get(timeout=1) + self._my_member_data = data + return True except: logger.exception('touch_member') + return False def take_leader(self): @@ -233,17 +270,17 @@ class ZooKeeper(AbstractDCS): def write_leader_optime(self, last_operation): last_operation = last_operation.encode('utf-8') if last_operation != self._last_leader_operation: - self._last_leader_operation = last_operation - path = self.leader_optime_path try: - self._client.retry(self._client.set, path, last_operation) + self._client.set_async(self.leader_optime_path, last_operation).get(timeout=1) + self._last_leader_operation = last_operation except NoNodeError: try: - self._client.retry(self._client.create, path, last_operation, makepath=True) + self._client.create_async(self.leader_optime_path, last_operation, makepath=True).get(timeout=1) + self._last_leader_operation = last_operation except: - logger.exception('Failed to create %s', path) + logger.exception('Failed to create %s', self.leader_optime_path) except: - logger.exception('Failed to update %s', path) + logger.exception('Failed to update %s', self.leader_optime_path) def update_leader(self): return True diff --git a/patroni/ha.py b/patroni/ha.py index cf3c4876..79cc8b04 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -296,11 +296,11 @@ class Ha(object): try: delta = (scheduled_at - now).total_seconds() - if delta > self.patroni.nap_time: + if delta > self.dcs.loop_wait: logger.info('Awaiting %s at %s (in %.0f seconds)', action_name, scheduled_at.isoformat(), delta) return False - elif delta < - int(self.patroni.nap_time * 1.5): + elif delta < - int(self.dcs.loop_wait * 1.5): logger.warning('Found a stale %s value, cleaning up: %s', action_name, scheduled_at.isoformat()) cleanup_fn() diff --git a/tests/test_api.py b/tests/test_api.py index 51bdf9b3..ea68486c 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -37,7 +37,6 @@ class MockPostgresql(object): class MockHa(object): - dcs = Mock() state_handler = MockPostgresql() @staticmethod @@ -67,10 +66,9 @@ class MockHa(object): class MockPatroni(object): - nap_time = 10 - config = Mock() - postgresql = MockPostgresql() ha = MockHa() + config = Mock() + postgresql = ha.state_handler dcs = Mock() tags = {} version = '0.00' @@ -138,14 +136,14 @@ class TestRestApiHandler(unittest.TestCase): self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'POST /restart HTTP/1.0')) MockRestApiServer(RestApiHandler, 'POST /restart HTTP/1.0\nAuthorization:') - @patch.object(MockHa, 'dcs') + @patch.object(MockPatroni, 'dcs') def test_do_GET_config(self, mock_dcs): mock_dcs.cluster.config.data = {} self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /config')) mock_dcs.cluster.config = None self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /config')) - @patch.object(MockHa, 'dcs') + @patch.object(MockPatroni, 'dcs') def test_do_PATCH_config(self, mock_dcs): config = {'postgresql': {'use_slots': False, 'use_pg_rewind': True, 'parameters': {'wal_level': 'logical'}}} mock_dcs.get_cluster.return_value.config = ClusterConfig.from_node(1, json.dumps(config)) @@ -161,7 +159,7 @@ class TestRestApiHandler(unittest.TestCase): mock_dcs.set_config_value.return_value = False MockRestApiServer(RestApiHandler, request) - @patch.object(MockHa, 'dcs') + @patch.object(MockPatroni, 'dcs') def test_do_PUT_config(self, mock_dcs): mock_dcs.get_cluster.return_value.config = ClusterConfig.from_node(1, '{}') request = 'PUT /config HTTP/1.0' + self._authorization + '\nContent-Length: ' @@ -181,9 +179,11 @@ class TestRestApiHandler(unittest.TestCase): MockRestApiServer(RestApiHandler, 'POST /reload HTTP/1.0' + self._authorization) self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'POST /reload HTTP/1.0' + self._authorization)) + #@patch.object(MockPatroni, 'dcs') def test_do_POST_restart(self): request = 'POST /restart HTTP/1.0' + self._authorization self.assertIsNotNone(MockRestApiServer(RestApiHandler, request)) + with patch.object(MockHa, 'restart', Mock(side_effect=Exception)): MockRestApiServer(RestApiHandler, request) @@ -221,13 +221,14 @@ class TestRestApiHandler(unittest.TestCase): request = make_request('{"role": "master", "postgres_version": "9.5.2"}') MockRestApiServer(RestApiHandler, request) + #@patch.object(MockPatroni, 'dcs') def test_do_DELETE_restart(self): for retval in (True, False): with patch.object(MockHa, 'delete_future_restart', Mock(return_value=retval)): request = 'DELETE /restart HTTP/1.0' + self._authorization self.assertIsNotNone(MockRestApiServer(RestApiHandler, request)) - @patch.object(MockHa, 'dcs') + @patch.object(MockPatroni, 'dcs') def test_do_POST_reinitialize(self, dcs): cluster = dcs.get_cluster.return_value request = 'POST /reinitialize HTTP/1.0' + self._authorization @@ -247,8 +248,9 @@ class TestRestApiHandler(unittest.TestCase): self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /patroni')) @patch('time.sleep', Mock()) - @patch.object(MockHa, 'dcs') + @patch.object(MockPatroni, 'dcs') def test_do_POST_failover(self, dcs): + dcs.loop_wait = 10 cluster = dcs.get_cluster.return_value post = 'POST /failover HTTP/1.0' + self._authorization + '\nContent-Length: ' @@ -273,19 +275,27 @@ class TestRestApiHandler(unittest.TestCase): cluster.members = [Member(0, 'postgresql0', 30, {'api_url': 'http'}), Member(0, 'postgresql2', 30, {'api_url': 'http'})] MockRestApiServer(RestApiHandler, request) - with patch.object(MockPatroni, 'dcs') as d: - cluster = d.get_cluster.return_value - cluster.leader.name = 'postgresql0' - MockRestApiServer(RestApiHandler, request) - cluster.leader.name = 'postgresql2' - MockRestApiServer(RestApiHandler, request) - cluster.leader.name = 'postgresql1' - cluster.failover = None - MockRestApiServer(RestApiHandler, request) - d.get_cluster = Mock(side_effect=Exception) - MockRestApiServer(RestApiHandler, request) - d.manual_failover.return_value = False - MockRestApiServer(RestApiHandler, request) + + cluster.failover = None + MockRestApiServer(RestApiHandler, request) + + dcs.get_cluster.side_effect = [cluster] + MockRestApiServer(RestApiHandler, request) + + cluster2 = cluster.copy() + cluster2.leader.name = 'postgresql0' + dcs.get_cluster.side_effect = [cluster, cluster2] + MockRestApiServer(RestApiHandler, request) + + cluster2.leader.name = 'postgresql2' + dcs.get_cluster.side_effect = [cluster, cluster2] + MockRestApiServer(RestApiHandler, request) + + dcs.get_cluster.side_effect = None + dcs.manual_failover.return_value = False + MockRestApiServer(RestApiHandler, request) + dcs.manual_failover.return_value = True + with patch.object(MockHa, 'fetch_nodes_statuses', Mock(return_value=[])): MockRestApiServer(RestApiHandler, request) diff --git a/tests/test_ha.py b/tests/test_ha.py index 0c3c2d0f..1b7f937e 100644 --- a/tests/test_ha.py +++ b/tests/test_ha.py @@ -82,7 +82,6 @@ zookeeper: self.api = Mock() self.tags = {'foo': 'bar'} self.nofailover = None - self.nap_time = 10 self.replicatefrom = None self.api.connection_string = 'http://127.0.0.1:8008' self.clonefrom = None diff --git a/tests/test_patroni.py b/tests/test_patroni.py index af8b36b0..30c1b585 100644 --- a/tests/test_patroni.py +++ b/tests/test_patroni.py @@ -74,7 +74,7 @@ class TestPatroni(unittest.TestCase): def test_schedule_next_run(self): self.p.ha.dcs.watch = Mock(return_value=True) self.p.schedule_next_run() - self.p.next_run = time.time() - self.p.nap_time - 1 + self.p.next_run = time.time() - self.p.dcs.loop_wait - 1 self.p.schedule_next_run() def test_noloadbalance(self): diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py index a406ec82..d7ee7ad5 100644 --- a/tests/test_zookeeper.py +++ b/tests/test_zookeeper.py @@ -58,11 +58,16 @@ class MockKazooClient(Mock): raise TypeError("Invalid type for 'path' (string expected)") if not isinstance(value, (six.binary_type,)): raise TypeError("Invalid type for 'value' (must be a byte string)") + if value == b'Exception': + raise Exception if path.endswith('/initialize') or path == '/service/test/optime/leader': raise Exception elif value == b'retry' or (value == b'exists' and self.exists): raise NodeExistsError + def create_async(self, path, value=b"", acl=None, ephemeral=False, sequence=False, makepath=False): + return self.create(path, value, acl, ephemeral, sequence, makepath) or Mock() + @staticmethod def set(path, value, version=-1): if not isinstance(path, six.string_types): @@ -80,6 +85,9 @@ class MockKazooClient(Mock): return raise NoNodeError + def set_async(self, path, value, version=-1): + return self.set(path, value, version) or Mock() + def delete(self, path, version=-1, recursive=False): if not isinstance(path, six.string_types): raise TypeError("Invalid type for 'path' (string expected)") @@ -92,6 +100,9 @@ class MockKazooClient(Mock): elif path.endswith('/') or path.endswith('/initialize') or path == '/service/test/members/bar': raise NoNodeError + def delete_async(self, path, version=-1, recursive=False): + return self.delete(path, version, recursive) or Mock() + class TestPatroniSequentialThreadingHandler(unittest.TestCase): @@ -109,16 +120,14 @@ class TestZooKeeper(unittest.TestCase): @patch('patroni.dcs.zookeeper.KazooClient', MockKazooClient) def setUp(self): self.zk = ZooKeeper({'hosts': ['localhost:2181'], 'scope': 'test', - 'name': 'foo', 'ttl': 30, 'retry_timeout': 10}) + 'name': 'foo', 'ttl': 30, 'retry_timeout': 10, 'loop_wait': 10}) def test_session_listener(self): self.zk.session_listener(KazooState.SUSPENDED) - def test_set_ttl(self): - self.zk.set_ttl(20) - - def test_set_retry_timeout(self): - self.zk.set_retry_timeout(10) + def test_reload_config(self): + self.zk.reload_config({'ttl': 20, 'retry_timeout': 10, 'loop_wait': 10}) + self.zk.reload_config({'ttl': 20, 'retry_timeout': 10, 'loop_wait': 5}) def test_get_node(self): self.assertIsNone(self.zk.get_node('/no_node')) @@ -165,7 +174,7 @@ class TestZooKeeper(unittest.TestCase): self.zk.touch_member('new') self.zk._name = 'na' self.zk._client.exists = 1 - self.zk.touch_member('exists') + self.zk.touch_member('Exception') self.zk._name = 'bar' self.zk.touch_member('retry') self.zk._fetch_cluster = True @@ -183,8 +192,12 @@ class TestZooKeeper(unittest.TestCase): def test_write_leader_optime(self): self.zk.last_leader_operation = '0' self.zk.write_leader_optime('1') + with patch.object(MockKazooClient, 'create_async', Mock()): + self.zk.write_leader_optime('1') + with patch.object(MockKazooClient, 'set_async', Mock()): + self.zk.write_leader_optime('2') self.zk._base_path = self.zk._base_path.replace('test', 'bla') - self.zk.write_leader_optime('2') + self.zk.write_leader_optime('3') def test_delete_cluster(self): self.assertTrue(self.zk.delete_cluster()) @@ -193,3 +206,8 @@ class TestZooKeeper(unittest.TestCase): self.zk.watch(0) self.zk.event.isSet = lambda: True self.zk.watch(0) + + def test__kazoo_connect(self): + self.zk._client._retry.deadline = 1 + self.zk._orig_kazoo_connect = Mock(return_value=(0, 0)) + self.zk._kazoo_connect(None, None)