DeviceServiceServicerImpl.py

# Copyright 2022-2024 ETSI SDG TeraFlowSDN (TFS) (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.

import grpc, logging, os, time
from typing import Dict
from prometheus_client import Histogram
from common.Constants import ServiceNameEnum
from common.Settings import ENVVAR_SUFIX_SERVICE_HOST, get_env_var_name
from common.method_wrappers.Decorator import MetricTypeEnum, 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,
    OpticalConfig, OpticalConfigId
)
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')

METRICS_POOL_DETAILS = MetricsPool('Device', 'execution', labels={
    'driver': '', 'operation': '', 'step': '',
})

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:
        t0 = time.time()

        device_uuid = request.device_id.device_uuid.uuid

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

        t1 = time.time()

        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

            if request.HasField('controller_id'):
                controller_id = request.controller_id
                if controller_id.HasField('device_uuid'):
                    controller_device_uuid = controller_id.device_uuid.uuid
                    device.controller_id.device_uuid.uuid = controller_device_uuid

            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
        device_name = device.name

        t2 = time.time()

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

            t4 = time.time()

            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()
            
            #----- Experimental ------------
            new_optical_configs : Dict[str, OpticalConfig] = dict()

            if len(device.device_endpoints) == 0:
                t5 = time.time()
                # created from request, populate endpoints using driver
                errors.extend(populate_endpoints(
                    device, driver, self.monitoring_loops, new_sub_devices, new_sub_links,
                    new_optical_configs
                ))
                t6 = time.time()
                t_pop_endpoints = t6 - t5
            else:
                t_pop_endpoints = None

            is_optical_device = request.device_drivers[0] == DeviceDriverEnum.DEVICEDRIVER_OC
            if len(device.device_config.config_rules) == len(connection_config_rules) and not is_optical_device:
                # created from request, populate config rules using driver
                t7 = time.time()
                errors.extend(populate_config_rules(device, driver))
                t8 = time.time()
                t_pop_config_rules = t8 - t7
            else:
                t_pop_config_rules = None

            # TODO: populate components

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

            t9 = time.time()

            ztp_service_host = get_env_var_name(ServiceNameEnum.ZTP, ENVVAR_SUFIX_SERVICE_HOST)
            environment_variables = set(os.environ.keys())
            if ztp_service_host in environment_variables:
                # ZTP component is deployed; leave devices disabled. ZTP will enable them.
                device.device_operational_status = DeviceOperationalStatusEnum.DEVICEOPERATIONALSTATUS_DISABLED
            else:
                # ZTP is not deployed; assume the device is ready while onboarding and set them as enabled.
                device.device_operational_status = DeviceOperationalStatusEnum.DEVICEOPERATIONALSTATUS_ENABLED

            # temporary line
            if  is_optical_device:
                #for endpoint in request.device_endpoints:
                #    #endpoint.endpoint_id.device_id.CopyFrom(device.device_id)
                #    pass

                if 'new_optical_config' in new_optical_configs and 'opticalconfig' in new_optical_configs["new_optical_config"]:
                    context_client.SetOpticalConfig(new_optical_configs["new_optical_config"]['opticalconfig'])

            device_id = context_client.SetDevice(device)

            t10 = time.time()

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

            t11 = time.time()

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

            t12 = time.time()

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

            t13 = time.time()

            context_client.close()

            t14 = time.time()

            metrics_labels = dict(driver=driver.name, operation='add_device')

            histogram_duration : Histogram = METRICS_POOL_DETAILS.get_or_create(
                'details', MetricTypeEnum.HISTOGRAM_DURATION)
            histogram_duration.labels(step='total'              , **metrics_labels).observe(t14-t0)
            histogram_duration.labels(step='execution'          , **metrics_labels).observe(t14-t3)
            histogram_duration.labels(step='endpoint_checks'    , **metrics_labels).observe(t1-t0)
            histogram_duration.labels(step='get_device'         , **metrics_labels).observe(t2-t1)
            histogram_duration.labels(step='wait_queue'         , **metrics_labels).observe(t3-t2)
            histogram_duration.labels(step='get_driver'         , **metrics_labels).observe(t4-t3)
            histogram_duration.labels(step='set_device'         , **metrics_labels).observe(t10-t9)
            histogram_duration.labels(step='populate_monit_rsrc', **metrics_labels).observe(t13-t12)

            if t_pop_endpoints is not None:
                histogram_duration.labels(step='populate_endpoints', **metrics_labels).observe(t_pop_endpoints)

            if t_pop_config_rules is not None:
                histogram_duration.labels(step='populate_config_rules', **metrics_labels).observe(t_pop_config_rules)

            if len(new_sub_devices) > 0:
                histogram_duration.labels(step='set_sub_devices', **metrics_labels).observe(t11-t10)

            if len(new_sub_links) > 0:
                histogram_duration.labels(step='set_sub_links', **metrics_labels).observe(t12-t11)

            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:
        t0 = time.time()
        device_id = request.device_id
        device_uuid = device_id.device_uuid.uuid

        self.mutex_queues.wait_my_turn(device_uuid)
        t1 = time.time()
        try:
            context_client = ContextClient()
            t2 = time.time()
            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')

            t3 = time.time()
            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)

            ztp_service_host = get_env_var_name(ServiceNameEnum.ZTP, ENVVAR_SUFIX_SERVICE_HOST)
            environment_variables = set(os.environ.keys())
            if ztp_service_host in environment_variables:
                # ZTP component is deployed; accept status updates
                if request.device_operational_status != DeviceOperationalStatusEnum.DEVICEOPERATIONALSTATUS_UNDEFINED:
                    device.device_operational_status = request.device_operational_status
            else:
                # ZTP is not deployed; activated during AddDevice and not modified
                pass

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

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

            t7 = time.time()
            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[:]

            t8 = time.time()
            # Note: Rules are updated by configure_rules() and deconfigure_rules() methods.
            device_id = context_client.SetDevice(device)

            t9 = time.time()

            metrics_labels = dict(driver=driver.name, operation='configure_device')

            histogram_duration : Histogram = METRICS_POOL_DETAILS.get_or_create(
                'details', MetricTypeEnum.HISTOGRAM_DURATION)
            histogram_duration.labels(step='total'            , **metrics_labels).observe(t9-t0)
            histogram_duration.labels(step='wait_queue'       , **metrics_labels).observe(t1-t0)
            histogram_duration.labels(step='execution'        , **metrics_labels).observe(t9-t1)
            histogram_duration.labels(step='get_device'       , **metrics_labels).observe(t3-t2)
            histogram_duration.labels(step='split_rules'      , **metrics_labels).observe(t5-t4)
            histogram_duration.labels(step='configure_rules'  , **metrics_labels).observe(t6-t5)
            histogram_duration.labels(step='deconfigure_rules', **metrics_labels).observe(t7-t6)
            histogram_duration.labels(step='set_device'       , **metrics_labels).observe(t9-t8)

            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)