From dd2025c65267a37a1ea0ccaa45a2920ab15ecaaf Mon Sep 17 00:00:00 2001 From: PawelPlesniak Date: Tue, 23 Jun 2026 11:45:23 +0200 Subject: [PATCH 01/13] Add logging calls --- .../process_manager/ssh_process_manager.py | 22 +++++++++++++++++-- .../ssh_process_lifetime_manager_shell.py | 19 +++++++++++++++- src/drunc/unified_shell/shell.py | 12 ++++++++++ 3 files changed, 50 insertions(+), 3 deletions(-) diff --git a/src/drunc/process_manager/ssh_process_manager.py b/src/drunc/process_manager/ssh_process_manager.py index e0bfb7aa4..836c10a54 100644 --- a/src/drunc/process_manager/ssh_process_manager.py +++ b/src/drunc/process_manager/ssh_process_manager.py @@ -362,6 +362,9 @@ def __boot(self, boot_request: BootRequest, uuid: str) -> ProcessInstance: # Update hostname in boot request for this attempt self.boot_request[uuid].process_description.metadata.hostname = host + self.log.critical( + f"Attempting to start process {uuid} on host {host} via SSH lifetime manager" + ) # Start the process via SSH manager self.ssh_lifetime_manager.start_process( uuid=uuid, boot_request=self.boot_request[uuid] @@ -440,6 +443,10 @@ def _ps_impl(self, query: ProcessQuery) -> ProcessInstanceList: boot_request_dict=self.boot_request, order_by="random", ) + if query.session: + self.log.warning( + f"{self.name} found {len(process_uuids)} processes matching {query} for ps" + ) # Iterate through all processes matching the query for proc_uuid in process_uuids: @@ -487,13 +494,24 @@ def _ps_impl(self, query: ProcessQuery) -> ProcessInstanceList: else: pi.remote_pid = remote_pid_result.reason ret += [pi] - - return ProcessInstanceList( + ret_fmt = ProcessInstanceList( name=self.name, token=None, values=ret, flag=ResponseFlag.EXECUTED_SUCCESSFULLY, ) + if query.session: + self.log.critical( + f"{self.name} returning {len(ret)} processes from ps query {query}" + ) + self.log.critical(ret_fmt) + self.log.critical(f"TEST: {ret_fmt=}") + else: + self.log.warning( + f"{self.name} returning {len(ret)} processes from ps query {query}" + ) + self.log.warning(ret) + return ret_fmt def _boot_impl(self, boot_request: BootRequest) -> ProcessInstanceList: self.log.debug(f"{self.name} running boot command") diff --git a/src/drunc/processes/ssh_process_lifetime_manager_shell.py b/src/drunc/processes/ssh_process_lifetime_manager_shell.py index 7cda71d08..a68d466bd 100644 --- a/src/drunc/processes/ssh_process_lifetime_manager_shell.py +++ b/src/drunc/processes/ssh_process_lifetime_manager_shell.py @@ -318,7 +318,8 @@ def __init__( """ self.disable_host_key_check = disable_host_key_check self.disable_localhost_host_key_check = disable_localhost_host_key_check - self.log = logger if logger else get_logger(__name__) + # self.log = logger if logger else get_logger(__name__) + self.log = get_logger("PM_LMS_TEST", rich_handler=True) self._on_process_exit = on_process_exit # Create SSH command wrapper @@ -422,6 +423,9 @@ def start_process(self, uuid: str, boot_request: BootRequest) -> None: hostname = boot_request.process_description.metadata.hostname user = boot_request.process_description.metadata.user log_file = boot_request.process_description.process_logs_path + self.log.critical( + f"Starting process {uuid} on {hostname} as {user} with log file {log_file}" + ) # Extract environment variables from boot request env_vars = ( @@ -442,6 +446,8 @@ def start_process(self, uuid: str, boot_request: BootRequest) -> None: if cmd.endswith(";"): cmd = cmd[:-1] + self.log.critical(f"Built command for {uuid}: {cmd}: {boot_request}") + # Execute the command via SSH self._execute_bootrequest_via_ssh( uuid=uuid, @@ -1027,8 +1033,16 @@ def read_process_metadata( ) arguments.append(remote_command) + # Execute SSH command to wait for and read file (single round-trip) + self.log.critical( + f"Attempting to read metadata for {uuid} from {hostname} with timeout {timeout}s" + ) result = self.ssh(*arguments) + self.log.critical( + f"DEBUG - Raw metadata content for {uuid} from {hostname}: {result}" + ) json_content = str(result).strip() + self.log.critical("Attempt successful?|???") self.log.debug(f"Metadata content for {uuid}: {json_content!r}") @@ -1134,6 +1148,9 @@ def _execute_bootrequest_via_ssh( remote_cmd += f"cd {boot_request.process_description.process_execution_directory} ; " metadata_file = SSHProcessLifetimeManagerShell.get_metadata_file_path(uuid) + self.log.critical( + f"Metadata file for {uuid} will be written to {metadata_file} on remote host" + ) tree_id = boot_request.process_description.metadata.tree_id name = boot_request.process_description.metadata.name is_controller = any( diff --git a/src/drunc/unified_shell/shell.py b/src/drunc/unified_shell/shell.py index 576201d3c..9b958dfbd 100644 --- a/src/drunc/unified_shell/shell.py +++ b/src/drunc/unified_shell/shell.py @@ -157,10 +157,16 @@ def unified_shell( ctx.obj.log.debug("Setting up the [green]unified_shell[/green] logger") # Parse the process manager argument to determine if it's a config or an address + ctx.obj.log.critical( + f"Parsing the process manager argument: {process_manager}" + ) process_manager_url: ParseResult = urlparse(process_manager) internal_pm: bool = True if process_manager_url.scheme == "grpc": # i.e. if it's an address internal_pm = False + ctx.obj.log.critical( + f"Process manager argument parsed, internal_pm set to {internal_pm}" + ) # If using a k8s process manager, validate the session name before proceeding if get_pm_type_from_name( @@ -173,6 +179,7 @@ def unified_shell( ) sys.exit(1) + ctx.obj.log.critical("TEST") # Setup configuration related context variables ctx.obj.configuration_file = f"oksconflibs:{configuration_file}" ctx.obj.configuration_id = configuration_id @@ -261,6 +268,9 @@ def unified_shell( ) else: # Connect to an existing process manager at the provided address + ctx.obj.log.critical( + "Connecting to an existing process manager at the provided address" + ) process_manager_address = process_manager.replace( "grpc://", "" ) # remove the grpc scheme @@ -276,6 +286,7 @@ def unified_shell( ctx.obj.reset(address_pm=process_manager_address) # Run a simple command (describe) to check the connection with the process manager + ctx.obj.log.critical("Getting driver") try: ctx.obj.get_driver().describe() except Exception as e: @@ -302,6 +313,7 @@ def unified_shell( ctx.obj.pm_process.join() sys.exit(1) + ctx.obj.log.critical("Process manager described successfully") # Add the unified shell Click commands to the CLI ctx.obj.log.debug("Adding [green]unified_shell[/green] commands") From 4da5ac4dc560c9ba6e9ca0223e2ab4305fc2a62e Mon Sep 17 00:00:00 2001 From: PawelPlesniak Date: Tue, 23 Jun 2026 11:47:27 +0200 Subject: [PATCH 02/13] Add multi user support --- .../ssh_process_lifetime_manager_shell.py | 52 +++++++++++++------ src/drunc/unified_shell/commands.py | 12 ++--- src/drunc/unified_shell/shell.py | 18 +++---- 3 files changed, 50 insertions(+), 32 deletions(-) diff --git a/src/drunc/processes/ssh_process_lifetime_manager_shell.py b/src/drunc/processes/ssh_process_lifetime_manager_shell.py index a68d466bd..a419b5263 100644 --- a/src/drunc/processes/ssh_process_lifetime_manager_shell.py +++ b/src/drunc/processes/ssh_process_lifetime_manager_shell.py @@ -439,7 +439,10 @@ def start_process(self, uuid: str, boot_request: BootRequest) -> None: for exe_arg in boot_request.process_description.executable_and_arguments: cmd += exe_arg.exec for arg in exe_arg.args: - cmd += f" {arg}" + if arg.endswith("daq_app_rte.sh"): + cmd += f" {os.getenv('DBT_AREA_ROOT')}/install/daq_app_rte.sh" + else: + cmd += f" {arg}" cmd += ";" # Remove trailing semicolon if present @@ -888,33 +891,45 @@ def _build_ssh_arguments( # Determine if host key checking should be disabled based on configuration and # target host - disable_host_key_check = self.disable_host_key_check or ( - self.disable_localhost_host_key_check - and hostname in ("localhost", "127.0.0.1", "::1") - ) + # disable_host_key_check = self.disable_host_key_check or ( + # self.disable_localhost_host_key_check + # and hostname in ("localhost", "127.0.0.1", "::1") + # ) + superuser_host = getpass.getuser() + "@" + user_host.split("@")[1] + # self.log.critical(f"Building SSH arguments for {user_host} with superuser host {superuser_host}") + arguments = [superuser_host, "-o", "StrictHostKeyChecking=no"] + # self.log.critical(f"SSH arguments after adding StrictHostKeyChecking for {user_host}%s", arguments) + # self.log.critical(f"{arguments=}") + # self.log.critical(f"Test list: {test_list_print}") + # self.log.critical(f"Test list 2: %s", test_list_print) # Base SSH arguments with user@host and strict host key checking disabled # StrictHostKeyChecking=no is set to as we have an nfs backed home directory and # the known_hosts file is not shared across hosts, so we cannot rely on it for # host key verification. - arguments = [user_host, "-o", "StrictHostKeyChecking=no"] + # arguments = [user_host, "-o", "StrictHostKeyChecking=no"] + # "-F /nfs/home/{user_host.split('@')[0]}/.ssh/config", if use_tty: arguments.append("-tt") # If host key checking is disabled, also disable known hosts file usage and # reduce log level to avoid cluttering logs with warnings about host key verification - if disable_host_key_check: - arguments.extend( - [ - "-o", - "LogLevel=error", - "-o", - "GlobalKnownHostsFile=/dev/null", - "-o", - "UserKnownHostsFile=/dev/null", - ] - ) + # if disable_host_key_check: + arguments.extend( + [ + "-o", + "LogLevel=info", + "-o", + "GlobalKnownHostsFile=/dev/null", + "-o", + "UserKnownHostsFile=/dev/null", + ] + ) + self.log.critical(f"SSH arguments for {user_host}: {arguments}") + self.log.critical( + f"PP: {getpass.getuser()} is running on {os.uname().nodename} with disable_host_key_check={self.disable_host_key_check} and disable_localhost_host_key_check={self.disable_localhost_host_key_check}" + ) return arguments @@ -1133,6 +1148,9 @@ def _execute_bootrequest_via_ssh( try: platform = os.uname().sysname.lower() is_macos = "darwin" in platform + hostname_for_gssapi = hostname + if hostname_for_gssapi == "localhost": + hostname_for_gssapi = os.uname().nodename user_host = f"{user}@{hostname}" # Build remote command with metadata file writing diff --git a/src/drunc/unified_shell/commands.py b/src/drunc/unified_shell/commands.py index f1c730126..45a617d91 100644 --- a/src/drunc/unified_shell/commands.py +++ b/src/drunc/unified_shell/commands.py @@ -45,12 +45,12 @@ def boot( override_logs_boot = obj.override_logs else: override_logs_boot = override_logs - if len(processes.values) > 0: - log.error( - f"Cannot boot: session {session_name} already has {len(processes.values)} processes running. " - "Please terminate the existing session first." - ) - return + # if len(processes.values) > 0: + # log.error( + # f"Cannot boot: session {session_name} already has {len(processes.values)} processes running. " + # "Please terminate the existing session first." + # ) + # return try: results = obj.get_driver("process_manager").boot( diff --git a/src/drunc/unified_shell/shell.py b/src/drunc/unified_shell/shell.py index 9b958dfbd..7109dbea7 100644 --- a/src/drunc/unified_shell/shell.py +++ b/src/drunc/unified_shell/shell.py @@ -169,15 +169,15 @@ def unified_shell( ) # If using a k8s process manager, validate the session name before proceeding - if get_pm_type_from_name( - process_manager - ) == ProcessManagerTypes.K8s and not validate_k8s_session_name(session_name): - ctx.obj.log.error( - f"[red]Invalid session/namespace name [bold]({session_name})[/bold][/red]. " - "Must match RFC1123 label: lowercase alphanumeric or '-', start/end with " - "alphanumeric, max 63 chars." - ) - sys.exit(1) + # if get_pm_type_from_name( + # process_manager + # ) == ProcessManagerTypes.K8s and not validate_k8s_session_name(session_name): + # ctx.obj.log.error( + # f"[red]Invalid session/namespace name [bold]({session_name})[/bold][/red]. " + # "Must match RFC1123 label: lowercase alphanumeric or '-', start/end with " + # "alphanumeric, max 63 chars." + # ) + # sys.exit(1) ctx.obj.log.critical("TEST") # Setup configuration related context variables From 93d4d770d281825046ed43c9ec972f27679a2c07 Mon Sep 17 00:00:00 2001 From: PawelPlesniak Date: Tue, 23 Jun 2026 11:48:08 +0200 Subject: [PATCH 03/13] Update terminate and ps commands --- src/drunc/unified_shell/commands.py | 38 +++++++++++++++++++++++++++++ src/drunc/unified_shell/shell.py | 21 ++++++++-------- 2 files changed, 49 insertions(+), 10 deletions(-) diff --git a/src/drunc/unified_shell/commands.py b/src/drunc/unified_shell/commands.py index 45a617d91..1bf87fb91 100644 --- a/src/drunc/unified_shell/commands.py +++ b/src/drunc/unified_shell/commands.py @@ -7,6 +7,7 @@ from drunc.controller.interface.shell_utils import controller_setup from drunc.exceptions import DruncSetupException from drunc.process_manager.interface.context import ProcessManagerContext +from drunc.process_manager.utils import tabulate_process_instance_list from drunc.unified_shell.context import UnifiedShellMode from drunc.utils.shell_utils import InterruptedCommand from drunc.utils.utils import get_logger @@ -124,6 +125,43 @@ def boot( sys.exit(1) +@click.command("terminate") +@click.pass_obj +@click.pass_context +def terminate(ctx, obj): + """ + Execute the process manager terminate command, but only do this for the current + session + """ + + log = get_logger("unified_shell.terminate") + session_query = ProcessQuery(session=ctx.obj.session_name) + log.info(f"Terminating session [green]{ctx.obj.session_name}[/]") + obj.get_driver("process_manager").kill(session_query) + + +@click.command("ps") +@click.pass_obj +@click.pass_context +def ps(ctx, obj): + """ + Execute the process manager terminate command, but only do this for the current + session + """ + + log = get_logger("unified_shell.ps") + session_query = ProcessQuery(session=ctx.obj.session_name) + log.info(f"Listing session [green]{ctx.obj.session_name}[/]") + results = obj.get_driver("process_manager").ps(session_query) + obj.print( + tabulate_process_instance_list( + results, title=f"Processes running in session {ctx.obj.session_name}" + ), + overflow="fold", + soft_wrap=True, + ) + + @click.command("start-shell") @click.pass_obj @click.pass_context diff --git a/src/drunc/unified_shell/shell.py b/src/drunc/unified_shell/shell.py index 7109dbea7..eeda0e748 100644 --- a/src/drunc/unified_shell/shell.py +++ b/src/drunc/unified_shell/shell.py @@ -44,7 +44,6 @@ from drunc.fsm.configuration import FSMConfHandler from drunc.fsm.utils import convert_fsm_transition from drunc.process_manager.configuration import ( - ProcessManagerTypes, get_process_manager_configuration, validate_pm_config, ) @@ -52,13 +51,12 @@ flush, kill, logs, - ps, + # ps, restart, - terminate, + # terminate, ) from drunc.process_manager.interface.process_manager import run_pm -from drunc.process_manager.utils import get_pm_type_from_name, validate_k8s_session_name -from drunc.unified_shell.commands import boot, start_shell +from drunc.unified_shell.commands import boot, ps, start_shell, terminate from drunc.unified_shell.context import UnifiedShellMode from drunc.unified_shell.shell_utils import generate_fsm_sequence_command from drunc.utils.configuration import ConfTypes, OKSKey @@ -317,18 +315,20 @@ def unified_shell( # Add the unified shell Click commands to the CLI ctx.obj.log.debug("Adding [green]unified_shell[/green] commands") - ctx.command.add_command(boot, "boot") - ctx.obj.dynamic_commands.add("boot") + unified_shell_commands = [boot, ps, terminate] + for cmd in unified_shell_commands: + ctx.command.add_command(cmd, format_name_for_cli(cmd.name)) + ctx.obj.dynamic_commands.add(format_name_for_cli(cmd.name)) # Add the process manager Click commands to the CLI ctx.obj.log.debug("Adding [green]process_manager[/green] commands") process_manager_commands: list[click.Command] = [ kill, - terminate, + # terminate, flush, logs, restart, - ps, + # ps, ] for cmd in process_manager_commands: ctx.command.add_command(cmd, format_name_for_cli(cmd.name)) @@ -469,7 +469,8 @@ def cleanup(): # Terminate any residual processes if ctx.obj.get_driver("process_manager"): - ctx.obj.get_driver("process_manager").terminate() + session_processes = ProcessQuery(session=ctx.obj.session_name) + ctx.obj.get_driver("process_manager").kill(session_processes) # Check if any processes are still running if ( From fed5b18184b9a02835ebf52aab641641c127d148 Mon Sep 17 00:00:00 2001 From: Emir Muhammad Date: Tue, 5 May 2026 11:42:51 +0200 Subject: [PATCH 04/13] Add split shell support --- src/drunc/controller/interface/context.py | 2 + .../process_manager/interface/cli_argument.py | 27 ++-- .../process_manager/interface/commands.py | 126 +++++++++++++----- .../process_manager/interface/context.py | 4 +- src/drunc/process_manager/interface/shell.py | 9 ++ .../process_manager/k8s_process_manager.py | 15 +++ src/drunc/process_manager/oks_parser.py | 9 +- src/drunc/process_manager/process_manager.py | 84 ++++++++++-- .../process_manager/process_manager_driver.py | 32 +++++ .../process_manager/ssh_process_manager.py | 52 +++++--- src/drunc/process_manager/utils.py | 3 + .../ssh_process_lifetime_manager_shell.py | 36 ++--- src/drunc/unified_shell/commands.py | 62 ++++++++- src/drunc/unified_shell/context.py | 5 +- src/drunc/unified_shell/shell.py | 52 ++++++-- src/drunc/utils/shell_utils.py | 34 +++++ src/drunc/utils/utils.py | 81 +++++++++++ .../interface/test_commands.py | 7 + .../process_manager_mock_impls.py | 10 ++ 19 files changed, 540 insertions(+), 110 deletions(-) diff --git a/src/drunc/controller/interface/context.py b/src/drunc/controller/interface/context.py index 713bcc7e6..07f367955 100644 --- a/src/drunc/controller/interface/context.py +++ b/src/drunc/controller/interface/context.py @@ -11,6 +11,8 @@ class ControllerContext(ShellContext): # boilerplatefest + shell_id = "controller_shell" + def __init__(self): self.status_receiver = None self.took_control = False diff --git a/src/drunc/process_manager/interface/cli_argument.py b/src/drunc/process_manager/interface/cli_argument.py index 27e84872f..abefd002c 100644 --- a/src/drunc/process_manager/interface/cli_argument.py +++ b/src/drunc/process_manager/interface/cli_argument.py @@ -7,15 +7,10 @@ def validate_conf_string(ctx, param, boot_configuration): return boot_configuration -def add_query_options(at_least_one: bool, all_processes_by_default: bool = False): - def wrapper(f0): - f1 = click.option( - "-s", - "--session", - type=str, - default=None, - help="Select the processes on a particular session", - )(f0) +def add_query_options_no_session( + at_least_one: bool, all_processes_by_default: bool = False +): + def wrapper(f1): f2 = click.option( "-n", "--name", @@ -41,3 +36,17 @@ def wrapper(f0): return generate_process_query(f4, at_least_one, all_processes_by_default) return wrapper + + +def add_query_options(at_least_one: bool, all_processes_by_default: bool = False): + def wrapper(f0): + f1 = click.option( + "-s", + "--session", + type=str, + default=None, + help="Select the processes on a particular session", + )(f0) + return add_query_options_no_session(at_least_one, all_processes_by_default)(f1) + + return wrapper diff --git a/src/drunc/process_manager/interface/commands.py b/src/drunc/process_manager/interface/commands.py index 129f3ea78..cbfde6250 100644 --- a/src/drunc/process_manager/interface/commands.py +++ b/src/drunc/process_manager/interface/commands.py @@ -1,4 +1,5 @@ import getpass +from time import sleep import click from druncschema.process_manager_pb2 import LogRequest, ProcessQuery @@ -11,7 +12,7 @@ ) from drunc.process_manager.interface.context import ProcessManagerContext from drunc.process_manager.utils import tabulate_process_instance_list -from drunc.utils.shell_utils import InterruptedCommand +from drunc.utils.shell_utils import InterruptedCommand, log_pm_cmd from drunc.utils.utils import get_logger @@ -43,6 +44,7 @@ def boot( override_logs: bool, ) -> None: log = get_logger("process_manager.shell") + log_pm_cmd(obj) processes = obj.get_driver("process_manager").ps(ProcessQuery(user=user)) if len(processes.values) > 0: @@ -129,6 +131,7 @@ def dummy_boot( session_name: str, ) -> None: log = get_logger("process_manager.shell") + log_pm_cmd(obj) log.debug( f"Running dummy_boot with {n_processes} processes for {sleep} seconds {n_sleeps} times, requested by user {user}" ) @@ -150,6 +153,16 @@ def dummy_boot( return +@click.command("wait") +@click.argument("sleep_time", type=int, default=1) +@click.pass_obj +def wait(obj: ProcessManagerContext, sleep_time: int) -> None: + log = get_logger("process_manager.wait") + log.info(f"Command [green]wait[/green] running for {sleep_time} seconds.") + sleep(sleep_time) # seconds + log.info(f"Command [green]wait[/green] ran for {sleep_time} seconds.") + + @click.command("terminate") @click.option( "-w", @@ -161,6 +174,7 @@ def dummy_boot( @click.pass_obj def terminate(obj: ProcessManagerContext, width: int | None) -> None: log = get_logger("process_manager.shell") + log_pm_cmd(obj) log.debug("Terminating") result = obj.get_driver("process_manager").terminate() if not result: @@ -171,23 +185,35 @@ def terminate(obj: ProcessManagerContext, width: int | None) -> None: obj.delete_driver("controller") +def kill_decorators(f): + f = click.pass_obj(f) + f = click.option( + "-w", + "--width", + type=int, + default=None, + help="Table width. Default is automatically calculated", + )(f) + f = click.option( + "--crash", + is_flag=True, + default=False, + help="Simulate a crash: send SIGKILL without any cleanup, leaving the process manager in an unexpected-death state.", + )(f) + return f + + @click.command("kill") -@click.option( - "-w", - "--width", - type=int, - default=None, - help="Table width. Default is automatically calculated", -) @add_query_options(at_least_one=True) -@click.option( - "--crash", - is_flag=True, - default=False, - help="Simulate a crash: send SIGKILL without any cleanup, leaving the process manager in an unexpected-death state.", -) -@click.pass_obj -def kill(obj: ProcessManagerContext, query: ProcessQuery, width: int | None) -> None: +@kill_decorators +def kill(obj, query, width): + log_pm_cmd(obj) + return kill_impl(obj, query, width) + + +def kill_impl( + obj: ProcessManagerContext, query: ProcessQuery, width: int | None +) -> None: log = get_logger("process_manager.shell") log.debug(f"Killing with query {query}") result = obj.get_driver("process_manager").kill(query) @@ -198,17 +224,27 @@ def kill(obj: ProcessManagerContext, query: ProcessQuery, width: int | None) -> ) # rich tables require console printing +def flush_decorators(f): + f = click.pass_obj(f) + f = click.option( + "-w", + "--width", + type=int, + default=None, + help="Table width. Default is automatically calculated", + )(f) + return f + + @click.command("flush") -@click.option( - "-w", - "--width", - type=int, - default=None, - help="Table width. Default is automatically calculated", -) @add_query_options(at_least_one=False, all_processes_by_default=True) -@click.pass_obj -def flush( +@flush_decorators +def flush(obj, query, width): + log_pm_cmd(obj) + return flush_impl(obj, query, width) + + +def flush_impl( obj: ProcessManagerContext, query: ProcessQuery, width: int | None, @@ -223,22 +259,34 @@ def flush( ) # rich tables require console printing +def logs_decorators(f): + f = click.pass_obj(f) + f = click.option("--grep", type=str, default=None)(f) + f = click.option( + "--how-far", + type=int, + show_default=True, + default=100, + help="How many lines one wants", + )(f) + return f + + @click.command("logs") @add_query_options(at_least_one=True) -@click.option( - "--how-far", - type=int, - show_default=True, - default=100, - help="How many lines one wants", -) -@click.option("--grep", type=str, default=None) -@click.pass_obj -def logs( +@logs_decorators +def logs(obj, how_far, grep, query): + log_pm_cmd(obj) + return logs_impl(obj, how_far, grep, query) + + +def logs_impl( obj: ProcessManagerContext, how_far: int, grep: str, query: ProcessQuery ) -> None: log = get_logger("process_manager.shell") - log.debug(f"Running logs with query {query}") + # TODO: MOVE BACK TO DEBUG BEFORE MERGE + # THIS IS USEFUL FOR TESTING THOUGH + log.error(f"Running logs with query {query}") log_req = LogRequest( how_far=how_far, query=query, @@ -276,6 +324,11 @@ def logs( @add_query_options(at_least_one=True) @click.pass_obj def restart(obj: ProcessManagerContext, query: ProcessQuery) -> None: + log_pm_cmd(obj) + return restart_impl(obj, query) + + +def restart_impl(obj: ProcessManagerContext, query: ProcessQuery) -> None: log = get_logger("process_manager.shell") log.debug(f"Restarting with query {query}") obj.get_driver("process_manager").restart(query) @@ -306,6 +359,7 @@ def ps( width: int | None, ) -> None: log = get_logger("process_manager.shell") + log_pm_cmd(obj) log.debug(f"Running ps with query {query}") results = obj.get_driver("process_manager").ps(query) if not results: diff --git a/src/drunc/process_manager/interface/context.py b/src/drunc/process_manager/interface/context.py index d6a30f592..24c0ff9df 100644 --- a/src/drunc/process_manager/interface/context.py +++ b/src/drunc/process_manager/interface/context.py @@ -10,7 +10,9 @@ from drunc.utils.utils import resolve_localhost_to_hostname -class ProcessManagerContext(ShellContext): +class ProcessManagerContext(ShellContext): # boilerplatefest + shell_id = "process_manager_shell" + def __init__(self, *args, **kwargs): self.status_receiver = None super(ProcessManagerContext, self).__init__(*args, **kwargs) diff --git a/src/drunc/process_manager/interface/shell.py b/src/drunc/process_manager/interface/shell.py index 18f7962e2..e1341df60 100644 --- a/src/drunc/process_manager/interface/shell.py +++ b/src/drunc/process_manager/interface/shell.py @@ -14,6 +14,7 @@ ps, restart, terminate, + wait, ) from drunc.utils.grpc_utils import ServerUnreachable from drunc.utils.utils import ( @@ -59,6 +60,10 @@ def process_manager_shell(ctx, process_manager_address: str, log_level: str) -> # process_manager_shell_log.error(e.message) # TODO: Keep this for production branch, remove this from dev branch exit(1) + ctx.obj.get_driver("process_manager").send_msg( + f"{getpass.getuser()} connected from {ctx.obj.shell_id}" + ) + # Manually add file handler to process manager log # Not possible to initialise logger immediately as it requires # knowledge of the log path @@ -73,6 +78,9 @@ def process_manager_shell(ctx, process_manager_address: str, log_level: str) -> ) def cleanup(): + ctx.obj.get_driver("process_manager").send_msg( + f"{getpass.getuser()} disconnected from {ctx.obj.shell_id}" + ) ctx.obj.terminate() process_manager_log.warning( f"[green]{getpass.getuser()}[/green] disconnected from the process manager through a [green]drunc-process-manager-shell[/green]" @@ -81,6 +89,7 @@ def cleanup(): ctx.call_on_close(cleanup) ctx.command.add_command(boot, "boot") + ctx.command.add_command(wait, "wait") ctx.command.add_command(terminate, "terminate") ctx.command.add_command(kill, "kill") ctx.command.add_command(flush, "flush") diff --git a/src/drunc/process_manager/k8s_process_manager.py b/src/drunc/process_manager/k8s_process_manager.py index e043ed13d..f2fc064ab 100644 --- a/src/drunc/process_manager/k8s_process_manager.py +++ b/src/drunc/process_manager/k8s_process_manager.py @@ -13,6 +13,7 @@ from time import sleep, time # Local Application Imports +from druncschema.generic_pb2 import OutcomeFlag, OutcomeStatus from druncschema.process_manager_pb2 import ( BootRequest, LogLines, @@ -39,6 +40,7 @@ from drunc.process_manager.configuration import ( PROCESS_SHUTDOWN_ORDERING, ProcessManagerConfHandler, + ProcessManagerTypes, ) from drunc.process_manager.process_manager import ProcessManager from drunc.process_manager.utils import ( @@ -172,6 +174,8 @@ def run(self) -> None: class K8sProcessManager(ProcessManager): + pm_type = ProcessManagerTypes.K8s + def __init__(self, configuration: ProcessManagerConfHandler, **kwargs) -> None: """ Manages processes as Kubernetes Pods. @@ -1962,6 +1966,17 @@ def _logs_impl(self, log_request: LogRequest) -> LogLines: lines=[f"Could not retrieve logs: {e.reason}"], ) + def _send_msg_impl(self, msg: str, peer: str) -> OutcomeStatus: + # Note: currently exact same implementation as ssh manager + # Although there is room here to change as necessary + try: + self.log.info(f"{msg}; from {peer}") + except Exception as e: + self.log.critical(f"Failed to receive message with exception {e}") + return OutcomeStatus(flag=OutcomeFlag.FAIL) + + return OutcomeStatus(flag=OutcomeFlag.SUCCESS) + def _boot_impl(self, boot_request: BootRequest) -> ProcessInstanceList: """ Handles the 'boot' command from the gRPC interface. diff --git a/src/drunc/process_manager/oks_parser.py b/src/drunc/process_manager/oks_parser.py index f679e7f1a..5feb6ba50 100644 --- a/src/drunc/process_manager/oks_parser.py +++ b/src/drunc/process_manager/oks_parser.py @@ -5,7 +5,7 @@ from drunc.exceptions import DruncException, DruncSetupException from drunc.process_manager.configuration import get_commandline_parameters -from drunc.utils.utils import get_logger +from drunc.utils.utils import file_is_read_only, get_logger if TYPE_CHECKING: import conffwk @@ -75,8 +75,13 @@ def get_full_db_path(db_path: str) -> str: err_str = f"No files found in DUNEDAQ_DB_PATH matching {db_path}." raise DruncSetupException(err_str) - # If multiple matches are found, take the first instance that matches. + # Prefer the first writable match; if every match is read-only, fall back to the first one. resolved_path = unique_matched_files[0] + for matched_file in unique_matched_files: + if not file_is_read_only(matched_file): + resolved_path = matched_file + break + log.debug(f"Path {db_path} resolved to {resolved_path}") return resolved_path diff --git a/src/drunc/process_manager/process_manager.py b/src/drunc/process_manager/process_manager.py index b4e979608..060625c08 100644 --- a/src/drunc/process_manager/process_manager.py +++ b/src/drunc/process_manager/process_manager.py @@ -7,9 +7,11 @@ from daqpytools.logging import LogHandlerConf, exceptions, setup_daq_ers_logger from druncschema.authoriser_pb2 import ActionType, SystemType from druncschema.description_pb2 import CommandDescription, Description +from druncschema.generic_pb2 import OutcomeStatus from druncschema.opmon.process_manager_pb2 import ProcessStatus from druncschema.process_manager_pb2 import ( BootRequest, + GenericNotificationMessage, LogLines, LogRequest, ProcessInstance, @@ -36,7 +38,7 @@ ProcessManagerTypes, ) from drunc.utils.configuration import ConfTypes -from drunc.utils.utils import get_logger, pid_info_str +from drunc.utils.utils import get_logger, pid_info_str, resolve_context_peer class BadQuery(DruncCommandException): @@ -45,6 +47,8 @@ def __init__(self, txt): class ProcessManager(abc.ABC, ProcessManagerServicer): + pm_type = ProcessManagerTypes.Unknown # Used for describe (and possibly others) + def __init__( self, configuration: ProcessManagerConfHandler, name: str, session: str ): @@ -424,7 +428,7 @@ def describe(self, request: Request, context: ServicerContext) -> Description: self.log.debug(f"{self.name} running describe") response = Description( - type="process_manager", + type=self.pm_type.name, name=self.name, info=self.get_log_path(), session="no_session" if not self.session else self.session, @@ -480,6 +484,60 @@ def logs(self, request: LogRequest, context: ServicerContext) -> LogLines: return response + @abc.abstractmethod + def _send_msg_impl( + self, msg: str | None = None, peer: str | None = None + ) -> OutcomeStatus: + raise NotImplementedError + + @broadcasted + @authentified_and_authorised( + action=ActionType.READ, system=SystemType.PROCESS_MANAGER + ) + def send_msg(self, request: Request, context: ServicerContext) -> OutcomeStatus: + self.log.debug(f"{self.name} running send_msg") + + try: + peer = context.peer() + peer_display = resolve_context_peer(peer) + except Exception: + self.log.warning("Could not determine caller peer", exc_info=True) + peer_display = "unknown" + + # Try to extract an optional GenericNotificationMessage from request.data + try: + if ( + request is not None + and hasattr(request, "data") + and request.data is not None + ): + gm = GenericNotificationMessage() + request.data.Unpack(gm) + msg_value = gm.message + except Exception as e: + self.log.debug( + f"Error while extracting send_msg payload: {e}", exc_info=True + ) + msg_value = "unknown payload" + + try: + response = self._send_msg_impl(msg_value, peer_display) + except NotImplementedError: + raise DruncNotImplementedException( + message="Implementation missing", + domain="ProcessManager.send_msg", + ) + except Exception as e: + context_msg = f"Unhandled exception in ProcessManager.send_msg: {e}" + self.log.exception(context_msg) + + raise DruncCommandException( + message=context_msg, + domain="ProcessManager.send_msg", + ) + + return response + def _ensure_one_process( self, uuids: list[str], in_boot_request: bool = False ) -> str: @@ -540,25 +598,27 @@ def _match_processes_against_query( # Filter processes based on query criteria processes = [] for uuid in available_uuids: - accepted = False + accepted = True meta = boot_request_dict[uuid].process_description.metadata # Check UUID match - if uuid in uuid_selector: - accepted = True + if uuid_selector and uuid not in uuid_selector: + accepted = False # Check name pattern match (regex) - for name_reg in name_selector: - if re.search(name_reg, meta.name): - accepted = True + + if name_selector and not any( + re.search(reg, meta.name) for reg in name_selector + ): + accepted = False # Check session match - if session_selector == meta.session: - accepted = True + if session_selector and session_selector != meta.session: + accepted = False # Check user match - if user_selector == meta.user: - accepted = True + if user_selector and user_selector != meta.user: + accepted = False if accepted: processes.append(uuid) diff --git a/src/drunc/process_manager/process_manager_driver.py b/src/drunc/process_manager/process_manager_driver.py index 1f9e50aeb..ea239f4d6 100644 --- a/src/drunc/process_manager/process_manager_driver.py +++ b/src/drunc/process_manager/process_manager_driver.py @@ -16,6 +16,7 @@ from druncschema.description_pb2 import Description from druncschema.process_manager_pb2 import ( BootRequest, + GenericNotificationMessage, LogLines, LogRequest, ProcessDescription, @@ -83,7 +84,35 @@ def close(self) -> None: except Exception as e: self.log.error(f"Error closing gRPC channel: {e}", exc_info=True) + def send_msg(self, msg): + request = Request(token=copy_token(self.token)) + + if msg is not None: + try: + gm = GenericNotificationMessage(message=str(msg)) + request.data.Pack(gm) + except Exception: + self.log.critical("Failed to pack send_msg payload", exc_info=True) + + timeout = 10 + + try: + response = self.stub.send_msg(request, timeout=timeout) + except grpc.RpcError as e: + try: + error_details = extract_grpc_rich_error(e) + self.log.error(error_details) + except Exception as extraction_error: + self.log.critical( + f"Could not extract rich error details from gRPC error: {extraction_error}", + exc_info=True, + ) + handle_grpc_error(e) + + return response + # ----- Boot workflow ----- + def boot( self, conf_file: str, @@ -100,6 +129,9 @@ def boot( ) -> Iterator[ProcessInstanceList] | None: self.log.info(f"Booting session [green]{session_name}[/green]") + # Assume oksconflibs if no framework is defined + conf_file = f"oksconflibs:{conf_file}" if ":" not in conf_file else conf_file + # Step 1 - consolidate configuration self._consolidate_config(session_name, conf_file) diff --git a/src/drunc/process_manager/ssh_process_manager.py b/src/drunc/process_manager/ssh_process_manager.py index 836c10a54..7f40d53d7 100644 --- a/src/drunc/process_manager/ssh_process_manager.py +++ b/src/drunc/process_manager/ssh_process_manager.py @@ -3,6 +3,8 @@ import uuid from typing import List, Optional + +from druncschema.generic_pb2 import OutcomeFlag, OutcomeStatus from druncschema.process_manager_pb2 import ( BootRequest, LogLines, @@ -17,12 +19,15 @@ from druncschema.request_response_pb2 import ResponseFlag from drunc.exceptions import DruncCommandException +from drunc.process_manager.configuration import ProcessManagerTypes from drunc.process_manager.process_manager import ProcessManager from drunc.processes.exit_status import ExitStatus from drunc.processes.ssh_process_lifetime_manager import ProcessLifetimeManager class SSHProcessManager(ProcessManager): + pm_type = ProcessManagerTypes.SSH_SHELL + def __init__( self, configuration, LifetimeManagerClass: ProcessLifetimeManager, **kwargs ): @@ -362,9 +367,9 @@ def __boot(self, boot_request: BootRequest, uuid: str) -> ProcessInstance: # Update hostname in boot request for this attempt self.boot_request[uuid].process_description.metadata.hostname = host - self.log.critical( - f"Attempting to start process {uuid} on host {host} via SSH lifetime manager" - ) + # self.log.critical( + # f"Attempting to start process {uuid} on host {host} via SSH lifetime manager" + # ) # Start the process via SSH manager self.ssh_lifetime_manager.start_process( uuid=uuid, boot_request=self.boot_request[uuid] @@ -443,10 +448,10 @@ def _ps_impl(self, query: ProcessQuery) -> ProcessInstanceList: boot_request_dict=self.boot_request, order_by="random", ) - if query.session: - self.log.warning( - f"{self.name} found {len(process_uuids)} processes matching {query} for ps" - ) + # if query.session: + # self.log.warning( + # f"{self.name} found {len(process_uuids)} processes matching {query} for ps" + # ) # Iterate through all processes matching the query for proc_uuid in process_uuids: @@ -500,19 +505,30 @@ def _ps_impl(self, query: ProcessQuery) -> ProcessInstanceList: values=ret, flag=ResponseFlag.EXECUTED_SUCCESSFULLY, ) - if query.session: - self.log.critical( - f"{self.name} returning {len(ret)} processes from ps query {query}" - ) - self.log.critical(ret_fmt) - self.log.critical(f"TEST: {ret_fmt=}") - else: - self.log.warning( - f"{self.name} returning {len(ret)} processes from ps query {query}" - ) - self.log.warning(ret) + # if query.session: + # self.log.critical( + # f"{self.name} returning {len(ret)} processes from ps query {query}" + # ) + # self.log.critical(ret_fmt) + # self.log.critical(f"TEST: {ret_fmt=}") + # else: + # self.log.warning( + # f"{self.name} returning {len(ret)} processes from ps query {query}" + # ) + # self.log.warning(ret) return ret_fmt + def _send_msg_impl(self, msg: str, peer: str) -> OutcomeStatus: + try: + # TODO: THIS IS CURRENTLY CRITICAL FOR EASIER TESTING + # DO _NOT_ MERGE UNTIL THIS IS BACK TO INFO! + self.log.critical(f"{msg}; from {peer}") + except Exception as e: + self.log.critical(f"Failed to receive message with exception {e}") + return OutcomeStatus(flag=OutcomeFlag.FAIL) + + return OutcomeStatus(flag=OutcomeFlag.SUCCESS) + def _boot_impl(self, boot_request: BootRequest) -> ProcessInstanceList: self.log.debug(f"{self.name} running boot command") this_uuid = str(uuid.uuid4()) diff --git a/src/drunc/process_manager/utils.py b/src/drunc/process_manager/utils.py index b7d881d7e..8d4b71826 100644 --- a/src/drunc/process_manager/utils.py +++ b/src/drunc/process_manager/utils.py @@ -334,6 +334,9 @@ def validate_k8s_session_name(session: str) -> bool: return True +#! Note for future developers +# This can probably be removed since we've added the +# pm_type attribute in each of the process managers def get_pm_type_from_name(pm_name: str) -> ProcessManagerTypes: """ Get the ProcessManagerTypes enum value from a string name. diff --git a/src/drunc/processes/ssh_process_lifetime_manager_shell.py b/src/drunc/processes/ssh_process_lifetime_manager_shell.py index a419b5263..0a9a3edbc 100644 --- a/src/drunc/processes/ssh_process_lifetime_manager_shell.py +++ b/src/drunc/processes/ssh_process_lifetime_manager_shell.py @@ -423,9 +423,9 @@ def start_process(self, uuid: str, boot_request: BootRequest) -> None: hostname = boot_request.process_description.metadata.hostname user = boot_request.process_description.metadata.user log_file = boot_request.process_description.process_logs_path - self.log.critical( - f"Starting process {uuid} on {hostname} as {user} with log file {log_file}" - ) + # self.log.critical( + # f"Starting process {uuid} on {hostname} as {user} with log file {log_file}" + # ) # Extract environment variables from boot request env_vars = ( @@ -449,7 +449,7 @@ def start_process(self, uuid: str, boot_request: BootRequest) -> None: if cmd.endswith(";"): cmd = cmd[:-1] - self.log.critical(f"Built command for {uuid}: {cmd}: {boot_request}") + # self.log.critical(f"Built command for {uuid}: {cmd}: {boot_request}") # Execute the command via SSH self._execute_bootrequest_via_ssh( @@ -926,10 +926,10 @@ def _build_ssh_arguments( "UserKnownHostsFile=/dev/null", ] ) - self.log.critical(f"SSH arguments for {user_host}: {arguments}") - self.log.critical( - f"PP: {getpass.getuser()} is running on {os.uname().nodename} with disable_host_key_check={self.disable_host_key_check} and disable_localhost_host_key_check={self.disable_localhost_host_key_check}" - ) + # self.log.critical(f"SSH arguments for {user_host}: {arguments}") + # self.log.critical( + # f"PP: {getpass.getuser()} is running on {os.uname().nodename} with disable_host_key_check={self.disable_host_key_check} and disable_localhost_host_key_check={self.disable_localhost_host_key_check}" + # ) return arguments @@ -1049,15 +1049,15 @@ def read_process_metadata( arguments.append(remote_command) # Execute SSH command to wait for and read file (single round-trip) - self.log.critical( - f"Attempting to read metadata for {uuid} from {hostname} with timeout {timeout}s" - ) + # self.log.critical( + # f"Attempting to read metadata for {uuid} from {hostname} with timeout {timeout}s" + # ) result = self.ssh(*arguments) - self.log.critical( - f"DEBUG - Raw metadata content for {uuid} from {hostname}: {result}" - ) + # self.log.critical( + # f"DEBUG - Raw metadata content for {uuid} from {hostname}: {result}" + # ) json_content = str(result).strip() - self.log.critical("Attempt successful?|???") + # self.log.critical("Attempt successful?|???") self.log.debug(f"Metadata content for {uuid}: {json_content!r}") @@ -1166,9 +1166,9 @@ def _execute_bootrequest_via_ssh( remote_cmd += f"cd {boot_request.process_description.process_execution_directory} ; " metadata_file = SSHProcessLifetimeManagerShell.get_metadata_file_path(uuid) - self.log.critical( - f"Metadata file for {uuid} will be written to {metadata_file} on remote host" - ) + # self.log.critical( + # f"Metadata file for {uuid} will be written to {metadata_file} on remote host" + # ) tree_id = boot_request.process_description.metadata.tree_id name = boot_request.process_description.metadata.name is_controller = any( diff --git a/src/drunc/unified_shell/commands.py b/src/drunc/unified_shell/commands.py index 1bf87fb91..be093c134 100644 --- a/src/drunc/unified_shell/commands.py +++ b/src/drunc/unified_shell/commands.py @@ -1,15 +1,26 @@ import getpass import sys +from functools import update_wrapper import click from druncschema.process_manager_pb2 import ProcessInstance, ProcessQuery from drunc.controller.interface.shell_utils import controller_setup from drunc.exceptions import DruncSetupException +from drunc.process_manager.interface.cli_argument import add_query_options_no_session +from drunc.process_manager.interface.commands import ( + flush_decorators, + flush_impl, + kill_decorators, + kill_impl, + logs_decorators, + logs_impl, + restart_impl, +) from drunc.process_manager.interface.context import ProcessManagerContext from drunc.process_manager.utils import tabulate_process_instance_list from drunc.unified_shell.context import UnifiedShellMode -from drunc.utils.shell_utils import InterruptedCommand +from drunc.utils.shell_utils import InterruptedCommand, log_pm_cmd from drunc.utils.utils import get_logger @@ -33,6 +44,7 @@ def boot( sleep_between_app_boot: int | float = 0, ) -> None: log = get_logger("unified_shell.boot") + log_pm_cmd(obj) session_name = obj.session_name user = getpass.getuser() processes = obj.get_driver("process_manager").ps( @@ -135,6 +147,7 @@ def terminate(ctx, obj): """ log = get_logger("unified_shell.terminate") + log_pm_cmd(obj) session_query = ProcessQuery(session=ctx.obj.session_name) log.info(f"Terminating session [green]{ctx.obj.session_name}[/]") obj.get_driver("process_manager").kill(session_query) @@ -150,6 +163,7 @@ def ps(ctx, obj): """ log = get_logger("unified_shell.ps") + log_pm_cmd(obj) session_query = ProcessQuery(session=ctx.obj.session_name) log.info(f"Listing session [green]{ctx.obj.session_name}[/]") results = obj.get_driver("process_manager").ps(session_query) @@ -162,6 +176,51 @@ def ps(ctx, obj): ) +def session_injector(f): + @click.pass_context + def wrapper(ctx, *args, **kwargs): + kwargs["session"] = ctx.obj.session_name + return ctx.invoke(f, *args, **kwargs) + + return update_wrapper(wrapper, f) + + +@click.command("logs") +@session_injector +@add_query_options_no_session(at_least_one=True) +@logs_decorators +def logs(obj, how_far, grep, query): + log_pm_cmd(obj) + return logs_impl(obj, how_far, grep, query) + + +@click.command("kill") +@session_injector +@add_query_options_no_session(at_least_one=True) +@kill_decorators +def kill(obj, query, width): + log_pm_cmd(obj) + return kill_impl(obj, query, width) + + +@click.command("flush") +@session_injector +@add_query_options_no_session(at_least_one=True) +@flush_decorators +def flush(obj, query, width): + log_pm_cmd(obj) + return flush_impl(obj, query, width) + + +@click.command("restart") +@session_injector +@add_query_options_no_session(at_least_one=True) +@click.pass_obj +def restart(obj, query): + log_pm_cmd(obj) + return restart_impl(obj, query) + + @click.command("start-shell") @click.pass_obj @click.pass_context @@ -173,6 +232,7 @@ def start_shell(ctx, obj): allowing you to execute commands interactively. """ log = get_logger("unified_shell.start_shell") + log_pm_cmd(obj) obj.running_mode = UnifiedShellMode.SEMIBATCH log.info("Switching to interactive mode...") diff --git a/src/drunc/unified_shell/context.py b/src/drunc/unified_shell/context.py index af1c37df4..5fbf4a2a0 100644 --- a/src/drunc/unified_shell/context.py +++ b/src/drunc/unified_shell/context.py @@ -7,6 +7,7 @@ from drunc.utils.grpc_utils import ServerTimeout, ServerUnreachable from drunc.utils.shell_utils import ShellContext +from drunc.utils.utils import resolve_localhost_to_hostname class UnifiedShellMode(Enum): @@ -16,6 +17,8 @@ class UnifiedShellMode(Enum): class UnifiedShellContext(ShellContext): # boilerplatefest + shell_id = "unified_shell" + def __init__(self): self.log = None self.status_receiver_pm = None @@ -33,7 +36,7 @@ def __init__(self): super(UnifiedShellContext, self).__init__() def reset(self, address_pm: str = ""): - self.address_pm = address_pm + self.address_pm = resolve_localhost_to_hostname(address_pm) super(UnifiedShellContext, self)._reset(name="unified_shell") def create_drivers(self, **kwargs) -> Mapping[str, object]: diff --git a/src/drunc/unified_shell/shell.py b/src/drunc/unified_shell/shell.py index eeda0e748..e61ba358c 100644 --- a/src/drunc/unified_shell/shell.py +++ b/src/drunc/unified_shell/shell.py @@ -47,16 +47,26 @@ get_process_manager_configuration, validate_pm_config, ) -from drunc.process_manager.interface.commands import ( +from drunc.process_manager.interface.process_manager import run_pm +from drunc.unified_shell.commands import ( + boot, flush, kill, logs, - # ps, + ps, restart, - # terminate, + start_shell, + terminate, ) -from drunc.process_manager.interface.process_manager import run_pm -from drunc.unified_shell.commands import boot, ps, start_shell, terminate + +#! Note with boot. We should discuss this +# When you run boot in the process manager with nothing else running, it works just fine +# however when you have a running session already, boot asks 'are you sure you want to do this? +# i bet you this is causing the behaviour that we are seeing with the unified shell +# when you log into an empty PM with the US, and you start-run, it works just fine +# but when you start-run from scratch with a running session in the PM, it doesn't work +# Theres also a whole thing about things not flushing correctly.. +# Aha! in the pm, my terminate doesn't flush properly! from drunc.unified_shell.context import UnifiedShellMode from drunc.unified_shell.shell_utils import generate_fsm_sequence_command from drunc.utils.configuration import ConfTypes, OKSKey @@ -155,9 +165,9 @@ def unified_shell( ctx.obj.log.debug("Setting up the [green]unified_shell[/green] logger") # Parse the process manager argument to determine if it's a config or an address - ctx.obj.log.critical( - f"Parsing the process manager argument: {process_manager}" - ) + # ctx.obj.log.critical( + # f"Parsing the process manager argument: {process_manager}" + # ) process_manager_url: ParseResult = urlparse(process_manager) internal_pm: bool = True if process_manager_url.scheme == "grpc": # i.e. if it's an address @@ -177,9 +187,15 @@ def unified_shell( # ) # sys.exit(1) - ctx.obj.log.critical("TEST") + # ctx.obj.log.critical("TEST") # Setup configuration related context variables - ctx.obj.configuration_file = f"oksconflibs:{configuration_file}" + # Assume oksconflibs if no framework is defined + ctx.obj.configuration_file = ( + lambda path: ( + path if path.startswith("oksconflibs:") else f"oksconflibs:{path}" + ) + )(configuration_file) + ctx.obj.configuration_id = configuration_id ctx.obj.session_name = session_name @@ -292,7 +308,7 @@ def unified_shell( f"[red]Could not connect to the process manager at the address: [/red]" f"[green]{process_manager_address}[/green]" ) - ctx.obj.log.debug(f"Reason: {e}") + ctx.obj.log.critical(f"Reason: {e}") if type(e) == ServerUnreachable: ctx.obj.log.error( @@ -311,7 +327,16 @@ def unified_shell( ctx.obj.pm_process.join() sys.exit(1) - ctx.obj.log.critical("Process manager described successfully") + # ctx.obj.log.critical("Process manager described successfully") + + ctx.obj.log.info( + f"[green]unified_shell[/green] connected to the [green]process_manager" + f"[/green] at address [green]{process_manager_address}[/green]" + ) + + ctx.obj.get_driver("process_manager").send_msg( + f"{getpass.getuser()} connected from unified shell" + ) # Add the unified shell Click commands to the CLI ctx.obj.log.debug("Adding [green]unified_shell[/green] commands") @@ -509,6 +534,9 @@ def cleanup(): ) # Remove the connection to the process manager + ctx.obj.get_driver("process_manager").send_msg( + f"{getpass.getuser()} disconnected from unified shell" + ) ctx.obj.get_driver("process_manager").close() ctx.obj.delete_driver("process_manager") diff --git a/src/drunc/utils/shell_utils.py b/src/drunc/utils/shell_utils.py index 02c9ab46f..b0a6a0b16 100644 --- a/src/drunc/utils/shell_utils.py +++ b/src/drunc/utils/shell_utils.py @@ -185,6 +185,11 @@ def __str__(self) -> str: class ShellContext: """Base class for shell contexts.""" + shell_id = None # used for logging if its a PM shell or Unified shell etc + + def get_shell_id(self): + return self.shell_id + def _reset( self, name: str, @@ -365,3 +370,32 @@ def print_status_summary(self) -> None: log.info( f"Current FSM status is [green]{current_state}[/green]. Available transitions are [green]{'[/green], [green]'.join(available_actions)}[/green]. Available sequence commands are [green]{'[/green], [green]'.join(available_sequences)}[/green]." ) + + +def log_pm_cmd(obj: ShellContext): + """Log a process-manager shell command with only explicitly provided arguments. + + The current Click command context is inspected and only parameters whose source is + ``COMMANDLINE`` are included in the log message. This keeps defaulted values out + of the message while still recording the command name, optional session name, and + shell identity. + + These are sent over via send_msg so that it can be displayed in the process manager + shell + + Args: + obj (ShellContext): Active shell context used to send the log message. + """ + + ctx_cmd = click.get_current_context(silent=True) + cmd_name = ctx_cmd.command.name if ctx_cmd else None + parms_dict = {} + for param in ctx_cmd.command.params: + name = param.name + if ctx_cmd.get_parameter_source(name) == click.core.ParameterSource.COMMANDLINE: + parms_dict[name] = f"{ctx_cmd.params[name]!r}" + + args = f" with arguments {parms_dict}" if parms_dict else "" + session = f" for session {obj.session_name}" if hasattr(obj, "session_name") else "" + msg = f"{getpass.getuser()} sent {cmd_name}{args}{session} via {obj.get_shell_id()}" + obj.get_driver("process_manager").send_msg(msg) diff --git a/src/drunc/utils/utils.py b/src/drunc/utils/utils.py index 71b07ad58..643431e15 100644 --- a/src/drunc/utils/utils.py +++ b/src/drunc/utils/utils.py @@ -1,6 +1,7 @@ """A set of utility functions for drunc.""" import ctypes +import ipaddress import logging import os import random @@ -706,6 +707,86 @@ def resolve_target_ip(host: str) -> str | None: return None +def resolve_context_peer(peer: str) -> str: + """Resolve a transport-qualified peer string to a display-friendly address. + + The input is expected to look like ``transport:address``. If the address contains + an IP literal, it is reverse-resolved where possible and IPv6 addresses are + re-wrapped in brackets. + + Example: + ``ipv4:10.73.136.70:41750`` -> ``np04-srv-028.cern.ch:41750`` + + Args: + peer (str): Transport-qualified peer string. + + Returns: + str: The original peer string, or a resolved ``host:port`` representation. + """ + + match = re.match(r"^(?P[^:]+):(?P
.+)$", peer) + if not match: + return peer + + parsed = _parse_host_port(match.group("address")) + if parsed is None: + return peer + + host, port = parsed + resolved_host = _resolve_host(host) + return f"{resolved_host}:{port}" + + +def _parse_host_port(address: str) -> tuple[str, str] | None: + """Extract a host and port from a peer address string. + + Supports bracketed IPv6 addresses such as ``[::1]:1234`` and unbracketed + ``host:port`` or ``ipv4:port`` forms. + + Args: + address (str): Address portion of a transport-qualified peer string. + + Returns: + tuple[str, str] | None: ``(host, port)`` when parsing succeeds, otherwise + ``None``. + """ + + bracket_match = re.match(r"^\[(?P[^\]]+)\]:(?P\d+)$", address) + if bracket_match: + return bracket_match.group("host"), bracket_match.group("port") + + host, sep, port = address.rpartition(":") + if sep and port.isdigit(): + return host, port + + return None + + +def _resolve_host(host: str) -> str: + """Reverse-resolve an IP host and keep IPv6 output bracketed. + + Args: + host (str): Hostname or IP literal to resolve. + + Returns: + str: The resolved hostname, or the original host if resolution fails. + """ + + try: + ip_obj = ipaddress.ip_address(host) + except ValueError: + return host + + try: + resolved_host, _, _ = socket.gethostbyaddr(str(ip_obj)) + except (socket.herror, socket.gaierror, socket.timeout, OSError): + resolved_host = host + + if ":" in resolved_host and not resolved_host.startswith("["): + return f"[{resolved_host}]" + return resolved_host + + def is_port_available(host: str, port: int, timeout: int = 2) -> bool: """Check if the given port number on a specified host is available. diff --git a/tests/process_manager/interface/test_commands.py b/tests/process_manager/interface/test_commands.py index 3571690e2..6ba1e10ad 100644 --- a/tests/process_manager/interface/test_commands.py +++ b/tests/process_manager/interface/test_commands.py @@ -102,6 +102,10 @@ def logs(self, log_request): mock_result.lines = [] return mock_result + def send_msg(self, msg: str) -> None: + # simulate sending a message; tests don't assert on this, so store it + self._last_sent_msg = msg + class MockContext: """ @@ -115,6 +119,9 @@ def __init__(self, driver=None): def get_driver(self, name): return self.driver + def get_shell_id(self): + return "mock-shell" + def print(self, msg, justify=None, overflow=None, soft_wrap=None): self.output.append(str(msg)) diff --git a/tests/process_manager/process_manager_mock_impls.py b/tests/process_manager/process_manager_mock_impls.py index d7f83c9c8..54b8b236f 100644 --- a/tests/process_manager/process_manager_mock_impls.py +++ b/tests/process_manager/process_manager_mock_impls.py @@ -9,6 +9,7 @@ from typing import Optional from unittest.mock import Mock +from druncschema.generic_pb2 import OutcomeFlag, OutcomeStatus from druncschema.process_manager_pb2 import ( BootRequest, LogLines, @@ -99,3 +100,12 @@ def _logs_impl(self, log_request: LogRequest) -> LogLines: def _flush_impl(self, query: ProcessQuery) -> ProcessInstanceList: return self._not_implemented_response() + + def _send_msg_impl( + self, msg: str | None = None, peer: str | None = None + ) -> OutcomeStatus: + """ + Returns an empty response to indicate communication is working. + Accepts an optional message parameter for compatibility with new API. + """ + return OutcomeStatus(flag=OutcomeFlag.SUCCESS) From 58fc2ce3ab2cd1ae7760088af6f0ed9f17c70373 Mon Sep 17 00:00:00 2001 From: Emir Muhammad Date: Mon, 1 Jun 2026 11:46:29 +0200 Subject: [PATCH 05/13] Add multiuser support --- src/drunc/controller/controller.py | 20 +++++- src/drunc/process_manager/configuration.py | 5 +- .../process_manager/interface/commands.py | 12 ++-- .../process_manager/process_manager_driver.py | 72 +++++++++++++------ .../process_manager/ssh_process_manager.py | 2 +- .../ssh_process_lifetime_manager_shell.py | 34 +++++++++ src/drunc/unified_shell/commands.py | 13 ++-- src/drunc/utils/utils.py | 24 +++++++ tests/process_manager/conftest.py | 6 +- .../interface/test_commands.py | 5 +- tests/utils/test_utils.py | 7 ++ 11 files changed, 160 insertions(+), 40 deletions(-) diff --git a/src/drunc/controller/controller.py b/src/drunc/controller/controller.py index daaa5ace3..c97602359 100644 --- a/src/drunc/controller/controller.py +++ b/src/drunc/controller/controller.py @@ -1,4 +1,6 @@ import multiprocessing +import os +import socket import threading import time from concurrent.futures import ThreadPoolExecutor @@ -129,9 +131,17 @@ def __init__(self, configuration, name: str, session: str, token: Token): self.connectivity_service_thread = None self.uri = "" if self.configuration.session.connectivity_service: - connection_server = self.configuration.session.connectivity_service.host - connection_port = ( - self.configuration.session.connectivity_service.service.port + # Remaps the localhost into the correct server + # and also grabs the correct port from the right environment from the config + + connection_server_host = ( + self.configuration.session.connectivity_service.host + ) + connection_port = os.getenv("CONNECTION_PORT") + connection_server = ( + socket.gethostname() + if connection_server_host == "localhost" + else connection_server_host ) log_init.info( f"Connectivity server {connection_server}:{connection_port} is enabled" @@ -326,6 +336,10 @@ def advertise_control_address(self, address): if not self.connectivity_service: return + if not self.connectivity_service.is_ready(timeout=10): + raise ValueError( + "Connectivity service unavailable for control address advertising." + ) self.log.info( f"Registering {self.name} ({address}) to the connectivity service at {self.connectivity_service.address}" ) diff --git a/src/drunc/process_manager/configuration.py b/src/drunc/process_manager/configuration.py index 5172fa219..32d3c7889 100644 --- a/src/drunc/process_manager/configuration.py +++ b/src/drunc/process_manager/configuration.py @@ -15,7 +15,7 @@ from drunc.exceptions import DruncCommandException from drunc.process_manager.exceptions import UnknownProcessManagerType from drunc.utils.configuration import ConfHandler -from drunc.utils.utils import get_logger +from drunc.utils.utils import get_logger, touch_and_chmod if TYPE_CHECKING: import conffwk @@ -104,6 +104,9 @@ def _parse_dict(self, data): self.opmon_conf, ) else: + # ensures users can access the opmon files (permissions) + # This is for the PM opmon file + touch_and_chmod(self.opmon_conf.path) new_data.opmon_publisher = OpMonPublisher( conf=self.opmon_conf, rich_handler=True ) diff --git a/src/drunc/process_manager/interface/commands.py b/src/drunc/process_manager/interface/commands.py index cbfde6250..703e5b2ae 100644 --- a/src/drunc/process_manager/interface/commands.py +++ b/src/drunc/process_manager/interface/commands.py @@ -13,7 +13,7 @@ from drunc.process_manager.interface.context import ProcessManagerContext from drunc.process_manager.utils import tabulate_process_instance_list from drunc.utils.shell_utils import InterruptedCommand, log_pm_cmd -from drunc.utils.utils import get_logger +from drunc.utils.utils import get_logger, resolve_context_peer @click.command("boot") @@ -45,11 +45,14 @@ def boot( ) -> None: log = get_logger("process_manager.shell") log_pm_cmd(obj) - processes = obj.get_driver("process_manager").ps(ProcessQuery(user=user)) + processes = obj.get_driver("process_manager").ps( + ProcessQuery(user=user, session=session_name) + ) + # The run control will validate this in the session manager in the future if len(processes.values) > 0: click.confirm( - f"You already have {len(processes.values)} processes running, are you sure you want to boot a session?", + f"You already have {len(processes.values)} processes running for {session_name}, are you sure you want to boot a session?", abort=True, ) @@ -78,6 +81,7 @@ def boot( raise e controller_address = obj.get_driver("process_manager").controller_address + controller_address = resolve_context_peer(controller_address) if controller_address: obj.print( Panel( @@ -286,7 +290,7 @@ def logs_impl( log = get_logger("process_manager.shell") # TODO: MOVE BACK TO DEBUG BEFORE MERGE # THIS IS USEFUL FOR TESTING THOUGH - log.error(f"Running logs with query {query}") + log.debug(f"Running logs with query {query}") log_req = LogRequest( how_far=how_far, query=query, diff --git a/src/drunc/process_manager/process_manager_driver.py b/src/drunc/process_manager/process_manager_driver.py index ea239f4d6..32d9a1040 100644 --- a/src/drunc/process_manager/process_manager_driver.py +++ b/src/drunc/process_manager/process_manager_driver.py @@ -7,6 +7,7 @@ from collections.abc import Iterator from time import sleep from typing import Dict, List +from urllib.parse import urlparse import conffwk import grpc @@ -49,6 +50,7 @@ resolve_localhost_and_127_ip_to_network_ip, resolve_localhost_to_hostname, strip_non_drunc_loggers, + touch_and_chmod, ) @@ -141,6 +143,9 @@ def boot( # Step 3 - check for port conflicts and update configuration/DAL as needed db, session_dal = self.check_port_conflicts(db, session_dal) + # step 3.5 update localhost mapping + session_dal = self.resolve_localhost(session_dal) + # Step 4 - connect to the connection service csc, connection_server, connection_port = self._connect_to_service( session_dal, session_name @@ -190,6 +195,24 @@ def boot( previous_host = this_host last_boot_on_host_at[this_host] = time.time() + # ensures users can access the opmon files (permissions) + # This is for the opmon files of the apps + + if session_dal.opmon_uri.type == "file": + # For future, this should probably be taken from the metadata + opmon_file = ( + f"{request.process_description.process_execution_directory}/info." + + request.process_description.metadata.session + + "." + + request.process_description.metadata.name + + ".json" + ) + + self.log.debug( + f"Touching and changing permissions for {opmon_file} because opmon is of type {session_dal.opmon_uri.type}" + ) + touch_and_chmod(opmon_file) + try: response = self.stub.boot(request, timeout=timeout) yield response @@ -287,7 +310,11 @@ def _build_boot_request( override_logs: bool, pwd: str, ) -> BootRequest: - host = format_hostname(app["restriction"]) + # Run mapping to physical hostname to enable multi host usage + host = resolve_localhost_to_hostname(format_hostname(app["restriction"])) + self.log.info(f"boot resolve {host}") # keep this until big PR gets merged + + # this is one of the two minimal changes needed to get this working in general? name = app["name"] exe = app["type"] args = app["args"] @@ -403,19 +430,29 @@ def _consolidate_config(self, session_name, conf_file: str) -> str | None: ) return - def update_connectivity_port_dal( - self, - env_variables: list["conffwk.dal.Variable | conffwk.dal.VariableSet"], - new_port: int, - ) -> None: - """Process a dal::Variable object, placing key/value pairs in a dictionary""" - for item in env_variables: - if item.className() == "VariableSet": - self.update_connectivity_port_dal(item.contains, new_port) - else: - if item.className() == "Variable": - if item.name == "CONNECTION_PORT": - item.value = new_port + def resolve_localhost(self, session_dal): + def dal_localhost_mapping(dal_host: str): + if dal_host != "localhost": + return dal_host + + resolved_address = resolve_localhost_to_hostname(dal_host) + if "://" not in resolved_address: + resolved_address = "grpc://" + resolved_address + + resolved_server = urlparse(resolved_address).hostname + self.log.debug( + f"Resolved connection server 'localhost' to '{resolved_server}' to avoid K8s hairpinning." + ) + return resolved_server + + session_dal.connectivity_service.host = dal_localhost_mapping( + session_dal.connectivity_service.host + ) + session_dal.segment.controller.runs_on.runs_on.id = dal_localhost_mapping( + session_dal.segment.controller.runs_on.runs_on.id + ) + + return session_dal def check_port_conflicts( self, db: conffwk.Configuration, session_dal: "conffwk.dal.Session" @@ -545,13 +582,6 @@ def _connect_to_service( connection_server = session_dal.connectivity_service.host connection_port = session_dal.connectivity_service.service.port - if connection_server == "localhost": - resolved_server = resolve_localhost_to_hostname(connection_server) - self.log.debug( - f"Resolved connection server 'localhost' to '{resolved_server}' to avoid K8s hairpinning." - ) - connection_server = resolved_server - client = ConnectivityServiceClient( session_name, f"{connection_server}:{connection_port}" ) diff --git a/src/drunc/process_manager/ssh_process_manager.py b/src/drunc/process_manager/ssh_process_manager.py index 7f40d53d7..413ba2c9f 100644 --- a/src/drunc/process_manager/ssh_process_manager.py +++ b/src/drunc/process_manager/ssh_process_manager.py @@ -390,7 +390,7 @@ def __boot(self, boot_request: BootRequest, uuid: str) -> ProcessInstance: self.log.info( f"Booted '{boot_request.process_description.metadata.name}' " f"from session '{boot_request.process_description.metadata.session}' " - f"with UUID {uuid}" + f"with UUID {uuid} in host '{hostname}" ) # Query current process status diff --git a/src/drunc/processes/ssh_process_lifetime_manager_shell.py b/src/drunc/processes/ssh_process_lifetime_manager_shell.py index 0a9a3edbc..61e1bc694 100644 --- a/src/drunc/processes/ssh_process_lifetime_manager_shell.py +++ b/src/drunc/processes/ssh_process_lifetime_manager_shell.py @@ -1191,6 +1191,7 @@ def _execute_bootrequest_via_ssh( remote_cmd += ( f"mkdir -p ${{XDG_RUNTIME_DIR:-/tmp}}/drunc ; " + f"rm {log_file}; " # delete log file so no issues on ovewriting in th next line f"{command} &> {log_file} & PID=$! ; " f"trap 'kill -HUP $PID 2>/dev/null || true; wait $PID 2>/dev/null || true' HUP TERM INT QUIT ; " f"echo '{remote_metadata_json}' > {metadata_file} ; " @@ -1200,6 +1201,39 @@ def _execute_bootrequest_via_ssh( arguments = self._build_ssh_arguments(hostname, user_host) arguments.append(remote_cmd) + # Test access to CMD + cd_path = f"{boot_request.process_description.process_execution_directory}" + touch_cmd = [ + arguments[0], # assume first arg is username@host + f"touch {cd_path}/.write_test && rm {cd_path}/.write_test", + ] + self.log.debug(f"running {touch_cmd} for CMD access test") + try: + access = self.ssh( + *touch_cmd, + _out=self.log.warning, + _err=self.log.error, + _bg=True, + _bg_exc=False, + _new_session=True, + _preexec_fn=on_parent_exit(signal.SIGTERM) + if not is_macos + else None, + ) + + access.wait() + if access.exit_code != 0: + raise RuntimeError("SSH error fails to finish successfully") + except Exception as e: + err_msg = ( + f"No access to {cd_path}" + "for multiusers to work, the above path needs elevated permissions for" + " the PM superuser to cd and write into. " + "Please change the permissions to allow for this." + ) + self.log.error(err_msg) + raise RuntimeError from e + process = self.ssh( *arguments, _out=self.log.debug, diff --git a/src/drunc/unified_shell/commands.py b/src/drunc/unified_shell/commands.py index be093c134..a39e3274d 100644 --- a/src/drunc/unified_shell/commands.py +++ b/src/drunc/unified_shell/commands.py @@ -58,12 +58,13 @@ def boot( override_logs_boot = obj.override_logs else: override_logs_boot = override_logs - # if len(processes.values) > 0: - # log.error( - # f"Cannot boot: session {session_name} already has {len(processes.values)} processes running. " - # "Please terminate the existing session first." - # ) - # return + # The run control will validate this in the session manager in the future + if len(processes.values) > 0: + log.error( + f"Cannot boot: session {session_name} already has {len(processes.values)} processes running. " + "Please terminate the existing session first." + ) + return try: results = obj.get_driver("process_manager").boot( diff --git a/src/drunc/utils/utils.py b/src/drunc/utils/utils.py index 643431e15..d24711b2c 100644 --- a/src/drunc/utils/utils.py +++ b/src/drunc/utils/utils.py @@ -364,6 +364,18 @@ def parent_death_pact(signal: int = signal.SIGHUP) -> None: raise Exception("prctl() returned nonzero retcode %d" % retcode) +# 777 PERMISSIONS ARE COMPLETELY TEMPORARY +# An established procedure for multi users will need to be discussed with sysadmins +# will be removed when done +def touch_and_chmod(filepath: str, mode=0o777): + """Makes and sets the permissions of a file. + This is used to ensure multiuser support when accessing files etc.""" + + with open(filepath, "a"): + os.utime(filepath, None) + os.chmod(filepath, mode) + + class IncorrectAddress(DruncException): """Exception raised when an address is invalid.""" @@ -724,6 +736,18 @@ def resolve_context_peer(peer: str) -> str: str: The original peer string, or a resolved ``host:port`` representation. """ + if not peer: + return peer + + # Some callers pass a plain host:port string without a transport prefix. + # Handle those directly instead of assuming the first token is always a transport. + if peer.startswith("[") or peer.count(":") == 1: + parsed = _parse_host_port(peer) + if parsed is not None: + host, port = parsed + resolved_host = _resolve_host(host) + return f"{resolved_host}:{port}" + match = re.match(r"^(?P[^:]+):(?P
.+)$", peer) if not match: return peer diff --git a/tests/process_manager/conftest.py b/tests/process_manager/conftest.py index 6a2ba4bfd..78aed2525 100644 --- a/tests/process_manager/conftest.py +++ b/tests/process_manager/conftest.py @@ -7,6 +7,8 @@ to be back in line with druncschema definitions. """ +import socket + import google.protobuf.any_pb2 import pytest from druncschema.description_pb2 import Description @@ -36,7 +38,7 @@ def app_data(): Provides a mock application dictionary with required keys. """ return { - "restriction": "localhost", + "restriction": socket.gethostname(), "name": "TestApp", "type": "binary", "args": ["--arg1"], @@ -55,7 +57,7 @@ def bootrequest(app_data): user="test_user", session="session1", name=app_data["name"], - hostname="localhost", + hostname=socket.gethostname(), tree_id=app_data["tree_id"], ), executable_and_arguments=[{"exec": "binary", "args": ["--arg1"]}], diff --git a/tests/process_manager/interface/test_commands.py b/tests/process_manager/interface/test_commands.py index 6ba1e10ad..f1a5c5e4b 100644 --- a/tests/process_manager/interface/test_commands.py +++ b/tests/process_manager/interface/test_commands.py @@ -237,8 +237,9 @@ def test_boot_exiting_processes_abort(boot_arguments): # check that 'boot' was never called mock_driver.boot.assert_not_called() + # conf-id-123 from session name in this file assert ( - "You already have 2 processes running, are you sure you want to boot a session?" + "You already have 2 processes running for conf-id-123, are you sure you want to boot a session?" in result.output ) @@ -267,7 +268,7 @@ def test_boot_exiting_processes_user_confirm(boot_arguments): mock_driver.boot.assert_called() assert ( - "You already have 2 processes running, are you sure you want to boot a session?" + "You already have 2 processes running for conf-id-123, are you sure you want to boot a session?" in result.output ) diff --git a/tests/utils/test_utils.py b/tests/utils/test_utils.py index 94b3573d6..5094824b0 100644 --- a/tests/utils/test_utils.py +++ b/tests/utils/test_utils.py @@ -20,6 +20,7 @@ now_str, parent_death_pact, regex_match, + resolve_context_peer, resolve_localhost_and_127_ip_to_network_ip, resolve_localhost_to_hostname, validate_command_facility, @@ -157,6 +158,12 @@ def test_resolve_localhost_and_127_ip_to_network_ip(): assert resolved == generate_address(this_ip) +def test_resolve_context_peer(): + assert resolve_context_peer("grpc:np04-srv-028:50000") == "np04-srv-028:50000" + assert resolve_context_peer("np04-srv-028:50000") == "np04-srv-028:50000" + assert resolve_context_peer("") == "" + + def test_host_is_local(): this_ip = socket.gethostbyname(socket.gethostname()) hostname = socket.gethostname() From f042c4e825f3340302ae857faf13289aac4ca026 Mon Sep 17 00:00:00 2001 From: Emir Muhammad Date: Tue, 23 Jun 2026 12:56:13 +0200 Subject: [PATCH 06/13] remove broadcaster again --- src/drunc/process_manager/process_manager.py | 1 - 1 file changed, 1 deletion(-) diff --git a/src/drunc/process_manager/process_manager.py b/src/drunc/process_manager/process_manager.py index 060625c08..efa92ee83 100644 --- a/src/drunc/process_manager/process_manager.py +++ b/src/drunc/process_manager/process_manager.py @@ -490,7 +490,6 @@ def _send_msg_impl( ) -> OutcomeStatus: raise NotImplementedError - @broadcasted @authentified_and_authorised( action=ActionType.READ, system=SystemType.PROCESS_MANAGER ) From ca8f16d935f1c57a5e0318bf8fc4b5c2d48441b4 Mon Sep 17 00:00:00 2001 From: PawelPlesniak Date: Mon, 29 Jun 2026 14:00:14 +0200 Subject: [PATCH 07/13] Ruff correction --- src/drunc/process_manager/ssh_process_manager.py | 1 - 1 file changed, 1 deletion(-) diff --git a/src/drunc/process_manager/ssh_process_manager.py b/src/drunc/process_manager/ssh_process_manager.py index 413ba2c9f..8064858b8 100644 --- a/src/drunc/process_manager/ssh_process_manager.py +++ b/src/drunc/process_manager/ssh_process_manager.py @@ -3,7 +3,6 @@ import uuid from typing import List, Optional - from druncschema.generic_pb2 import OutcomeFlag, OutcomeStatus from druncschema.process_manager_pb2 import ( BootRequest, From ee715314cdb5725d5a93d5a977f41e7bd87914b8 Mon Sep 17 00:00:00 2001 From: PawelPlesniak Date: Mon, 29 Jun 2026 17:19:25 +0200 Subject: [PATCH 08/13] Cleanup --- .../process_manager/interface/commands.py | 2 - src/drunc/process_manager/interface/shell.py | 4 +- .../process_manager/k8s_process_manager.py | 6 ++- .../process_manager/ssh_process_manager.py | 30 +++--------- .../ssh_process_lifetime_manager_shell.py | 49 +++++++------------ src/drunc/unified_shell/commands.py | 20 +++++--- src/drunc/unified_shell/shell.py | 18 +------ 7 files changed, 48 insertions(+), 81 deletions(-) diff --git a/src/drunc/process_manager/interface/commands.py b/src/drunc/process_manager/interface/commands.py index 703e5b2ae..0d2cd3c43 100644 --- a/src/drunc/process_manager/interface/commands.py +++ b/src/drunc/process_manager/interface/commands.py @@ -288,8 +288,6 @@ def logs_impl( obj: ProcessManagerContext, how_far: int, grep: str, query: ProcessQuery ) -> None: log = get_logger("process_manager.shell") - # TODO: MOVE BACK TO DEBUG BEFORE MERGE - # THIS IS USEFUL FOR TESTING THOUGH log.debug(f"Running logs with query {query}") log_req = LogRequest( how_far=how_far, diff --git a/src/drunc/process_manager/interface/shell.py b/src/drunc/process_manager/interface/shell.py index e1341df60..b933ffcc1 100644 --- a/src/drunc/process_manager/interface/shell.py +++ b/src/drunc/process_manager/interface/shell.py @@ -79,10 +79,10 @@ def process_manager_shell(ctx, process_manager_address: str, log_level: str) -> def cleanup(): ctx.obj.get_driver("process_manager").send_msg( - f"{getpass.getuser()} disconnected from {ctx.obj.shell_id}" + f"{getpass.getuser()} disconnecting from {ctx.obj.shell_id}" ) ctx.obj.terminate() - process_manager_log.warning( + process_manager_log.info( f"[green]{getpass.getuser()}[/green] disconnected from the process manager through a [green]drunc-process-manager-shell[/green]" ) diff --git a/src/drunc/process_manager/k8s_process_manager.py b/src/drunc/process_manager/k8s_process_manager.py index f2fc064ab..d4eea8e70 100644 --- a/src/drunc/process_manager/k8s_process_manager.py +++ b/src/drunc/process_manager/k8s_process_manager.py @@ -1430,7 +1430,9 @@ def _get_pod_host_aliases( self.log.info( f"Pod '{podname}' will resolve localhost to connection server IP {connection_server_ip}" ) - return [client.V1HostAlias(ip=connection_server_ip, hostnames=["localhost"])] + return [ + client.V1HostAlias(ip=connection_server_ip, hostnames=["localhost"]) + ] self.log.warning( f"Could not get connection server ClusterIP for pod '{podname}'" @@ -1972,7 +1974,7 @@ def _send_msg_impl(self, msg: str, peer: str) -> OutcomeStatus: try: self.log.info(f"{msg}; from {peer}") except Exception as e: - self.log.critical(f"Failed to receive message with exception {e}") + self.log.error(f"Failed to receive message with exception {e}") return OutcomeStatus(flag=OutcomeFlag.FAIL) return OutcomeStatus(flag=OutcomeFlag.SUCCESS) diff --git a/src/drunc/process_manager/ssh_process_manager.py b/src/drunc/process_manager/ssh_process_manager.py index 8064858b8..2cd3a24e5 100644 --- a/src/drunc/process_manager/ssh_process_manager.py +++ b/src/drunc/process_manager/ssh_process_manager.py @@ -366,9 +366,9 @@ def __boot(self, boot_request: BootRequest, uuid: str) -> ProcessInstance: # Update hostname in boot request for this attempt self.boot_request[uuid].process_description.metadata.hostname = host - # self.log.critical( - # f"Attempting to start process {uuid} on host {host} via SSH lifetime manager" - # ) + self.log.debug( + f"Attempting to start process {uuid} on host {host} via SSH lifetime manager" + ) # Start the process via SSH manager self.ssh_lifetime_manager.start_process( uuid=uuid, boot_request=self.boot_request[uuid] @@ -447,10 +447,6 @@ def _ps_impl(self, query: ProcessQuery) -> ProcessInstanceList: boot_request_dict=self.boot_request, order_by="random", ) - # if query.session: - # self.log.warning( - # f"{self.name} found {len(process_uuids)} processes matching {query} for ps" - # ) # Iterate through all processes matching the query for proc_uuid in process_uuids: @@ -504,26 +500,16 @@ def _ps_impl(self, query: ProcessQuery) -> ProcessInstanceList: values=ret, flag=ResponseFlag.EXECUTED_SUCCESSFULLY, ) - # if query.session: - # self.log.critical( - # f"{self.name} returning {len(ret)} processes from ps query {query}" - # ) - # self.log.critical(ret_fmt) - # self.log.critical(f"TEST: {ret_fmt=}") - # else: - # self.log.warning( - # f"{self.name} returning {len(ret)} processes from ps query {query}" - # ) - # self.log.warning(ret) + self.log.debug( + f"{self.name} returning {len(ret)} processes from ps query {query}" + ) return ret_fmt def _send_msg_impl(self, msg: str, peer: str) -> OutcomeStatus: try: - # TODO: THIS IS CURRENTLY CRITICAL FOR EASIER TESTING - # DO _NOT_ MERGE UNTIL THIS IS BACK TO INFO! - self.log.critical(f"{msg}; from {peer}") + self.log.info(f"{msg}; from {peer}") except Exception as e: - self.log.critical(f"Failed to receive message with exception {e}") + self.log.error(f"Failed to receive message with exception {e}") return OutcomeStatus(flag=OutcomeFlag.FAIL) return OutcomeStatus(flag=OutcomeFlag.SUCCESS) diff --git a/src/drunc/processes/ssh_process_lifetime_manager_shell.py b/src/drunc/processes/ssh_process_lifetime_manager_shell.py index 396967d7b..d68f537eb 100644 --- a/src/drunc/processes/ssh_process_lifetime_manager_shell.py +++ b/src/drunc/processes/ssh_process_lifetime_manager_shell.py @@ -426,9 +426,9 @@ def start_process(self, uuid: str, boot_request: BootRequest) -> None: hostname = boot_request.process_description.metadata.hostname user = boot_request.process_description.metadata.user log_file = boot_request.process_description.process_logs_path - # self.log.critical( - # f"Starting process {uuid} on {hostname} as {user} with log file {log_file}" - # ) + self.log.debug( + f"Starting process {uuid} on {hostname} as {user} with log file {log_file}" + ) # Extract environment variables from boot request env_vars = ( @@ -452,7 +452,7 @@ def start_process(self, uuid: str, boot_request: BootRequest) -> None: if cmd.endswith(";"): cmd = cmd[:-1] - # self.log.critical(f"Built command for {uuid}: {cmd}: {boot_request}") + self.log.debug(f"Built command for {uuid}: {cmd}: {boot_request}") # Execute the command via SSH self._execute_bootrequest_via_ssh( @@ -894,45 +894,37 @@ def _build_ssh_arguments( # Determine if host key checking should be disabled based on configuration and # target host - # disable_host_key_check = self.disable_host_key_check or ( - # self.disable_localhost_host_key_check - # and hostname in ("localhost", "127.0.0.1", "::1") - # ) superuser_host = getpass.getuser() + "@" + user_host.split("@")[1] - # self.log.critical(f"Building SSH arguments for {user_host} with superuser host {superuser_host}") - arguments = [superuser_host, "-o", "StrictHostKeyChecking=no"] - # self.log.critical(f"SSH arguments after adding StrictHostKeyChecking for {user_host}%s", arguments) - # self.log.critical(f"{arguments=}") - # self.log.critical(f"Test list: {test_list_print}") - # self.log.critical(f"Test list 2: %s", test_list_print) + self.log.debug( + f"Building SSH arguments for {user_host} with superuser host {superuser_host}" + ) # Base SSH arguments with user@host and strict host key checking disabled # StrictHostKeyChecking=no is set to as we have an nfs backed home directory and # the known_hosts file is not shared across hosts, so we cannot rely on it for # host key verification. - # arguments = [user_host, "-o", "StrictHostKeyChecking=no"] - # "-F /nfs/home/{user_host.split('@')[0]}/.ssh/config", + arguments = [superuser_host, "-o", "StrictHostKeyChecking=no"] + # If TTY allocation is requested, add the -tt flag to force allocation. This is + # needed as SSH permissions are different for general users and for np04daq if use_tty: arguments.append("-tt") # If host key checking is disabled, also disable known hosts file usage and - # reduce log level to avoid cluttering logs with warnings about host key verification - # if disable_host_key_check: + # reduce log level to avoid cluttering logs with warnings about host key + # verification arguments.extend( [ "-o", - "LogLevel=info", + "LogLevel=error", "-o", "GlobalKnownHostsFile=/dev/null", "-o", "UserKnownHostsFile=/dev/null", ] ) - # self.log.critical(f"SSH arguments for {user_host}: {arguments}") - # self.log.critical( - # f"PP: {getpass.getuser()} is running on {os.uname().nodename} with disable_host_key_check={self.disable_host_key_check} and disable_localhost_host_key_check={self.disable_localhost_host_key_check}" - # ) + + self.log.debug(f"SSH arguments for {user_host}: {arguments}") return arguments @@ -1052,15 +1044,12 @@ def read_process_metadata( arguments.append(remote_command) # Execute SSH command to wait for and read file (single round-trip) - # self.log.critical( - # f"Attempting to read metadata for {uuid} from {hostname} with timeout {timeout}s" - # ) + self.log.debug( + f"Attempting to read metadata for {uuid} from {hostname} with timeout {timeout}s" + ) result = self.ssh(*arguments) - # self.log.critical( - # f"DEBUG - Raw metadata content for {uuid} from {hostname}: {result}" - # ) + self.log.debug(f"Raw metadata content for {uuid} from {hostname}: {result}") json_content = str(result).strip() - # self.log.critical("Attempt successful?|???") self.log.debug(f"Metadata content for {uuid}: {json_content!r}") diff --git a/src/drunc/unified_shell/commands.py b/src/drunc/unified_shell/commands.py index a39e3274d..b42c191a2 100644 --- a/src/drunc/unified_shell/commands.py +++ b/src/drunc/unified_shell/commands.py @@ -168,13 +168,19 @@ def ps(ctx, obj): session_query = ProcessQuery(session=ctx.obj.session_name) log.info(f"Listing session [green]{ctx.obj.session_name}[/]") results = obj.get_driver("process_manager").ps(session_query) - obj.print( - tabulate_process_instance_list( - results, title=f"Processes running in session {ctx.obj.session_name}" - ), - overflow="fold", - soft_wrap=True, - ) + + # If there are processes running, tabulate them, otherwise log that there are no + # processes running. + if results.values: + obj.print( + tabulate_process_instance_list( + results, title=f"Processes running in session {ctx.obj.session_name}" + ), + overflow="fold", + soft_wrap=True, + ) + else: + log.info(f"No processes running in session [green]{ctx.obj.session_name}[/]") def session_injector(f): diff --git a/src/drunc/unified_shell/shell.py b/src/drunc/unified_shell/shell.py index e61ba358c..093f7f51c 100644 --- a/src/drunc/unified_shell/shell.py +++ b/src/drunc/unified_shell/shell.py @@ -58,15 +58,6 @@ start_shell, terminate, ) - -#! Note with boot. We should discuss this -# When you run boot in the process manager with nothing else running, it works just fine -# however when you have a running session already, boot asks 'are you sure you want to do this? -# i bet you this is causing the behaviour that we are seeing with the unified shell -# when you log into an empty PM with the US, and you start-run, it works just fine -# but when you start-run from scratch with a running session in the PM, it doesn't work -# Theres also a whole thing about things not flushing correctly.. -# Aha! in the pm, my terminate doesn't flush properly! from drunc.unified_shell.context import UnifiedShellMode from drunc.unified_shell.shell_utils import generate_fsm_sequence_command from drunc.utils.configuration import ConfTypes, OKSKey @@ -165,17 +156,15 @@ def unified_shell( ctx.obj.log.debug("Setting up the [green]unified_shell[/green] logger") # Parse the process manager argument to determine if it's a config or an address - # ctx.obj.log.critical( - # f"Parsing the process manager argument: {process_manager}" - # ) process_manager_url: ParseResult = urlparse(process_manager) internal_pm: bool = True if process_manager_url.scheme == "grpc": # i.e. if it's an address internal_pm = False - ctx.obj.log.critical( + ctx.obj.log.debug( f"Process manager argument parsed, internal_pm set to {internal_pm}" ) + #! TODO - reintroduce this before this PR is merged. # If using a k8s process manager, validate the session name before proceeding # if get_pm_type_from_name( # process_manager @@ -300,7 +289,6 @@ def unified_shell( ctx.obj.reset(address_pm=process_manager_address) # Run a simple command (describe) to check the connection with the process manager - ctx.obj.log.critical("Getting driver") try: ctx.obj.get_driver().describe() except Exception as e: @@ -349,11 +337,9 @@ def unified_shell( ctx.obj.log.debug("Adding [green]process_manager[/green] commands") process_manager_commands: list[click.Command] = [ kill, - # terminate, flush, logs, restart, - # ps, ] for cmd in process_manager_commands: ctx.command.add_command(cmd, format_name_for_cli(cmd.name)) From 323137edf4dd6feaa1ac9a000528a4f55e801968 Mon Sep 17 00:00:00 2001 From: PawelPlesniak Date: Tue, 30 Jun 2026 14:24:08 +0200 Subject: [PATCH 09/13] K8s session name check re-introduced, LCS configs working --- src/drunc/controller/controller.py | 24 ++++++++++++------- .../process_manager/process_manager_driver.py | 8 ++++++- .../process_manager/ssh_process_manager.py | 2 +- src/drunc/unified_shell/commands.py | 4 ++++ src/drunc/unified_shell/shell.py | 24 +++++++++---------- 5 files changed, 39 insertions(+), 23 deletions(-) diff --git a/src/drunc/controller/controller.py b/src/drunc/controller/controller.py index c97602359..9bda0b034 100644 --- a/src/drunc/controller/controller.py +++ b/src/drunc/controller/controller.py @@ -1,6 +1,5 @@ import multiprocessing import os -import socket import threading import time from concurrent.futures import ThreadPoolExecutor @@ -138,18 +137,22 @@ def __init__(self, configuration, name: str, session: str, token: Token): self.configuration.session.connectivity_service.host ) connection_port = os.getenv("CONNECTION_PORT") - connection_server = ( - socket.gethostname() - if connection_server_host == "localhost" - else connection_server_host - ) + if connection_server_host == "localhost": + injected_hostname = os.getenv("DRUNC_HOST_NAME") + if not injected_hostname: + raise ValueError("DRUNC_HOST_NAME environment variable is not set.") + self.log.debug( + f"Remapping connectivity service host from 'localhost' to '{injected_hostname}'" + ) + connection_server_host = injected_hostname + log_init.info( - f"Connectivity server {connection_server}:{connection_port} is enabled" + f"Connectivity server {connection_server_host}:{connection_port} is enabled" ) self.connectivity_service = ConnectivityServiceClient( session=self.session, - address=f"{connection_server}:{connection_port}", + address=f"{connection_server_host}:{connection_port}", ) def init_controller(self) -> None: @@ -336,7 +339,10 @@ def advertise_control_address(self, address): if not self.connectivity_service: return - if not self.connectivity_service.is_ready(timeout=10): + self.log.debug( + f"Looking for connectivity service at address {self.connectivity_service.address}" + ) + if not self.connectivity_service.is_ready(timeout=20): raise ValueError( "Connectivity service unavailable for control address advertising." ) diff --git a/src/drunc/process_manager/process_manager_driver.py b/src/drunc/process_manager/process_manager_driver.py index 32d9a1040..48c37b1c9 100644 --- a/src/drunc/process_manager/process_manager_driver.py +++ b/src/drunc/process_manager/process_manager_driver.py @@ -312,7 +312,6 @@ def _build_boot_request( ) -> BootRequest: # Run mapping to physical hostname to enable multi host usage host = resolve_localhost_to_hostname(format_hostname(app["restriction"])) - self.log.info(f"boot resolve {host}") # keep this until big PR gets merged # this is one of the two minimal changes needed to get this working in general? name = app["name"] @@ -324,6 +323,13 @@ def _build_boot_request( env["DUNE_DAQ_BASE_RELEASE"] = os.getenv("DUNE_DAQ_BASE_RELEASE") env["SPACK_RELEASES_DIR"] = os.getenv("SPACK_RELEASES_DIR") tree_id = app["tree_id"] + + # The following line is required to provide an independent method of injecting + # the hostname into the environment for applications that need it. This is the + # case for containerized applications, for which the hostname is not + # automatically injected into the environment, and standard methods like + # socket.gethostname() do not return the expected value. + env["DRUNC_HOST_NAME"] = host self.log.debug(f"{name}:\n{json.dumps(app, indent=4)}") try: diff --git a/src/drunc/process_manager/ssh_process_manager.py b/src/drunc/process_manager/ssh_process_manager.py index 2cd3a24e5..25d351006 100644 --- a/src/drunc/process_manager/ssh_process_manager.py +++ b/src/drunc/process_manager/ssh_process_manager.py @@ -389,7 +389,7 @@ def __boot(self, boot_request: BootRequest, uuid: str) -> ProcessInstance: self.log.info( f"Booted '{boot_request.process_description.metadata.name}' " f"from session '{boot_request.process_description.metadata.session}' " - f"with UUID {uuid} in host '{hostname}" + f"with UUID {uuid} on host {hostname}" ) # Query current process status diff --git a/src/drunc/unified_shell/commands.py b/src/drunc/unified_shell/commands.py index b42c191a2..6285ad3bf 100644 --- a/src/drunc/unified_shell/commands.py +++ b/src/drunc/unified_shell/commands.py @@ -153,6 +153,10 @@ def terminate(ctx, obj): log.info(f"Terminating session [green]{ctx.obj.session_name}[/]") obj.get_driver("process_manager").kill(session_query) + # As the session is now terminated, we can delete the controller driver, as it is no + # longer needed. + obj.delete_driver("controller") + @click.command("ps") @click.pass_obj diff --git a/src/drunc/unified_shell/shell.py b/src/drunc/unified_shell/shell.py index 093f7f51c..8feb17331 100644 --- a/src/drunc/unified_shell/shell.py +++ b/src/drunc/unified_shell/shell.py @@ -44,10 +44,12 @@ from drunc.fsm.configuration import FSMConfHandler from drunc.fsm.utils import convert_fsm_transition from drunc.process_manager.configuration import ( + ProcessManagerTypes, get_process_manager_configuration, validate_pm_config, ) from drunc.process_manager.interface.process_manager import run_pm +from drunc.process_manager.utils import get_pm_type_from_name, validate_k8s_session_name from drunc.unified_shell.commands import ( boot, flush, @@ -166,17 +168,16 @@ def unified_shell( #! TODO - reintroduce this before this PR is merged. # If using a k8s process manager, validate the session name before proceeding - # if get_pm_type_from_name( - # process_manager - # ) == ProcessManagerTypes.K8s and not validate_k8s_session_name(session_name): - # ctx.obj.log.error( - # f"[red]Invalid session/namespace name [bold]({session_name})[/bold][/red]. " - # "Must match RFC1123 label: lowercase alphanumeric or '-', start/end with " - # "alphanumeric, max 63 chars." - # ) - # sys.exit(1) - - # ctx.obj.log.critical("TEST") + if get_pm_type_from_name( + process_manager + ) == ProcessManagerTypes.K8s and not validate_k8s_session_name(session_name): + ctx.obj.log.error( + f"[red]Invalid session/namespace name [bold]({session_name})[/bold][/red]. " + "Must match RFC1123 label: lowercase alphanumeric or '-', start/end with " + "alphanumeric, max 63 chars." + ) + sys.exit(1) + # Setup configuration related context variables # Assume oksconflibs if no framework is defined ctx.obj.configuration_file = ( @@ -476,7 +477,6 @@ def cleanup(): ctx.obj.log.error( f"Could not retrieve the controller status, reason: {e}" ) - ctx.obj.delete_driver("controller") # Terminate any residual processes if ctx.obj.get_driver("process_manager"): From bc2ca0f11891a5f362d3b54a5f8f90a6cb84e358 Mon Sep 17 00:00:00 2001 From: PawelPlesniak Date: Tue, 30 Jun 2026 16:15:02 +0200 Subject: [PATCH 10/13] Pytest working --- src/drunc/controller/controller.py | 9 --------- tests/conftest.py | 4 ++++ tests/process_manager/conftest.py | 1 + tests/process_manager/test_process_manager_driver.py | 2 +- .../test_ssh_process_lifetime_manager_common.py | 7 +++++++ 5 files changed, 13 insertions(+), 10 deletions(-) diff --git a/src/drunc/controller/controller.py b/src/drunc/controller/controller.py index 9bda0b034..75edf24bf 100644 --- a/src/drunc/controller/controller.py +++ b/src/drunc/controller/controller.py @@ -1,4 +1,3 @@ -import multiprocessing import os import threading import time @@ -404,14 +403,6 @@ def terminate(self): except Exception as e: self.log.warning(f"Error stopping opmon publisher: {e}") - self.log.debug("Threading threads") - for t in threading.enumerate(): - self.log.debug(f"{t.name} TID: {t.native_id} is_alive: {t.is_alive}") - - with multiprocessing.Manager() as manager: - self.log.debug("Multiprocess threads") - self.log.debug(manager.list()) - def __del__(self): self.terminate() diff --git a/tests/conftest.py b/tests/conftest.py index 9ba6b9df5..e2e8705a0 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,6 +1,7 @@ import getpass import logging import os +import socket import time from pathlib import Path from subprocess import Popen @@ -103,6 +104,9 @@ def boot_session( # Prepare environment variables env = os.environ.copy() env["DUNEDAQ_SESSION"] = session_name + env["DRUNC_HOST_NAME"] = ( + socket.gethostname() + ) # This works as the process is a subprocess that does not run in a container. # Load the configuration, get the DAL configuration_consolidated_file = f"oksconflibs:{configuration_consolidated_file}" diff --git a/tests/process_manager/conftest.py b/tests/process_manager/conftest.py index 78aed2525..f4aa11dd8 100644 --- a/tests/process_manager/conftest.py +++ b/tests/process_manager/conftest.py @@ -65,6 +65,7 @@ def bootrequest(app_data): **app_data["env"], "DUNE_DAQ_BASE_RELEASE": "release1", "SPACK_RELEASES_DIR": "spack_release", + "DRUNC_HOST_NAME": socket.gethostname(), }, process_execution_directory="/pwd", process_logs_path=app_data["log_path"], diff --git a/tests/process_manager/test_process_manager_driver.py b/tests/process_manager/test_process_manager_driver.py index f200fba22..415afef0e 100644 --- a/tests/process_manager/test_process_manager_driver.py +++ b/tests/process_manager/test_process_manager_driver.py @@ -488,7 +488,7 @@ def test_connect_to_service_success(mock_client_class, mock_driver): pytest_hostname = socket.gethostname() mock_session_dal = MagicMock() - mock_session_dal.connectivity_service.host = "localhost" + mock_session_dal.connectivity_service.host = socket.gethostname() mock_session_dal.connectivity_service.service.port = 1234 result_localhost = mock_driver._connect_to_service(mock_session_dal, "session1") diff --git a/tests/processes/test_ssh_process_lifetime_manager_common.py b/tests/processes/test_ssh_process_lifetime_manager_common.py index 0a0bab2b6..ad9b86740 100644 --- a/tests/processes/test_ssh_process_lifetime_manager_common.py +++ b/tests/processes/test_ssh_process_lifetime_manager_common.py @@ -37,6 +37,8 @@ def create_boot_request(process_name, tree_id, log_file, test_file_path): """ simple_process_script = test_file_path.parent / "simple_process.py" + work_area_root = os.getenv("DBT_AREA_ROOT", "/tmp") + boot_request = BootRequest( process_description=ProcessDescription( metadata=ProcessMetadata( @@ -46,10 +48,15 @@ def create_boot_request(process_name, tree_id, log_file, test_file_path): hostname="localhost", tree_id=tree_id, ), + process_execution_directory=work_area_root, process_logs_path=log_file, ) ) + boot_request.process_description.executable_and_arguments.add( + exec="cd", args=[work_area_root] + ) + boot_request.process_description.executable_and_arguments.add( exec=f"python3 {simple_process_script}", args=[process_name] ) From a8179c7b4b47624266d5a77a17dcf39c63a41241 Mon Sep 17 00:00:00 2001 From: PawelPlesniak Date: Wed, 8 Jul 2026 12:20:15 +0200 Subject: [PATCH 11/13] Reordering the position to parse the address --- src/drunc/unified_shell/shell.py | 52 ++++++++++++++------------------ 1 file changed, 22 insertions(+), 30 deletions(-) diff --git a/src/drunc/unified_shell/shell.py b/src/drunc/unified_shell/shell.py index 8feb17331..f497ac2a6 100644 --- a/src/drunc/unified_shell/shell.py +++ b/src/drunc/unified_shell/shell.py @@ -155,29 +155,33 @@ def unified_shell( # Set up the drunc and unified_shell loggers get_root_logger(log_level) ctx.obj.log = get_logger("unified_shell", rich_handler=True) - ctx.obj.log.debug("Setting up the [green]unified_shell[/green] logger") + ctx.obj.log.info(f"[green]{getpass.getuser()} starting the unified_shell[/green]") # Parse the process manager argument to determine if it's a config or an address process_manager_url: ParseResult = urlparse(process_manager) - internal_pm: bool = True if process_manager_url.scheme == "grpc": # i.e. if it's an address internal_pm = False + ctx.obj.reset(address_pm=process_manager_url.netloc) + pm_type = ProcessManagerTypes[ + ctx.obj.get_driver("process_manager").describe().type + ] + else: + internal_pm = True + pm_type = get_pm_type_from_name(process_manager) + ctx.obj.log.debug( f"Process manager argument parsed, internal_pm set to {internal_pm}" ) - #! TODO - reintroduce this before this PR is merged. # If using a k8s process manager, validate the session name before proceeding - if get_pm_type_from_name( - process_manager - ) == ProcessManagerTypes.K8s and not validate_k8s_session_name(session_name): + if not validate_k8s_session_name(session_name) and (pm_type == ProcessManagerTypes.K8s): ctx.obj.log.error( f"[red]Invalid session/namespace name [bold]({session_name})[/bold][/red]. " "Must match RFC1123 label: lowercase alphanumeric or '-', start/end with " "alphanumeric, max 63 chars." ) sys.exit(1) - + # Setup configuration related context variables # Assume oksconflibs if no framework is defined ctx.obj.configuration_file = ( @@ -194,14 +198,9 @@ def unified_shell( session_dal = db.get_dal(class_name="Session", uid=ctx.obj.configuration_id) app_log_path = session_dal.log_path - ctx.obj.log.info( - f"[green]Setting up to use the process manager[/green] with configuration " - f"[green]{process_manager}[/green] and configuration id [green]" - f'"{configuration_id}"[/green] from [green]{ctx.obj.configuration_file}[/green]' - ) - # Establish communication with the process manager, spawning it if needed if internal_pm: # Spawn the Process Manager + ctx.obj.log.info(f"[green]Setting up the {pm_type.name} process manager[/]") ctx.obj.log.debug( f"Spawning process_manager with configuration {process_manager}" ) @@ -270,24 +269,22 @@ def unified_shell( process_manager_address = resolve_localhost_and_127_ip_to_network_ip( f"localhost:{port.value}" ) + ctx.obj.reset(address_pm=process_manager_address) + ctx.obj.log.debug( + f"[green]process_manager[/green] started at address [green]" + f"{process_manager_address}[/green]" + ) else: # Connect to an existing process manager at the provided address - ctx.obj.log.critical( - "Connecting to an existing process manager at the provided address" + ctx.obj.log.info( + f"[green]Connecting to the up the {pm_type.name} process manager[/] at " + f"address [green]{process_manager}[/]" ) process_manager_address = process_manager.replace( "grpc://", "" ) # remove the grpc scheme - ctx.obj.log.info( - f"[green]unified_shell[/green] connected to the [green]process_manager" - f"[/green] at address [green]{process_manager_address}[/green]" - ) - ctx.obj.log.debug( - f"[green]process_manager[/green] started, communicating through address [green]" - f"{process_manager_address}[/green]" - ) - ctx.obj.reset(address_pm=process_manager_address) + ctx.obj.log.info("[green]Setting up the controller interface[/green]") # Run a simple command (describe) to check the connection with the process manager try: @@ -316,12 +313,7 @@ def unified_shell( ctx.obj.pm_process.join() sys.exit(1) - # ctx.obj.log.critical("Process manager described successfully") - - ctx.obj.log.info( - f"[green]unified_shell[/green] connected to the [green]process_manager" - f"[/green] at address [green]{process_manager_address}[/green]" - ) + ctx.obj.log.debug("Communication with the process manager verified successfully") ctx.obj.get_driver("process_manager").send_msg( f"{getpass.getuser()} connected from unified shell" From 6260fa476348f95b1c286ab4cdf2fb98e6e6dd67 Mon Sep 17 00:00:00 2001 From: PawelPlesniak Date: Wed, 8 Jul 2026 12:47:19 +0200 Subject: [PATCH 12/13] The k8s-specific naming format check now does not cause the PMaaS operaiton to fail --- src/drunc/unified_shell/shell.py | 20 +++++++------------- 1 file changed, 7 insertions(+), 13 deletions(-) diff --git a/src/drunc/unified_shell/shell.py b/src/drunc/unified_shell/shell.py index f497ac2a6..e05ea6e37 100644 --- a/src/drunc/unified_shell/shell.py +++ b/src/drunc/unified_shell/shell.py @@ -155,16 +155,19 @@ def unified_shell( # Set up the drunc and unified_shell loggers get_root_logger(log_level) ctx.obj.log = get_logger("unified_shell", rich_handler=True) - ctx.obj.log.info(f"[green]{getpass.getuser()} starting the unified_shell[/green]") + ctx.obj.log.info(f"User {getpass.getuser()} [green]starting the unified_shell[/green]") # Parse the process manager argument to determine if it's a config or an address + # If the process manager is already running, connect to it. process_manager_url: ParseResult = urlparse(process_manager) if process_manager_url.scheme == "grpc": # i.e. if it's an address + ctx.obj.log.info(f"[green]Connecting to an existing process manager[/] at the address {process_manager_url.netloc}") internal_pm = False ctx.obj.reset(address_pm=process_manager_url.netloc) pm_type = ProcessManagerTypes[ ctx.obj.get_driver("process_manager").describe().type ] + ctx.obj.log.info(f"[green]Connected to the {pm_type.name} process manager[/] running at address {process_manager_url.netloc}") else: internal_pm = True pm_type = get_pm_type_from_name(process_manager) @@ -198,9 +201,9 @@ def unified_shell( session_dal = db.get_dal(class_name="Session", uid=ctx.obj.configuration_id) app_log_path = session_dal.log_path - # Establish communication with the process manager, spawning it if needed + # Start the process manager if it's an internal one if internal_pm: # Spawn the Process Manager - ctx.obj.log.info(f"[green]Setting up the {pm_type.name} process manager[/]") + ctx.obj.log.info(f"[green]Setting up the {pm_type.name} process manager[/] with configuration [green]{process_manager}[/green]") ctx.obj.log.debug( f"Spawning process_manager with configuration {process_manager}" ) @@ -275,16 +278,7 @@ def unified_shell( f"{process_manager_address}[/green]" ) - else: # Connect to an existing process manager at the provided address - ctx.obj.log.info( - f"[green]Connecting to the up the {pm_type.name} process manager[/] at " - f"address [green]{process_manager}[/]" - ) - process_manager_address = process_manager.replace( - "grpc://", "" - ) # remove the grpc scheme - - ctx.obj.log.info("[green]Setting up the controller interface[/green]") + ctx.obj.log.info("Setting up the controller interface") # Run a simple command (describe) to check the connection with the process manager try: From c9f565b91c187775ac0ed1c68e03f38c09656d1a Mon Sep 17 00:00:00 2001 From: Emir Muhammad Date: Thu, 16 Jul 2026 15:39:19 +0200 Subject: [PATCH 13/13] minor cleanup --- src/drunc/processes/ssh_process_lifetime_manager_shell.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/src/drunc/processes/ssh_process_lifetime_manager_shell.py b/src/drunc/processes/ssh_process_lifetime_manager_shell.py index d68f537eb..1d53203dc 100644 --- a/src/drunc/processes/ssh_process_lifetime_manager_shell.py +++ b/src/drunc/processes/ssh_process_lifetime_manager_shell.py @@ -1158,9 +1158,6 @@ def _execute_bootrequest_via_ssh( remote_cmd += f"cd {boot_request.process_description.process_execution_directory} ; " metadata_file = SSHProcessLifetimeManagerShell.get_metadata_file_path(uuid) - # self.log.critical( - # f"Metadata file for {uuid} will be written to {metadata_file} on remote host" - # ) tree_id = boot_request.process_description.metadata.tree_id name = boot_request.process_description.metadata.name is_controller = any(