diff --git a/features/environment.py b/features/environment.py index 8fe41064..e0e2af58 100644 --- a/features/environment.py +++ b/features/environment.py @@ -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']) diff --git a/features/patroni_api.feature b/features/patroni_api.feature index 67b886e7..dfdf878b 100644 --- a/features/patroni_api.feature +++ b/features/patroni_api.feature @@ -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 diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index b80c0387..36984da7 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -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=''): diff --git a/tests/test_consul.py b/tests/test_consul.py index d176a6c8..e65d691b 100644 --- a/tests/test_consul.py +++ b/tests/test_consul.py @@ -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()