From 3625ecd63cf6ef2d7f498f53e33c4883b700778d Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Wed, 13 Feb 2019 10:43:47 +0100 Subject: [PATCH 01/43] Add utility script for listing tooz / service registry groups and members with their capabilities. --- tools/list_group_members.py | 70 +++++++++++++++++++++++++++++++++++++ 1 file changed, 70 insertions(+) create mode 100755 tools/list_group_members.py diff --git a/tools/list_group_members.py b/tools/list_group_members.py new file mode 100755 index 0000000000..63c78c294d --- /dev/null +++ b/tools/list_group_members.py @@ -0,0 +1,70 @@ +#!/usr/bin/env python +# 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. + +from __future__ import absolute_import + +import six + +from oslo_config import cfg + +from st2common import config +from st2common.services import coordination + +""" +Tool which lists all the services registered in the service registry and their capabilities. +""" + +def main(group_id=None): + coordinator = coordination.get_coordinator() + + if not group_id: + group_ids = list(coordinator.get_groups().get()) + + print('Available groups:') + for group_id in group_ids: + print(' - %s' % (group_id)) + print('') + else: + group_ids = [group_id] + + for group_id in group_ids: + print('Members in group "%s":' % (group_id)) + member_ids = list(coordinator.get_members(group_id).get()) + + for member_id in member_ids: + capabilities = coordinator.get_member_capabilities(group_id, member_id).get() + print(' - %s (capabilities=%s)' % (member_id, str(capabilities))) + + +def do_register_cli_opts(opts, ignore_errors=False): + for opt in opts: + try: + cfg.CONF.register_cli_opt(opt) + except: + if not ignore_errors: + raise + + +if __name__ == '__main__': + cli_opts = [ + cfg.StrOpt('group-id', default=None, + help='If provided, only list members for that group.'), + + ] + do_register_cli_opts(cli_opts) + config.parse_args() + + main(group_id=cfg.CONF.group_id) From b95e7c8003a7b33ce605a7c51433ffeb7aa2e1cd Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Wed, 13 Feb 2019 11:28:04 +0100 Subject: [PATCH 02/43] Add support for registering a service in service registry inside "service_setup" function. --- st2common/st2common/service_setup.py | 40 +++++++++++++++++++- st2common/st2common/services/coordination.py | 22 ++++++++--- 2 files changed, 55 insertions(+), 7 deletions(-) diff --git a/st2common/st2common/service_setup.py b/st2common/st2common/service_setup.py index bf82e00192..7909d78231 100644 --- a/st2common/st2common/service_setup.py +++ b/st2common/st2common/service_setup.py @@ -24,7 +24,9 @@ import traceback import logging as stdlib_logging +import six from oslo_config import cfg +from tooz.coordination import GroupAlreadyExist from st2common import log as logging from st2common.constants.logging import DEFAULT_LOGGING_CONF_PATH @@ -37,6 +39,8 @@ from st2common import triggers from st2common.rbac.migrations import run_all as run_all_rbac_migrations from st2common.logging.filters import LogLevelFilter +from st2common.util import system_info +from st2common.services import coordination # Note: This is here for backward compatibility. # Function has been moved in a standalone module to avoid expensive in-direct @@ -59,7 +63,8 @@ def setup(service, config, setup_db=True, register_mq_exchanges=True, register_signal_handlers=True, register_internal_trigger_types=False, - run_migrations=True, register_runners=True, config_args=None): + run_migrations=True, register_runners=True, service_registry=False, + capabilities=None, config_args=None): """ Common setup function. @@ -73,10 +78,13 @@ def setup(service, config, setup_db=True, register_mq_exchanges=True, 5. Registers common signal handlers 6. Register internal trigger types 7. Register all the runners which are installed inside StackStorm virtualenv. + 8. Register service in the service registry with the provided capabilities :param service: Name of the service. :param config: Config object to use to parse args. """ + capabilities = capabilities or {} + # Set up logger which logs everything which happens during and before config # parsing to sys.stdout logging.setup(DEFAULT_LOGGING_CONF_PATH, excludes=None) @@ -165,9 +173,39 @@ def setup(service, config, setup_db=True, register_mq_exchanges=True, metrics_initialize() + # Register service in the service registry + if service_registry: + # NOTE: It's important that we pass start_heart=True to start the hearbeat process + coordinator = coordination.get_coordinator(start_heart=True) + + member_id = coordination.get_member_id() + + # 1. Create a group with the name of the service + group_id = six.binary_type(six.text_type(service).encode('ascii')) + + try: + coordinator.create_group(group_id).get() + except GroupAlreadyExist: + pass + + # Include common capabilities such as hostname and process ID + proc_info = system_info.get_process_info() + capabilities['hostname'] = proc_info['hostname'] + capabilities['pid'] = proc_info['pid'] + + # 1. Join the group as a member + LOG.debug('Joining service registry group "%s" as member_id "%s" with capabilities "%s"' % + (group_id, member_id, capabilities)) + coordinator.join_group(group_id, capabilities=capabilities).get() + def teardown(): """ Common teardown function. """ + # 1. Tear down the database db_teardown() + + # 2. Tear down the coordinator + coordinator = coordination.get_coordinator() + coordination.coordinator_teardown(coordinator) diff --git a/st2common/st2common/services/coordination.py b/st2common/st2common/services/coordination.py index e7dcd41937..13f93f74c0 100644 --- a/st2common/st2common/services/coordination.py +++ b/st2common/st2common/services/coordination.py @@ -30,7 +30,9 @@ __all__ = [ 'configured', + 'get_coordinator', + 'get_memeber_id', 'coordinator_setup', 'coordinator_teardown' @@ -137,7 +139,7 @@ def configured(): return backend_configured and not mock_backend -def coordinator_setup(): +def coordinator_setup(start_heart=False): """ Sets up the client for the coordination service. @@ -149,8 +151,7 @@ def coordinator_setup(): """ url = cfg.CONF.coordination.url lock_timeout = cfg.CONF.coordination.lock_timeout - proc_info = system_info.get_process_info() - member_id = six.b('%s_%d' % (proc_info['hostname'], proc_info['pid'])) + member_id = get_member_id() if url: coordinator = coordination.get_coordinator(url, member_id, lock_timeout=lock_timeout) @@ -161,7 +162,7 @@ def coordinator_setup(): # to work coordinator = NoOpDriver(member_id) - coordinator.start() + coordinator.start(start_heart=start_heart) return coordinator @@ -169,7 +170,7 @@ def coordinator_teardown(coordinator): coordinator.stop() -def get_coordinator(): +def get_coordinator(start_heart=False): global COORDINATOR if not configured(): @@ -177,6 +178,15 @@ def get_coordinator(): 'service will use best effort approach and race conditions are possible.') if not COORDINATOR: - COORDINATOR = coordinator_setup() + COORDINATOR = coordinator_setup(start_heart=start_heart) return COORDINATOR + + +def get_member_id(): + """ + Retrieve member if for the current process. + """ + proc_info = system_info.get_process_info() + member_id = six.b('%s_%d' % (proc_info['hostname'], proc_info['pid'])) + return member_id From eb56273a8f7b8d1b3a9f40c8f2fa2e8520a7342b Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Wed, 13 Feb 2019 11:48:57 +0100 Subject: [PATCH 03/43] Fix lint and Python 3 compatibility. --- tools/list_group_members.py | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/tools/list_group_members.py b/tools/list_group_members.py index 63c78c294d..d03d116672 100755 --- a/tools/list_group_members.py +++ b/tools/list_group_members.py @@ -16,8 +16,6 @@ from __future__ import absolute_import -import six - from oslo_config import cfg from st2common import config @@ -27,13 +25,15 @@ Tool which lists all the services registered in the service registry and their capabilities. """ + def main(group_id=None): coordinator = coordination.get_coordinator() if not group_id: group_ids = list(coordinator.get_groups().get()) + group_ids = [group_id.decode('utf-8') for group_id in group_ids] - print('Available groups:') + print('Available groups (%s):' % (len(group_ids))) for group_id in group_ids: print(' - %s' % (group_id)) print('') @@ -41,8 +41,10 @@ def main(group_id=None): group_ids = [group_id] for group_id in group_ids: - print('Members in group "%s":' % (group_id)) member_ids = list(coordinator.get_members(group_id).get()) + member_ids = [member_id.decode('utf-8') for member_id in member_ids] + + print('Members in group "%s" (%s):' % (group_id, len(member_ids))) for member_id in member_ids: capabilities = coordinator.get_member_capabilities(group_id, member_id).get() @@ -61,7 +63,7 @@ def do_register_cli_opts(opts, ignore_errors=False): if __name__ == '__main__': cli_opts = [ cfg.StrOpt('group-id', default=None, - help='If provided, only list members for that group.'), + help='If provided, only list members for that group.'), ] do_register_cli_opts(cli_opts) From 293757a8566389bd1e0110a30ed90ce5b25eb9c9 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Wed, 13 Feb 2019 12:03:35 +0100 Subject: [PATCH 04/43] Make sure we register all the services in the service registry. --- st2actions/st2actions/cmd/actionrunner.py | 7 ++++++- st2actions/st2actions/cmd/scheduler.py | 7 ++++++- st2actions/st2actions/cmd/st2notifier.py | 6 +++++- st2actions/st2actions/cmd/st2resultstracker.py | 8 ++++++-- st2actions/st2actions/cmd/workflow_engine.py | 6 ++++++ st2api/st2api/app.py | 9 +++++++++ st2api/st2api/cmd/api.py | 10 +++++++++- st2auth/st2auth/app.py | 12 +++++++++++- st2auth/st2auth/cmd/api.py | 9 ++++++++- st2reactor/st2reactor/cmd/garbagecollector.py | 8 +++++++- st2reactor/st2reactor/cmd/rulesengine.py | 6 +++++- st2reactor/st2reactor/cmd/sensormanager.py | 6 +++++- st2reactor/st2reactor/cmd/timersengine.py | 6 +++++- st2stream/st2stream/app.py | 8 ++++++++ st2stream/st2stream/cmd/api.py | 8 +++++++- 15 files changed, 103 insertions(+), 13 deletions(-) diff --git a/st2actions/st2actions/cmd/actionrunner.py b/st2actions/st2actions/cmd/actionrunner.py index d02e6ccd9a..9f1163d25f 100644 --- a/st2actions/st2actions/cmd/actionrunner.py +++ b/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() diff --git a/st2actions/st2actions/cmd/scheduler.py b/st2actions/st2actions/cmd/scheduler.py index 1ae0096084..40b33cb3ad 100644 --- a/st2actions/st2actions/cmd/scheduler.py +++ b/st2actions/st2actions/cmd/scheduler.py @@ -31,8 +31,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() diff --git a/st2actions/st2actions/cmd/st2notifier.py b/st2actions/st2actions/cmd/st2notifier.py index b0146ea591..718a4ceefd 100644 --- a/st2actions/st2actions/cmd/st2notifier.py +++ b/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(): diff --git a/st2actions/st2actions/cmd/st2resultstracker.py b/st2actions/st2actions/cmd/st2resultstracker.py index 312a2cd73a..5af5e151c2 100644 --- a/st2actions/st2actions/cmd/st2resultstracker.py +++ b/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(): diff --git a/st2actions/st2actions/cmd/workflow_engine.py b/st2actions/st2actions/cmd/workflow_engine.py index 71a4ec6025..be8475310a 100644 --- a/st2actions/st2actions/cmd/workflow_engine.py +++ b/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 + service_registry=True, + capabilities=capabilities ) setup_sigterm_handler() diff --git a/st2api/st2api/app.py b/st2api/st2api/app.py index 27c9a815ff..e127d91426 100644 --- a/st2api/st2api/app.py +++ b/st2api/st2api/app.py @@ -45,6 +45,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. @@ -53,6 +60,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 diff --git a/st2api/st2api/cmd/api.py b/st2api/st2api/cmd/api.py index c412dd3c33..6ae6ec877d 100644 --- a/st2api/st2api/cmd/api.py +++ b/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() diff --git a/st2auth/st2auth/app.py b/st2auth/st2auth/app.py index 0dd154bb11..8fd6e6d610 100644 --- a/st2auth/st2auth/app.py +++ b/st2auth/st2auth/app.py @@ -43,15 +43,25 @@ def setup_app(config={}): # including shutdown monkey_patch() + st2auth_config.register_opts() + capabilities = { + 'name': 'auth', + 'listen_host': cfg.CONF.auth.host, + 'listen_port': cfg.CONF.auth.port, + 'listen_ssl': cfg.CONF.auth.use_ssl, + '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. - st2auth_config.register_opts() common_setup(service='auth', config=st2auth_config, setup_db=True, register_mq_exchanges=False, register_signal_handlers=True, register_internal_trigger_types=False, run_migrations=False, + service_registry=True, + capabilities=capabilities, config_args=config.get('config_args', None)) # Additional pre-run time checks diff --git a/st2auth/st2auth/cmd/api.py b/st2auth/st2auth/cmd/api.py index 74d1536252..01fee95007 100644 --- a/st2auth/st2auth/cmd/api.py +++ b/st2auth/st2auth/cmd/api.py @@ -41,9 +41,16 @@ def _setup(): + capabilities = { + 'name': 'auth', + 'listen_host': cfg.CONF.auth.host, + 'listen_port': cfg.CONF.auth.port, + 'listen_ssl': cfg.CONF.auth.use_ssl, + 'type': 'active' + } common_setup(service='auth', config=config, setup_db=True, register_mq_exchanges=False, register_signal_handlers=True, register_internal_trigger_types=False, - run_migrations=False) + run_migrations=False, service_registry=True, capabilities=capabilities) # Additional pre-run time checks validate_auth_backend_is_correctly_configured() diff --git a/st2reactor/st2reactor/cmd/garbagecollector.py b/st2reactor/st2reactor/cmd/garbagecollector.py index b425172e33..3e42c0b629 100644 --- a/st2reactor/st2reactor/cmd/garbagecollector.py +++ b/st2reactor/st2reactor/cmd/garbagecollector.py @@ -14,6 +14,7 @@ # limitations under the License. from __future__ import absolute_import + import os import sys @@ -40,9 +41,14 @@ def _setup(): + capabilities = { + 'name': 'garbagecollector', + 'type': 'passive' + } common_setup(service='garbagecollector', config=config, setup_db=True, register_mq_exchanges=True, register_signal_handlers=True, - register_runners=False) + register_runners=False, service_registry=True, + capabilities=capabilities) def _teardown(): diff --git a/st2reactor/st2reactor/cmd/rulesengine.py b/st2reactor/st2reactor/cmd/rulesengine.py index 22ae356648..a8a9c2347c 100644 --- a/st2reactor/st2reactor/cmd/rulesengine.py +++ b/st2reactor/st2reactor/cmd/rulesengine.py @@ -33,9 +33,13 @@ def _setup(): + capabilities = { + 'name': 'rulesengine', + 'type': 'passive' + } common_setup(service='rulesengine', config=config, setup_db=True, register_mq_exchanges=True, register_signal_handlers=True, register_internal_trigger_types=True, - register_runners=False) + register_runners=False, service_registry=True, capabilities=capabilities) def _teardown(): diff --git a/st2reactor/st2reactor/cmd/sensormanager.py b/st2reactor/st2reactor/cmd/sensormanager.py index 0fbada701e..d066cff228 100644 --- a/st2reactor/st2reactor/cmd/sensormanager.py +++ b/st2reactor/st2reactor/cmd/sensormanager.py @@ -42,9 +42,13 @@ def _setup(): + capabilities = { + 'name': 'sensorcontainer', + 'type': 'passive' + } common_setup(service='sensorcontainer', config=config, setup_db=True, register_mq_exchanges=True, register_signal_handlers=True, - register_runners=False) + register_runners=False, service_registry=True, capabilities=capabilities) def _teardown(): diff --git a/st2reactor/st2reactor/cmd/timersengine.py b/st2reactor/st2reactor/cmd/timersengine.py index acf1e6932e..998eeaecc0 100644 --- a/st2reactor/st2reactor/cmd/timersengine.py +++ b/st2reactor/st2reactor/cmd/timersengine.py @@ -36,8 +36,12 @@ def _setup(): + capabilities = { + 'name': 'timerengine', + 'type': 'passive' + } common_setup(service='timer_engine', config=config, setup_db=True, register_mq_exchanges=True, - register_signal_handlers=True) + register_signal_handlers=True, service_registry=True, capabilities=capabilities) def _teardown(): diff --git a/st2stream/st2stream/app.py b/st2stream/st2stream/app.py index 588aaf026a..83579ec32e 100644 --- a/st2stream/st2stream/app.py +++ b/st2stream/st2stream/app.py @@ -53,6 +53,12 @@ def setup_app(config={}): monkey_patch() st2stream_config.register_opts() + capabilities = { + 'name': 'stream', + 'listen_host': cfg.CONF.stream.host, + 'listen_port': cfg.CONF.stream.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. @@ -61,6 +67,8 @@ def setup_app(config={}): register_signal_handlers=True, register_internal_trigger_types=False, run_migrations=False, + service_registry=True, + capabilities=capabilities, config_args=config.get('config_args', None)) router = Router(debug=cfg.CONF.stream.debug, auth=cfg.CONF.auth.enable, diff --git a/st2stream/st2stream/cmd/api.py b/st2stream/st2stream/cmd/api.py index 55e48f4648..ab9741de3c 100644 --- a/st2stream/st2stream/cmd/api.py +++ b/st2stream/st2stream/cmd/api.py @@ -49,9 +49,15 @@ def _setup(): + capabilities = { + 'name': 'stream', + 'listen_host': cfg.CONF.stream.host, + 'listen_port': cfg.CONF.stream.port, + 'type': 'active' + } common_setup(service='stream', config=config, setup_db=True, register_mq_exchanges=True, register_signal_handlers=True, register_internal_trigger_types=False, - run_migrations=False) + run_migrations=False, service_registry=True, capabilities=capabilities) def _run_server(): From 079448c1bc21abca55dd1310cc632788f504725b Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Wed, 13 Feb 2019 12:04:22 +0100 Subject: [PATCH 05/43] Update config with sample configuration for coordination section. --- conf/st2.dev.conf | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/conf/st2.dev.conf b/conf/st2.dev.conf index 52f9ac61dc..798aad32ca 100644 --- a/conf/st2.dev.conf +++ b/conf/st2.dev.conf @@ -77,6 +77,16 @@ 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 + [webui] # webui_base_url = https://mywebhost.domain From 6d2d7d971764f892f0e49a09decff66fb5d674cf Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Wed, 13 Feb 2019 12:07:16 +0100 Subject: [PATCH 06/43] Update NoOpCoordinationBackend so it correctly works with group functionality. --- st2common/st2common/services/coordination.py | 22 +++++++++++++------- 1 file changed, 15 insertions(+), 7 deletions(-) diff --git a/st2common/st2common/services/coordination.py b/st2common/st2common/services/coordination.py index 13f93f74c0..3053376183 100644 --- a/st2common/st2common/services/coordination.py +++ b/st2common/st2common/services/coordination.py @@ -32,7 +32,7 @@ 'configured', 'get_coordinator', - 'get_memeber_id', + 'get_member_id', 'coordinator_setup', 'coordinator_teardown' @@ -53,6 +53,14 @@ def heartbeat(self): return True +class NoOpAsyncResult(object): + def __init__(self, result=None): + self._result = result + + def get(self): + return self._result + + class NoOpDriver(coordination.CoordinationDriver): """ Tooz driver where each operation is a no-op. @@ -87,27 +95,27 @@ def stand_down_group_leader(group_id): @staticmethod def create_group(group_id): - return None + return NoOpAsyncResult() @staticmethod def get_groups(): - return None + return NoOpAsyncResult(result=[]) @staticmethod def join_group(group_id, capabilities=''): - return None + return NoOpAsyncResult() @staticmethod def leave_group(group_id): - return None + return NoOpAsyncResult() @staticmethod def delete_group(group_id): - return None + return NoOpAsyncResult() @staticmethod def get_members(group_id): - return None + return NoOpAsyncResult(result=[]) @staticmethod def get_member_capabilities(group_id, member_id): From 46d7f564950a913712c204bba598db8782b98958 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Wed, 13 Feb 2019 12:47:50 +0100 Subject: [PATCH 07/43] Fix lint. --- st2actions/st2actions/cmd/workflow_engine.py | 2 +- tools/list_group_members.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/st2actions/st2actions/cmd/workflow_engine.py b/st2actions/st2actions/cmd/workflow_engine.py index be8475310a..8720b77479 100644 --- a/st2actions/st2actions/cmd/workflow_engine.py +++ b/st2actions/st2actions/cmd/workflow_engine.py @@ -60,7 +60,7 @@ def setup(): config=config, setup_db=True, register_mq_exchanges=True, - register_signal_handlers=True + register_signal_handlers=True, service_registry=True, capabilities=capabilities ) diff --git a/tools/list_group_members.py b/tools/list_group_members.py index d03d116672..5bd0ec6fec 100755 --- a/tools/list_group_members.py +++ b/tools/list_group_members.py @@ -31,7 +31,7 @@ def main(group_id=None): if not group_id: group_ids = list(coordinator.get_groups().get()) - group_ids = [group_id.decode('utf-8') for group_id in group_ids] + group_ids = [group_id_.decode('utf-8') for group_id_ in group_ids] print('Available groups (%s):' % (len(group_ids))) for group_id in group_ids: From b7e618fb8fec40b320d8af0b1faaf22102dd480e Mon Sep 17 00:00:00 2001 From: W Chan Date: Wed, 13 Feb 2019 00:54:08 +0000 Subject: [PATCH 08/43] Fix scheduler test configs in unit tests Add or move the parsing of test configs to the top of affected test modules and make sure the scheduler default config options do not conflict with test configs. --- .../tests/unit/test_actionchain_cancel.py | 3 +++ .../tests/unit/test_actionchain_pause_resume.py | 3 +++ st2actions/st2actions/scheduler/config.py | 9 ++++++++- st2actions/tests/unit/test_scheduler.py | 6 +++--- 4 files changed, 17 insertions(+), 4 deletions(-) diff --git a/contrib/runners/action_chain_runner/tests/unit/test_actionchain_cancel.py b/contrib/runners/action_chain_runner/tests/unit/test_actionchain_cancel.py index 1e663f3dab..e53a1673ad 100644 --- a/contrib/runners/action_chain_runner/tests/unit/test_actionchain_cancel.py +++ b/contrib/runners/action_chain_runner/tests/unit/test_actionchain_cancel.py @@ -19,6 +19,9 @@ import os import tempfile +from st2tests import config as test_config +test_config.parse_args() + from st2common.bootstrap import actionsregistrar from st2common.bootstrap import runnersregistrar diff --git a/contrib/runners/action_chain_runner/tests/unit/test_actionchain_pause_resume.py b/contrib/runners/action_chain_runner/tests/unit/test_actionchain_pause_resume.py index 189d74ec34..0b25930a2f 100644 --- a/contrib/runners/action_chain_runner/tests/unit/test_actionchain_pause_resume.py +++ b/contrib/runners/action_chain_runner/tests/unit/test_actionchain_pause_resume.py @@ -19,6 +19,9 @@ import os import tempfile +from st2tests import config as test_config +test_config.parse_args() + from st2common.bootstrap import actionsregistrar from st2common.bootstrap import runnersregistrar diff --git a/st2actions/st2actions/scheduler/config.py b/st2actions/st2actions/scheduler/config.py index ed7d7477be..0dac7c2045 100644 --- a/st2actions/st2actions/scheduler/config.py +++ b/st2actions/st2actions/scheduler/config.py @@ -19,6 +19,10 @@ from st2common import config as common_config from st2common.constants import system as sys_constants +from st2common import log as logging + + +LOG = logging.getLogger(__name__) def parse_args(args=None): @@ -62,4 +66,7 @@ def _register_service_opts(): cfg.CONF.register_opts(scheduler_opts, group='scheduler') -register_opts() +try: + register_opts() +except cfg.DuplicateOptError: + LOG.exception('The scheduler configuration options are already parsed and loaded.') diff --git a/st2actions/tests/unit/test_scheduler.py b/st2actions/tests/unit/test_scheduler.py index 96620ea5d8..41c0b437f3 100644 --- a/st2actions/tests/unit/test_scheduler.py +++ b/st2actions/tests/unit/test_scheduler.py @@ -19,6 +19,9 @@ import mock import eventlet +from st2tests import config as test_config +test_config.parse_args() + import st2common from st2tests import ExecutionDbTestCase from st2tests.fixturesloader import FixturesLoader @@ -39,9 +42,6 @@ from st2common.services import executions as execution_service from st2common.exceptions import db as db_exc -from st2tests import config as test_config -test_config.parse_args() - LIVE_ACTION = { 'parameters': { From 09ed1b5bb907e4ceef3da8a24ebdfa5fbe8e42e4 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Wed, 13 Feb 2019 15:51:51 +0100 Subject: [PATCH 09/43] Add a changelog entry. --- CHANGELOG.rst | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 7fb1e76a65..7bf846ff3a 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -26,6 +26,11 @@ Added NOTE: Those options are only supported when using a default and officially supported AMQP backend with RabbitMQ server. (new feature) #4541 +* 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 Changed ~~~~~~~ From 4660d973d8ec93f6e46660574a462a38ddf319d7 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Thu, 14 Feb 2019 17:13:43 +0100 Subject: [PATCH 10/43] Add two new service registry related admin API endpoint. 1. ``GET /v1/service_registry/groups`` - list available groups. 2. ``GET /v1/service_registry/groups//members`` - list available / active members in a particular group. NOTE: This functionality is only accessible to RBAC admins. --- .../st2api/controllers/v1/service_registry.py | 71 +++++++++++++++++++ st2common/st2common/openapi.yaml.j2 | 43 +++++++++++ 2 files changed, 114 insertions(+) create mode 100644 st2api/st2api/controllers/v1/service_registry.py diff --git a/st2api/st2api/controllers/v1/service_registry.py b/st2api/st2api/controllers/v1/service_registry.py new file mode 100644 index 0000000000..21906f9c84 --- /dev/null +++ b/st2api/st2api/controllers/v1/service_registry.py @@ -0,0 +1,71 @@ +# 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. + +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 = [group_id_.decode('utf-8') for group_id_ 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() + + try: + member_ids = list(coordinator.get_members(group_id).get()) + except GroupNotCreated: + msg = ('Group with ID "%s" not found.' % (group_id)) + raise StackStormDBObjectNotFoundError(msg) + member_ids = [member_id.decode('utf-8') for member_id in member_ids] + + result = { + 'members': [] + } + + for member_id in member_ids: + capabilities = coordinator.get_member_capabilities(group_id, member_id).get() + item = { + 'member_id': member_id, + 'capabilities': capabilities + } + result['members'].append(item) + + return result + +groups_controller = ServiceRegistryGroupsController() +members_controller = ServiceRegistryGroupMembersController() diff --git a/st2common/st2common/openapi.yaml.j2 b/st2common/st2common/openapi.yaml.j2 index b7f1d56768..b3b4592fc1 100644 --- a/st2common/st2common/openapi.yaml.j2 +++ b/st2common/st2common/openapi.yaml.j2 @@ -4296,6 +4296,49 @@ paths: description: Unexpected error schema: $ref: '#/definitions/Error' + /api/v1/service_registry/groups: + get: + operationId: st2api.controllers.v1.service_registry:groups_controller.get_all + description: Retrieve available service registry groups. + x-parameters: + - name: user + in: context + x-as: requester_user + description: User performing the operation. + responses: + '200': + description: List of available groups. + schema: + type: object + default: + description: Unexpected error + schema: + $ref: '#/definitions/Error' + /api/v1/service_registry/groups/{group_id}/members: + get: + operationId: st2api.controllers.v1.service_registry:members_controller.get_one + description: Retrieve active / available members for a particular group. + parameters: + - name: group_id + in: path + description: Group ID. + type: string + required: true + x-parameters: + - name: user + in: context + x-as: requester_user + description: User performing the operation. + responses: + '200': + description: List of available members. + schema: + type: object + default: + description: Unexpected error + schema: + $ref: '#/definitions/Error' + /auth/v1/tokens: post: From 5141fc78c1d93d969272d8aa77ead78293698a94 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Fri, 15 Feb 2019 11:13:06 +0100 Subject: [PATCH 11/43] Refactor it in a separate function so it's more self contained and easier to test. --- st2common/st2common/service_setup.py | 61 ++++++++++++++++++---------- 1 file changed, 39 insertions(+), 22 deletions(-) diff --git a/st2common/st2common/service_setup.py b/st2common/st2common/service_setup.py index 7909d78231..0b6b4061a9 100644 --- a/st2common/st2common/service_setup.py +++ b/st2common/st2common/service_setup.py @@ -55,7 +55,9 @@ 'teardown', 'db_setup', - 'db_teardown' + 'db_teardown', + + 'register_service_in_service_registry' ] LOG = logging.getLogger(__name__) @@ -176,27 +178,8 @@ def setup(service, config, setup_db=True, register_mq_exchanges=True, # Register service in the service registry if service_registry: # NOTE: It's important that we pass start_heart=True to start the hearbeat process - coordinator = coordination.get_coordinator(start_heart=True) - - member_id = coordination.get_member_id() - - # 1. Create a group with the name of the service - group_id = six.binary_type(six.text_type(service).encode('ascii')) - - try: - coordinator.create_group(group_id).get() - except GroupAlreadyExist: - pass - - # Include common capabilities such as hostname and process ID - proc_info = system_info.get_process_info() - capabilities['hostname'] = proc_info['hostname'] - capabilities['pid'] = proc_info['pid'] - - # 1. Join the group as a member - LOG.debug('Joining service registry group "%s" as member_id "%s" with capabilities "%s"' % - (group_id, member_id, capabilities)) - coordinator.join_group(group_id, capabilities=capabilities).get() + register_service_in_service_registry(service=service, capabilities=capabilities, + start_heart=True) def teardown(): @@ -209,3 +192,37 @@ def teardown(): # 2. Tear down the coordinator coordinator = coordination.get_coordinator() coordination.coordinator_teardown(coordinator) + + +def register_service_in_service_registry(service, capabilities=None, start_heart=True): + """ + Register provided service in the service registry and start the heartbeat process. + + :param service: Service name which will also be used for a group name (e.g. "api"). + :type service: ``str`` + + :param capabilities: Optional metadata associated with the service. + :type capabilities: ``dict`` + """ + # NOTE: It's important that we pass start_heart=True to start the hearbeat process + coordinator = coordination.get_coordinator(start_heart=start_heart) + + member_id = coordination.get_member_id() + + # 1. Create a group with the name of the service + group_id = six.binary_type(six.text_type(service).encode('ascii')) + + try: + coordinator.create_group(group_id).get() + except GroupAlreadyExist: + pass + + # Include common capabilities such as hostname and process ID + proc_info = system_info.get_process_info() + capabilities['hostname'] = proc_info['hostname'] + capabilities['pid'] = proc_info['pid'] + + # 1. Join the group as a member + LOG.debug('Joining service registry group "%s" as member_id "%s" with capabilities "%s"' % + (group_id, member_id, capabilities)) + return coordinator.join_group(group_id, capabilities=capabilities).get() From b4f1f1398772d26c8eb20a80c4df76b601e406dd Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Fri, 15 Feb 2019 11:20:09 +0100 Subject: [PATCH 12/43] Add test cases for new service registry related API endpoint. --- .../controllers/v1/test_service_registry.py | 79 +++++++++++++++++++ 1 file changed, 79 insertions(+) create mode 100644 st2api/tests/unit/controllers/v1/test_service_registry.py diff --git a/st2api/tests/unit/controllers/v1/test_service_registry.py b/st2api/tests/unit/controllers/v1/test_service_registry.py new file mode 100644 index 0000000000..b5ffe14b1f --- /dev/null +++ b/st2api/tests/unit/controllers/v1/test_service_registry.py @@ -0,0 +1,79 @@ +# 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. + +from st2common.service_setup import teardown as common_teardown +from st2common.service_setup import register_service_in_service_registry +from st2common.util import system_info +from st2common.services.coordination import get_member_id + +from st2tests import config as tests_config + +from tests.base import FunctionalTest + +__all__ = [ + 'ServiceyRegistryControllerTestCase' +] + + +class ServiceyRegistryControllerTestCase(FunctionalTest): + + @classmethod + def setUpClass(cls): + super(ServiceyRegistryControllerTestCase, cls).setUpClass() + + tests_config.parse_args() + + # NOTE: We mock call common_setup to emulate service being registered in the service + # registry during bootstrap phase + register_service_in_service_registry(service='mock_service', + capabilities={'key1': 'value1', + 'name': 'mock_service'}, + start_heart=True) + + @classmethod + def tearDownClass(cls): + super(ServiceyRegistryControllerTestCase, cls).tearDownClass() + common_teardown() + + def test_get_groups(self): + list_resp = self.app.get('/v1/service_registry/groups') + self.assertEqual(list_resp.status_int, 200) + self.assertEqual(list_resp.json, {'groups': ['mock_service']}) + + def test_get_group_members(self): + proc_info = system_info.get_process_info() + member_id = get_member_id() + + # 1. Group doesn't exist + resp = self.app.get('/v1/service_registry/groups/doesnt-exist/members', expect_errors=True) + self.assertEqual(resp.status_int, 404) + self.assertEqual(resp.json['faultstring'], 'Group with ID "doesnt-exist" not found.') + + # 2. Group exists and has a single member + resp = self.app.get('/v1/service_registry/groups/mock_service/members') + self.assertEqual(resp.status_int, 200) + self.assertEqual(resp.json, { + 'members': [ + { + 'member_id': member_id, + 'capabilities': { + 'key1': 'value1', + 'name': 'mock_service', + 'hostname': proc_info['hostname'], + 'pid': proc_info['pid'] + } + } + ] + }) From ed9b270fcaab423ad3260663ce1bc1f4b7979f5e Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Fri, 15 Feb 2019 11:23:27 +0100 Subject: [PATCH 13/43] Add changelog entry. --- CHANGELOG.rst | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index d7569468aa..c878fd4a3a 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -31,6 +31,12 @@ Added 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//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 Changed ~~~~~~~ From af231a06dd0032f0628c4cef22ecc6cf3901e7d3 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Fri, 15 Feb 2019 11:27:11 +0100 Subject: [PATCH 14/43] Add new service registry endpoints to RBAC API endpoint test cases. --- .../v1/test_rbac_for_supported_endpoints.py | 29 +++++++++++++++++++ 1 file changed, 29 insertions(+) diff --git a/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py b/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py index 57f1c2bcdf..f6d823b01f 100644 --- a/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py +++ b/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py @@ -23,6 +23,8 @@ 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 teardown as common_teardown +from st2common.service_setup import register_service_in_service_registry from st2tests.fixturesloader import FixturesLoader from tests.base import APIControllerWithRBACTestCase @@ -110,6 +112,22 @@ class APIControllersRBACTestCase(APIControllerWithRBACTestCase): register_packs = True fixtures_loader = FixturesLoader() + @classmethod + def setUpClass(cls): + super(APIControllersRBACTestCase, cls).setUpClass() + + # NOTE: We mock call common_setup to emulate service being registered in the service + # registry during bootstrap phase + register_service_in_service_registry(service='mock_service', + capabilities={'key1': 'value1', + 'name': 'mock_service'}, + start_heart=True) + + @classmethod + def tearDownClass(cls): + super(APIControllersRBACTestCase, cls).tearDownClass() + common_teardown() + def setUp(self): super(APIControllersRBACTestCase, self).setUp() @@ -450,6 +468,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 } ] From cead8424cb87fc9a48dc9cf937cf8baa2230558a Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Fri, 15 Feb 2019 11:43:15 +0100 Subject: [PATCH 15/43] Only tear down the coordination service. --- .../controllers/v1/test_rbac_for_supported_endpoints.py | 6 ++++-- st2api/tests/unit/controllers/v1/test_service_registry.py | 6 ++++-- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py b/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py index f6d823b01f..77e006cbf0 100644 --- a/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py +++ b/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py @@ -23,8 +23,8 @@ 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 teardown as common_teardown from st2common.service_setup import register_service_in_service_registry +from st2common.services import coordination from st2tests.fixturesloader import FixturesLoader from tests.base import APIControllerWithRBACTestCase @@ -126,7 +126,9 @@ def setUpClass(cls): @classmethod def tearDownClass(cls): super(APIControllersRBACTestCase, cls).tearDownClass() - common_teardown() + + coordinator = coordination.get_coordinator() + coordination.coordinator_teardown(coordinator) def setUp(self): super(APIControllersRBACTestCase, self).setUp() diff --git a/st2api/tests/unit/controllers/v1/test_service_registry.py b/st2api/tests/unit/controllers/v1/test_service_registry.py index b5ffe14b1f..226a9ea7fb 100644 --- a/st2api/tests/unit/controllers/v1/test_service_registry.py +++ b/st2api/tests/unit/controllers/v1/test_service_registry.py @@ -13,10 +13,10 @@ # See the License for the specific language governing permissions and # limitations under the License. -from st2common.service_setup import teardown as common_teardown from st2common.service_setup import register_service_in_service_registry from st2common.util import system_info from st2common.services.coordination import get_member_id +from st2common.services import coordination from st2tests import config as tests_config @@ -45,7 +45,9 @@ def setUpClass(cls): @classmethod def tearDownClass(cls): super(ServiceyRegistryControllerTestCase, cls).tearDownClass() - common_teardown() + + coordinator = coordination.get_coordinator() + coordination.coordinator_teardown(coordinator) def test_get_groups(self): list_resp = self.app.get('/v1/service_registry/groups') From de605d849cccfe1237792a427d75c9e2d15c37b1 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Fri, 15 Feb 2019 13:03:33 +0100 Subject: [PATCH 16/43] Add RBAC test cases for the new service regstry API endpoints. --- .../v1/test_service_registry_rbac.py | 82 +++++++++++++++++++ 1 file changed, 82 insertions(+) create mode 100644 st2api/tests/unit/controllers/v1/test_service_registry_rbac.py diff --git a/st2api/tests/unit/controllers/v1/test_service_registry_rbac.py b/st2api/tests/unit/controllers/v1/test_service_registry_rbac.py new file mode 100644 index 0000000000..7381e3b58b --- /dev/null +++ b/st2api/tests/unit/controllers/v1/test_service_registry_rbac.py @@ -0,0 +1,82 @@ +# 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 st2common.service_setup import register_service_in_service_registry +from st2common.services import coordination + +from tests.base import APIControllerWithRBACTestCase + +http_client = six.moves.http_client + +__all__ = [ + 'ServiceRegistryControllerRBACTestCase' +] + + +class ServiceRegistryControllerRBACTestCase(APIControllerWithRBACTestCase): + @classmethod + def setUpClass(cls): + super(ServiceRegistryControllerRBACTestCase, cls).setUpClass() + # Register mock service in the service registry for testing purposes + register_service_in_service_registry(service='mock_service', + capabilities={'key1': 'value1', + 'name': 'mock_service'}, + start_heart=True) + + @classmethod + def tearDownClass(cls): + super(ServiceRegistryControllerRBACTestCase, cls).tearDownClass() + + coordinator = coordination.get_coordinator() + coordination.coordinator_teardown(coordinator) + + def test_get_groups(self): + # Non admin users can't access that API endpoint + for user_db in [self.users['no_permissions'], self.users['observer']]: + self.use_user(user_db) + + resp = self.app.get('/v1/service_registry/groups', expect_errors=True) + expected_msg = ('Administrator access required') + self.assertEqual(resp.status_code, http_client.FORBIDDEN) + self.assertEqual(resp.json['faultstring'], expected_msg) + + # Admin user can access it + user_db = self.users['admin'] + self.use_user(user_db) + + resp = self.app.get('/v1/service_registry/groups') + self.assertEqual(resp.status_code, http_client.OK) + self.assertEqual(resp.json, {'groups': ['mock_service']}) + + def test_get_group_members(self): + # Non admin users can't access that API endpoint + for user_db in [self.users['no_permissions'], self.users['observer']]: + self.use_user(user_db) + + resp = self.app.get('/v1/service_registry/groups/mock_service/members', + expect_errors=True) + expected_msg = ('Administrator access required') + self.assertEqual(resp.status_code, http_client.FORBIDDEN) + self.assertEqual(resp.json['faultstring'], expected_msg) + + # Admin user can access it + user_db = self.users['admin'] + self.use_user(user_db) + + resp = self.app.get('/v1/service_registry/groups/mock_service/members') + self.assertEqual(resp.status_code, http_client.OK) + self.assertTrue('members' in resp.json) From 76406ff337e265dcba4821ae6918d31878786594 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Fri, 15 Feb 2019 13:04:41 +0100 Subject: [PATCH 17/43] Re-generate openapi config. --- st2common/st2common/openapi.yaml | 43 ++++++++++++++++++++++++++++++++ 1 file changed, 43 insertions(+) diff --git a/st2common/st2common/openapi.yaml b/st2common/st2common/openapi.yaml index 82e914f8bf..aa92aace65 100644 --- a/st2common/st2common/openapi.yaml +++ b/st2common/st2common/openapi.yaml @@ -4300,6 +4300,49 @@ paths: description: Unexpected error schema: $ref: '#/definitions/Error' + /api/v1/service_registry/groups: + get: + operationId: st2api.controllers.v1.service_registry:groups_controller.get_all + description: Retrieve available service registry groups. + x-parameters: + - name: user + in: context + x-as: requester_user + description: User performing the operation. + responses: + '200': + description: List of available groups. + schema: + type: object + default: + description: Unexpected error + schema: + $ref: '#/definitions/Error' + /api/v1/service_registry/groups/{group_id}/members: + get: + operationId: st2api.controllers.v1.service_registry:members_controller.get_one + description: Retrieve active / available members for a particular group. + parameters: + - name: group_id + in: path + description: Group ID. + type: string + required: true + x-parameters: + - name: user + in: context + x-as: requester_user + description: User performing the operation. + responses: + '200': + description: List of available members. + schema: + type: object + default: + description: Unexpected error + schema: + $ref: '#/definitions/Error' + /auth/v1/tokens: post: From 4963266de93212d76d09c13845eca4b73e963871 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Fri, 15 Feb 2019 13:06:13 +0100 Subject: [PATCH 18/43] Update comment, fix lint. --- st2api/st2api/controllers/v1/service_registry.py | 1 + .../unit/controllers/v1/test_rbac_for_supported_endpoints.py | 4 +--- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/st2api/st2api/controllers/v1/service_registry.py b/st2api/st2api/controllers/v1/service_registry.py index 21906f9c84..42a55f91e6 100644 --- a/st2api/st2api/controllers/v1/service_registry.py +++ b/st2api/st2api/controllers/v1/service_registry.py @@ -67,5 +67,6 @@ def get_one(self, group_id, requester_user): return result + groups_controller = ServiceRegistryGroupsController() members_controller = ServiceRegistryGroupMembersController() diff --git a/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py b/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py index 77e006cbf0..9b8c9f56c0 100644 --- a/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py +++ b/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py @@ -115,9 +115,7 @@ class APIControllersRBACTestCase(APIControllerWithRBACTestCase): @classmethod def setUpClass(cls): super(APIControllersRBACTestCase, cls).setUpClass() - - # NOTE: We mock call common_setup to emulate service being registered in the service - # registry during bootstrap phase + # Register mock service in the service registry for testing purposes register_service_in_service_registry(service='mock_service', capabilities={'key1': 'value1', 'name': 'mock_service'}, From 57ad1297177460d362096b159b64eb46cb0c783d Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Sat, 2 Mar 2019 20:34:22 -0800 Subject: [PATCH 19/43] Always refer to the same coordinator instance. --- .../controllers/v1/test_rbac_for_supported_endpoints.py | 8 ++++++-- .../tests/unit/controllers/v1/test_service_registry.py | 7 +++++-- .../unit/controllers/v1/test_service_registry_rbac.py | 9 +++++++-- 3 files changed, 18 insertions(+), 6 deletions(-) diff --git a/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py b/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py index 9b8c9f56c0..3593c8d862 100644 --- a/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py +++ b/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py @@ -112,9 +112,14 @@ class APIControllersRBACTestCase(APIControllerWithRBACTestCase): register_packs = True fixtures_loader = FixturesLoader() + coordinator = None + @classmethod def setUpClass(cls): super(APIControllersRBACTestCase, cls).setUpClass() + + cls.coordinator = coordination.get_coordinator() + # Register mock service in the service registry for testing purposes register_service_in_service_registry(service='mock_service', capabilities={'key1': 'value1', @@ -125,8 +130,7 @@ def setUpClass(cls): def tearDownClass(cls): super(APIControllersRBACTestCase, cls).tearDownClass() - coordinator = coordination.get_coordinator() - coordination.coordinator_teardown(coordinator) + coordination.coordinator_teardown(cls.coordinator) def setUp(self): super(APIControllersRBACTestCase, self).setUp() diff --git a/st2api/tests/unit/controllers/v1/test_service_registry.py b/st2api/tests/unit/controllers/v1/test_service_registry.py index 226a9ea7fb..3ba99110d3 100644 --- a/st2api/tests/unit/controllers/v1/test_service_registry.py +++ b/st2api/tests/unit/controllers/v1/test_service_registry.py @@ -29,12 +29,16 @@ class ServiceyRegistryControllerTestCase(FunctionalTest): + coordinator = None + @classmethod def setUpClass(cls): super(ServiceyRegistryControllerTestCase, cls).setUpClass() tests_config.parse_args() + cls.coordinator = coordination.get_coordinator() + # NOTE: We mock call common_setup to emulate service being registered in the service # registry during bootstrap phase register_service_in_service_registry(service='mock_service', @@ -46,8 +50,7 @@ def setUpClass(cls): def tearDownClass(cls): super(ServiceyRegistryControllerTestCase, cls).tearDownClass() - coordinator = coordination.get_coordinator() - coordination.coordinator_teardown(coordinator) + coordination.coordinator_teardown(cls.coordinator) def test_get_groups(self): list_resp = self.app.get('/v1/service_registry/groups') diff --git a/st2api/tests/unit/controllers/v1/test_service_registry_rbac.py b/st2api/tests/unit/controllers/v1/test_service_registry_rbac.py index 7381e3b58b..60335973fe 100644 --- a/st2api/tests/unit/controllers/v1/test_service_registry_rbac.py +++ b/st2api/tests/unit/controllers/v1/test_service_registry_rbac.py @@ -28,9 +28,15 @@ class ServiceRegistryControllerRBACTestCase(APIControllerWithRBACTestCase): + + coordinator = None + @classmethod def setUpClass(cls): super(ServiceRegistryControllerRBACTestCase, cls).setUpClass() + + cls.coordinator = coordination.get_coordinator() + # Register mock service in the service registry for testing purposes register_service_in_service_registry(service='mock_service', capabilities={'key1': 'value1', @@ -41,8 +47,7 @@ def setUpClass(cls): def tearDownClass(cls): super(ServiceRegistryControllerRBACTestCase, cls).tearDownClass() - coordinator = coordination.get_coordinator() - coordination.coordinator_teardown(coordinator) + coordination.coordinator_teardown(cls.coordinator) def test_get_groups(self): # Non admin users can't access that API endpoint From 7fa0934f3f65560011b6c568c8450789560245c2 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Sun, 3 Mar 2019 13:49:53 +0800 Subject: [PATCH 20/43] Use NoOp driver for tests since zake one causes some tests go hang. --- .../v1/test_rbac_for_supported_endpoints.py | 4 ++ .../controllers/v1/test_service_registry.py | 2 +- .../v1/test_service_registry_rbac.py | 4 ++ st2common/st2common/services/coordination.py | 58 +++++++++++++------ st2tests/st2tests/config.py | 4 +- 5 files changed, 51 insertions(+), 21 deletions(-) diff --git a/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py b/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py index 3593c8d862..58c27b9e93 100644 --- a/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py +++ b/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py @@ -26,7 +26,9 @@ 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 @@ -116,6 +118,8 @@ class APIControllersRBACTestCase(APIControllerWithRBACTestCase): @classmethod def setUpClass(cls): + tests_config.parse_args(coordinator_noop=True) + super(APIControllersRBACTestCase, cls).setUpClass() cls.coordinator = coordination.get_coordinator() diff --git a/st2api/tests/unit/controllers/v1/test_service_registry.py b/st2api/tests/unit/controllers/v1/test_service_registry.py index 3ba99110d3..7cf115fedb 100644 --- a/st2api/tests/unit/controllers/v1/test_service_registry.py +++ b/st2api/tests/unit/controllers/v1/test_service_registry.py @@ -35,7 +35,7 @@ class ServiceyRegistryControllerTestCase(FunctionalTest): def setUpClass(cls): super(ServiceyRegistryControllerTestCase, cls).setUpClass() - tests_config.parse_args() + tests_config.parse_args(coordinator_noop=True) cls.coordinator = coordination.get_coordinator() diff --git a/st2api/tests/unit/controllers/v1/test_service_registry_rbac.py b/st2api/tests/unit/controllers/v1/test_service_registry_rbac.py index 60335973fe..7e5c763477 100644 --- a/st2api/tests/unit/controllers/v1/test_service_registry_rbac.py +++ b/st2api/tests/unit/controllers/v1/test_service_registry_rbac.py @@ -18,6 +18,8 @@ from st2common.service_setup import register_service_in_service_registry from st2common.services import coordination +from st2tests import config as tests_config + from tests.base import APIControllerWithRBACTestCase http_client = six.moves.http_client @@ -33,6 +35,8 @@ class ServiceRegistryControllerRBACTestCase(APIControllerWithRBACTestCase): @classmethod def setUpClass(cls): + tests_config.parse_args(coordinator_noop=True) + super(ServiceRegistryControllerRBACTestCase, cls).setUpClass() cls.coordinator = coordination.get_coordinator() diff --git a/st2common/st2common/services/coordination.py b/st2common/st2common/services/coordination.py index 3053376183..5109d3dd5e 100644 --- a/st2common/st2common/services/coordination.py +++ b/st2common/st2common/services/coordination.py @@ -16,9 +16,11 @@ from __future__ import absolute_import import six + from oslo_config import cfg from tooz import coordination from tooz import locking +from tooz.coordination import GroupNotCreated from st2common import log as logging from st2common.util import system_info @@ -68,9 +70,15 @@ class NoOpDriver(coordination.CoordinationDriver): This driver is used if coordination service is not configured. """ + groups = {} + def __init__(self, member_id, parsed_url=None, options=None): super(NoOpDriver, self).__init__(member_id, parsed_url, options) + @classmethod + def stop(cls): + cls.groups = {} + def watch_join_group(self, group_id, callback): self._hooks_join_group[group_id].append(callback) @@ -93,33 +101,47 @@ def unwatch_elected_as_leader(self, group_id, callback): def stand_down_group_leader(group_id): return None - @staticmethod - def create_group(group_id): + @classmethod + def create_group(cls, group_id): + cls.groups[group_id] = {'members': {}} return NoOpAsyncResult() - @staticmethod - def get_groups(): - return NoOpAsyncResult(result=[]) + @classmethod + def get_groups(cls): + return NoOpAsyncResult(result=cls.groups.keys()) - @staticmethod - def join_group(group_id, capabilities=''): + @classmethod + def join_group(cls, group_id, capabilities=''): + member_id = get_member_id() + + cls.groups[group_id]['members'][member_id] = {'capabilities': capabilities} return NoOpAsyncResult() - @staticmethod - def leave_group(group_id): + @classmethod + def leave_group(cls, group_id): + member_id = get_member_id() + + del cls.groups[group_id]['members'][member_id] return NoOpAsyncResult() - @staticmethod - def delete_group(group_id): + @classmethod + def delete_group(cls, group_id): + del cls.groups[group_id] return NoOpAsyncResult() - @staticmethod - def get_members(group_id): - return NoOpAsyncResult(result=[]) + @classmethod + def get_members(cls, group_id): + try: + member_ids = cls.groups[group_id]['members'].keys() + except KeyError: + raise GroupNotCreated('Group doesnt exist') - @staticmethod - def get_member_capabilities(group_id, member_id): - return None + return NoOpAsyncResult(result=member_ids) + + @classmethod + def get_member_capabilities(cls, group_id, member_id): + member_capabiliteis = cls.groups[group_id]['members'][member_id]['capabilities'] + return NoOpAsyncResult(result=member_capabiliteis) @staticmethod def update_capabilities(group_id, capabilities): @@ -186,7 +208,7 @@ def get_coordinator(start_heart=False): 'service will use best effort approach and race conditions are possible.') if not COORDINATOR: - COORDINATOR = coordinator_setup(start_heart=start_heart) + COORDINATOR = coordinator_setup(start_heart=False) return COORDINATOR diff --git a/st2tests/st2tests/config.py b/st2tests/st2tests/config.py index c8cda2cd86..3c1f07b263 100644 --- a/st2tests/st2tests/config.py +++ b/st2tests/st2tests/config.py @@ -28,12 +28,12 @@ LOG = logging.getLogger(__name__) -def parse_args(coordinator_noop=False): +def parse_args(coordinator_noop=True): _setup_config_opts(coordinator_noop=coordinator_noop) CONF(args=[]) -def _setup_config_opts(coordinator_noop=False): +def _setup_config_opts(coordinator_noop=True): cfg.CONF.reset() try: From fd63345ca440e5dda8434580d16d6e96434075fa Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Mon, 11 Mar 2019 10:39:03 +0100 Subject: [PATCH 21/43] Remove extra whitespace. --- CHANGELOG.rst | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index cf48af2155..81470d82d2 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -41,13 +41,13 @@ Changed to make it easier for developers to understand. (improvement) * Update Python runner code so it prioritizes libraries from pack virtual environment over StackStorm system dependencies. - + For example, if pack depends on ``six==1.11.0`` in pack ``requirements.txt``, but StackStorm depends on ``six==1.10.0``, ``six==1.11.0`` will be used when running Python actions from that pack. - + Keep in mind that will not work correctly if pack depends on a library which brakes functionality used by Python action wrapper code. - + Contributed by Hiroyasu OHYAMA (@userlocalhost). #4571 Fixed @@ -71,7 +71,7 @@ Fixed with ``null`` for the ``Access-Control-Allow-Origin`` header. The fix returns the first of our allowed origins if the requesting origin is not a supported origin. Reported by Barak Tawily. (bug fix) - + 2.9.3 - March 06, 2019 ----------------------- @@ -80,7 +80,7 @@ Fixed * Fix improper CORS where request from an origin not listed in ``allowed_origins`` will be responded with ``null`` for the ``Access-Control-Allow-Origin`` header. The fix returns the first of our - allowed origins if the requesting origin is not a supported origin. Reported by Barak Tawily. + Bllowed origins if the requesting origin is not a supported origin. Reported by Barak Tawily. (bug fix) 2.10.2 - February 21, 2019 @@ -145,7 +145,7 @@ Fixed Reported by @johandahlberg (bug fix) #4533 * Fix ``core.sendmail`` action so it specifies ``charset=UTF-8`` in the ``Content-Type`` email header. This way it works correctly when an email subject and / or body contains unicode data. - + Reported by @johandahlberg (bug fix) #4533 4534 * Fix CLI ``st2 apikey load`` not being idempotent and API endpoint ``/api/v1/apikeys`` not From ed861531991a52da6e2f8e623b7d489a13f17901 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Mon, 11 Mar 2019 11:30:49 +0100 Subject: [PATCH 22/43] Remove debug code. --- st2common/st2common/services/coordination.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/st2common/st2common/services/coordination.py b/st2common/st2common/services/coordination.py index 5109d3dd5e..e798a4eb26 100644 --- a/st2common/st2common/services/coordination.py +++ b/st2common/st2common/services/coordination.py @@ -208,7 +208,7 @@ def get_coordinator(start_heart=False): 'service will use best effort approach and race conditions are possible.') if not COORDINATOR: - COORDINATOR = coordinator_setup(start_heart=False) + COORDINATOR = coordinator_setup(start_heart=start_heart) return COORDINATOR From b0e48e2ecff1f46b2ac8969b55f9d52dcf89b297 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Mon, 11 Mar 2019 11:32:15 +0100 Subject: [PATCH 23/43] Add new CLI commands for listing service registry groups and members: * st2 service-registry groups list * st2 service-registry members list [--group-id=] --- st2client/st2client/client.py | 11 ++ .../st2client/commands/service_registry.py | 112 ++++++++++++++++++ st2client/st2client/models/__init__.py | 1 + st2client/st2client/models/core.py | 48 ++++++++ .../st2client/models/service_registry.py | 48 ++++++++ st2client/st2client/shell.py | 6 + 6 files changed, 226 insertions(+) create mode 100644 st2client/st2client/commands/service_registry.py create mode 100644 st2client/st2client/models/service_registry.py diff --git a/st2client/st2client/client.py b/st2client/st2client/client.py index a176658056..4297453957 100644 --- a/st2client/st2client/client.py +++ b/st2client/st2client/client.py @@ -34,6 +34,8 @@ from st2client.models.core import WebhookManager from st2client.models.core import StreamManager from st2client.models.core import WorkflowManager +from st2client.models.core import ServiceRegistryGroupsManager +from st2client.models.core import ServiceRegistryMembersManager from st2client.models.core import add_auth_token_to_kwargs_from_env @@ -168,6 +170,15 @@ def __init__(self, base_url=None, auth_url=None, api_url=None, stream_url=None, self.managers['Workflow'] = WorkflowManager( self.endpoints['api'], cacert=self.cacert, debug=self.debug) + # Service Registry + self.managers['ServiceRegistryGroups'] = ServiceRegistryGroupsManager( + models.ServiceRegistryGroup, self.endpoints['api'], cacert=self.cacert, + debug=self.debug) + + self.managers['ServiceRegistryMembers'] = ServiceRegistryMembersManager( + models.ServiceRegistryMember, self.endpoints['api'], cacert=self.cacert, + debug=self.debug) + # RBAC self.managers['Role'] = ResourceManager( models.Role, self.endpoints['api'], cacert=self.cacert, debug=self.debug) diff --git a/st2client/st2client/commands/service_registry.py b/st2client/st2client/commands/service_registry.py new file mode 100644 index 0000000000..02ba52778c --- /dev/null +++ b/st2client/st2client/commands/service_registry.py @@ -0,0 +1,112 @@ +# 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. + +from __future__ import absolute_import + +from st2client.models.service_registry import ServiceRegistryGroup +from st2client.models.service_registry import ServiceRegistryMember +from st2client import commands +from st2client.commands.noop import NoopCommand +from st2client.commands import resource + + +class ServiceRegistryBranch(commands.Branch): + def __init__(self, description, app, subparsers, parent_parser=None): + super(ServiceRegistryBranch, self).__init__( + 'service-registry', description, + app, subparsers, parent_parser=parent_parser) + + self.subparsers = self.parser.add_subparsers( + help=('List of commands for managing service registry.')) + + # Instantiate commands + args_groups = ['Manage service registry groups', self.app, self.subparsers] + args_members = ['Manage service registry members', self.app, self.subparsers] + + self.commands['groups'] = ServiceRegistryGroupsBranch(*args_groups) + self.commands['members'] = ServiceRegistryMembersBranch(*args_members) + + +class ServiceRegistryGroupsBranch(resource.ResourceBranch): + def __init__(self, description, app, subparsers, parent_parser=None): + super(ServiceRegistryGroupsBranch, self).__init__( + ServiceRegistryGroup, description, app, subparsers, + parent_parser=parent_parser, + read_only=True, + commands={ + 'list': ServiceRegistryListGroupsCommand, + 'get': NoopCommand + }) + + del self.commands['get'] + + +class ServiceRegistryMembersBranch(resource.ResourceBranch): + def __init__(self, description, app, subparsers, parent_parser=None): + super(ServiceRegistryMembersBranch, self).__init__( + ServiceRegistryMember, description, app, subparsers, + parent_parser=parent_parser, + read_only=True, + commands={ + 'list': ServiceRegistryListMembersCommand, + 'get': NoopCommand + }) + + del self.commands['get'] + + +class ServiceRegistryListGroupsCommand(resource.ResourceListCommand): + display_attributes = ['group_id'] + attribute_display_order = ['group_id'] + + @resource.add_auth_token_to_kwargs_from_cli + def run(self, args, **kwargs): + manager = self.app.client.managers['ServiceRegistryGroups'] + + groups = manager.list() + return groups + + +class ServiceRegistryListMembersCommand(resource.ResourceListCommand): + display_attributes = ['group_id', 'member_id', 'capabilities'] + attribute_display_order = ['group_id', 'member_id', 'capabilities'] + + def __init__(self, resource, *args, **kwargs): + super(ServiceRegistryListMembersCommand, self).__init__( + resource, *args, **kwargs + ) + + self.parser.add_argument('--group-id', dest='group_id', default=None, + help='If provided only retrieve members for the specified group.') + + @resource.add_auth_token_to_kwargs_from_cli + def run(self, args, **kwargs): + groups_manager = self.app.client.managers['ServiceRegistryGroups'] + members_manager = self.app.client.managers['ServiceRegistryMembers'] + + # If group ID is provided only retrieve members for that group, otherwise retrieve members + # for all groups + if args.group_id: + members = members_manager.list(args.group_id) + return members + else: + groups = groups_manager.list() + + result = [] + for group in groups: + members = members_manager.list(group.group_id) + result.extend(members) + + return result diff --git a/st2client/st2client/models/__init__.py b/st2client/st2client/models/__init__.py index dabce91510..0ce9c7b4c6 100644 --- a/st2client/st2client/models/__init__.py +++ b/st2client/st2client/models/__init__.py @@ -29,4 +29,5 @@ from st2client.models.trace import * # noqa from st2client.models.webhook import * # noqa from st2client.models.timer import * # noqa +from st2client.models.service_registry import * # noqa from st2client.models.rbac import * # noqa diff --git a/st2client/st2client/models/core.py b/st2client/st2client/models/core.py index 892942440f..9fb946939b 100644 --- a/st2client/st2client/models/core.py +++ b/st2client/st2client/models/core.py @@ -682,3 +682,51 @@ def inspect(self, definition, **kwargs): self.handle_error(response) return response.json() + + +class ServiceRegistryGroupsManager(ResourceManager): + @add_auth_token_to_kwargs_from_env + def list(self, **kwargs): + url = '/service_registry/groups' + + headers = {} + response = self.client.get(url, headers=headers, **kwargs) + + if response.status_code != http_client.OK: + self.handle_error(response) + + groups = response.json()['groups'] + + result = [] + for group in groups: + item = self.resource.deserialize({'group_id': group}) + result.append(item) + + return result + + +class ServiceRegistryMembersManager(ResourceManager): + + @add_auth_token_to_kwargs_from_env + def list(self, group_id, **kwargs): + url = '/service_registry/groups/%s/members' % (group_id) + + headers = {} + response = self.client.get(url, headers=headers, **kwargs) + + if response.status_code != http_client.OK: + self.handle_error(response) + + members = response.json()['members'] + + result = [] + for member in members: + data = { + 'group_id': group_id, + 'member_id': member['member_id'], + 'capabilities': member['capabilities'] + } + item = self.resource.deserialize(data) + result.append(item) + + return result diff --git a/st2client/st2client/models/service_registry.py b/st2client/st2client/models/service_registry.py new file mode 100644 index 0000000000..7dbed43485 --- /dev/null +++ b/st2client/st2client/models/service_registry.py @@ -0,0 +1,48 @@ +# 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. + +from __future__ import absolute_import + +from st2client.models import core + +__all__ = [ + 'ServiceRegistry', + + 'ServiceRegistryGroup', + 'ServiceRegistryMember' +] + + +class ServiceRegistry(core.Resource): + _alias = 'service-registry' + _display_name = 'service registry' + _plural = 'service registry' + _plural_display_name = 'service registry' + + +class ServiceRegistryGroup(core.Resource): + _alias = 'group' + _display_name = 'Group' + _plural = 'Groups' + _plural_display_name = 'Groups' + _repr_attributes = ['group_id'] + + +class ServiceRegistryMember(core.Resource): + _alias = 'member' + _display_name = 'Group Member' + _plural = 'Group Members' + _plural_display_name = 'Group Members' + _repr_attributes = ['group_id', 'member_id'] diff --git a/st2client/st2client/shell.py b/st2client/st2client/shell.py index fdf0a0c160..2bf62d9ed8 100755 --- a/st2client/st2client/shell.py +++ b/st2client/st2client/shell.py @@ -58,6 +58,7 @@ from st2client.commands import rule_enforcement from st2client.commands import rbac from st2client.commands import workflow +from st2client.commands import service_registry from st2client.config import set_config from st2client.exceptions.operations import OperationFailureException from st2client.utils.logging import LogLevelFilter, set_log_level_for_all_loggers @@ -338,6 +339,11 @@ def __init__(self): 'Only orquesta workflows are supported.', self, self.subparsers) + # Service Registry + self.commands['service-registry'] = service_registry.ServiceRegistryBranch( + 'Service registry group and membership related commands.', + self, self.subparsers) + # RBAC self.commands['role'] = rbac.RoleBranch( 'RBAC roles.', From 33a3cedd249ac48836b8d1dd1844d3ce4bc4d366 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Mon, 11 Mar 2019 12:30:31 +0100 Subject: [PATCH 24/43] Update changelog. --- CHANGELOG.rst | 3 +++ 1 file changed, 3 insertions(+) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 81470d82d2..1710b906b5 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -29,6 +29,9 @@ Added 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=]`` + * Adding ``Cache-Control`` header to all API responses so clients will favor refresh from API instead of using cached version. From 825778d5cd589990ff299cccd4d111271cbebd02 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Mon, 11 Mar 2019 12:47:14 +0100 Subject: [PATCH 25/43] Update more affected tests. --- st2common/tests/unit/services/test_synchronization.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/st2common/tests/unit/services/test_synchronization.py b/st2common/tests/unit/services/test_synchronization.py index 5337860ba8..202dee5555 100644 --- a/st2common/tests/unit/services/test_synchronization.py +++ b/st2common/tests/unit/services/test_synchronization.py @@ -29,7 +29,7 @@ class SynchronizationTest(unittest2.TestCase): @classmethod def setUpClass(cls): super(SynchronizationTest, cls).setUpClass() - tests_config.parse_args() + tests_config.parse_args(coordinator_noop=False) cls.coordinator = coordination.get_coordinator() @classmethod From 4208251bc052768a683b2cf2adb78b7e38e84deb Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Mon, 11 Mar 2019 14:27:20 +0100 Subject: [PATCH 26/43] Don't use cached coordinator instances for tests. This way we avoid potential cross test pollution issues. --- .../v1/test_rbac_for_supported_endpoints.py | 2 +- .../unit/controllers/v1/test_service_registry.py | 2 +- .../controllers/v1/test_service_registry_rbac.py | 2 +- st2common/st2common/services/coordination.py | 13 ++++++++++++- .../tests/unit/services/test_synchronization.py | 2 +- 5 files changed, 16 insertions(+), 5 deletions(-) diff --git a/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py b/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py index 58c27b9e93..70e0ca3aa8 100644 --- a/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py +++ b/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py @@ -122,7 +122,7 @@ def setUpClass(cls): super(APIControllersRBACTestCase, cls).setUpClass() - cls.coordinator = coordination.get_coordinator() + cls.coordinator = coordination.get_coordinator(use_cache=False) # Register mock service in the service registry for testing purposes register_service_in_service_registry(service='mock_service', diff --git a/st2api/tests/unit/controllers/v1/test_service_registry.py b/st2api/tests/unit/controllers/v1/test_service_registry.py index 7cf115fedb..6cab2496da 100644 --- a/st2api/tests/unit/controllers/v1/test_service_registry.py +++ b/st2api/tests/unit/controllers/v1/test_service_registry.py @@ -37,7 +37,7 @@ def setUpClass(cls): tests_config.parse_args(coordinator_noop=True) - cls.coordinator = coordination.get_coordinator() + cls.coordinator = coordination.get_coordinator(use_cache=False) # NOTE: We mock call common_setup to emulate service being registered in the service # registry during bootstrap phase diff --git a/st2api/tests/unit/controllers/v1/test_service_registry_rbac.py b/st2api/tests/unit/controllers/v1/test_service_registry_rbac.py index 7e5c763477..c83de9749e 100644 --- a/st2api/tests/unit/controllers/v1/test_service_registry_rbac.py +++ b/st2api/tests/unit/controllers/v1/test_service_registry_rbac.py @@ -39,7 +39,7 @@ def setUpClass(cls): super(ServiceRegistryControllerRBACTestCase, cls).setUpClass() - cls.coordinator = coordination.get_coordinator() + cls.coordinator = coordination.get_coordinator(use_cache=False) # Register mock service in the service registry for testing purposes register_service_in_service_registry(service='mock_service', diff --git a/st2common/st2common/services/coordination.py b/st2common/st2common/services/coordination.py index e798a4eb26..c9d270fc68 100644 --- a/st2common/st2common/services/coordination.py +++ b/st2common/st2common/services/coordination.py @@ -200,13 +200,24 @@ def coordinator_teardown(coordinator): coordinator.stop() -def get_coordinator(start_heart=False): +def get_coordinator(start_heart=False, use_cache=True): + """ + :param start_heart: True to start heartbeating process. + :type start_heart: ``bool`` + + :param use_cache: True to use cached coordinator instance. False should only be used in tests. + :type use_cache: ``bool`` + """ global COORDINATOR if not configured(): LOG.warn('Coordination backend is not configured. Code paths which use coordination ' 'service will use best effort approach and race conditions are possible.') + if not use_cache: + coordinator = coordinator_setup(start_heart=start_heart) + return coordinator + if not COORDINATOR: COORDINATOR = coordinator_setup(start_heart=start_heart) diff --git a/st2common/tests/unit/services/test_synchronization.py b/st2common/tests/unit/services/test_synchronization.py index 202dee5555..a68adf9c51 100644 --- a/st2common/tests/unit/services/test_synchronization.py +++ b/st2common/tests/unit/services/test_synchronization.py @@ -30,7 +30,7 @@ class SynchronizationTest(unittest2.TestCase): def setUpClass(cls): super(SynchronizationTest, cls).setUpClass() tests_config.parse_args(coordinator_noop=False) - cls.coordinator = coordination.get_coordinator() + cls.coordinator = coordination.get_coordinator(use_cache=False) @classmethod def tearDownClass(cls): From dd27733796998b19a8131b97f87f8b8aa4a83306 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Mon, 11 Mar 2019 14:51:43 +0100 Subject: [PATCH 27/43] Make sure we correctly tear down the coordinator. --- st2actions/tests/unit/policies/test_concurrency.py | 1 + st2actions/tests/unit/policies/test_concurrency_by_attr.py | 1 + 2 files changed, 2 insertions(+) diff --git a/st2actions/tests/unit/policies/test_concurrency.py b/st2actions/tests/unit/policies/test_concurrency.py index 301fae2b7d..f90fae2664 100644 --- a/st2actions/tests/unit/policies/test_concurrency.py +++ b/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() diff --git a/st2actions/tests/unit/policies/test_concurrency_by_attr.py b/st2actions/tests/unit/policies/test_concurrency_by_attr.py index 0056b33a4f..81c722ed7a 100644 --- a/st2actions/tests/unit/policies/test_concurrency_by_attr.py +++ b/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() From 2c113b9e1e81dc8b25ae0e4fba9f2fb101568227 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Mon, 11 Mar 2019 14:53:23 +0100 Subject: [PATCH 28/43] Fix Python 3 related test failures. --- st2api/st2api/controllers/v1/service_registry.py | 6 ++++-- .../controllers/v1/test_rbac_for_supported_endpoints.py | 3 ++- st2common/st2common/service_setup.py | 5 ++++- 3 files changed, 10 insertions(+), 4 deletions(-) diff --git a/st2api/st2api/controllers/v1/service_registry.py b/st2api/st2api/controllers/v1/service_registry.py index 42a55f91e6..335cd87b3b 100644 --- a/st2api/st2api/controllers/v1/service_registry.py +++ b/st2api/st2api/controllers/v1/service_registry.py @@ -13,6 +13,8 @@ # 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 @@ -47,11 +49,11 @@ def get_one(self, group_id, requester_user): coordinator = coordination.get_coordinator() try: + group_id = six.binary_type(six.text_type(group_id).encode('ascii')) member_ids = list(coordinator.get_members(group_id).get()) except GroupNotCreated: msg = ('Group with ID "%s" not found.' % (group_id)) raise StackStormDBObjectNotFoundError(msg) - member_ids = [member_id.decode('utf-8') for member_id in member_ids] result = { 'members': [] @@ -60,7 +62,7 @@ def get_one(self, group_id, requester_user): for member_id in member_ids: capabilities = coordinator.get_member_capabilities(group_id, member_id).get() item = { - 'member_id': member_id, + 'member_id': member_id.decode('utf-8'), 'capabilities': capabilities } result['members'].append(item) diff --git a/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py b/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py index 70e0ca3aa8..d7afcd718e 100644 --- a/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py +++ b/st2api/tests/unit/controllers/v1/test_rbac_for_supported_endpoints.py @@ -125,7 +125,8 @@ def setUpClass(cls): cls.coordinator = coordination.get_coordinator(use_cache=False) # Register mock service in the service registry for testing purposes - register_service_in_service_registry(service='mock_service', + 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) diff --git a/st2common/st2common/service_setup.py b/st2common/st2common/service_setup.py index f98bfa922f..83c4524e71 100644 --- a/st2common/st2common/service_setup.py +++ b/st2common/st2common/service_setup.py @@ -215,7 +215,10 @@ def register_service_in_service_registry(service, capabilities=None, start_heart member_id = coordination.get_member_id() # 1. Create a group with the name of the service - group_id = six.binary_type(six.text_type(service).encode('ascii')) + if not isinstance(service, six.binary_type): + group_id = service.encode('utf-8') + else: + group_id = service try: coordinator.create_group(group_id).get() From 32a4d46053b867f7efc5a1aa71775d5b0130965d Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Mon, 11 Mar 2019 15:14:47 +0100 Subject: [PATCH 29/43] Use full absolute paths in launchdev script. --- tools/launchdev.sh | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tools/launchdev.sh b/tools/launchdev.sh index 765104ae87..b883cd545e 100755 --- a/tools/launchdev.sh +++ b/tools/launchdev.sh @@ -68,6 +68,7 @@ function init(){ fi VIRTUALENV=${VIRTUALENV_DIR:-${ST2_REPO}/virtualenv} + VIRTUALENV=$(realpath ${VIRTUALENV}) PY=${VIRTUALENV}/bin/python PYTHON_VERSION=$(${PY} --version 2>&1) @@ -77,6 +78,8 @@ function init(){ if [ -z "$ST2_CONF" ]; then ST2_CONF=${ST2_REPO}/conf/st2.dev.conf fi + + ST2_CONF=$(realpath ${ST2_CONF}) echo "Using st2 config file: $ST2_CONF" if [ ! -f "$ST2_CONF" ]; then From 98b77d13a86dea846ba2040ee0da0a3be6885287 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Mon, 11 Mar 2019 16:06:01 +0100 Subject: [PATCH 30/43] Make sure we also start coordinator heartbeat process in places where we use coordinator backend for locking purpose. This way it's safe and also works correctly with drivers which are timeout based. --- st2actions/st2actions/scheduler/handler.py | 2 +- st2common/st2common/policies/concurrency.py | 2 +- st2common/st2common/service_setup.py | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/st2actions/st2actions/scheduler/handler.py b/st2actions/st2actions/scheduler/handler.py index 74027b88ed..08491beb6a 100644 --- a/st2actions/st2actions/scheduler/handler.py +++ b/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 diff --git a/st2common/st2common/policies/concurrency.py b/st2common/st2common/policies/concurrency.py index 883721b628..52289f002d 100644 --- a/st2common/st2common/policies/concurrency.py +++ b/st2common/st2common/policies/concurrency.py @@ -30,7 +30,7 @@ def __init__(self, policy_ref, policy_type, threshold=0, action='delay'): self.threshold = threshold self.policy_action = action - self.coordinator = coordination.get_coordinator() + self.coordinator = coordination.get_coordinator(start_heart=True) def _get_status_for_policy_action(self, action): if action == 'delay': diff --git a/st2common/st2common/service_setup.py b/st2common/st2common/service_setup.py index 83c4524e71..83a692a57f 100644 --- a/st2common/st2common/service_setup.py +++ b/st2common/st2common/service_setup.py @@ -195,7 +195,7 @@ def teardown(): db_teardown() # 2. Tear down the coordinator - coordinator = coordination.get_coordinator() + coordinator = coordination.get_coordinator(start_heart=False) coordination.coordinator_teardown(coordinator) From f5f5116d273bd99ee19bab728617275c164f992f Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Mon, 11 Mar 2019 17:03:58 +0100 Subject: [PATCH 31/43] Use more compatible function. --- tools/launchdev.sh | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tools/launchdev.sh b/tools/launchdev.sh index b883cd545e..67bb908146 100755 --- a/tools/launchdev.sh +++ b/tools/launchdev.sh @@ -68,7 +68,7 @@ function init(){ fi VIRTUALENV=${VIRTUALENV_DIR:-${ST2_REPO}/virtualenv} - VIRTUALENV=$(realpath ${VIRTUALENV}) + VIRTUALENV=$(readlink -f ${VIRTUALENV}) PY=${VIRTUALENV}/bin/python PYTHON_VERSION=$(${PY} --version 2>&1) @@ -79,7 +79,7 @@ function init(){ ST2_CONF=${ST2_REPO}/conf/st2.dev.conf fi - ST2_CONF=$(realpath ${ST2_CONF}) + ST2_CONF=$(readlink -f ${ST2_CONF}) echo "Using st2 config file: $ST2_CONF" if [ ! -f "$ST2_CONF" ]; then From f530cbea6ab7f3f03a46f3241be95879aeb8d45a Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Mon, 11 Mar 2019 17:28:48 +0100 Subject: [PATCH 32/43] Revert "Make sure we also start coordinator heartbeat process in places where we" This reverts commit 98b77d13a86dea846ba2040ee0da0a3be6885287. --- st2actions/st2actions/scheduler/handler.py | 2 +- st2common/st2common/policies/concurrency.py | 2 +- st2common/st2common/service_setup.py | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/st2actions/st2actions/scheduler/handler.py b/st2actions/st2actions/scheduler/handler.py index 08491beb6a..74027b88ed 100644 --- a/st2actions/st2actions/scheduler/handler.py +++ b/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(start_heart=True) + self._coordinator = coordination_service.get_coordinator() self._main_thread = None self._cleanup_thread = None diff --git a/st2common/st2common/policies/concurrency.py b/st2common/st2common/policies/concurrency.py index 52289f002d..883721b628 100644 --- a/st2common/st2common/policies/concurrency.py +++ b/st2common/st2common/policies/concurrency.py @@ -30,7 +30,7 @@ def __init__(self, policy_ref, policy_type, threshold=0, action='delay'): self.threshold = threshold self.policy_action = action - self.coordinator = coordination.get_coordinator(start_heart=True) + self.coordinator = coordination.get_coordinator() def _get_status_for_policy_action(self, action): if action == 'delay': diff --git a/st2common/st2common/service_setup.py b/st2common/st2common/service_setup.py index 83a692a57f..83c4524e71 100644 --- a/st2common/st2common/service_setup.py +++ b/st2common/st2common/service_setup.py @@ -195,7 +195,7 @@ def teardown(): db_teardown() # 2. Tear down the coordinator - coordinator = coordination.get_coordinator(start_heart=False) + coordinator = coordination.get_coordinator() coordination.coordinator_teardown(coordinator) From 5ccf4ba0fbcc6f396a3004e7b896d172346cd9b4 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Mon, 11 Mar 2019 18:53:33 +0100 Subject: [PATCH 33/43] Update affected test. --- st2api/st2api/controllers/v1/service_registry.py | 7 +++++-- st2api/tests/unit/controllers/v1/test_service_registry.py | 3 ++- st2common/st2common/services/coordination.py | 2 ++ 3 files changed, 9 insertions(+), 3 deletions(-) diff --git a/st2api/st2api/controllers/v1/service_registry.py b/st2api/st2api/controllers/v1/service_registry.py index 335cd87b3b..9384a73480 100644 --- a/st2api/st2api/controllers/v1/service_registry.py +++ b/st2api/st2api/controllers/v1/service_registry.py @@ -48,11 +48,13 @@ def get_one(self, group_id, requester_user): coordinator = coordination.get_coordinator() + if not isinstance(group_id, six.binary_type): + group_id = group_id.encode('utf-8') + try: - group_id = six.binary_type(six.text_type(group_id).encode('ascii')) member_ids = list(coordinator.get_members(group_id).get()) except GroupNotCreated: - msg = ('Group with ID "%s" not found.' % (group_id)) + msg = ('Group with ID "%s" not found.' % (group_id.decode('utf-8'))) raise StackStormDBObjectNotFoundError(msg) result = { @@ -62,6 +64,7 @@ def get_one(self, group_id, requester_user): 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 } diff --git a/st2api/tests/unit/controllers/v1/test_service_registry.py b/st2api/tests/unit/controllers/v1/test_service_registry.py index 6cab2496da..a31f0bd31e 100644 --- a/st2api/tests/unit/controllers/v1/test_service_registry.py +++ b/st2api/tests/unit/controllers/v1/test_service_registry.py @@ -72,7 +72,8 @@ def test_get_group_members(self): self.assertEqual(resp.json, { 'members': [ { - 'member_id': member_id, + 'group_id': 'mock_service', + 'member_id': member_id.decode('utf-8'), 'capabilities': { 'key1': 'value1', 'name': 'mock_service', diff --git a/st2common/st2common/services/coordination.py b/st2common/st2common/services/coordination.py index c9d270fc68..c4dad1f078 100644 --- a/st2common/st2common/services/coordination.py +++ b/st2common/st2common/services/coordination.py @@ -227,6 +227,8 @@ def get_coordinator(start_heart=False, use_cache=True): def get_member_id(): """ Retrieve member if for the current process. + + :rtype: ``bytes`` """ proc_info = system_info.get_process_info() member_id = six.b('%s_%d' % (proc_info['hostname'], proc_info['pid'])) From 4ebcdc32535e5b892e6c6f3afd132ceae3ad4097 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Mon, 11 Mar 2019 19:00:41 +0100 Subject: [PATCH 34/43] Update sample config. --- conf/st2.dev.conf | 1 + 1 file changed, 1 insertion(+) diff --git a/conf/st2.dev.conf b/conf/st2.dev.conf index 798aad32ca..8043dffb8f 100644 --- a/conf/st2.dev.conf +++ b/conf/st2.dev.conf @@ -86,6 +86,7 @@ protocol = udp # 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 From ed37c020b53487c9af21849c6490f180ff448416 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Tue, 12 Mar 2019 18:17:57 +0100 Subject: [PATCH 35/43] Revert "Revert "Make sure we also start coordinator heartbeat process in places where we"" This reverts commit f530cbea6ab7f3f03a46f3241be95879aeb8d45a. --- st2actions/st2actions/scheduler/handler.py | 2 +- st2common/st2common/policies/concurrency.py | 2 +- st2common/st2common/service_setup.py | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/st2actions/st2actions/scheduler/handler.py b/st2actions/st2actions/scheduler/handler.py index 74027b88ed..08491beb6a 100644 --- a/st2actions/st2actions/scheduler/handler.py +++ b/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 diff --git a/st2common/st2common/policies/concurrency.py b/st2common/st2common/policies/concurrency.py index 883721b628..52289f002d 100644 --- a/st2common/st2common/policies/concurrency.py +++ b/st2common/st2common/policies/concurrency.py @@ -30,7 +30,7 @@ def __init__(self, policy_ref, policy_type, threshold=0, action='delay'): self.threshold = threshold self.policy_action = action - self.coordinator = coordination.get_coordinator() + self.coordinator = coordination.get_coordinator(start_heart=True) def _get_status_for_policy_action(self, action): if action == 'delay': diff --git a/st2common/st2common/service_setup.py b/st2common/st2common/service_setup.py index 83c4524e71..83a692a57f 100644 --- a/st2common/st2common/service_setup.py +++ b/st2common/st2common/service_setup.py @@ -195,7 +195,7 @@ def teardown(): db_teardown() # 2. Tear down the coordinator - coordinator = coordination.get_coordinator() + coordinator = coordination.get_coordinator(start_heart=False) coordination.coordinator_teardown(coordinator) From 2f481e035a75bd5cb4bb24c73a33ef2096aab328 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Mon, 18 Mar 2019 14:29:17 +0100 Subject: [PATCH 36/43] Fix typo. --- CHANGELOG.rst | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 632948ce75..5782c58096 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -113,7 +113,7 @@ Fixed * Fix improper CORS where request from an origin not listed in ``allowed_origins`` will be responded with ``null`` for the ``Access-Control-Allow-Origin`` header. The fix returns the first of our - Bllowed origins if the requesting origin is not a supported origin. Reported by Barak Tawily. + allowed origins if the requesting origin is not a supported origin. Reported by Barak Tawily. (bug fix) 2.10.2 - February 21, 2019 From 893cfccbd990000c150bb22e2fa06cb0d732e1d5 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Mon, 18 Mar 2019 14:30:27 +0100 Subject: [PATCH 37/43] Add docstring which clarifies the behavior. --- st2common/st2common/services/coordination.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/st2common/st2common/services/coordination.py b/st2common/st2common/services/coordination.py index c4dad1f078..009794aef8 100644 --- a/st2common/st2common/services/coordination.py +++ b/st2common/st2common/services/coordination.py @@ -56,6 +56,10 @@ def heartbeat(self): class NoOpAsyncResult(object): + """ + In most scenarios, tooz library returns an async result, a future and this + class wrapper is here to correctly mimic tooz API and behavior. + """ def __init__(self, result=None): self._result = result From e4cfa82915502ca57785ea487c028cef37f7a89c Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Mon, 18 Mar 2019 15:13:33 +0100 Subject: [PATCH 38/43] Only teardown coordinator if one has been initialized. --- st2common/st2common/service_setup.py | 2 +- st2common/st2common/services/coordination.py | 16 ++++++++++++++-- 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/st2common/st2common/service_setup.py b/st2common/st2common/service_setup.py index 599be40ff3..3c1002a0db 100644 --- a/st2common/st2common/service_setup.py +++ b/st2common/st2common/service_setup.py @@ -195,7 +195,7 @@ def teardown(): db_teardown() # 2. Tear down the coordinator - coordinator = coordination.get_coordinator(start_heart=False) + coordinator = coordination.get_coordinator_if_set() coordination.coordinator_teardown(coordinator) diff --git a/st2common/st2common/services/coordination.py b/st2common/st2common/services/coordination.py index 009794aef8..11ed241acd 100644 --- a/st2common/st2common/services/coordination.py +++ b/st2common/st2common/services/coordination.py @@ -34,6 +34,7 @@ 'configured', 'get_coordinator', + 'get_coordinator_if_set', 'get_member_id', 'coordinator_setup', @@ -200,8 +201,9 @@ def coordinator_setup(start_heart=False): return coordinator -def coordinator_teardown(coordinator): - coordinator.stop() +def coordinator_teardown(coordinator=None): + if coordinator: + coordinator.stop() def get_coordinator(start_heart=False, use_cache=True): @@ -224,7 +226,17 @@ def get_coordinator(start_heart=False, use_cache=True): if not COORDINATOR: COORDINATOR = coordinator_setup(start_heart=start_heart) + else: + LOG.debug('Using cached coordinator instance: %s' % (str(COORDINATOR))) + + return COORDINATOR + +def get_coordinator_if_set(): + """ + Return a coordinator instance if one has been initialized, None otherwise. + """ + global COORDINATOR return COORDINATOR From 639e708e1a5b7517a534dd29cd87f3e64c89503f Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Mon, 18 Mar 2019 15:16:25 +0100 Subject: [PATCH 39/43] Make sure we start heartbeat process everywhere we retrieve coordinator instance. --- st2common/st2common/services/workflows.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/st2common/st2common/services/workflows.py b/st2common/st2common/services/workflows.py index f69c963f91..ddc8128ee1 100644 --- a/st2common/st2common/services/workflows.py +++ b/st2common/st2common/services/workflows.py @@ -727,7 +727,7 @@ def handle_action_execution_completion(ac_ex_db): task_ex_id = ac_ex_db.context['orquesta']['task_execution_id'] # Acquire lock before write operations. - with coord_svc.get_coordinator().get_lock(wf_ex_id): + with coord_svc.get_coordinator(start_heart=True).get_lock(wf_ex_id): # Get execution records for logging purposes. wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_id) task_ex_db = wf_db_access.TaskExecution.get_by_id(task_ex_id) From 3c82ff4a0a723ebdb35725454dc3195e71d41cf6 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Mon, 18 Mar 2019 16:35:08 +0100 Subject: [PATCH 40/43] To be on the safe side default start_heart to True and log a message when we create a new instance. --- st2common/st2common/services/coordination.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/st2common/st2common/services/coordination.py b/st2common/st2common/services/coordination.py index 11ed241acd..31138ac8df 100644 --- a/st2common/st2common/services/coordination.py +++ b/st2common/st2common/services/coordination.py @@ -174,7 +174,7 @@ def configured(): return backend_configured and not mock_backend -def coordinator_setup(start_heart=False): +def coordinator_setup(start_heart=True): """ Sets up the client for the coordination service. @@ -206,7 +206,7 @@ def coordinator_teardown(coordinator=None): coordinator.stop() -def get_coordinator(start_heart=False, use_cache=True): +def get_coordinator(start_heart=True, use_cache=True): """ :param start_heart: True to start heartbeating process. :type start_heart: ``bool`` @@ -226,6 +226,7 @@ def get_coordinator(start_heart=False, use_cache=True): if not COORDINATOR: COORDINATOR = coordinator_setup(start_heart=start_heart) + LOG.debug('Initializing and caching new coordinator instance: %s' % (str(COORDINATOR))) else: LOG.debug('Using cached coordinator instance: %s' % (str(COORDINATOR))) From 5c6cc5db888a19be50194d05108f8ac7391156a4 Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Mon, 18 Mar 2019 21:15:34 +0100 Subject: [PATCH 41/43] Use consistent casing. --- st2client/st2client/models/service_registry.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/st2client/st2client/models/service_registry.py b/st2client/st2client/models/service_registry.py index 7dbed43485..6cf7206de0 100644 --- a/st2client/st2client/models/service_registry.py +++ b/st2client/st2client/models/service_registry.py @@ -27,9 +27,9 @@ class ServiceRegistry(core.Resource): _alias = 'service-registry' - _display_name = 'service registry' - _plural = 'service registry' - _plural_display_name = 'service registry' + _display_name = 'Service Registry' + _plural = 'Service Registry' + _plural_display_name = 'Service Registry' class ServiceRegistryGroup(core.Resource): From 100ad9fb405eeb29ff57d6600d8bc5eccd517c9b Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Mon, 18 Mar 2019 21:39:22 +0100 Subject: [PATCH 42/43] Remove assignment. --- st2common/st2common/services/coordination.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/st2common/st2common/services/coordination.py b/st2common/st2common/services/coordination.py index 31138ac8df..477330bdda 100644 --- a/st2common/st2common/services/coordination.py +++ b/st2common/st2common/services/coordination.py @@ -221,8 +221,7 @@ def get_coordinator(start_heart=True, use_cache=True): 'service will use best effort approach and race conditions are possible.') if not use_cache: - coordinator = coordinator_setup(start_heart=start_heart) - return coordinator + return coordinator_setup(start_heart=start_heart) if not COORDINATOR: COORDINATOR = coordinator_setup(start_heart=start_heart) From 7fffa523df38dc180e3d5d043d43e3ec393c12af Mon Sep 17 00:00:00 2001 From: Tomaz Muraus Date: Tue, 19 Mar 2019 08:39:37 +0100 Subject: [PATCH 43/43] Use item variable name instead of group_id_. --- st2api/st2api/controllers/v1/service_registry.py | 2 +- tools/list_group_members.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/st2api/st2api/controllers/v1/service_registry.py b/st2api/st2api/controllers/v1/service_registry.py index 9384a73480..91acb36b60 100644 --- a/st2api/st2api/controllers/v1/service_registry.py +++ b/st2api/st2api/controllers/v1/service_registry.py @@ -34,7 +34,7 @@ def get_all(self, requester_user): coordinator = coordination.get_coordinator() group_ids = list(coordinator.get_groups().get()) - group_ids = [group_id_.decode('utf-8') for group_id_ in group_ids] + group_ids = [item.decode('utf-8') for item in group_ids] result = { 'groups': group_ids diff --git a/tools/list_group_members.py b/tools/list_group_members.py index 5bd0ec6fec..c5db38acea 100755 --- a/tools/list_group_members.py +++ b/tools/list_group_members.py @@ -31,7 +31,7 @@ def main(group_id=None): if not group_id: group_ids = list(coordinator.get_groups().get()) - group_ids = [group_id_.decode('utf-8') for group_id_ in group_ids] + group_ids = [item.decode('utf-8') for item in group_ids] print('Available groups (%s):' % (len(group_ids))) for group_id in group_ids: