K8s Process manager update (after cluster changes)

python/daqconf/set_connectivity_service_port.py

@@ -2,47 +2,67 @@
 import confmodel_dal
 from daqconf.utils import find_free_port
 
-import re
-
 def set_connectivity_service_port(oksfile, session_name, connsvc_port=0):
     """Script to set the value of the Connectivity Service port in the specified Session of the specified
-    OKS database file. If the new port is not specified, it is set to a random available port number."""
+    OKS database file. If the new port is not specified, it is set to a random available k8s NodePort."""
     db = conffwk.Configuration("oksconflibs:" + oksfile)
-    if session_name == "":
-        print(f"Error: the session name needs to be specified")
+    if not session_name:
+        print("Error: the session name needs to be specified")
         return 0
     else:
         try:
             session = db.get_dal("Session", session_name)
-        except:
-            print(f"Error could not find Session {session_name} in file {oksfile}")
+        except Exception:
+            print(f"Error: could not find Session {session_name} in file {oksfile}")
             return 0
 
-    schemafiles = [
-        "schema/confmodel/dunedaq.schema.xml"
-    ]
-    dal = conffwk.dal.module("dal", schemafiles)
-
+    k8s_min_port, k8s_max_port = 30000, 32767
     if connsvc_port == 0:
-        new_port = find_free_port()
+        new_port = find_free_port(k8s_min_port, k8s_max_port)
+        print(f"Found free Kubernetes NodePort: {new_port}")
     else:
         new_port = connsvc_port
+        if not (k8s_min_port <= new_port <= k8s_max_port):
+            print(f"Warning: Port {new_port} is outside the standard k8s NodePort range ({k8s_min_port}-{k8s_max_port}).")
 
+    # Update the Service
     if session.connectivity_service is not None:
         session.connectivity_service.service.port = new_port
         db.update_dal(session.connectivity_service.service)
+        print(f"Updated Connectivity Service '{session.connectivity_service.service.id}' to use port {new_port}")
+    else:
+        print(f"Error: Session '{session_name}' has no connectivity_service defined. Skipping Service object update.")
+        return 0
+
+    # Update the env var
+    if hasattr(session, 'environment') and session.environment is not None:
+        found_var = False
+        for item in session.environment:
+            if item.className() == 'VariableSet':
+                for var in item.contains:
+                    if var.name == "CONNECTION_PORT":
+                        var.value = str(new_port)
+                        db.update_dal(var)
+                        print(f"Updated runtime environment variable '{var.id}' to '{new_port}'")
+                        found_var = True
+                        break
+            elif item.className() == 'Variable':
+                if item.name == "CONNECTION_PORT":
+                    item.value = str(new_port)
+                    db.update_dal(item)
+                    print(f"Updated runtime environment variable '{item.id}' to '{new_port}'")
+                    found_var = True
 
-    for app in session.infrastructure_applications:
-        if app.className() == "ConnectionService":
-            index = 0
-            for clparam in app.commandline_parameters:
-                if "gunicorn" in clparam:
-                    pattern = re.compile(r'(.*0\.0\.0\.0)\:\d+(.*)')
-                    app.commandline_parameters[index] = pattern.sub(f'\\1:{new_port}\\2', clparam)
-                    #print(f"{app}")
-                    db.update_dal(app)
-                    break
-                index += 1
+            if found_var:
+                break
+
+        if not found_var:
+            print("Error: Could not find a 'CONNECTION_PORT' variable in the session's environment.")
+            return 0
+    else:
+        print("Error: Session has no 'environment' configured. Cannot update CONNECTION_PORT variable.")
+        return 0
 
     db.commit()
+    print(f"Successfully configured connectivity service port for session '{session_name}'.")
     return new_port

python/daqconf/set_rc_controller_port.py

@@ -0,0 +1,86 @@
+import conffwk
+import confmodel_dal
+from daqconf.utils import find_free_port
+import sys
+
+def set_rc_controller_port(oksfile, session_name, rc_port=0):
+    """
+    Script to set the value of the RC Controller Service port used by the specified Session
+    in the specified OKS database file. If the new port is not specified,
+    it is set to a random available k8s NodePort.
+    """
+    try:
+        db = conffwk.Configuration("oksconflibs:" + oksfile)
+    except Exception as e:
+        print(f"Error: Could not open OKS file {oksfile}. Make sure OKS_CONFLIBS_PATH is set.")
+        print(f"Details: {e}")
+        return 0
+
+    if not session_name:
+        print("Error: The session name needs to be specified")
+        return 0
+    else:
+        try:
+            session = db.get_dal("Session", session_name)
+        except Exception:
+            print(f"Error: Could not find Session '{session_name}' in file '{oksfile}'")
+            return 0
+
+    # Find a new port if one isn't specified
+    k8s_min_port, k8s_max_port = 30000, 32767
+    if rc_port == 0:
+        new_port = find_free_port(k8s_min_port, k8s_max_port)
+        print(f"Found free Kubernetes NodePort: {new_port}")
+    else:
+        new_port = rc_port
+        if not (k8s_min_port <= new_port <= k8s_max_port):
+            print(f"Warning: Port {new_port} is outside the standard k8s NodePort range ({k8s_min_port}-{k8s_max_port}).")
+
+    # Traverse from Session -> Segment -> Controller -> Service
+    service_to_update = None
+    try:
+        if not hasattr(session, 'segment') or session.segment is None:
+            print(f"Error: Session '{session_name}' has no 'segment' defined.")
+            return 0
+
+        segment = session.segment
+        if not hasattr(segment, 'controller') or segment.controller is None:
+            print(f"Error: Segment '{segment.id}' has no 'controller' defined.")
+            return 0
+
+        controller = segment.controller
+        if not hasattr(controller, 'exposes_service') or not controller.exposes_service:
+            # Check if the attribute exists AND if the list is not empty
+            print(f"Error: Controller '{controller.id}' has no 'exposes_service' defined or the list is empty.")
+            return 0
+
+        service_list = controller.exposes_service
+
+        if len(service_list) > 1:
+            print(f"Warning: Controller '{controller.id}' exposes multiple services. Only updating the first one ('{service_list[0].id}').")
+
+        service_to_update = service_list[0]
+
+        if service_to_update is None:
+            print(f"Error: Controller '{controller.id}' has a null service in its 'exposes_service' list.")
+            return 0
+
+    except Exception as e:
+        print(f"Error: Failed to navigate object graph from Session '{session_name}'.")
+        print(f"Details: {e}")
+        return 0
+
+    # Update the Service
+    if service_to_update:
+        service_to_update.port = new_port
+        db.update_dal(service_to_update)
+        print(f"Updated RC Controller Service '{service_to_update.id}' to use port {new_port}")
+
+        db.commit()
+        print(f"Successfully configured RC controller port for session '{session_name}'.")
+        return new_port
+    else:
+        # This case is mostly covered by the checks above, but serves as a fallback.
+        print(f"Error: Could not find the RC Controller Service for session '{session_name}'.")
+        return 0
+

python/daqconf/utils.py

@@ -1,6 +1,7 @@
 import glob
 import logging
 import os
+import random
 import socket
 from rich.logging import RichHandler
 
@@ -87,10 +88,27 @@ def find_oksincludes(includes:list[str], extra_dirs:list[str] = []):
 
     return [True, includefiles]
 
-def find_free_port():
-    with socket.socket() as s:
-        s.bind(("", 0))
-        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-        port = s.getsockname()[1]
-        s.close()
-        return port
+# This function returns a random available network port.  Users can optionally
+# specify a range that should be used.
+def find_free_port(min_port_num:int=0, max_port_num:int=65535):
+    # If the user didn't specify a minimum port number (or deliberately specified
+    # zero), we can simply ask the system for an available port.
+    if min_port_num == 0:
+        with socket.socket() as s:
+            s.bind(("", 0))
+            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+            port = s.getsockname()[1]
+            s.close()
+            return port
+    # If the user specified a minimum port number, use the specified range.
+    else:
+        if min_port_num < 1024:
+            min_port_num = 1024
+        while True:
+            port = random.randint(min_port_num, max_port_num)
+            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+                try:
+                    s.bind(("0.0.0.0", port))
+                    return port
+                except OSError:
+                    continue

scripts/daqconf_set_rc_controller_port

@@ -0,0 +1,30 @@
+#!/bin/env python3
+import click
+import sys
+
+# Assuming the new script is saved as daqconf/set_rc_controller_port.py
+from daqconf.set_rc_controller_port import set_rc_controller_port
+
+@click.command()
+@click.argument('oks_session_name', type=str)
+@click.argument('oksfile', type=str)
+@click.argument('rc_port', type=int, required=False, default=0)
+def daqconf_set_rc_controller_port(oks_session_name, oksfile, rc_port):
+    """Script to set the value of the RC Controller Service port used by the specified Session
+    in the specified OKS database file. If the new port is not specified, it is set to a 
+    random available k8s NodePort."""
+
+    # The set_rc_controller_port function (oksfile, session_name, rc_port)
+    # prints its own status messages and returns the new port, or 0 on failure.
+    new_port = set_rc_controller_port(oksfile, oks_session_name, rc_port)
+
+    if new_port == 0:
+        click.echo(f"Error: Failed to set RC controller port for session '{oks_session_name}'.", err=True)
+        sys.exit(1)
+    else:
+        # Print the final port to stdout for scripting (e.g., port=$(...))
+        print(new_port)
+        sys.exit(0)
+
+if __name__ == '__main__':
+    daqconf_set_rc_controller_port()