diff --git a/src/drunc/controller/controller.py b/src/drunc/controller/controller.py index 7e076fbcc..de3ac41c2 100644 --- a/src/drunc/controller/controller.py +++ b/src/drunc/controller/controller.py @@ -1,4 +1,4 @@ -import multiprocessing +import os import threading import time from concurrent.futures import ThreadPoolExecutor @@ -129,17 +129,29 @@ 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") + 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: @@ -326,6 +338,13 @@ def advertise_control_address(self, address): if not self.connectivity_service: return + 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." + ) self.log.info( f"Registering {self.name} ({address}) to the connectivity service at {self.connectivity_service.address}" ) @@ -384,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/src/drunc/controller/interface/context.py b/src/drunc/controller/interface/context.py index 8da69f38e..3468edce6 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/configuration.py b/src/drunc/process_manager/configuration.py index 5eedbaa7c..bcbefeb0d 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 @@ -110,6 +110,7 @@ def _post_process_oks(self) -> None: opmon_conf, ) else: + touch_and_chmod(opmon_conf.path) self.opmon_publisher = OpMonPublisher( conf=opmon_conf, rich_handler=True ) 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..0d2cd3c43 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,8 +12,8 @@ ) 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.utils import get_logger +from drunc.utils.shell_utils import InterruptedCommand, log_pm_cmd +from drunc.utils.utils import get_logger, resolve_context_peer @click.command("boot") @@ -43,11 +44,15 @@ def boot( override_logs: bool, ) -> None: log = get_logger("process_manager.shell") - processes = obj.get_driver("process_manager").ps(ProcessQuery(user=user)) + log_pm_cmd(obj) + 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, ) @@ -76,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( @@ -129,6 +135,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 +157,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 +178,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 +189,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 +228,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,18 +263,28 @@ 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") @@ -276,6 +326,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 +361,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..b933ffcc1 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,14 +78,18 @@ 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()} 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]" ) 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 c237b0725..fbd5d9405 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. @@ -1964,6 +1968,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.error(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 524f6b800..4bd2bf4ee 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, @@ -35,7 +37,7 @@ ProcessManagerConfHandler, ProcessManagerTypes, ) -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): @@ -44,6 +46,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 ): @@ -422,7 +426,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, @@ -478,6 +482,59 @@ 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 + + @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: @@ -538,25 +595,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..48c37b1c9 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 @@ -16,6 +17,7 @@ from druncschema.description_pb2 import Description from druncschema.process_manager_pb2 import ( BootRequest, + GenericNotificationMessage, LogLines, LogRequest, ProcessDescription, @@ -48,6 +50,7 @@ resolve_localhost_and_127_ip_to_network_ip, resolve_localhost_to_hostname, strip_non_drunc_loggers, + touch_and_chmod, ) @@ -83,7 +86,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 +131,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) @@ -109,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 @@ -158,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 @@ -255,7 +310,10 @@ 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"])) + + # this is one of the two minimal changes needed to get this working in general? name = app["name"] exe = app["type"] args = app["args"] @@ -265,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: @@ -371,19 +436,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" @@ -513,13 +588,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 24b14356e..7de746666 100644 --- a/src/drunc/process_manager/ssh_process_manager.py +++ b/src/drunc/process_manager/ssh_process_manager.py @@ -3,6 +3,7 @@ import uuid from typing import List, Optional +from druncschema.generic_pb2 import OutcomeFlag, OutcomeStatus from druncschema.process_manager_pb2 import ( BootRequest, LogLines, @@ -17,12 +18,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 ): @@ -360,6 +364,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.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] @@ -380,7 +387,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} on host {hostname}" ) # Query current process status @@ -485,13 +492,25 @@ 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, ) + 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: + self.log.info(f"{msg}; from {peer}") + except Exception as e: + self.log.error(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") diff --git a/src/drunc/process_manager/utils.py b/src/drunc/process_manager/utils.py index 8793cc8d4..f314af441 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 7d25a3ff1..1d53203dc 100644 --- a/src/drunc/processes/ssh_process_lifetime_manager_shell.py +++ b/src/drunc/processes/ssh_process_lifetime_manager_shell.py @@ -426,6 +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.debug( + f"Starting process {uuid} on {hostname} as {user} with log file {log_file}" + ) # Extract environment variables from boot request env_vars = ( @@ -439,13 +442,18 @@ 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 if cmd.endswith(";"): cmd = cmd[:-1] + self.log.debug(f"Built command for {uuid}: {cmd}: {boot_request}") + # Execute the command via SSH self._execute_bootrequest_via_ssh( uuid=uuid, @@ -886,33 +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.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"] + 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: - arguments.extend( - [ - "-o", - "LogLevel=error", - "-o", - "GlobalKnownHostsFile=/dev/null", - "-o", - "UserKnownHostsFile=/dev/null", - ] - ) + # reduce log level to avoid cluttering logs with warnings about host key + # verification + arguments.extend( + [ + "-o", + "LogLevel=error", + "-o", + "GlobalKnownHostsFile=/dev/null", + "-o", + "UserKnownHostsFile=/dev/null", + ] + ) + + self.log.debug(f"SSH arguments for {user_host}: {arguments}") return arguments @@ -1031,7 +1043,12 @@ def read_process_metadata( ) arguments.append(remote_command) + # Execute SSH command to wait for and read file (single round-trip) + self.log.debug( + f"Attempting to read metadata for {uuid} from {hostname} with timeout {timeout}s" + ) result = self.ssh(*arguments) + self.log.debug(f"Raw metadata content for {uuid} from {hostname}: {result}") json_content = str(result).strip() self.log.debug(f"Metadata content for {uuid}: {json_content!r}") @@ -1123,6 +1140,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 @@ -1160,6 +1180,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} ; " @@ -1169,6 +1190,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 f1c730126..6285ad3bf 100644 --- a/src/drunc/unified_shell/commands.py +++ b/src/drunc/unified_shell/commands.py @@ -1,14 +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 @@ -32,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( @@ -45,6 +58,7 @@ def boot( override_logs_boot = obj.override_logs else: override_logs_boot = override_logs + # 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. " @@ -124,6 +138,100 @@ 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") + 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) + + # 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 +@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") + 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) + + # 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): + @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 @@ -135,6 +243,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 87963e605..1e28ed084 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 eb8d67b5d..c939ffa33 100644 --- a/src/drunc/unified_shell/shell.py +++ b/src/drunc/unified_shell/shell.py @@ -48,17 +48,18 @@ 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.process_manager.utils import get_pm_type_from_name, validate_k8s_session_name +from drunc.unified_shell.commands import ( + boot, flush, kill, logs, ps, restart, + start_shell, 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.context import UnifiedShellMode from drunc.unified_shell.shell_utils import generate_fsm_sequence_command from drunc.utils.configuration import OKSKey @@ -154,18 +155,37 @@ 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"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) - internal_pm: bool = True 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) + + ctx.obj.log.debug( + 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( - 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 " @@ -174,7 +194,13 @@ def unified_shell( sys.exit(1) # 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 @@ -183,14 +209,11 @@ 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 + # 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[/] with configuration [green]{process_manager}[/green]" + ) ctx.obj.log.debug( f"Spawning process_manager with configuration {process_manager}" ) @@ -259,21 +282,13 @@ def unified_shell( process_manager_address = resolve_localhost_and_127_ip_to_network_ip( f"localhost:{port.value}" ) - - else: # Connect to an existing process manager at the provided address - 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.reset(address_pm=process_manager_address) + ctx.obj.log.debug( + f"[green]process_manager[/green] started at address [green]" + f"{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("Setting up the controller interface") # Run a simple command (describe) to check the connection with the process manager try: @@ -283,7 +298,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( @@ -302,21 +317,26 @@ def unified_shell( ctx.obj.pm_process.join() sys.exit(1) + 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" + ) # 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, flush, logs, restart, - ps, ] for cmd in process_manager_commands: ctx.command.add_command(cmd, format_name_for_cli(cmd.name)) @@ -452,11 +472,11 @@ 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"): - 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 ( @@ -495,6 +515,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..d24711b2c 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 @@ -363,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.""" @@ -706,6 +719,98 @@ 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. + """ + + 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 + + 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/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 6a2ba4bfd..f4aa11dd8 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"]}], @@ -63,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/interface/test_commands.py b/tests/process_manager/interface/test_commands.py index 3571690e2..f1a5c5e4b 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)) @@ -230,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 ) @@ -260,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/process_manager/process_manager_mock_impls.py b/tests/process_manager/process_manager_mock_impls.py index 831ac2c38..2ab0773b4 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, @@ -100,3 +101,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) 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] ) 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()