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}"
    raw_channel = grpc.insecure_channel(target_address, options=options)
    rich_interceptor = RichErrorClientInterceptor(logger=self.log)
    self.channel = grpc.intercept_channel(raw_channel, rich_interceptor)
    self.stub = ControllerStub(self.channel)
    self.token = Token()
    self.token.CopyFrom(token)

Methods:

send_log(text, severity='INFO', logger=LoggerTarget.MAIN, target='', execute_along_path=False, execute_on_all_subsequent_children_in_path=True, timeout=60)

Sends a log message to the controller and optionally its children.

Parameters:

Name	Type	Description	Default
text	str	The message to log.	required
severity	str	The log severity, such as INFO or ERROR.	'INFO'
logger	int	The server logger target.	MAIN
target	str	The target node for the message.	''
execute_along_path	bool	Whether to execute along the target path.	False
execute_on_all_subsequent_children_in_path	bool	Whether to execute on all subsequent children in the target path.	True
timeout	int | float	The gRPC request timeout in seconds.	60

Returns:

Type	Description
SendLogResponse	The response from the controller.

Raises:

Type	Description
RpcError	If the gRPC request fails.

Source code in drunc/controller/controller_driver.py

def send_log(
    self,
    text: str,
    severity: str = "INFO",
    logger: int = LoggerTarget.MAIN,
    target: str = "",
    execute_along_path: bool = False,
    execute_on_all_subsequent_children_in_path: bool = True,
    timeout: int | float = 60,
) -> SendLogResponse:
    """
    Sends a log message to the controller and optionally its children.

    Args:
        text: The message to log.
        severity: The log severity, such as ``INFO`` or ``ERROR``.
        logger: The server logger target.
        target: The target node for the message.
        execute_along_path: Whether to execute along the target path.
        execute_on_all_subsequent_children_in_path: Whether to execute on all
            subsequent children in the target path.
        timeout: The gRPC request timeout in seconds.

    Returns:
        The response from the controller.

    Raises:
        grpc.RpcError: If the gRPC request fails.
    """
    request = SendLogRequest(
        token=self.token,
        text=text,
        severity=severity,
        logger=logger,
        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)
    response: SendLogResponse = self.stub.send_log(request, timeout=timeout)
    return response

Functions: