Deregister services from service registry on shutdown

st2actions/st2actions/cmd/actionrunner.py

@@ -30,6 +30,7 @@
 from st2common import log as logging
 from st2common.service_setup import setup as common_setup
 from st2common.service_setup import teardown as common_teardown
+from st2common.service_setup import deregister_service
 
 __all__ = ["main"]
 
@@ -75,6 +76,7 @@ def _run_worker():
         errors = False
 
         try:
+            deregister_service(service="actionrunner")
             action_worker.shutdown()
         except:
             LOG.exception("Unable to shutdown worker.")

st2actions/st2actions/cmd/scheduler.py

@@ -28,6 +28,7 @@
 from st2common import log as logging
 from st2common.service_setup import teardown as common_teardown
 from st2common.service_setup import setup as common_setup
+from st2common.service_setup import deregister_service
 
 __all__ = ["main"]
 
@@ -101,6 +102,7 @@ def _run_scheduler():
         errors = False
 
         try:
+            deregister_service(service="scheduler")
             handler.shutdown()
             entrypoint.shutdown()
         except:

st2actions/st2actions/cmd/st2notifier.py

@@ -25,6 +25,7 @@
 from st2common import log as logging
 from st2common.service_setup import setup as common_setup
 from st2common.service_setup import teardown as common_teardown
+from st2common.service_setup import deregister_service
 from st2actions.notifier import config
 from st2actions.notifier import notifier
 
@@ -53,6 +54,7 @@ def _run_worker():
         actions_notifier.start(wait=True)
     except (KeyboardInterrupt, SystemExit):
         LOG.info("(PID=%s) Actions notifier stopped.", os.getpid())
+        deregister_service(service="notifier")
         actions_notifier.shutdown()
     return 0
 

st2actions/st2actions/cmd/workflow_engine.py

@@ -32,6 +32,7 @@
 from st2common import log as logging
 from st2common.service_setup import setup as common_setup
 from st2common.service_setup import teardown as common_teardown
+from st2common.service_setup import deregister_service as deregister
 
 __all__ = ["main"]
 
@@ -72,6 +73,7 @@ def run_server():
         engine.start(wait=True)
     except (KeyboardInterrupt, SystemExit):
         LOG.info("(PID=%s) Workflow engine stopped.", os.getpid())
+        deregister(service="workflow_engine")
         engine.shutdown()
     except:
         LOG.exception("(PID=%s) Workflow engine unexpectedly stopped.", os.getpid())

st2common/st2common/service_setup.py

@@ -29,6 +29,8 @@
 import eventlet.debug
 from oslo_config import cfg
 from tooz.coordination import GroupAlreadyExist
+from tooz.coordination import GroupNotCreated
+from tooz.coordination import MemberNotJoined
 
 from st2common import log as logging
 from st2common.constants.logging import DEFAULT_LOGGING_CONF_PATH
@@ -62,6 +64,7 @@
     "db_setup",
     "db_teardown",
     "register_service_in_service_registry",
+    "deregister_service",
 ]
 
 # Message which is logged if non utf-8 locale is detected on startup.
@@ -339,3 +342,22 @@ def register_service_in_service_registry(service, capabilities=None, start_heart
         % (group_id, member_id, capabilities)
     )
     return coordinator.join_group(group_id, capabilities=capabilities).get()
+
+
+def deregister_service(service, start_heart=True):
+
+    if not isinstance(service, six.binary_type):
+        group_id = service.encode("utf-8")
+    else:
+        group_id = service
+
+    coordinator = coordination.get_coordinator(start_heart=start_heart)
+
+    member_id = coordination.get_member_id()
+    LOG.debug(
+        'Leaving service registry group "%s" as member_id "%s"' % (group_id, member_id)
+    )
+    try:
+        coordinator.leave_group(group_id).get()
+    except (GroupNotCreated, MemberNotJoined):
+        pass