Support dynamically registering/deregistering as a consul service and changing tags (#1993)

Close #1988

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

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()