Skip to content

context

drunc.unified_shell.context

Classes

UnifiedShellContext()

Bases: ShellContext

Source code in drunc/unified_shell/context.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def __init__(self):
    self.log = None
    self.status_receiver_pm = None
    self.status_receiver_controller = None
    self.took_control = False
    self.pm_process = None
    self.address_pm = ""
    self.address_controller = ""
    self.configuration_file = ""
    self.configuration_id = ""
    self.session_name = ""
    self.override_logs = True
    self.running_mode = UnifiedShellMode.INTERACTIVE
    self.batch_commands: list(str) = []
    super(UnifiedShellContext, self).__init__()
Methods:
get_endpoint_display_host_overrides()

Return a mapping of process name -> preferred display hostname for endpoint rendering in the UI.

These values are cosmetic only. The controller's advertised endpoint remains the authoritative connect address.

Returns:

Type Description
dict[str, str]

dict[str, str]: Mapping from process name to preferred display hostname.
Source code in drunc/unified_shell/context.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def get_endpoint_display_host_overrides(self) -> dict[str, str]:
    """
    Return a mapping of process name -> preferred display hostname for endpoint
    rendering in the UI.

    These values are cosmetic only. The controller's advertised endpoint remains
    the authoritative connect address.

    Returns:
        dict[str, str]: Mapping from process name to preferred display hostname.
    """
    # The PM driver may not be registered if the user connected directly to a
    # controller without going through the process manager (e.g. standalone boot).
    # In that case hostname overrides are unavailable and we fall back to
    # get_hostname_smart in the endpoint rendering path.
    pm_driver = self.get_driver("process_manager", quiet_fail=True)
    if not pm_driver:
        return {}

    if not self.session_name:
        raise RuntimeError("session name must be set before querying process list")
    query = ProcessQuery(names=[".*"], session=self.session_name)
    try:
        proc_list = pm_driver.ps(query)
    except (ServerUnreachable, ServerTimeout, grpc.RpcError):
        return {}

    overrides: dict[str, str] = {}

    for proc in proc_list.values:
        metadata = proc.process_description.metadata
        proc_name = getattr(metadata, "name", "")
        host_name = getattr(metadata, "hostname", "")

        if not proc_name or not host_name:
            continue

        overrides[proc_name] = host_name

    return overrides