Link shutdown routine and sigterm handler to main thread

.github/workflows/ci.yaml

@@ -21,6 +21,9 @@ on:
     - cron:  '0 0 * * *'
 
 jobs:
+  # TODO: Fix the required checks!
+  #       When the pre_job triggers and skips builds, it prevents merging the PR because
+  #       the required checks are reported as skipped instead of passed.
   # Special job which automatically cancels old runs for the same branch, prevents runs for the
   # same file set which has already passed, etc.
   pre_job:
@@ -40,7 +43,7 @@ jobs:
     needs: pre_job
     # NOTE: We always want to run job on master since we run some additional checks there (code
     # coverage, etc)
-    if: ${{ needs.pre_job.outputs.should_skip != 'true' || github.ref == 'refs/heads/master' }}
+    # if: ${{ needs.pre_job.outputs.should_skip != 'true' || github.ref == 'refs/heads/master' }}
     name: '${{ matrix.name }} - Python ${{ matrix.python-version-short }}'
     runs-on: ubuntu-latest
     strategy:
@@ -135,7 +138,8 @@ jobs:
     needs: pre_job
     # NOTE: We always want to run job on master since we run some additional checks there (code
     # coverage, etc)
-    if: ${{ needs.pre_job.outputs.should_skip != 'true' || github.ref == 'refs/heads/master' }}
+    # NB: disabled. See TODO above pre_job
+    # if: ${{ needs.pre_job.outputs.should_skip != 'true' || github.ref == 'refs/heads/master' }}
     name: '${{ matrix.name }} - Python ${{ matrix.python-version-short }}'
     runs-on: ubuntu-latest
     strategy:
@@ -304,7 +308,7 @@ jobs:
     needs: pre_job
     # NOTE: We always want to run job on master since we run some additional checks there (code
     # coverage, etc)
-    if: ${{ needs.pre_job.outputs.should_skip != 'true' || github.ref == 'refs/heads/master' }}
+    # if: ${{ needs.pre_job.outputs.should_skip != 'true' || github.ref == 'refs/heads/master' }}
     name: '${{ matrix.name }} - Python ${{ matrix.python-version-short }}'
     runs-on: ubuntu-latest
     strategy:

.github/workflows/orquesta-integration-tests.yaml

@@ -25,6 +25,9 @@ on:
     - cron:  '0 0 * * *'
 
 jobs:
+  # TODO: Fix the required checks!
+  #       When the pre_job triggers and skips builds, it prevents merging the PR because
+  #       the required checks are reported as skipped instead of passed.
   # Special job which automatically cancels old runs for the same branch, prevents runs for the
   # same file set which has already passed, etc.
   pre_job:
@@ -43,7 +46,7 @@ jobs:
     needs: pre_job
     # NOTE: We always want to run job on master since we run some additional checks there (code
     # coverage, etc)
-    if: ${{ needs.pre_job.outputs.should_skip != 'true' || github.ref == 'refs/heads/master' }}
+    # if: ${{ needs.pre_job.outputs.should_skip != 'true' || github.ref == 'refs/heads/master' }}
     name: '${{ matrix.name }} - Python ${{ matrix.python-version-short }}'
     runs-on: ubuntu-latest
     strategy:

st2actions/st2actions/cmd/actionrunner.py

@@ -38,10 +38,10 @@
 ACTIONRUNNER = "actionrunner"
 
 
-def _setup_sigterm_handler():
+def _setup_sigterm_handler(action_worker):
     def sigterm_handler(signum=None, frame=None):
         # This will cause SystemExit to be throw and allow for component cleanup.
-        sys.exit(0)
+        action_worker.kill()
 
     # Register a SIGTERM signal handler which calls sys.exit which causes SystemExit to
     # be thrown. We catch SystemExit and handle cleanup there.
@@ -60,14 +60,12 @@ def _setup():
         capabilities=capabilities,
     )
 
-    _setup_sigterm_handler()
-
 
 def _run_worker():
     LOG.info("(PID=%s) Worker started.", os.getpid())
 
     action_worker = worker.get_worker()
-
+    _setup_sigterm_handler(action_worker)
     try:
         action_worker.start()
         action_worker.wait()

st2actions/st2actions/cmd/workflow_engine.py

@@ -40,10 +40,10 @@
 WORKFLOW_ENGINE = "workflow_engine"
 
 
-def setup_sigterm_handler():
+def setup_sigterm_handler(engine):
     def sigterm_handler(signum=None, frame=None):
         # This will cause SystemExit to be throw and allow for component cleanup.
-        sys.exit(0)
+        engine.kill()
 
     # Register a SIGTERM signal handler which calls sys.exit which causes SystemExit to
     # be thrown. We catch SystemExit and handle cleanup there.
@@ -62,14 +62,12 @@ def setup():
         capabilities=capabilities,
     )
 
-    setup_sigterm_handler()
-
 
 def run_server():
     LOG.info("(PID=%s) Workflow engine started.", os.getpid())
 
     engine = workflows.get_engine()
-
+    setup_sigterm_handler(engine)
     try:
         engine.start(wait=True)
     except (KeyboardInterrupt, SystemExit):
@@ -79,7 +77,6 @@ def run_server():
     except:
         LOG.exception("(PID=%s) Workflow engine unexpectedly stopped.", os.getpid())
         return 1
-
     return 0
 
 

st2common/st2common/transport/consumers.py

@@ -205,6 +205,9 @@ def shutdown(self):
         LOG.info("Shutting down %s...", self.__class__.__name__)
         self._queue_consumer.shutdown()
 
+    def kill(self):
+        self._consumer_thread.kill(SystemExit())
+
     @abc.abstractmethod
     def process(self, message):
         pass