src/moblab_common/moblab_configuration_connector.py

# -*- coding: utf-8 -*-
# Copyright 2021 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Simple layer to abstract the grpc implementation from users."""


import logging
import grpc

from moblab_common.proto import moblab_configuration_rpc_pb2
from moblab_common.proto import moblab_configuration_rpc_pb2_grpc as grpc_stub

from moblab_common import moblabrpc_connector

from google.protobuf import empty_pb2

DEFAULT_SERVER = "moblab-rpcserver:6002"

channel = None
stub = None


class MoblabConfigurationRpcConnector(object):
    """Abstract out the grpc details of connecting the moblab rpc
    configuration server.
    """

    channel = None
    stub = None

    @classmethod
    def connect(cls):
        """Connect to the grpc service, caching the connection.

        Raises:
            MoblabRpcConnectorError: On a connection issue.
        """
        if not cls.channel or not cls.stub:
            cls.channel = grpc.insecure_channel(DEFAULT_SERVER)
            if cls.channel:
                cls.stub = grpc_stub.MoblabConfigurationServiceStub(
                    cls.channel
                )
                if not cls.stub:
                    raise moblabrpc_connector.MoblabRpcConnectorError(
                        "Unable to connect to server %s" % DEFAULT_SERVER
                    )
            else:
                raise moblabrpc_connector.MoblabRpcConnectorError(
                    "No server found %s" % DEFAULT_SERVER
                )

    @classmethod
    def disconnect(cls):
        """Invalidate the cached grpc server connection."""
        cls.channel = None
        cls.stub = None

    @classmethod
    def get_stable_build_version(cls, board):
        """Read stable build version configuration.

        Raises:
            MoblabRpcConnectorError: If rpc call failed.

        Returns: StableVersion
        """
        cls.connect()
        try:
            request = (
                moblab_configuration_rpc_pb2.GetStableBuildVersionRequest(
                    board=board
                )
            )
            response = cls.stub.get_stable_build_version(request)
            return response.stable_version if response else None
        except grpc.RpcError:
            cls.disconnect()  # Force reconnect on retry
            logging.exception("Error getting the host identifier")
            raise moblabrpc_connector.MoblabRpcConnectorError(
                "Error getting the host identifier"
            )

    @classmethod
    def get_stable_build_version_resolved(cls, board):
        """Read stable build version resolved between default and override.

        Raises:
            MoblabRpcConnectorError: If rpc call fails.

        Returns:
            Configured Build Version string
        """
        response = cls.get_stable_build_version(board)
        if response.override_version:
            return response.override_version
        return response.default_version

    @classmethod
    def get_local_results_retention_duration(cls):
        """Returns the local retention period taht is configured in the system

        Raises:
            MoblabRpcConnectorError: [description]
        """
        cls.connect()
        try:
            return cls.stub.get_local_results_retention_duration(
                empty_pb2.Empty()
            ).duration.seconds
        except grpc.RpcError:
            cls.disconnect()  # Force reconnect on retry
            logging.exception("Error getting jobs.")
            raise moblabrpc_connector.MoblabRpcConnectorError(
                "Error getting getting jobs."
            )