Merge pull request #941 from DUNE-DAQ/emmuhamm/fix-lcs

Add multi host and multi user support to the process manager

Author: PawelPlesniak

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}"
         )

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
                 )

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,

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}"
             )

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

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,

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(

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<transport>[^:]+):(?P<address>.+)$", peer)
     if not match:
         return peer

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"]}],

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
     )