diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py index e2a0f749..8b4c25bc 100644 --- a/patroni/dcs/consul.py +++ b/patroni/dcs/consul.py @@ -227,12 +227,11 @@ class Consul(AbstractDCS): self._last_session_refresh = 0 self.__session_checks = config.get('checks', []) self._register_service = config.get('register_service', False) + self._previous_loop_register_service = self._register_service + self._service_tags = sorted(config.get('service_tags', [])) + self._previous_loop_service_tags = self._service_tags if self._register_service: - self._service_tags = config.get('service_tags', []) - self._service_name = service_name_from_scope_name(self._scope) - if self._scope != self._service_name: - logger.warning('Using %s as consul service name instead of scope name %s', self._service_name, - self._scope) + self._set_service_name() self._service_check_interval = config.get('service_check_interval', '5s') if not self._ctl: self.create_session() @@ -250,7 +249,18 @@ class Consul(AbstractDCS): def reload_config(self, config): super(Consul, self).reload_config(config) - self._client.reload_config(config.get('consul', {})) + + consul_config = config.get('consul', {}) + self._client.reload_config(consul_config) + self._previous_loop_service_tags = self._service_tags + self._service_tags = sorted(consul_config.get('service_tags', [])) + + should_register_service = consul_config.get('register_service', False) + if should_register_service and not self._register_service: + self._set_service_name() + + self._previous_loop_register_service = self._register_service + self._register_service = should_register_service def set_ttl(self, ttl): if self._client.http.set_ttl(ttl/2.0): # Consul multiplies the TTL by 2x @@ -402,14 +412,18 @@ class Consul(AbstractDCS): self._client.kv.delete(self.member_path) create_member = True + if self._register_service or self._previous_loop_register_service: + try: + self.update_service(not create_member and member and member.data or {}, data) + except Exception: + logger.exception('update_service') + if not create_member and member and deep_compare(data, member.data): return True try: args = {} if permanent else {'acquire': self._session} self._client.kv.put(self.member_path, json.dumps(data, separators=(',', ':')), **args) - if self._register_service: - self.update_service(not create_member and member and member.data or {}, data) return True except InvalidSession: self._session = None @@ -418,6 +432,11 @@ class Consul(AbstractDCS): logger.exception('touch_member') return False + def _set_service_name(self): + self._service_name = service_name_from_scope_name(self._scope) + if self._scope != self._service_name: + logger.warning('Using %s as consul service name instead of scope name %s', self._service_name, self._scope) + @catch_consul_errors def register_service(self, service_name, **kwargs): logger.info('Register service %s, params %s', service_name, kwargs) @@ -441,17 +460,22 @@ class Consul(AbstractDCS): deregister='{0}s'.format(self._client.http.ttl * 10)) tags = self._service_tags[:] tags.append(role) + self._previous_loop_service_tags = self._service_tags + params = { 'service_id': '{0}/{1}'.format(self._scope, self._name), 'address': conn_parts.hostname, 'port': conn_parts.port, 'check': check, - 'tags': tags + 'tags': tags, + 'enable_tag_override': True, } - if state == 'stopped': + if state == 'stopped' or (not self._register_service and self._previous_loop_register_service): + self._previous_loop_register_service = self._register_service return self.deregister_service(params['service_id']) + self._previous_loop_register_service = self._register_service if role in ['master', 'replica', 'standby-leader']: if state != 'running': return @@ -470,7 +494,10 @@ class Consul(AbstractDCS): if old_data.get(key) != new_data[key]: update = True - if force or update: + if ( + force or update or self._register_service != self._previous_loop_register_service + or self._service_tags != self._previous_loop_service_tags + ): return self._update_service(new_data) @catch_consul_errors diff --git a/tests/test_consul.py b/tests/test_consul.py index 6b5f773b..bd503c75 100644 --- a/tests/test_consul.py +++ b/tests/test_consul.py @@ -215,4 +215,50 @@ class TestConsul(unittest.TestCase): self.assertIsNone(self.c.update_service({}, d)) def test_reload_config(self): - self.c.reload_config({'consul': {'token': 'foo'}, 'loop_wait': 10, 'ttl': 30, 'retry_timeout': 10}) + self.assertEqual([], self.c._service_tags) + self.c.reload_config({'consul': {'token': 'foo', 'register_service': True, 'service_tags': ['foo']}, + 'loop_wait': 10, 'ttl': 30, 'retry_timeout': 10}) + self.assertEqual(["foo"], self.c._service_tags) + + self.c.refresh_session = Mock(return_value=False) + + d = {'role': 'replica', 'api_url': 'http://a/t', 'conn_url': 'pg://c:1', 'state': 'running'} + + # Changing register_service from True to False calls deregister() + self.c.reload_config({'consul': {'register_service': False}, 'loop_wait': 10, 'ttl': 30, 'retry_timeout': 10}) + with patch('consul.Consul.Agent.Service.deregister') as mock_deregister: + self.c.touch_member(d) + mock_deregister.assert_called_once() + + self.assertEqual([], self.c._service_tags) + + # register_service staying False between reloads does not call deregister() + self.c.reload_config({'consul': {'register_service': False}, 'loop_wait': 10, 'ttl': 30, 'retry_timeout': 10}) + with patch('consul.Consul.Agent.Service.deregister') as mock_deregister: + self.c.touch_member(d) + self.assertFalse(mock_deregister.called) + + # Changing register_service from False to True calls register() + self.c.reload_config({'consul': {'register_service': True}, 'loop_wait': 10, 'ttl': 30, 'retry_timeout': 10}) + with patch('consul.Consul.Agent.Service.register') as mock_register: + self.c.touch_member(d) + mock_register.assert_called_once() + + # register_service staying True between reloads does not call register() + self.c.reload_config({'consul': {'register_service': True}, 'loop_wait': 10, 'ttl': 30, 'retry_timeout': 10}) + with patch('consul.Consul.Agent.Service.register') as mock_register: + self.c.touch_member(d) + self.assertFalse(mock_deregister.called) + + # register_service staying True between reloads does calls register() if other service data has changed + self.c.reload_config({'consul': {'register_service': True}, 'loop_wait': 10, 'ttl': 30, 'retry_timeout': 10}) + with patch('consul.Consul.Agent.Service.register') as mock_register: + self.c.touch_member(d) + mock_register.assert_called_once() + + # register_service staying True between reloads does calls register() if service_tags have changed + self.c.reload_config({'consul': {'register_service': True, 'service_tags': ['foo']}, 'loop_wait': 10, + 'ttl': 30, 'retry_timeout': 10}) + with patch('consul.Consul.Agent.Service.register') as mock_register: + self.c.touch_member(d) + mock_register.assert_called_once()