Skip to content

ssh_process_lifetime_manager

drunc.processes.ssh_process_lifetime_manager

Abstract base class for process lifetime management.

Defines the common interface for managing remote process lifecycles, including process startup, monitoring, termination, and output capture.

Classes

ProcessLifetimeManager(disable_host_key_check=False, disable_localhost_host_key_check=False, logger=None, on_process_exit=None)

Bases: ABC

Abstract base class for process lifetime management.

Provides a common interface for starting, monitoring, and terminating processes on remote hosts via SSH. Concrete implementations may use different underlying SSH libraries (e.g., Paramiko, sh library).

Initialise process lifetime manager.

Parameters:

Name Type Description Default
disable_host_key_check bool

Disable SSH host key verification for all hosts

 False
disable_localhost_host_key_check bool

Disable SSH host key verification for localhost

 False
logger Optional[Logger]

Logger instance for real-time output logging

 None
on_process_exit Optional[Callable[[str, Optional[int], Optional[Exception]], None]]

Optional callback function(uuid, exit_code, exception) invoked when process exits

 None
Source code in drunc/processes/ssh_process_lifetime_manager.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
@abstractmethod
def __init__(
    self,
    disable_host_key_check: bool = False,
    disable_localhost_host_key_check: bool = False,
    logger: Optional[logging.Logger] = None,
    on_process_exit: Optional[
        Callable[[str, Optional[int], Optional[Exception]], None]
    ] = None,
):
    """
    Initialise process lifetime manager.

    Args:
        disable_host_key_check: Disable SSH host key verification for all hosts
        disable_localhost_host_key_check: Disable SSH host key verification for localhost
        logger: Logger instance for real-time output logging
        on_process_exit: Optional callback function(uuid, exit_code, exception) invoked when process exits
    """
    pass
Functions
cleanup_all() abstractmethod

Clean up all processes and resources.

Terminates all managed processes and releases all associated resources.

Source code in drunc/processes/ssh_process_lifetime_manager.py 
148
149
150
151
152
153
154
155
@abstractmethod
def cleanup_all(self) -> None:
    """
    Clean up all processes and resources.

    Terminates all managed processes and releases all associated resources.
    """
    pass
cleanup_process(uuid) abstractmethod

Clean up process resources.

Terminates the process (if still running) and releases all associated resources.

Parameters:

Name Type Description Default
uuid str

Process UUID to clean up

 required
Source code in drunc/processes/ssh_process_lifetime_manager.py 
136
137
138
139
140
141
142
143
144
145
146
@abstractmethod
def cleanup_process(self, uuid: str) -> None:
    """
    Clean up process resources.

    Terminates the process (if still running) and releases all associated resources.

    Args:
        uuid: Process UUID to clean up
    """
    pass
get_active_process_keys() abstractmethod

Get list of active process UUIDs.

Returns:

Type Description
List[str]

List of active process UUID strings
Source code in drunc/processes/ssh_process_lifetime_manager.py 
45
46
47
48
49
50
51
52
53
@abstractmethod
def get_active_process_keys(self) -> List[str]:
    """
    Get list of active process UUIDs.

    Returns:
        List of active process UUID strings
    """
    pass
get_exit_code(uuid) abstractmethod

Get process exit code.

Parameters:

Name Type Description Default
uuid str

Process UUID

 required

Returns:

Type Description
Optional[int]

Exit code if process has terminated, None if still running or not found
Source code in drunc/processes/ssh_process_lifetime_manager.py 
86
87
88
89
90
91
92
93
94
95
96
97
@abstractmethod
def get_exit_code(self, uuid: str) -> Optional[int]:
    """
    Get process exit code.

    Args:
        uuid: Process UUID

    Returns:
        Exit code if process has terminated, None if still running or not found
    """
    pass
get_process_stderr(uuid) abstractmethod

Get stderr from process.

Parameters:

Name Type Description Default
uuid str

Process UUID

 required

Returns:

Type Description
Optional[str]

Accumulated stderr content as string, None if not found
Source code in drunc/processes/ssh_process_lifetime_manager.py 
123
124
125
126
127
128
129
130
131
132
133
134
@abstractmethod
def get_process_stderr(self, uuid: str) -> Optional[str]:
    """
    Get stderr from process.

    Args:
        uuid: Process UUID

    Returns:
        Accumulated stderr content as string, None if not found
    """
    pass
get_process_stdout(uuid) abstractmethod

Get stdout from process.

Parameters:

Name Type Description Default
uuid str

Process UUID

 required

Returns:

