Skip to content

utils

drunc.controller.utils

Classes

ControllerMonitoringMetrics(run_type='', trigger_rate=0.0, run_number=0, disable_data_storage=False, run_time_at_start=0, run_time_since_start=0) dataclass

Store the metrics that the OpMon Controller publishes

Functions:

count_processes_in_status_response(response)

Count the number of processes in the status table, including all children.

This function is recursive to allow for the counting of processes following a nested structure of StatusResponse objects through the child attribute.

Parameters:

Name Type Description Default
response StatusResponse

The StatusResponse object returrned from a controller servicer status request.

 required

Returns:

Name Type Description
int int

The total number of processes in the status table.
Source code in drunc/controller/utils.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def count_processes_in_status_response(response: StatusResponse) -> int:
    """
    Count the number of processes in the status table, including all children.

    This function is recursive to allow for the counting of processes following a nested
    structure of StatusResponse objects through the `child` attribute.

    Args:
        response (StatusResponse): The StatusResponse object returrned from a controller
            servicer status request.

    Returns:
        int: The total number of processes in the status table.

    Raises:
        None
    """
    processes_found = 0

    # 1. Count the processes in the current node
    if response.status:
        processes_found += 1

    for child in response.children:
        processes_found += count_processes_in_status_response(child)

    return processes_found

get_all_apps_with_named_substate(response, substate_query)

Recursively searches for app names with a specific substate in StatusResponse and its children.

Source code in drunc/controller/utils.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def get_all_apps_with_named_substate(
    response: StatusResponse, substate_query: str
) -> list[str]:
    """
    Recursively searches for app names with a specific substate in StatusResponse and its children.
    """
    matching_apps = []

    if response.status and response.status.sub_state == substate_query:
        if response.name:
            matching_apps.append(response.name)

    for child in response.children:
        matching_apps.extend(get_all_apps_with_named_substate(child, substate_query))

    return matching_apps

get_all_states(response)

Recursively extracts 'state' from StatusResponse and its children.

Source code in drunc/controller/utils.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def get_all_states(response: StatusResponse) -> list[str]:
    """
    Recursively extracts 'state' from StatusResponse and its children.
    """
    states = []

    # 1. Get the state of the current node
    if response.status:
        states.append(response.status.state)

    # 2. Recurse through all children
    for child in response.children:
        states.extend(get_all_states(child))

    return states