debug messages

Author: emmuhamm

src/drunc/connectivity_service/client.py

@@ -42,7 +42,7 @@ def is_ready(self, timeout: int = 10):
             elapsed = time.time() - start
 
             self.log.debug(f"Health check attempt {attempt} at {elapsed:.2f}s elapsed")
-            self.log.debug(f"Polling address: {self.address}")
+            self.log.warning(f"Polling address: {self.address}")
             try:
                 r = get(self.address)
             except Exception as e:

src/drunc/controller/controller.py

@@ -1,4 +1,6 @@
 import multiprocessing
+import os
+import socket
 import threading
 import time
 from concurrent.futures import ThreadPoolExecutor
@@ -151,6 +153,15 @@ def __init__(self, configuration, name: str, session: str, token: Token):
 
             connection_server_new = socket.gethostname()
             connection_port_new = os.getenv("CONNECTION_PORT")
+
+            # i mean the heasiest thing to do would be to do the same, if localhost change something
+
+            log_init.critical(
+                f"{connection_server_old=}, {connection_port_old=}, {connection_server_new=}, {connection_port_new=}"
+            )
+
+            ##########
+
             connection_server = (
                 socket.gethostname()
                 if connection_server_old == "localhost"
@@ -369,6 +380,10 @@ def advertise_control_address(self, address):
         if not self.connectivity_service:
             return
 
+        self.log.info(
+            f"Checking if connectivity service exists at {self.connectivity_service.address}"
+        )
+
         if not self.connectivity_service.is_ready(10):
             raise ValueError("failssss")
         self.log.info(

src/drunc/process_manager/configuration.py

@@ -21,6 +21,7 @@
 if TYPE_CHECKING:
     import conffwk
 
+
 def touch_and_chmod(filepath, mode=0o777):
     with open(filepath, "a"):
         os.utime(filepath, None)
@@ -96,6 +97,10 @@ def _parse_dict(self, data):
         )
 
         if self.opmon_conf.path == "./info.json":
+            self.log.error(
+                f"IN JSON MAKER,{self.opmon_conf.session=}, {self.opmon_conf.application=}"
+            )
+
             self.opmon_conf.path = (
                 "./info."
                 + self.opmon_conf.session
@@ -120,6 +125,8 @@ def _parse_dict(self, data):
                 new_data.opmon_publisher = OpMonPublisher(
                     conf=self.opmon_conf, rich_handler=True
                 )
+
+                # raise ValueError("should be here!")
                 self.log.debug(
                     "%s OpMonPublisher initialized with configuration %s",
                     self.opmon_conf.opmon_type,

src/drunc/process_manager/interface/process_manager.py

@@ -34,6 +34,12 @@
 _cleanup_coroutines = []
 
 
+def touch_and_chmod(filepath, mode=0o777):
+    with open(filepath, "a"):
+        os.utime(filepath, None)
+    os.chmod(filepath, mode)
+
+
 def run_pm(
     pm_conf: str,
     pm_address: str,
@@ -65,6 +71,12 @@ def run_pm(
         log_path=log_path, type=conf_type, data=conf_path.split(":")[1]
     )
 
+    # raise ValueError("Yote")
+
+    log.warning(f"{conf_path.split(":")[1]=}, {conf_type=}, {log_path=}")
+
+    # do we want to touch things here for the conf path?
+
     log_path = get_log_path(
         user=getpass.getuser(),
         session_name=pmch.data.type.name,
@@ -73,6 +85,12 @@ def run_pm(
         app_log_path=log_path,
     )
 
+    # touch_and_chmod(log_path)
+
+    # also the log path?
+
+    log.warning(f"{conf_path=}, {conf_type=}, {log_path=}")
+
     # Logger has been added to process_manager, so everything will be logged
     add_handler(log, HandlerType.File, True, path=log_path)
 

src/drunc/process_manager/oks_parser.py

@@ -183,6 +183,7 @@ def collect_apps(
     collect_variables(controller.application_environment, rc_env)
     rc_env["DUNEDAQ_APPLICATION_NAME"] = controller.id
     host = controller.runs_on.runs_on.id
+    log.info(f"main host is {host}")
 
     tree_id_str = ".".join(map(str, tree_prefix))
     apps.append(
@@ -260,6 +261,8 @@ def collect_apps(
         )
         log.debug(f"Collecting app {app.id} with args {args}")
 
+        log.info(f"app host is {host}")
+
         data_path = get_writer_directory_path(app, log)
         if not data_path:
             log.debug(f"No data path found for app {app.id}")

src/drunc/process_manager/process_manager.py

@@ -1,3 +1,4 @@
+from pprint import pprint
 import abc
 import re
 import sys
@@ -45,6 +46,17 @@
 from drunc.utils.utils import get_logger, pid_info_str, resolve_context_peer
 
 
+def to_dict(obj):
+    if hasattr(obj, "__dict__"):
+        return {k: to_dict(v) for k, v in obj.__dict__.items()}
+    elif isinstance(obj, list):
+        return [to_dict(i) for i in obj]
+    elif isinstance(obj, dict):
+        return {k: to_dict(v) for k, v in obj.items()}
+    else:
+        return obj
+
+
 class BadQuery(DruncCommandException):
     def __init__(self, txt):
         super(BadQuery, self).__init__(txt, code_pb2.INVALID_ARGUMENT)
@@ -94,6 +106,11 @@ def __init__(
         self.opmon_publisher = getattr(
             self.configuration.get_data(), "opmon_publisher", None
         )
+
+        obj = self.configuration.get_data().opmon_uri
+
+        self.log.error("some errors")
+        self.log.error(pprint(obj))
         interval_s = getattr(self.configuration.get_data(), "interval_s", 10.0)
         self.authoriser = DummyAuthoriser(dach, SystemType.PROCESS_MANAGER)
 
@@ -296,7 +313,7 @@ def _boot_impl(self, boot_request: BootRequest) -> ProcessInstanceList:
     def boot(
         self, request: BootRequest, context: ServicerContext
     ) -> ProcessInstanceList:
-        self.log.debug(
+        self.log.warning(
             f"{self.name} booting '{request.process_description.metadata.name}' "
             f"from session '{request.process_description.metadata.session}'"
         )

src/drunc/process_manager/process_manager_driver.py

@@ -5,6 +5,7 @@
 import tempfile
 import time
 from collections.abc import Iterator
+from pprint import pprint
 from time import sleep
 from typing import Dict, List
 from urllib.parse import urlparse
@@ -58,6 +59,18 @@ def touch_and_chmod(filepath, mode=0o777):
         os.utime(filepath, None)
     os.chmod(filepath, mode)
 
+
+def to_dict(obj):
+    if hasattr(obj, "__dict__"):
+        return {k: to_dict(v) for k, v in obj.__dict__.items()}
+    elif isinstance(obj, list):
+        return [to_dict(i) for i in obj]
+    elif isinstance(obj, dict):
+        return {k: to_dict(v) for k, v in obj.items()}
+    else:
+        return obj
+
+
 class ProcessManagerDriver:
     controller_address = ""
 
@@ -145,8 +158,13 @@ def boot(
         db, session_dal = self._initialise_session(conf_file, conf_id)
 
         # Step 3 - check for port conflicts and update configuration/DAL as needed
+        # Should remap the local host here actually
         db, session_dal = self.check_port_conflicts(db, session_dal)
 
+        # step 3.5 update localhost mapping
+
+        # session_dal = self.change_localhost(session_dal)
+
         # Step 4 - connect to the connection service
         csc, connection_server, connection_port = self._connect_to_service(
             session_dal, session_name
@@ -155,7 +173,7 @@ def boot(
         # Step 5 - track boot timings per host
         last_boot_on_host_at = {}
         previous_host = None
-
+        self.log.warning("step 6")
         # Step 6: iterate over boot requests
         for request in self._convert_oks_to_boot_request(
             oks_conf=conf_file,
@@ -175,7 +193,7 @@ def boot(
                     f"Skipping connectivity service readiness check for application {request.process_description.metadata.name}"
                 )
             else:
-                self.log.debug(
+                self.log.warning(
                     f"Checking connectivity service readiness before booting application {request.process_description.metadata.name}"
                 )
                 if csc and not csc.is_ready(timeout=10):
@@ -195,14 +213,35 @@ def boot(
 
             previous_host = this_host
             last_boot_on_host_at[this_host] = time.time()
+
+            self.log.critical(request)
+
+            self.log.critical(
+                f"info name: {request.process_description.metadata.name}, {request.process_description.metadata.session}, {request.process_description.process_execution_directory}"
+            )
+
             newfile = (
                 f"{request.process_description.process_execution_directory}/info."
                 + request.process_description.metadata.session
                 + "."
                 + request.process_description.metadata.name
                 + ".json"
             )
+
+            self.log.critical(f"{newfile=}")
+
             touch_and_chmod(newfile)
+            # touch_and_chmod(request.process_description.process_logs_path)
+
+            # we should also do the same for the log file if necessary
+
+            # although the CD should keep us all safe
+
+            # touch this and make this easily accessible
+
+            # BefORE YOU TRY BOOT PARSE OUT THE CD PATH AND PING IT
+
+            # ah wait you need to do it as the ssh-ee
 
             try:
                 response = self.stub.boot(request, timeout=timeout)
@@ -220,6 +259,7 @@ def boot(
                 handle_grpc_error(e)
 
         # Step 7: discover segment root controller
+        # maybe the problem is here?
         self._discover_controller(
             session_dal, session_name, csc, connection_server, connection_port
         )
@@ -253,7 +293,8 @@ def _collect_all_apps(
 
         apps = infra_apps + apps
 
-        self.log.debug(f"{json.dumps(apps, indent=4)}")
+        # self.log.warning("json dumps")
+        # self.log.debug(f"{json.dumps(apps, indent=4)}")
 
         return apps
 
@@ -362,6 +403,8 @@ def _build_boot_request(
             process_restriction=process_restriction,
         )
         self.log.debug(f"{breq=}\n\n")
+
+        # boot_request.process_description.metadata.name
         return breq
 
     def _convert_oks_to_boot_request(
@@ -421,7 +464,7 @@ def _consolidate_config(self, session_name, conf_file: str) -> str | None:
                 )
                 return
 
-    def update_connectivity_port_dal(
+    def update_connectivity_port_dal(  # why is th9is hanging?
         self,
         env_variables: list["conffwk.dal.Variable | conffwk.dal.VariableSet"],
         new_port: int,
@@ -435,6 +478,28 @@ def update_connectivity_port_dal(
                     if item.name == "CONNECTION_PORT":
                         item.value = new_port
 
+    def change_localhost(self, session_dal):
+        self.log.critical("in session dalling")
+
+        connectivity_service_host: str = session_dal.connectivity_service.host
+        if connectivity_service_host == "localhost":
+            # resolve localhost to hostname to enable multi host
+            resolved_address = resolve_localhost_to_hostname(connectivity_service_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."
+            )
+            session_dal.connectivity_service.host = resolved_server
+
+        output = to_dict(session_dal.connectivity_service)
+
+        self.log.critical(pprint(output))
+
+        return session_dal
+
     def check_port_conflicts(
         self, db: conffwk.Configuration, session_dal: "conffwk.dal.Session"
     ) -> tuple[conffwk.Configuration, "conffwk.dal.Session"]:
@@ -500,8 +565,13 @@ def check_port_conflicts(
         # Temporarily removed to allow integration tests to pass without restructuring
         # Note - if infrastructure applications outside of the connectivity service are spawned, this will need to be adjusted.
         if session_dal.infrastructure_applications:  # Check if the own application needs to be spawned, or if an externally managed one is in use (e.g. if using ehn1 connectivity service or integration tests.)
+            #! WE NEEED TO REPLACE THE CONNECTIVITY SERVICE HOST HERE
+
             connectivity_service_host: str = session_dal.connectivity_service.host
             connectivity_service_port = session_dal.connectivity_service.service.port
+            self.log.info(
+                f"infrastructure apps: {connectivity_service_host=}, {connectivity_service_port=}"
+            )
             if not is_port_available(
                 connectivity_service_host, connectivity_service_port
             ):
@@ -608,14 +678,14 @@ def get_controller_address(session_dal, session_name):
 
             # 1: Try dynamic lookup via Connectivity Service
             if csc:
-                self.log.debug(
+                self.log.critical(
                     f"Attempting to discover controller '{top_controller_name}' via connectivity service at {connection_server}:{connection_port}"
                 )
                 try:
                     timeout = (
                         get_segment_lookup_timeout(session_dal.segment, 60) + 60
                     )  # root-controller timeout to find all its children + 60s for the root controller to start itself
-                    self.log.debug(
+                    self.log.critical(
                         f"Using a timeout of {timeout}s to find the [green]{top_controller_name}[/] on the connectivity service"
                     )
                     _, uri = get_control_type_and_uri_from_connectivity_service(
@@ -628,7 +698,7 @@ def get_controller_address(session_dal, session_name):
                     )
 
                     address = uri.replace("grpc://", "")
-                    self.log.debug(
+                    self.log.warning(
                         f"Successfully discovered controller '{top_controller_name}' via connectivity service: {address}"
                     )
                     return address