Register services in service registry during the service bootstrap phase

CHANGELOG.rst

@@ -17,6 +17,20 @@ Added
 
    For backward compatibility reasons, if pack metadata file doesn't contain that attribute, it's
    assumed it only works with Python 2. (new feature) #4474
+* Update service bootstrap code and make sure all the services register in a service registry once
+  they come online and become available.
+
+  This functionality is only used internally and will only work if configuration backend is
+  correctly configured in ``st2.conf`` (new feature) #4548
+* Add new ``GET /v1/service_registry/groups`` and
+  ``GET /v1/service_registry/groups/<group_id>/members`` API endpoint for listing available service
+  registry groups and members.
+
+  NOTE: This API endpoint is behind a RBAC control check and can only be views by the admins.
+  (new feature) #4548
+
+  Also add corresponding CLI commands - ``st2 service-registry group list``,
+  ``st2 service registry member list [--group-id=<group id>]``
 
 * Adding ``Cache-Control`` header to all API responses so clients will favor
   refresh from API instead of using cached version.

conf/st2.dev.conf

@@ -77,6 +77,17 @@ port = 514
 facility = local7
 protocol = udp
 
+[coordination]
+# NOTE: Service registry / groups functionality is only available in the following drivers:
+# - zookeeper
+# - redis
+# - etcd3
+# - etcd3gw
+# Keep in mind that zake driver works in process so it won't work when testing cross process 
+# / cross server functionality
+#url = redis://localhost
+#url = kazoo://localhost
+
 [webui]
 # webui_base_url = https://mywebhost.domain
 

st2actions/st2actions/cmd/actionrunner.py

@@ -33,8 +33,13 @@ def sigterm_handler(signum=None, frame=None):
 
 
 def _setup():
+    capabilities = {
+        'name': 'actionrunner',
+        'type': 'passive'
+    }
     common_setup(service='actionrunner', config=config, setup_db=True, register_mq_exchanges=True,
-                 register_signal_handlers=True)
+                 register_signal_handlers=True, service_registry=True, capabilities=capabilities)
+
     _setup_sigterm_handler()
 
 

st2actions/st2actions/cmd/scheduler.py

@@ -46,8 +46,13 @@ def sigterm_handler(signum=None, frame=None):
 
 
 def _setup():
+    capabilities = {
+        'name': 'scheduler',
+        'type': 'passive'
+    }
     common_setup(service='scheduler', config=config, setup_db=True, register_mq_exchanges=True,
-                 register_signal_handlers=True)
+                 register_signal_handlers=True, service_registry=True, capabilities=capabilities)
+
     _setup_sigterm_handler()
 
 

st2actions/st2actions/cmd/st2notifier.py

@@ -19,8 +19,12 @@
 
 
 def _setup():
+    capabilities = {
+        'name': 'notifier',
+        'type': 'passive'
+    }
     common_setup(service='notifier', config=config, setup_db=True, register_mq_exchanges=True,
-                 register_signal_handlers=True)
+                 register_signal_handlers=True, service_registry=True, capabilities=capabilities)
 
 
 def _run_worker():

st2actions/st2actions/cmd/st2resultstracker.py

@@ -20,8 +20,12 @@
 
 
 def _setup():
-    common_setup(service='resultstracker', config=config, setup_db=True,
-                 register_mq_exchanges=True, register_signal_handlers=True)
+    capabilities = {
+        'name': 'resultstracker',
+        'type': 'passive'
+    }
+    common_setup(service='resultstracker', config=config, setup_db=True, register_mq_exchanges=True,
+                 register_signal_handlers=True, service_registry=True, capabilities=capabilities)
 
 
 def _run_worker():

st2actions/st2actions/cmd/workflow_engine.py

@@ -51,12 +51,18 @@ def sigterm_handler(signum=None, frame=None):
 
 
 def setup():
+    capabilities = {
+        'name': 'workflowengine',
+        'type': 'passive'
+    }
     common_setup(
         service='workflow_engine',
         config=config,
         setup_db=True,
         register_mq_exchanges=True,
-        register_signal_handlers=True
+        register_signal_handlers=True,
+        service_registry=True,
+        capabilities=capabilities
     )
 
     setup_sigterm_handler()

st2actions/st2actions/scheduler/handler.py

@@ -61,7 +61,7 @@ def __init__(self):
         self.message_type = LiveActionDB
         self._shutdown = False
         self._pool = eventlet.GreenPool(size=cfg.CONF.scheduler.pool_size)
-        self._coordinator = coordination_service.get_coordinator()
+        self._coordinator = coordination_service.get_coordinator(start_heart=True)
         self._main_thread = None
         self._cleanup_thread = None
 

st2actions/tests/unit/policies/test_concurrency.py

@@ -98,6 +98,7 @@ def setUpClass(cls):
     @classmethod
     def tearDownClass(cls):
         # Reset the coordinator.
+        coordination.coordinator_teardown(coordination.COORDINATOR)
         coordination.COORDINATOR = None
 
         super(ConcurrencyPolicyTestCase, cls).tearDownClass()

st2actions/tests/unit/policies/test_concurrency_by_attr.py

@@ -97,6 +97,7 @@ def setUpClass(cls):
     @classmethod
     def tearDownClass(cls):
         # Reset the coordinator.
+        coordination.coordinator_teardown(coordination.COORDINATOR)
         coordination.COORDINATOR = None
 
         super(ConcurrencyByAttributePolicyTestCase, cls).tearDownClass()

st2api/st2api/app.py

