DeviceServiceServicerImpl.py

from typing import List, Tuple
import grpc, logging
from prometheus_client import Counter, Histogram
from common.Checkers import chk_options, chk_string
from common.database.api.context.Constants import DEFAULT_CONTEXT_ID, DEFAULT_TOPOLOGY_ID
from common.database.api.Database import Database
from common.database.api.context.topology.device.OperationalStatus import OperationalStatus, \
    operationalstatus_enum_values, to_operationalstatus_enum
from common.exceptions.ServiceException import ServiceException
from device.proto.context_pb2 import DeviceId, Device, Empty
from device.proto.device_pb2_grpc import DeviceServiceServicer
from device.service.Tools import check_device_id_request, check_device_request

LOGGER = logging.getLogger(__name__)

ADDDEVICE_COUNTER_STARTED    = Counter  ('device_adddevice_counter_started',
                                          'Device:AddDevice counter of requests started'  )
ADDDEVICE_COUNTER_COMPLETED  = Counter  ('device_adddevice_counter_completed',
                                          'Device:AddDevice counter of requests completed')
ADDDEVICE_COUNTER_FAILED     = Counter  ('device_adddevice_counter_failed',
                                          'Device:AddDevice counter of requests failed'   )
ADDDEVICE_HISTOGRAM_DURATION = Histogram('device_adddevice_histogram_duration',
                                          'Device:AddDevice histogram of request duration')

CONFIGUREDEVICE_COUNTER_STARTED    = Counter  ('device_configuredevice_counter_started',
                                               'Device:ConfigureDevice counter of requests started'  )
CONFIGUREDEVICE_COUNTER_COMPLETED  = Counter  ('device_configuredevice_counter_completed',
                                               'Device:ConfigureDevice counter of requests completed')
CONFIGUREDEVICE_COUNTER_FAILED     = Counter  ('device_configuredevice_counter_failed',
                                               'Device:ConfigureDevice counter of requests failed'   )
CONFIGUREDEVICE_HISTOGRAM_DURATION = Histogram('device_configuredevice_histogram_duration',
                                               'Device:ConfigureDevice histogram of request duration')

DELETEDEVICE_COUNTER_STARTED    = Counter  ('device_deletedevice_counter_started',
                                            'Device:DeleteDevice counter of requests started'  )
DELETEDEVICE_COUNTER_COMPLETED  = Counter  ('device_deletedevice_counter_completed',
                                            'Device:DeleteDevice counter of requests completed')
DELETEDEVICE_COUNTER_FAILED     = Counter  ('device_deletedevice_counter_failed',
                                            'Device:DeleteDevice counter of requests failed'   )
DELETEDEVICE_HISTOGRAM_DURATION = Histogram('device_deletedevice_histogram_duration',
                                            'Device:DeleteDevice histogram of request duration')

