grpc_child

drunc.controller.children_interface.grpc_child

Classes

gRCPChildConfHandler

Bases: ConfHandler

Handler for gRPC child node configuration.

gRPCChildNode(name, configuration, uri, connectivity_service, init_token=None)

Bases: ChildNode

Source code in drunc/controller/children_interface/grpc_child.py

def __init__(
    self,
    name: str,
    configuration: gRCPChildConfHandler,
    uri: str,
    connectivity_service,
    init_token: Token | None = None,
):
    super().__init__(name, ControlType.gRPC)

    self.configuration = configuration
    self.connectivity_service = connectivity_service
    self._lock = threading.Lock()
    self.init_token = init_token

    host, port = uri.split(":")
    port = int(port)

    if port == 0:
        raise DruncSetupException(
            f"Application {name} does not expose a control service in the configuration, or has not advertised itself to the application registry service, or the application registry service is not reachable."
        )

    self.uri = f"{host}:{port}"
    self._setup_connection()

Methods:

check_connection()

Probe child connectivity and retry once after reconnecting if needed.

Use the describe endpoint to check if the child is reachable. If not, the node attempts to resolve a fresh endpoint from the connectivity service, rebuild the gRPC channel, and retry the probe once.

Returns:

Type	Description
bool	True if the child is reachable either immediately or after a successful
bool	reconnection attempt, otherwise False.

Source code in drunc/controller/children_interface/grpc_child.py

def check_connection(self) -> bool:
    """Probe child connectivity and retry once after reconnecting if needed.

    Use the describe endpoint to check if the child is reachable. If not,
    the node attempts to resolve a fresh endpoint from the connectivity
    service, rebuild the gRPC channel, and retry the probe once.

    Returns:
        True if the child is reachable either immediately or after a successful
        reconnection attempt, otherwise False.
    """
    request = DescribeRequest(
        token=None,
        target="",
        execute_along_path=False,
        execute_on_all_subsequent_children_in_path=False,
    )

    try:
        self.stub.describe(request)
        return True
    except grpc.RpcError as error:
        try:
            self.handle_child_grpc_error(error)
        except ServerUnreachable:
            self.log.info(
                f"Connection to {self.name} at {self.uri} failed during connectivity check, attempting to reconnect..."
            )
            try:
                self._attempt_reconnection(lambda: self.stub.describe(request))
                return True
            except Exception as reconnect_error:
                self.log.warning(
                    f"Connection check failed for {self.name}: {reconnect_error}"
                )
                return False
        except Exception as unexpected_error:
            self.log.warning(
                f"Connection check failed for {self.name}: {unexpected_error}"
            )
            return False

    return False

handle_child_grpc_error(error)

Handle gRPC errors from sending commands to the child controller.

Parameters:

Name	Type	Description	Default
error	RpcError	The gRPC error to handle.	required

Source code in drunc/controller/children_interface/grpc_child.py

def handle_child_grpc_error(self, error: grpc.RpcError) -> NoReturn:
    """Handle gRPC errors from sending commands to the child controller.

    Args:
        error: The gRPC error to handle.
    """
    rethrow_if_unreachable_server(error)

    # RpcError is also a subclass of Call, and can be used in from_call.
    # The type stubs in types-grpcio do not reflect this, so we must cast.
    # See https://github.com/grpc/grpc/issues/10885.
    status = rpc_status.from_call(cast(grpc.Call, error))

    self.log.error(f"Error sending command to child node {self.name} at {self.uri}")

    if hasattr(status, "message"):
        self.log.error(status.message)

    if hasattr(status, "details"):
        for detail in status.details:
            if detail.Is(Stacktrace.DESCRIPTOR):
                text = "Stacktrace on remote server!\n"
                stack = unpack_any(detail, Stacktrace)
                for l in stack.text:
                    text += l + "\n"
                self.log.error(text)
            elif detail.Is(PlainText.DESCRIPTOR):
                text = unpack_any(detail, PlainText)
                self.log.error(text)

    raise error

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

Log a message on the server with the specified severity.

Parameters:

Name	Type	Description	Default
text	str	The message to log.	required
severity	str	The severity level of the log message (default: "INFO").	'INFO'
target	str	The target for the log message (default: "").	''
execute_along_path	bool	Whether to execute along the path (default: False).	False
execute_on_all_subsequent_children_in_path	bool	Whether to execute on all subsequent children in the path (default: True).	True

Returns:

Name	Type	Description
LogOnServerResponse	LogOnServerResponse	The response from the server after logging the message.

Source code in drunc/controller/children_interface/grpc_child.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,
) -> LogOnServerResponse:
    """
    Log a message on the server with the specified severity.

    Args:
        text (str): The message to log.
        severity (str): The severity level of the log message (default: "INFO").
        target (str): The target for the log message (default: "").
        execute_along_path (bool): Whether to execute along the path (default: False).
        execute_on_all_subsequent_children_in_path (bool): Whether to execute on all subsequent children in the path (default: True).

    Returns:
        LogOnServerResponse: The response from the server after logging the message.
    """
    request = LogOnServerRequest(
        token=None,
        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,
    )

    try:
        response = self.stub.log_on_server(request)
    except grpc.RpcError as e:
        try:
            self.handle_child_grpc_error(e)
        except ServerUnreachable:
            self.log.info(
                f"Connection to {self.name} at {self.uri} failed during log_on_server, attempting to reconnect..."
            )
            response = self._attempt_reconnection(
                lambda: self.stub.log_on_server(request)
            )

    return response

Functions: