utils

drunc.process_manager.utils

Classes

Functions

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

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