class DeviceServiceServicerImpl(DeviceServiceServicer):
    def __init__(self, database : Database):
        LOGGER.debug('Creating Servicer...')
        self.database = database
        LOGGER.debug('Servicer Created')

    @ADDDEVICE_HISTOGRAM_DURATION.time()
    def AddDevice(self, request : Device, grpc_context : grpc.ServicerContext) -> DeviceId:
        ADDDEVICE_COUNTER_STARTED.inc()
        try:
            LOGGER.debug('AddDevice request: {}'.format(str(request)))

            # ----- Validate request data and pre-conditions -----------------------------------------------------------
            device_id, device_type, device_config, device_opstat, db_endpoints_ports = \
                check_device_request('AddDevice', request, self.database, LOGGER)

            # ----- Implement changes in the database ------------------------------------------------------------------
            db_context = self.database.context(DEFAULT_CONTEXT_ID).create()
            db_topology = db_context.topology(DEFAULT_TOPOLOGY_ID).create()
            db_device = db_topology.device(device_id).create(device_type, device_config, device_opstat)
            for db_endpoint,port_type in db_endpoints_ports:
                db_endpoint.create(port_type)

            # ----- Compose reply --------------------------------------------------------------------------------------
            reply = DeviceId(**db_device.dump_id())
            LOGGER.debug('AddDevice reply: {}'.format(str(reply)))
            ADDDEVICE_COUNTER_COMPLETED.inc()
            return reply
        except ServiceException as e:
            LOGGER.exception('AddDevice exception')
            ADDDEVICE_COUNTER_FAILED.inc()
            grpc_context.abort(e.code, e.details)
        except Exception as e:                                      # pragma: no cover
            LOGGER.exception('AddDevice exception')
            ADDDEVICE_COUNTER_FAILED.inc()
            grpc_context.abort(grpc.StatusCode.INTERNAL, str(e))

    @CONFIGUREDEVICE_HISTOGRAM_DURATION.time()
    def ConfigureDevice(self, request : Device, grpc_context : grpc.ServicerContext) -> DeviceId:
        CONFIGUREDEVICE_COUNTER_STARTED.inc()
        try:
            LOGGER.debug('ConfigureDevice request: {}'.format(str(request)))

            # ----- Validate request data and pre-conditions -----------------------------------------------------------
            try:
                device_id     = chk_string ('device.device_id.device_id.uuid',
                                            request.device_id.device_id.uuid,
                                            allow_empty=False)
                device_type   = chk_string ('device.device_type',
                                            request.device_type,
                                            allow_empty=True)
                device_config = chk_string ('device.device_config.device_config',
                                            request.device_config.device_config,
                                            allow_empty=True)
                device_opstat = chk_options('device.devOperationalStatus',
                                            request.devOperationalStatus,
                                            operationalstatus_enum_values())
            except Exception as e:
                LOGGER.exception('Invalid arguments:')
                raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, str(e))

            device_opstat = to_operationalstatus_enum(device_opstat)
            # should not happen because gRPC limits accepted values in enums
            if device_opstat is None:                                           # pragma: no cover
                msg = 'Unsupported OperationalStatus({}).'                      # pragma: no cover
                msg = msg.format(request.devOperationalStatus)                  # pragma: no cover
                raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, msg)   # pragma: no cover

            db_context = self.database.context(DEFAULT_CONTEXT_ID).create()
            db_topology = db_context.topology(DEFAULT_TOPOLOGY_ID).create()

            if not db_topology.devices.contains(device_id):
                msg = 'Device({}) does not exist in the database.'
                msg = msg.format(device_id)
                raise ServiceException(grpc.StatusCode.NOT_FOUND, msg)

            db_device = db_topology.device(device_id)
            db_device_attributes = db_device.attributes.get(attributes=['device_type'])
            # should not happen, device creation through Database API ensures all fields are always present
            if len(db_device_attributes) == 0:                                                  # pragma: no cover
                msg = 'Attribute device_type for Device({}) does not exist in the database.'    # pragma: no cover
                msg = msg.format(device_id)                                                     # pragma: no cover
                raise ServiceException(grpc.StatusCode.FAILED_PRECONDITION, msg)                # pragma: no cover

            db_device_type = db_device_attributes.get('device_type')
            # should not happen, device creation through Database API ensures all fields are always present
            if len(db_device_type) == 0:                                                # pragma: no cover
                msg = 'Attribute device_type for Device({}) is empty in the database.'  # pragma: no cover
                msg = msg.format(device_id)                                             # pragma: no cover
                raise ServiceException(grpc.StatusCode.FAILED_PRECONDITION, msg)        # pragma: no cover

            if db_device_type != device_type:
                msg = 'Device({}) has Type({}) in the database. Cannot be changed to Type({}).'
                msg = msg.format(device_id, db_device_type, device_type)
                raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, msg)

            if len(request.endpointList) > 0:
                msg = 'Endpoints belonging to Device({}) cannot be modified.'
                msg = msg.format(device_id)
                raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, msg)

            update_attributes = {}

            if len(device_config) > 0:
                update_attributes['device_config'] = device_config
            
            if device_opstat != OperationalStatus.KEEP_STATE:
                update_attributes['device_operational_status'] = device_opstat

            if len(update_attributes) == 0:
                msg = ' '.join([
                    'Any change has been requested for Device({}).',
                    'Either specify a new configuration or a new device operational status.',
                ])
                msg = msg.format(device_id)
                raise ServiceException(grpc.StatusCode.ABORTED, msg)

            # ----- Implement changes in the database ------------------------------------------------------------------
            db_device.update(update_attributes=update_attributes)

            # ----- Compose reply --------------------------------------------------------------------------------------
            reply = DeviceId(**db_device.dump_id())
            LOGGER.debug('ConfigureDevice reply: {}'.format(str(reply)))
            CONFIGUREDEVICE_COUNTER_COMPLETED.inc()
            return reply
        except ServiceException as e:
            LOGGER.exception('ConfigureDevice exception')
            CONFIGUREDEVICE_COUNTER_FAILED.inc()
            grpc_context.abort(e.code, e.details)
        except Exception as e:                                      # pragma: no cover
            LOGGER.exception('ConfigureDevice exception')
            CONFIGUREDEVICE_COUNTER_FAILED.inc()
            grpc_context.abort(grpc.StatusCode.INTERNAL, str(e))

    @DELETEDEVICE_HISTOGRAM_DURATION.time()
    def DeleteDevice(self, request : DeviceId, grpc_context : grpc.ServicerContext) -> Empty:
        DELETEDEVICE_COUNTER_STARTED.inc()
        try:
            LOGGER.debug('DeleteDevice request: {}'.format(str(request)))

            # ----- Validate request data and pre-conditions -----------------------------------------------------------
            device_id = check_device_id_request('DeleteDevice', request, self.database, LOGGER)

            # ----- Implement changes in the database ------------------------------------------------------------------
            db_context = self.database.context(DEFAULT_CONTEXT_ID).create()
            db_topology = db_context.topology(DEFAULT_TOPOLOGY_ID).create()
            db_topology.device(device_id).delete()

            # ----- Compose reply --------------------------------------------------------------------------------------
            reply = Empty()
            LOGGER.debug('DeleteDevice reply: {}'.format(str(reply)))
            DELETEDEVICE_COUNTER_COMPLETED.inc()
            return reply
        except ServiceException as e:
            LOGGER.exception('DeleteDevice exception')
            DELETEDEVICE_COUNTER_FAILED.inc()
            grpc_context.abort(e.code, e.details)
        except Exception as e:                                      # pragma: no cover
            LOGGER.exception('DeleteDevice exception')
            DELETEDEVICE_COUNTER_FAILED.inc()
            grpc_context.abort(grpc.StatusCode.INTERNAL, str(e))