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)