DeviceServiceServicerImpl.py

# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import grpc, logging
from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method
from common.method_wrappers.ServiceExceptions import NotFoundException, OperationFailedException
from common.proto.context_pb2 import Device, DeviceConfig, DeviceId, Empty
from common.proto.device_pb2 import MonitoringSettings
from common.proto.device_pb2_grpc import DeviceServiceServicer
from common.tools.context_queries.Device import get_device
from common.tools.mutex_queues.MutexQueues import MutexQueues
from context.client.ContextClient import ContextClient
from .driver_api._Driver import _Driver
from .driver_api.DriverInstanceCache import DriverInstanceCache, get_driver
from .monitoring.MonitoringLoops import MonitoringLoops
from .Tools import (
    check_connect_rules, check_no_endpoints, compute_rules_to_add_delete, configure_rules, deconfigure_rules,
    populate_config_rules, populate_endpoints, populate_initial_config_rules, subscribe_kpi, unsubscribe_kpi)

LOGGER = logging.getLogger(__name__)

METRICS_POOL = MetricsPool('Device', 'RPC')

ERROR_MISSING_DRIVER = 'Device({:s}) has not been added to this Device instance'

class DeviceServiceServicerImpl(DeviceServiceServicer):
    def __init__(self, driver_instance_cache : DriverInstanceCache, monitoring_loops : MonitoringLoops) -> None:
        LOGGER.debug('Creating Servicer...')
        self.driver_instance_cache = driver_instance_cache
        self.monitoring_loops = monitoring_loops
        self.mutex_queues = MutexQueues()
        LOGGER.debug('Servicer Created')

    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
    def AddDevice(self, request : Device, context : grpc.ServicerContext) -> DeviceId:
        device_id = request.device_id
        device_uuid = device_id.device_uuid.uuid

        connection_config_rules = check_connect_rules(request.device_config)
        check_no_endpoints(request.device_endpoints)

        context_client = ContextClient()
        device = get_device(context_client, device_uuid, rw_copy=True)
        if device is None:
            # not in context, create from request
            device = Device()
            device.CopyFrom(request)

        self.mutex_queues.wait_my_turn(device_uuid)
        try:
            driver : _Driver = get_driver(self.driver_instance_cache, device)

            errors = []

            if len(device.device_endpoints) == 0:
                # created from request, populate endpoints using driver
                errors.extend(populate_endpoints(device, driver, self.monitoring_loops))

            if len(device.device_config.config_rules) == len(connection_config_rules):
                # created from request, populate config rules using driver
                errors.extend(populate_config_rules(device, driver))

            if len(errors) > 0:
                for error in errors: LOGGER.error(error)
                raise OperationFailedException('AddDevice', extra_details=errors)

            device_id = context_client.SetDevice(device)
            return device_id
        finally:
            self.mutex_queues.signal_done(device_uuid)

    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
    def ConfigureDevice(self, request : Device, context : grpc.ServicerContext) -> DeviceId:
        device_id = request.device_id
        device_uuid = device_id.device_uuid.uuid

        self.mutex_queues.wait_my_turn(device_uuid)
        try:
            context_client = ContextClient()
            device = get_device(context_client, device_uuid, rw_copy=True)
            if device is None:
                raise NotFoundException('Device', device_uuid, extra_details='loading in ConfigureDevice')

            driver : _Driver = self.driver_instance_cache.get(device_uuid)
            if driver is None:
                msg = ERROR_MISSING_DRIVER.format(str(device_uuid))
                raise OperationFailedException('ConfigureDevice', extra_details=msg)

            # TODO: use of datastores (might be virtual ones) to enable rollbacks
            resources_to_set, resources_to_delete = compute_rules_to_add_delete(device, request)

            errors = []
            errors.extend(configure_rules(device, driver, resources_to_set))
            errors.extend(deconfigure_rules(device, driver, resources_to_delete))

            if len(errors) > 0:
                for error in errors: LOGGER.error(error)
                raise OperationFailedException('ConfigureDevice', extra_details=errors)

            # Rules updated by configure_rules() and deconfigure_rules() methods.
            # Code to be removed soon if not needed.
            #running_config_rules = driver.GetConfig()
            #for config_rule in running_config_rules:
            #    if isinstance(config_rule[1], Exception): continue
            #    config_rule = device.device_config.config_rules.add()
            #    config_rule.action = ConfigActionEnum.CONFIGACTION_SET
            #    config_rule.custom.resource_key = config_rule[0]
            #    config_rule.custom.resource_value = json.dumps(config_rule[1], sort_keys=True)

            device_id = context_client.SetDevice(device)
            return device_id
        finally:
            self.mutex_queues.signal_done(device_uuid)

    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
    def DeleteDevice(self, request : DeviceId, context : grpc.ServicerContext) -> Empty:
        device_uuid = request.device_uuid.uuid

        self.mutex_queues.wait_my_turn(device_uuid)
        try:
            context_client = ContextClient()
            self.monitoring_loops.remove_device(device_uuid)
            self.driver_instance_cache.delete(device_uuid)
            context_client.RemoveDevice(request)
            return Empty()
        finally:
            self.mutex_queues.signal_done(device_uuid)

    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
    def GetInitialConfig(self, request : DeviceId, context : grpc.ServicerContext) -> DeviceConfig:
        device_uuid = request.device_uuid.uuid

        self.mutex_queues.wait_my_turn(device_uuid)
        try:
            driver : _Driver = self.driver_instance_cache.get(device_uuid)
            if driver is None:
                msg = ERROR_MISSING_DRIVER.format(str(device_uuid))
                raise OperationFailedException('GetInitialConfig', extra_details=msg)

            device_config = DeviceConfig()
            errors = populate_initial_config_rules(device_uuid, device_config, driver)

            if len(errors) > 0:
                for error in errors: LOGGER.error(error)
                raise OperationFailedException('GetInitialConfig', extra_details=errors)

            return device_config
        finally:
            self.mutex_queues.signal_done(device_uuid)

    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
    def MonitorDeviceKpi(self, request : MonitoringSettings, context : grpc.ServicerContext) -> Empty:
        device_uuid = request.kpi_descriptor.device_id.device_uuid.uuid
        subscribe = (request.sampling_duration_s > 0.0) and (request.sampling_interval_s > 0.0)
        manage_kpi_method = subscribe_kpi if subscribe else unsubscribe_kpi

        self.mutex_queues.wait_my_turn(device_uuid)
        try:
            driver : _Driver = self.driver_instance_cache.get(device_uuid)
            if driver is None:
                msg = ERROR_MISSING_DRIVER.format(str(device_uuid))
                raise OperationFailedException('MonitorDeviceKpi', extra_details=msg)

            errors = manage_kpi_method(request, driver, self.monitoring_loops)
            if len(errors) > 0: raise OperationFailedException('MonitorDeviceKpi', extra_details=errors)

            return Empty()
        finally:
            self.mutex_queues.signal_done(device_uuid)