DeviceServiceServicerImpl.py

# Copyright 2022-2023 ETSI TeraFlowSDN - TFS OSG (https://tfs.etsi.org/)
#
# 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.

from typing import Dict
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, DeviceDriverEnum, DeviceId, DeviceOperationalStatusEnum, Empty, Link)
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 .ErrorMessages import ERROR_MISSING_DRIVER, ERROR_MISSING_KPI
from .Tools import (
    check_connect_rules, check_no_endpoints, compute_rules_to_add_delete, configure_rules, deconfigure_rules,
    get_device_controller_uuid, populate_config_rules, populate_endpoint_monitoring_resources, populate_endpoints,
    populate_initial_config_rules, subscribe_kpi, unsubscribe_kpi, update_endpoints)

LOGGER = logging.getLogger(__name__)

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

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_uuid = request.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 blank one to get UUID, and populate it below
            device = Device()
            device.device_id.CopyFrom(request.device_id)            # pylint: disable=no-member
            device.name = request.name
            device.device_type = request.device_type
            device.device_operational_status = DeviceOperationalStatusEnum.DEVICEOPERATIONALSTATUS_UNDEFINED
            device.device_drivers.extend(request.device_drivers)    # pylint: disable=no-member
            device.device_config.CopyFrom(request.device_config)    # pylint: disable=no-member
            device_id = context_client.SetDevice(device)
            device = get_device(context_client, device_id.device_uuid.uuid, rw_copy=True)

        # update device_uuid to honor UUID provided by Context
        device_uuid = device.device_id.device_uuid.uuid

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

            errors = []

            # Sub-devices and sub-links are exposed by intermediate controllers or represent mgmt links.
            # They are used to assist in path computation algorithms, and/or to identify dependencies
            # (which controller is in charge of which sub-device).
            new_sub_devices : Dict[str, Device] = dict()
            new_sub_links : Dict[str, Link] = dict()

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

            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))

            # TODO: populate components

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

            device.device_operational_status = DeviceOperationalStatusEnum.DEVICEOPERATIONALSTATUS_DISABLED
            device_id = context_client.SetDevice(device)

            for sub_device in new_sub_devices.values():
                context_client.SetDevice(sub_device)

            for sub_links in new_sub_links.values():
                context_client.SetLink(sub_links)

            # Update endpoint monitoring resources with UUIDs
            device_with_uuids = get_device(
                context_client, device_id.device_uuid.uuid, rw_copy=False, include_endpoints=True,
                include_components=False, include_config_rules=False)
            populate_endpoint_monitoring_resources(device_with_uuids, self.monitoring_loops)

            context_client.close()
            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, include_endpoints=False, include_components=False,
                include_config_rules=True)
            if device is None:
                raise NotFoundException('Device', device_uuid, extra_details='loading in ConfigureDevice')

            device_controller_uuid = get_device_controller_uuid(device)
            if device_controller_uuid is not None:
                device = get_device(
                    context_client, device_controller_uuid, rw_copy=True, include_endpoints=False,
                    include_components=False, include_config_rules=True)
                if device is None:
                    raise NotFoundException(
                        'Device', device_controller_uuid, extra_details='loading in ConfigureDevice')

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

            if DeviceDriverEnum.DEVICEDRIVER_P4 in device.device_drivers:
                device = get_device(
                    context_client, device_uuid, rw_copy=False, include_endpoints=True, include_components=False,
                    include_config_rules=True)
                # P4 Driver, by now, has no means to retrieve endpoints
                # We allow defining the endpoints manually
                update_endpoints(request, device)

                # Update endpoints to get UUIDs
                device_id = context_client.SetDevice(device)
                device = context_client.GetDevice(device_id)

            if request.device_operational_status != DeviceOperationalStatusEnum.DEVICEOPERATIONALSTATUS_UNDEFINED:
                device.device_operational_status = request.device_operational_status

            # 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)

            # Context Performance+Scalability enhancement:
            # This method, besides P4 logic, does not add/update/delete endpoints.
            # Remove endpoints to reduce number of inserts done by Context.
            # TODO: Add logic to inspect endpoints and keep only those ones modified with respect to Context.
            del device.device_endpoints[:]

            # Note: Rules are updated by configure_rules() and deconfigure_rules() methods.
            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()
            device = get_device(
                context_client, device_uuid, rw_copy=False, include_endpoints=False, include_config_rules=False,
                include_components=False)
            if device is None:
                raise NotFoundException('Device', device_uuid, extra_details='loading in DeleteDevice')
            device_uuid = device.device_id.device_uuid.uuid

            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:
            context_client = ContextClient()
            device = get_device(
                context_client, device_uuid, rw_copy=False, include_endpoints=False, include_components=False,
                include_config_rules=True)
            if device is None:
                raise NotFoundException('Device', device_uuid, extra_details='loading in DeleteDevice')

            driver : _Driver = get_driver(self.driver_instance_cache, device)
            if driver is None:
                msg = ERROR_MISSING_DRIVER.format(device_uuid=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:
        subscribe = (request.sampling_duration_s > 0.0) and (request.sampling_interval_s > 0.0)
        manage_kpi_method = subscribe_kpi if subscribe else unsubscribe_kpi

        if subscribe:
            device_uuid = request.kpi_descriptor.device_id.device_uuid.uuid
        else:
            # unsubscribe only carries kpi_uuid; take device_uuid from recorded KPIs
            kpi_uuid = request.kpi_id.kpi_id.uuid
            kpi_details = self.monitoring_loops.get_kpi_by_uuid(kpi_uuid)
            if kpi_details is None:
                msg = ERROR_MISSING_KPI.format(kpi_uuid=str(kpi_uuid))
                raise OperationFailedException('MonitorDeviceKpi', extra_details=msg)
            device_uuid = kpi_details[0]

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

            driver : _Driver = get_driver(self.driver_instance_cache, device)
            if driver is None:
                msg = ERROR_MISSING_DRIVER.format(device_uuid=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)