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

Do session/renew call to Consul when update_leader is called (#336)

Browse Source
This commit is contained in:
Alexander Kukushkin
2016-10-10 10:05:55 +02:00
committed by GitHub
parent 6c9a870f09
commit 1e573aec8f
4 changed files with 72 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
features/environment.py
+16 -7
View File
@@ -192,9 +192,9 @@ class AbstractDcsController(AbstractController):
def _is_accessible(self):
return self._is_running()
def stop_and_remove_work_directory(self, timeout=15):
def stop(self, kill=False, timeout=15):
""" terminate process and wipe out the temp work directory, but only if we actually started it"""
self.stop(timeout=timeout)
super(AbstractDcsController, self).stop(kill=kill, timeout=timeout)
if self._work_directory:
shutil.rmtree(self._work_directory)
@@ -233,8 +233,16 @@ class ConsulController(AbstractDcsController):
self._client = consul.Consul()
def _start(self):
return subprocess.Popen(['consul', 'agent', '-server', '-bootstrap', '-advertise=127.0.0.1',
'-data-dir', self._work_directory], stdout=self._log, stderr=subprocess.STDOUT)
config_file = self._work_directory + '.json'
with open(config_file, 'wb') as f:
f.write(b'{"session_ttl_min":"5s","server":true,"bootstrap":true,"advertise_addr":"127.0.0.1"}')
return subprocess.Popen(['consul', 'agent', '-config-file', config_file, '-data-dir', self._work_directory],
stdout=self._log, stderr=subprocess.STDOUT)
def stop(self, kill=False, timeout=15):
super(ConsulController, self).stop(kill=kill, timeout=timeout)
if self._work_directory:
os.unlink(self._work_directory + '.json')
def _is_running(self):
try:
@@ -405,19 +413,20 @@ class PatroniPoolController(object):
# actions to execute on start/stop of the tests and before running invidual features
def before_all(context):
context.timeout_multiplier = 2 if 'TRAVIS_BUILD_NUMBER' in os.environ or 'BUILD_NUMBER' in os.environ else 1
context.ci = 'TRAVIS_BUILD_NUMBER' in os.environ or 'BUILD_NUMBER' in os.environ
context.timeout_multiplier = 2 if context.ci else 1
context.pctl = PatroniPoolController(context)
context.dcs_ctl = context.pctl.known_dcs[context.pctl.dcs](context)
context.dcs_ctl.start()
try:
context.dcs_ctl.cleanup_service_tree()
except AssertionError: # after_all handlers won't be executed in before_all
context.dcs_ctl.stop_and_remove_work_directory()
context.dcs_ctl.stop()
raise
def after_all(context):
context.dcs_ctl.stop_and_remove_work_directory()
context.dcs_ctl.stop()
subprocess.call(['coverage', 'combine'])
subprocess.call(['coverage', 'report'])
features/patroni_api.feature
+1 -1
View File
@@ -34,7 +34,7 @@ Scenario: check local configuration reload
Then I receive a response code 202
Scenario: check dynamic configuration change via DCS
Given I issue a PATCH request to http://127.0.0.1:8008/config with {"ttl": 20, "loop_wait": 2, "postgresql": {"parameters": {"max_connections": 101}}}
Given I issue a PATCH request to http://127.0.0.1:8008/config with {"ttl": 10, "loop_wait": 2, "postgresql": {"parameters": {"max_connections": 101}}}
Then I receive a response code 200
And I receive a response loop_wait 2
And Response on GET http://127.0.0.1:8008/patroni contains pending_restart after 11 seconds
patroni/dcs/consul.py
+37 -15
View File
@@ -34,15 +34,22 @@ class HTTPClient(object):
self.set_read_timeout(timeout)
self.base_uri = '{0}://{1}:{2}'.format(self.scheme, self.host, self.port)
self.http = urllib3.PoolManager(num_pools=10)
self._ttl = None
def set_read_timeout(self, timeout):
self._read_timeout = timeout/3.0
def set_ttl(self, ttl):
ret = self._ttl != ttl
self._ttl = ttl
return ret
@staticmethod
def response(response):
data = response.data.decode('utf-8')
if response.status == 500:
raise ConsulInternalError()
return base.Response(response.status, response.headers, response.data.decode('utf-8'))
raise ConsulInternalError('{0} {1}'.format(response.status, data))
return base.Response(response.status, response.headers, data)
def uri(self, path, params=None):
return '{0}{1}{2}'.format(self.base_uri, path, params and '?' + urlencode(params) or '')
@@ -52,6 +59,14 @@ class HTTPClient(object):
raise AttributeError("HTTPClient instance has no attribute '{0}'".format(method))
def wrapper(callback, path, params=None, data=''):
# python-consul doesn't allow to specify ttl smaller then 10 seconds
# because session_ttl_min defaults to 10s, so we have to do this ugly dirty hack...
if method == 'put' and path == '/v1/session/create':
ttl = '"ttl": "{0}s"'.format(self._ttl)
if not data or data == '{}':
data = '{' + ttl + '}'
else:
data = data[:-1] + ', ' + ttl + '}'
kwargs = {'retries': 0, 'preload_content': False, 'body': data}
if method == 'get' and isinstance(params, dict) and 'index' in params:
kwargs['timeout'] = (float(params['wait'][:-1]) if 'wait' in params else 300) + 1
@@ -83,17 +98,17 @@ class Consul(AbstractDCS):
super(Consul, self).__init__(config)
self._scope = config['scope']
self._session = None
self._ttl = None
self.__do_not_watch = False
self._retry = Retry(deadline=config['retry_timeout'], max_delay=1, max_tries=-1,
retry_exceptions=(ConsulInternalError, HTTPException,
HTTPError, socket.error, socket.timeout))
self._my_member_data = None
self.set_ttl(config.get('ttl') or 30)
host, port = config.get('host', '127.0.0.1:8500').split(':')
self._client = ConsulClient(host=host, port=port)
self.set_retry_timeout(config['retry_timeout'])
self.set_ttl(config.get('ttl') or 30)
self._last_session_refresh = 0
if not self._ctl:
self.create_session()
@@ -109,11 +124,9 @@ class Consul(AbstractDCS):
sleep(5)
def set_ttl(self, ttl):
ttl = ttl/2.0 # My experiments have shown that session expires after 2*ttl time
if self._ttl != ttl:
if self._client.http.set_ttl(ttl/2.0): # Consul multiplies the TTL by 2x
self._session = None
self.__do_not_watch = True
self._ttl = ttl
def set_retry_timeout(self, retry_timeout):
self._retry.deadline = retry_timeout
@@ -121,15 +134,20 @@ class Consul(AbstractDCS):
def _do_refresh_session(self):
""":returns: `!True` if it had to create new session"""
if self._session and self._last_session_refresh + self._loop_wait > time.time():
return False
if self._session:
try:
return self._client.session.renew(self._session) is None
self._client.session.renew(self._session)
except NotFound:
self._session = None
if not self._session:
name = self._scope + '-' + self._name
self._session = self._client.session.create(name=name, lock_delay=0, behavior='delete', ttl=self._ttl)
return True
ret = not self._session
if ret:
self._session = self._client.session.create(name=self._scope + '-' + self._name,
lock_delay=0.001, behavior='delete')
self._last_session_refresh = time.time()
return ret
def refresh_session(self):
try:
@@ -202,6 +220,7 @@ class Consul(AbstractDCS):
cluster = self.cluster
member = cluster and cluster.get_member(self._name, fallback_to_leader=False)
create_member = self.refresh_session()
if member and (create_member or member.session != self._session):
try:
self._client.kv.delete(self.member_path)
@@ -247,9 +266,12 @@ class Consul(AbstractDCS):
def _write_leader_optime(self, last_operation):
return self._client.kv.put(self.leader_optime_path, last_operation)
@staticmethod
def update_leader():
return True
@catch_consul_errors
def update_leader(self):
if self._session:
self.retry(self._client.session.renew, self._session)
self._last_session_refresh = time.time()
return bool(self._session)
@catch_consul_errors
def initialize(self, create_new=True, sysid=''):
tests/test_consul.py
+18 -8
View File
@@ -37,19 +37,27 @@ def kv_get(self, key, **kwargs):
class TestHTTPClient(unittest.TestCase):
def setUp(self):
self.client = HTTPClient('127.0.0.1', '8500', 'http', False)
self.client.http.request = Mock()
def test_get(self):
client = HTTPClient('127.0.0.1', '8500', 'http', False)
client.http.request = Mock()
client.get(Mock(), '')
client.get(Mock(), '', {'wait': '1s', 'index': 1})
client.http.request.return_value.status = 500
self.assertRaises(ConsulInternalError, client.get, Mock(), '')
self.client.get(Mock(), '')
self.client.get(Mock(), '', {'wait': '1s', 'index': 1})
self.client.http.request.return_value.status = 500
self.assertRaises(ConsulInternalError, self.client.get, Mock(), '')
def test_unknown_method(self):
try:
client.bla(Mock(), '')
self.client.bla(Mock(), '')
self.assertFail()
except Exception as e:
self.assertTrue(isinstance(e, AttributeError))
def test_put(self):
self.client.put(Mock(), '/v1/session/create')
self.client.put(Mock(), '/v1/session/create', data='{"foo": "bar"}')
@patch.object(consul.Consul.KV, 'get', kv_get)
class TestConsul(unittest.TestCase):
@@ -73,7 +81,8 @@ class TestConsul(unittest.TestCase):
@patch.object(consul.Consul.Session, 'create', Mock(side_effect=ConsulException))
def test_referesh_session(self):
self.c._session = '1'
self.c._name = ''
self.assertFalse(self.c.refresh_session())
self.c._last_session_refresh = 0
self.assertRaises(ConsulError, self.c.refresh_session)
@patch.object(consul.Consul.KV, 'delete', Mock())
@@ -115,6 +124,7 @@ class TestConsul(unittest.TestCase):
def test_write_leader_optime(self):
self.c.write_leader_optime('1')
@patch.object(consul.Consul.Session, 'renew', Mock())
def test_update_leader(self):
self.c.update_leader()