@@ -46,6 +46,13 @@ def setup_app(config={}):
         monkey_patch()
 
         st2api_config.register_opts()
+        capabilities = {
+            'name': 'api',
+            'listen_host': cfg.CONF.api.host,
+            'listen_port': cfg.CONF.api.port,
+            'type': 'active'
+        }
+
         # This should be called in gunicorn case because we only want
         # workers to connect to db, rabbbitmq etc. In standalone HTTP
         # server case, this setup would have already occurred.
@@ -54,6 +61,8 @@ def setup_app(config={}):
                      register_signal_handlers=True,
                      register_internal_trigger_types=True,
                      run_migrations=True,
+                     service_registry=True,
+                     capabilities=capabilities,
                      config_args=config.get('config_args', None))
 
     # Additional pre-run time checks

st2api/st2api/cmd/api.py

@@ -43,8 +43,16 @@
 
 
 def _setup():
+    capabilities = {
+        'name': 'api',
+        'listen_host': cfg.CONF.api.host,
+        'listen_port': cfg.CONF.api.port,
+        'type': 'active'
+    }
+
     common_setup(service='api', config=config, setup_db=True, register_mq_exchanges=True,
-                 register_signal_handlers=True, register_internal_trigger_types=True)
+                 register_signal_handlers=True, register_internal_trigger_types=True,
+                 service_registry=True, capabilities=capabilities)
 
     # Additional pre-run time checks
     validate_rbac_is_correctly_configured()

st2api/st2api/controllers/v1/service_registry.py

@@ -0,0 +1,77 @@
+# Licensed to the StackStorm, Inc ('StackStorm') under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import six
+
+from tooz.coordination import GroupNotCreated
+
+from st2common.services import coordination
+from st2common.exceptions.db import StackStormDBObjectNotFoundError
+from st2common.rbac import utils as rbac_utils
+
+__all__ = [
+    'ServiceRegistryGroupsController',
+    'ServiceRegistryGroupMembersController',
+]
+
+
+class ServiceRegistryGroupsController(object):
+    def get_all(self, requester_user):
+        rbac_utils.assert_user_is_admin(user_db=requester_user)
+
+        coordinator = coordination.get_coordinator()
+
+        group_ids = list(coordinator.get_groups().get())
+        group_ids = [item.decode('utf-8') for item in group_ids]
+
+        result = {
+            'groups': group_ids
+        }
+        return result
+
+
+class ServiceRegistryGroupMembersController(object):
+    def get_one(self, group_id, requester_user):
+        rbac_utils.assert_user_is_admin(user_db=requester_user)
+
+        coordinator = coordination.get_coordinator()
+
+        if not isinstance(group_id, six.binary_type):
+            group_id = group_id.encode('utf-8')
+
+        try:
+            member_ids = list(coordinator.get_members(group_id).get())
+        except GroupNotCreated:
+            msg = ('Group with ID "%s" not found.' % (group_id.decode('utf-8')))
+            raise StackStormDBObjectNotFoundError(msg)
+
+        result = {
+            'members': []
+        }
+
+        for member_id in member_ids:
+            capabilities = coordinator.get_member_capabilities(group_id, member_id).get()
+            item = {
+                'group_id': group_id.decode('utf-8'),
+                'member_id': member_id.decode('utf-8'),
+                'capabilities': capabilities
+            }
+            result['members'].append(item)
+
+        return result
+
+
+groups_controller = ServiceRegistryGroupsController()
+members_controller = ServiceRegistryGroupMembersController()

st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py

@@ -23,8 +23,12 @@
     from st2api.controllers.v1.webhooks import HooksHolder
 from st2common.persistence.rbac import UserRoleAssignment
 from st2common.models.db.rbac import UserRoleAssignmentDB
+from st2common.service_setup import register_service_in_service_registry
+from st2common.services import coordination
 
+from st2tests import config as tests_config
 from st2tests.fixturesloader import FixturesLoader
+
 from tests.base import APIControllerWithRBACTestCase
 from tests.unit.controllers.v1.test_webhooks import DUMMY_TRIGGER_DICT
 
@@ -110,6 +114,29 @@ class APIControllersRBACTestCase(APIControllerWithRBACTestCase):
     register_packs = True
     fixtures_loader = FixturesLoader()
 
+    coordinator = None
+
+    @classmethod
+    def setUpClass(cls):
+        tests_config.parse_args(coordinator_noop=True)
+
+        super(APIControllersRBACTestCase, cls).setUpClass()
+
+        cls.coordinator = coordination.get_coordinator(use_cache=False)
+
+        # Register mock service in the service registry for testing purposes
+        service = six.binary_type(six.text_type('mock_service').encode('ascii'))
+        register_service_in_service_registry(service=service,
+                                             capabilities={'key1': 'value1',
+                                                           'name': 'mock_service'},
+                                             start_heart=True)
+
+    @classmethod
+    def tearDownClass(cls):
+        super(APIControllersRBACTestCase, cls).tearDownClass()
+
+        coordination.coordinator_teardown(cls.coordinator)
+
     def setUp(self):
         super(APIControllersRBACTestCase, self).setUp()
 
@@ -450,6 +477,17 @@ def test_api_endpoints_behind_rbac_wall(self):
                 'path': '/v1/rules/views',
                 'method': 'GET',
                 'is_getall': True
+            },
+            # Service registry
+            {
+                'path': '/v1/service_registry/groups',
+                'method': 'GET',
+                'is_getall': True
+            },
+            {
+                'path': '/v1/service_registry/groups/mock_service/members',
+                'method': 'GET',
+                'is_getall': True
             }
         ]