gitops/patroni
1
0
Fork 0
mirror of https://github.com/outbackdingo/patroni.git synced 2026-08-25 14:53:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Make different kazoo timeouts depend on loop_wait (#243)

Browse Source 
* 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
This commit is contained in:
Alexander Kukushkin
2016-08-10 10:15:09 +02:00
committed by GitHub
parent a47a2bceff
commit 5fe74bec3b
9 changed files with 153 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
patroni/__init__.py
+2 -5
View File
@@ -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:
patroni/api.py
+11 -11
View File
@@ -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",
patroni/dcs/__init__.py
+15 -2
View File
@@ -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
patroni/dcs/zookeeper.py
+64 -27
View File
@@ -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
patroni/ha.py
+2 -2
View File
@@ -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()
tests/test_api.py
+32 -22
View File
@@ -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)
tests/test_ha.py
-1
View File
@@ -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
tests/test_patroni.py
+1 -1
View File
@@ -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):
tests/test_zookeeper.py
+26 -8
View File
@@ -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)