utils

drunc.process_manager.utils

Classes

Functions:

compute_role_from_boot_request(boot_request)

Determine the process role from a BootRequest.

Extracts tree_id from the process metadata and checks executable_and_arguments for the drunc-controller executable, then delegates to ProcessMetadata.compute_role_from_tree_id.

Parameters:

Name	Type	Description	Default
boot_request	BootRequest	The BootRequest describing the process to launch.	required

Returns:

Type	Description
str	Role string: "root-controller", "segment-controller", "application", "infrastructure-applications", or "unknown".

Source code in drunc/process_manager/utils.py

def compute_role_from_boot_request(boot_request: BootRequest) -> str:
    """
    Determine the process role from a BootRequest.

    Extracts tree_id from the process metadata and checks
    executable_and_arguments for the drunc-controller executable,
    then delegates to ProcessMetadata.compute_role_from_tree_id.

    Args:
        boot_request: The BootRequest describing the process to launch.

    Returns:
        Role string: "root-controller", "segment-controller", "application",
                     "infrastructure-applications", or "unknown".
    """
    tree_id = boot_request.process_description.metadata.tree_id
    is_controller = any(
        e.exec == "drunc-controller"
        for e in boot_request.process_description.executable_and_arguments
    )
    return ProcessMetadata.compute_role_from_tree_id(
        tree_id, is_controller=is_controller
    )

format_hostname(hostname)

Format the host name to truly reflect what the host name is, removing any extensions that do not reflect the true host alias.

Parameters:

Name	Type	Description	Default
hostname	str	The hostname to format.	required

Returns:

Name	Type	Description
str	str	The formatted hostname.

Raises:

Type	Description
DruncCommandException	If the hostname is empty or None.

Example

If the input hostname is "np02-srv-005-1", the output will be "np02-srv-005".

Source code in drunc/process_manager/utils.py

def format_hostname(hostname: str) -> str:
    """
    Format the host name to truly reflect what the host name is, removing any extensions
    that do not reflect the true host alias.

    Args:
        hostname (str): The hostname to format.

    Returns:
        str: The formatted hostname.

    Raises:
        DruncCommandException: If the hostname is empty or None.

    Example:
        If the input hostname is "np02-srv-005-1", the output will be "np02-srv-005".
    """
    # Validate that the hostname is not empty or None
    if not hostname:
        raise DruncCommandException("Hostname cannot be empty or None.")

    # Make a copy of the hostname to modify
    formatted_hostname = hostname

    # Strip common suffixes that do not reflect the true host alias
    if hostname.endswith("-1"):
        formatted_hostname = hostname[:-2]

    return formatted_hostname

get_pm_type_from_name(pm_name)

Get the ProcessManagerTypes enum value from a string name.

Parameters:

Name	Type	Description	Default
pm_name	str	The name of the process manager type (e.g., "SSH", "K8s").	required

Returns:

Name	Type	Description
ProcessManagerTypes	ProcessManagerTypes	The corresponding enum value.

Source code in drunc/process_manager/utils.py

def get_pm_type_from_name(pm_name: str) -> ProcessManagerTypes:
    """
    Get the ProcessManagerTypes enum value from a string name.

    Args:
        pm_name (str): The name of the process manager type (e.g., "SSH", "K8s").

    Returns:
        ProcessManagerTypes: The corresponding enum value.
    """
    pm_conf_file = get_process_manager_configuration(pm_name)

    conf_path, conf_type = parse_conf_url(pm_conf_file)
    pmch = ProcessManagerConfHandler(
        log_path="./", type=conf_type, data=conf_path.split(":")[1]
    )

    return pmch.data.type

on_parent_exit(signum)

Return a function to be run in a child process which will trigger SIGNAME to be sent when the parent process dies

Source code in drunc/process_manager/utils.py

def on_parent_exit(signum):
    """Return a function to be run in a child process which will trigger
    SIGNAME to be sent when the parent process dies
    """

    def set_parent_exit_signal():
        from ctypes import cdll

        # http://linux.die.net/man/2/prctl
        result = cdll["libc.so.6"].prctl(PR_SET_PDEATHSIG, signum)
        if result != 0:
            raise PrCtlError("prctl failed with error code %s" % result)

    return set_parent_exit_signal

order_process_by_name(processes)

Given a list of processes, perform a tiered order by the name

Source code in drunc/process_manager/utils.py

def order_process_by_name(processes: list[ProcessInstance]):
    """Given a list of processes, perform a tiered order by the name"""
    by_session = {}
    for process in processes:
        m = process.process_description.metadata
        by_session.setdefault(m.session, []).append(process)

    ordered = []
    for session in sorted(by_session.keys()):
        session_processes = by_session[session]
        node_by_id = {}
        children = {}
        roots = []

        for process in session_processes:
            tree_id = process.process_description.metadata.tree_id or ""
            node_by_id.setdefault(tree_id, []).append(process)

        for tree_id, processes in node_by_id.items():
            node_by_id[tree_id] = sorted(
                processes,
                key=lambda p: (
                    p.process_description.metadata.name,
                    p.uuid.uuid,
                ),
            )

        for tree_id in node_by_id.keys():
            parent_id = tree_id.rsplit(".", 1)[0] if "." in tree_id else None
            if not parent_id or parent_id not in node_by_id:
                roots.append(tree_id)
            else:
                children.setdefault(parent_id, []).append(tree_id)

        def sort_key(tree_id):
            m = node_by_id[tree_id][0].process_description.metadata
            return (m.name, tree_id)

        def walk(tree_id):
            ordered.extend(node_by_id[tree_id])
            for child_id in sorted(children.get(tree_id, []), key=sort_key):
                walk(child_id)

        for root_id in sorted(roots, key=sort_key):
            walk(root_id)

    return ordered

validate_k8s_session_name(session)

Validate that the session/namespace name is valid according to RFC1123 label standard.

Parameters:

Name	Type	Description	Default
session	str	The session/namespace name to validate.	required

Returns:

Name	Type	Description
bool	bool	True if the session name is valid, False otherwise.

Source code in drunc/process_manager/utils.py

def validate_k8s_session_name(session: str) -> bool:
    """
    Validate that the session/namespace name is valid according to RFC1123 label standard.

    Args:
        session (str): The session/namespace name to validate.

    Returns:
        bool: True if the session name is valid, False otherwise.
    """
    session_re = re.compile(r"^[a-z0-9]([-a-z0-9]{0,61}[a-z0-9])?$")
    if not session_re.match(session):
        return False
    return True