controller_driver

drunc.controller.controller_driver

Classes

ControllerDriver(address, token)

Source code in drunc/controller/controller_driver.py

def __init__(self, address: str, token: Token):
    self.log = get_logger("controller.core.ControllerDriver")
    self.address = address
    options = [
        ("grpc.keepalive_time_ms", 60000)  # pings the server every 60 seconds
    ]

    try:
        resolved_address = self._resolve_address_to_ipv4(address)
    except ValueError as e:
        self.log.error(
            f"Failed to initialize ControllerDriver. Invalid address '{self.address}': {e}"
        )
        raise e

    target_address = f"ipv4:{resolved_address}"
    self.channel = grpc.insecure_channel(target_address, options=options)
    self.stub = ControllerStub(self.channel)
    self.token = Token()
    self.token.CopyFrom(token)

Methods:

log_on_server(text, severity='INFO', target='', execute_along_path=False, execute_on_all_subsequent_children_in_path=True, timeout=60)

Logs a message to the server's log system.

Parameters:

Name	Type	Description	Default
text	str	The message to log.	required
target	str	The target node for the log message. Defaults to "".	''
execute_along_path	bool	Whether to execute the log command along the path. Defaults to False.	False
execute_on_all_subsequent_children_in_path	bool	Whether to execute the log command on all subsequent children in the path. Defaults to True.	True
timeout	int | float	The timeout for the gRPC request in seconds. Defaults to 60.	60

Returns:

Type	Description
LogOnServerResponse	None

Raises:

Type	Description
RpcError	If the gRPC request fails.

Source code in drunc/controller/controller_driver.py

def log_on_server(
    self,
    text: str,
    severity: str = "INFO",
    target: str = "",
    execute_along_path: bool = False,
    execute_on_all_subsequent_children_in_path: bool = True,
    timeout: int | float = 60,
) -> LogOnServerResponse:
    """
    Logs a message to the server's log system.

    Args:
        text (str): The message to log.
        target (str, optional): The target node for the log message. Defaults to "".
        execute_along_path (bool, optional): Whether to execute the log command along the path. Defaults to False.
        execute_on_all_subsequent_children_in_path (bool, optional): Whether to execute the log command on all subsequent children in the path. Defaults to True.
        timeout (int | float, optional): The timeout for the gRPC request in seconds. Defaults to 60.

    Returns:
        None

    Raises:
        grpc.RpcError: If the gRPC request fails.
    """
    request = LogOnServerRequest(
        token=self.token,
        text=text,
        severity=severity,
        target=target,
        execute_along_path=execute_along_path,
        execute_on_all_subsequent_children_in_path=execute_on_all_subsequent_children_in_path,
    )
    request.token.CopyFrom(self.token)
    try:
        response = self.stub.log_on_server(request, timeout=timeout)
    except grpc.RpcError as e:
        handle_grpc_error(e)

    return response

Functions: