Subprocess process manager for CI workflows

.ruff.toml

@@ -5,4 +5,4 @@ ignore = ["ALL"]
 select = [
     "F",   # PyFlakes
     "I"   # Isort
-]
+]

pyproject.toml

@@ -54,6 +54,7 @@ drunc-fsm-tests = "drunc.tests.fsm:main"
 application-registry-service = "drunc.apps.app_connectivity_server:main"
 drunc-ssh-validator = "drunc.apps.ssh_validator:main"
 drunc-ssh-doctor = "drunc.apps.ssh_doctor:main"
+drunc-process-wrapper = "drunc.apps.process_wrapper:main"
 
 [tool.setuptools.packages.find]
 where = ["src"]
@@ -69,6 +70,10 @@ addopts = "-v --tb=short --cov=drunc --cov=src/drunc tests/"
 source = ["drunc"]
 omit = ["tests/*"]
 
+[tool.ruff]
+exclude = ["tests/"]
+
+
 # * See https://docs.astral.sh/ruff/rules/ for details on Ruff's linting options
 [tool.ruff.lint]
 select = [

src/drunc/apps/process_wrapper.py

@@ -0,0 +1,49 @@
+import os
+import signal
+import subprocess
+import time
+
+import click
+
+from drunc.process_manager.subprocess_process_manager import on_parent_exit
+
+
+def terminate_all(sig, frame):
+    pgrp = os.getpgid(os.getpid())
+    os.killpg(pgrp, signal.SIGKILL)
+
+
+@click.command()
+@click.argument("cmd")
+@click.option(
+    "-l",
+    "--log",
+    "log_path",
+    type=click.Path(file_okay=True, dir_okay=False),
+    required=True,
+)
+def main(cmd: str, log_path: str):
+    signal.signal(signal.SIGTERM, terminate_all)
+
+    with open(log_path, "w") as logfile:
+        proc = subprocess.Popen(
+            cmd,
+            shell=True,
+            stdout=logfile,
+            stderr=logfile,
+            preexec_fn=on_parent_exit(
+                signal.SIGTERM,  # Propagate SIGHUP to child processes, SIGKILL doesn't seem to kill gunicorn...
+                setsid=False,  # Don't create a new session, so that the process group can be killed
+            ),
+        )
+
+        return_code = None
+        while True:
+            return_code = proc.poll()
+            if return_code is not None:
+                return return_code
+            time.sleep(0.1)
+
+
+if __name__ == "__main__":
+    main()

src/drunc/broadcast/server/broadcast_sender.py

@@ -60,8 +60,7 @@ def __init__(
     def describe_broadcast(self):
         if self.implementation:
             return self.implementation.describe_broadcast()
-        else:
-            return None
+        return None
 
     def can_broadcast(self):
         if not self.implementation:

src/drunc/broadcast/utils.py

@@ -26,5 +26,4 @@ def get_broadcast_level_from_broadcast_type(
     bt = BroadcastType.Name(btype)
     if bt not in levels:
         return logger.info
-    else:
-        return getattr(logger, levels[bt].lower())
+    return getattr(logger, levels[bt].lower())

src/drunc/connectivity_service/client.py

@@ -194,11 +194,10 @@ def resolve(self, uid_regex: str, data_type: str, ntries=50) -> dict:
                 content = response.json()
                 if content:
                     return content
-                else:
-                    self.log.debug(
-                        f"Could not find the address of '{uid_regex}' on the application registry"
-                    )
-                    time.sleep(0.2)
+                self.log.debug(
+                    f"Could not find the address of '{uid_regex}' on the application registry"
+                )
+                time.sleep(0.2)
 
             except (HTTPError, ConnectionError, ReadTimeout) as e:
                 self.log.debug(e)

src/drunc/controller/children_interface/grpc_child.py

@@ -22,9 +22,7 @@
 
 from drunc.broadcast.client.broadcast_handler import BroadcastHandler
 from drunc.broadcast.client.configuration import BroadcastClientConfHandler
-from drunc.connectivity_service.exceptions import (
-    ApplicationLookupUnsuccessful,
-)
+from drunc.connectivity_service.exceptions import ApplicationLookupUnsuccessful
 from drunc.controller.children_interface.child_node import ChildNode
 from drunc.exceptions import DruncSetupException
 from drunc.utils.configuration import ConfHandler, ConfTypes
@@ -116,10 +114,8 @@ def _setup_connection(self):
                     if tries_remaining == 0:
                         raise server_unreachable_error
                     self.log.info(
-                        (
-                            f"Could not connect to the controller ({self.uri}). "
-                            f"Trying {tries_remaining} more times..."
-                        )
+                        f"Could not connect to the controller ({self.uri}). "
+                        f"Trying {tries_remaining} more times..."
                     )
                     time.sleep(5)
 

src/drunc/controller/children_interface/rest_api_child.py

@@ -294,8 +294,7 @@ def send_app_command(
                     raise CouldnotSendCommand(
                         f"Connection error to {self.app_url}"
                     ) from e
-                else:
-                    self.log.error("Trying again...")
+                self.log.error("Trying again...")
 
         self.log.debug(f"Ack to {self.app}: {ack.status_code}")
         self.sent_cmd = cmd_id
@@ -326,13 +325,12 @@ def check_response(self, timeout: int = 0) -> dict:
                 raise NoResponse(
                     f"No response available from {self.app} for command {self.sent_cmd}"
                 )
-            else:
-                self.log.error(
-                    f"Timeout while waiting for a reply from {self.app} for command {self.sent_cmd}"
-                )
-                raise ResponseTimeout(
-                    f"Timeout while waiting for a reply from {self.app} for command {self.sent_cmd}"
-                )
+            self.log.error(
+                f"Timeout while waiting for a reply from {self.app} for command {self.sent_cmd}"
+            )
+            raise ResponseTimeout(
+                f"Timeout while waiting for a reply from {self.app} for command {self.sent_cmd}"
+            )
         return r
 
 

src/drunc/controller/controller.py

@@ -3,9 +3,10 @@
 import threading
 import time
 import traceback
+from collections.abc import Callable
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from functools import wraps
-from typing import Callable, List, TypeVar
+from typing import TypeVar
 
 from druncschema.authoriser_pb2 import ActionType, SystemType
 from druncschema.broadcast_pb2 import BroadcastType
@@ -51,10 +52,7 @@
 from drunc.exceptions import DruncCommandException, DruncException
 from drunc.fsm.actions.utils import get_dotdrunc_json
 from drunc.fsm.configuration import FSMConfHandler
-from drunc.fsm.exceptions import (
-    DotDruncJsonIncorrectFormat,
-    DotDruncJsonNotFound,
-)
+from drunc.fsm.exceptions import DotDruncJsonIncorrectFormat, DotDruncJsonNotFound
 from drunc.fsm.utils import convert_fsm_transition
 from drunc.utils.grpc_utils import UnpackingError, pack_to_any, unpack_any
 from drunc.utils.utils import get_logger
@@ -208,7 +206,7 @@ def wrap(obj, request, context):
 
 
 class Controller(ControllerServicer):
-    children_nodes: List[ChildNode] = []
+    children_nodes: list[ChildNode] = []
 
     def __init__(self, configuration, name: str, session: str, token: Token):
         super().__init__()
@@ -629,7 +627,7 @@ def propagate_to_child(
                     )
 
                 self.log.error(
-                    f"Failed to propagate {command_name} to {child.name} ({child.name}) EXCEPTION THROWN: {str(e)}"
+                    f"Failed to propagate {command_name} to {child.name} ({child.name}) EXCEPTION THROWN: {e!s}"
                 )
 
         threads = []

src/drunc/controller/controller_actor.py

@@ -1,5 +1,4 @@
 import threading
-from typing import Optional
 
 from druncschema.token_pb2 import Token
 
@@ -8,7 +7,7 @@
 
 
 class ControllerActor:
-    def __init__(self, token: Optional[Token] = None):
+    def __init__(self, token: Token | None = None):
         self.log = get_logger("controller.actor")
         self._token = Token(token="", user_name="")
         if token is not None:
@@ -21,7 +20,7 @@ def get_token(self) -> Token:
     def get_user_name(self) -> str:
         return self._token.user_name
 
-    def _update_actor(self, token: Optional[Token] = None) -> None:
+    def _update_actor(self, token: Token | None = None) -> None:
         self._lock.acquire()
         self._token = Token(token="", user_name="")
         if token is not None:

src/drunc/controller/controller_driver.py

@@ -20,11 +20,7 @@
 from druncschema.token_pb2 import Token
 
 from drunc.exceptions import DruncServerSideError
-from drunc.utils.grpc_utils import (
-    UnpackingError,
-    handle_grpc_error,
-    unpack_any,
-)
+from drunc.utils.grpc_utils import UnpackingError, handle_grpc_error, unpack_any
 from drunc.utils.shell_utils import DecodedResponse
 from drunc.utils.utils import get_logger
 
@@ -297,7 +293,7 @@ def text(verb="not executed", reason=""):
 
             elif response.data.Is(PlainText.DESCRIPTOR):
                 txt = unpack_any(response.data, PlainText)
-                error_txt = txt.text  # noqa: F841  (might need to revisit this)
+                error_txt = txt.text
                 dr.data = error_txt
 
             if response.flag in [

src/drunc/controller/interface/commands.py

@@ -362,7 +362,7 @@ def expert_command(
         if string:
             data = json.loads(command)
         else:
-            with open(command, "r") as f:
+            with open(command) as f:
                 data = json.load(f)
 
     except FileNotFoundError:

src/drunc/controller/interface/context.py

@@ -6,10 +6,7 @@
 from drunc.broadcast.client.configuration import BroadcastClientConfHandler
 from drunc.controller.controller_driver import ControllerDriver
 from drunc.utils.configuration import ConfTypes
-from drunc.utils.shell_utils import (
-    ShellContext,
-    create_dummy_token_from_uname,
-)
+from drunc.utils.shell_utils import ShellContext, create_dummy_token_from_uname
 from drunc.utils.utils import resolve_localhost_to_hostname
 
 

src/drunc/controller/interface/shell_utils.py

@@ -297,7 +297,7 @@ def controller_setup(ctx, controller_address):
 
     if state == "initialising":
         log.error("Controller did not initialise in time")
-        return
+        return None
 
     log.debug(f"Taking control of the controller as {ctx.get_token()}")
     try:
@@ -375,12 +375,11 @@ def tree_prefix(i, n):
     last = "└── "
     if i == 0 and n == 1:
         return first_one
-    elif i == 0:
+    if i == 0:
         return first_many
-    elif i == n - 1:
+    if i == n - 1:
         return last
-    else:
-        return next
+    return next
 
 
 def validate_and_format_fsm_arguments(

src/drunc/controller/stateful_node.py

@@ -1,7 +1,8 @@
 from __future__ import annotations
 
 import abc
-from typing import Any, Callable, Optional
+from collections.abc import Callable
+from typing import Any
 
 from druncschema.opmon.FSM_pb2 import FSMStatus
 
@@ -32,9 +33,7 @@ def value(self, value):
             self.stateful_node.log.info(f"{self._name} changed to {value}")
             self.stateful_node.publish_state()
 
-    def __init__(
-        self, name: str, stateful_node=None, initial_value: Optional[str] = None
-    ):
+    def __init__(self, name: str, stateful_node=None, initial_value: str | None = None):
         self._name = name
         self.stateful_node = stateful_node
         self._value = initial_value
@@ -89,7 +88,7 @@ class StatefulNode(abc.ABC):
     def __init__(
         self,
         fsm_configuration,
-        publisher: Optional[Callable[[Any], None]] = None,
+        publisher: Callable[[Any], None] | None = None,
         init_state: str = "",
         session: str = "",
         name: str = "",