Type Description
Optional[str]

Accumulated stdout content as string, None if not found
Source code in drunc/processes/ssh_process_lifetime_manager.py 
110
111
112
113
114
115
116
117
118
119
120
121
@abstractmethod
def get_process_stdout(self, uuid: str) -> Optional[str]:
    """
    Get stdout from process.

    Args:
        uuid: Process UUID

    Returns:
        Accumulated stdout content as string, None if not found
    """
    pass
is_process_alive(uuid) abstractmethod

Check if process is alive.

Parameters:

Name Type Description Default
uuid str

Process UUID to check

 required

Returns:

Type Description
bool

True if process is alive, False otherwise
Source code in drunc/processes/ssh_process_lifetime_manager.py 
73
74
75
76
77
78
79
80
81
82
83
84
@abstractmethod
def is_process_alive(self, uuid: str) -> bool:
    """
    Check if process is alive.

    Args:
        uuid: Process UUID to check

    Returns:
        True if process is alive, False otherwise
    """
    pass
read_log_file(hostname, user, log_file, num_lines=100) abstractmethod

Read remote log file via SSH.

Creates a temporary SSH connection to read the log file and returns the last N lines.

Parameters:

Name Type Description Default
hostname str

Target hostname

 required
user str

SSH username

 required
log_file str

Remote log file path

 required
num_lines int

Number of lines to read from end of file

 100

Returns:

Type Description
List[str]

List of log lines
Source code in drunc/processes/ssh_process_lifetime_manager.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
@abstractmethod
def read_log_file(
    self, hostname: str, user: str, log_file: str, num_lines: int = 100
) -> List[str]:
    """
    Read remote log file via SSH.

    Creates a temporary SSH connection to read the log file and returns
    the last N lines.

    Args:
        hostname: Target hostname
        user: SSH username
        log_file: Remote log file path
        num_lines: Number of lines to read from end of file

    Returns:
        List of log lines
    """
    pass
start_process(uuid, boot_request) abstractmethod

Start a remote process using the boot request configuration.

Extracts all necessary parameters from the boot request and executes the process on the remote host via SSH.

Parameters:

Name Type Description Default
uuid str

Unique identifier for this process

 required
boot_request BootRequest

BootRequest containing process configuration, metadata, environment variables, and execution parameters

 required

Raises:

Type Description
RuntimeError

If SSH connection or process execution fails
Source code in drunc/processes/ssh_process_lifetime_manager.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@abstractmethod
def start_process(self, uuid: str, boot_request: BootRequest) -> None:
    """
    Start a remote process using the boot request configuration.

    Extracts all necessary parameters from the boot request and executes
    the process on the remote host via SSH.

    Args:
        uuid: Unique identifier for this process
        boot_request: BootRequest containing process configuration, metadata,
                    environment variables, and execution parameters

    Raises:
        RuntimeError: If SSH connection or process execution fails
    """
    pass
terminate_process(uuid, timeout=10.0) abstractmethod

Terminate process gracefully with optional timeout.

Parameters:

Name Type Description Default
uuid str

Process UUID to terminate

 required
timeout float

Timeout for graceful termination in seconds

 10.0
Source code in drunc/processes/ssh_process_lifetime_manager.py 
 99
100
101
102
103
104
105
106
107
108
@abstractmethod
def terminate_process(self, uuid: str, timeout: float = 10.0) -> None:
    """
    Terminate process gracefully with optional timeout.

    Args:
        uuid: Process UUID to terminate
        timeout: Timeout for graceful termination in seconds
    """
    pass
validate_host_connection(host, auth_method, user) abstractmethod

Validate SSH connection to the specified host.

Attempts to establish an SSH connection to the host and execute a simple command to verify connectivity. Used to validate access before starting processes.

Parameters:

Name Type Description Default
host str

Target hostname

 required
auth_method str

Authentication method to use (implementation-specific)

 required
user str

SSH username

 required

Raises:

Type Description
RuntimeError

If SSH connection or command execution fails
Source code in drunc/processes/ssh_process_lifetime_manager.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
@abstractmethod
def validate_host_connection(
    self,
    host: str,
    auth_method: str,
    user: str,
) -> None:
    """
    Validate SSH connection to the specified host.

    Attempts to establish an SSH connection to the host and execute a
    simple command to verify connectivity. Used to validate access before
    starting processes.

    Args:
        host: Target hostname
        auth_method: Authentication method to use (implementation-specific)
        user: SSH username

    Raises:
        RuntimeError: If SSH connection or command execution fails
    """
    pass