ServiceServiceServicerImpl.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.

from typing import Dict, List
import graphlib, grpc, json, logging
from common.orm.Database import Database
from common.orm.HighLevel import get_object
from common.orm.backend.Tools import key_to_str
from common.proto.context_pb2 import ConnectionId, Empty, Service, ServiceId
from common.proto.pathcomp_pb2 import PathCompRequest
from common.proto.service_pb2_grpc import ServiceServiceServicer
from common.rpc_method_wrapper.Decorator import create_metrics, safe_and_metered_rpc_method
from common.rpc_method_wrapper.ServiceExceptions import AlreadyExistsException, InvalidArgumentException, NotFoundException, NotImplementedException
from common.tools.grpc.Tools import grpc_message_to_json, grpc_message_to_json_string
from common.tools.object_factory.Connection import json_connection_id
from common.tools.object_factory.Service import json_service_id
from context.client.ContextClient import ContextClient
from device.client.DeviceClient import DeviceClient
from pathcomp.frontend.client.PathCompClient import PathCompClient
from service.service.DependencyResolver import ObjectType, resolve_dependencies
from service.service.database.DeviceModel import DeviceModel
from .database.DatabaseServiceTools import (
    sync_service_from_context, sync_service_to_context, update_service_in_local_database)
from .database.ServiceModel import ServiceModel
from .path_computation_element.PathComputationElement import PathComputationElement, dump_connectivity
from .service_handler_api.ServiceHandlerFactory import ServiceHandlerFactory
from .Tools import delete_service, get_connection, get_service, sync_devices_from_context, update_service

LOGGER = logging.getLogger(__name__)

SERVICE_NAME = 'Service'
METHOD_NAMES = ['CreateService', 'UpdateService', 'DeleteService']
METRICS = create_metrics(SERVICE_NAME, METHOD_NAMES)

class ServiceServiceServicerImpl(ServiceServiceServicer):
    def __init__(self, database : Database, service_handler_factory : ServiceHandlerFactory) -> None:
        LOGGER.debug('Creating Servicer...')
        self.database = database
        self.service_handler_factory = service_handler_factory
        LOGGER.debug('Servicer Created')

    @safe_and_metered_rpc_method(METRICS, LOGGER)
    def CreateService(self, request : Service, context : grpc.ServicerContext) -> ServiceId:
        LOGGER.info('[CreateService] begin ; request = {:s}'.format(grpc_message_to_json_string(request)))

        service_id = request.service_id
        service_uuid = service_id.service_uuid.uuid
        service_context_uuid = service_id.context_id.context_uuid.uuid

        if len(request.service_endpoint_ids) > 0:
            unexpected_endpoints = []
            for service_endpoint_id in request.service_endpoint_ids:
                unexpected_endpoints.append(grpc_message_to_json(service_endpoint_id))
            str_unexpected_endpoints = json.dumps(unexpected_endpoints, sort_keys=True)
            raise InvalidArgumentException(
                'service.service_endpoint_ids', str_unexpected_endpoints,
                extra_details='RPC method CreateService does not accept Endpoints. '\
                              'Endpoints should be configured after creating the service.')

        if len(request.service_constraints) > 0:
            unexpected_constraints = []
            for service_constraint in request.service_constraints:
                unexpected_constraints.append(grpc_message_to_json(service_constraint))
            str_unexpected_constraints = json.dumps(unexpected_constraints, sort_keys=True)
            raise InvalidArgumentException(
                'service.service_constraints', str_unexpected_constraints,
                extra_details='RPC method CreateService does not accept Constraints. '\
                              'Constraints should be configured after creating the service.')

        if len(request.service_config.config_rules) > 0:
            unexpected_config_rules = grpc_message_to_json(request.service_config)
            unexpected_config_rules = unexpected_config_rules['config_rules']
            str_unexpected_config_rules = json.dumps(unexpected_config_rules, sort_keys=True)
            raise InvalidArgumentException(
                'service.service_config.config_rules', str_unexpected_config_rules,
                extra_details='RPC method CreateService does not accept Config Rules. '\
                              'Config Rules should be configured after creating the service.')

        # check that service does not exist
        context_client = ContextClient()
        current_service = get_service(context_client, request.service_id)
        if current_service is not None:
            context_uuid = request.service_id.context_id.context_uuid.uuid
            service_uuid = request.service_id.service_uuid.uuid
            raise AlreadyExistsException(
                'Service', service_uuid, extra_details='context_uuid={:s}'.format(str(context_uuid)))

        # just create the service in the database to lock the service_id
        # update will perform changes on the resources
        service_id = context_client.SetService(request)
        return service_id

    @safe_and_metered_rpc_method(METRICS, LOGGER)
    def UpdateService(self, request : Service, context : grpc.ServicerContext) -> ServiceId:
        LOGGER.info('[UpdateService] begin ; request = {:s}'.format(grpc_message_to_json_string(request)))

        service_id = request.service_id
        service_uuid = service_id.service_uuid.uuid
        service_context_uuid = service_id.context_id.context_uuid.uuid

        pathcomp_request = PathCompRequest()
        pathcomp_request.services.append(request)
        pathcomp_request.services.k_shortest_path.k_inspection = 5
        pathcomp_request.services.k_shortest_path.k_return = 5

        pathcomp = PathCompClient()
        pathcomp_response = pathcomp.Compute(pathcomp_request)

        # convert from a unordered lists of services and connections to a list of ordered items
        # that fulfill interdependencies among them. E.g., a service cannot be created if connections
        # supporting that service still does not exist.
        resolution = resolve_dependencies(pathcomp_response)

        # implement changes
        context_client = ContextClient()
        device_client = DeviceClient()
        for (obj_type, obj_key), (grpc_objid, grpc_obj) in resolution:
            if grpc_obj is None:
                # check if the resource already exists
                if obj_type == ObjectType.CONNECTION.value:
                    connection = get_connection(context_client, grpc_objid)
                    if connection is None: raise NotFoundException('Connection', obj_key)
                elif obj_type == ObjectType.SERVICE.value:
                    service = get_service(context_client, grpc_objid)
                    if service is None: raise NotFoundException('Service', obj_key)
                else:
                    MSG_EXTRA_DETAILS = 'obj_type={:s} obj_key={:s} grpc_objid={:s} grpc_obj={:s}'
                    str_grpc_obj = 'None' if grpc_obj is None else grpc_message_to_json_string(grpc_obj)
                    str_grpc_objid = 'None' if grpc_objid is None else grpc_message_to_json_string(grpc_objid)
                    msg_extra_details = MSG_EXTRA_DETAILS.format(obj_type, obj_key, str_grpc_objid, str_grpc_obj)
                    raise NotImplementedException('Empty Dependency', extra_details=msg_extra_details)
            else:
                # create/update the resource
                if obj_type == ObjectType.CONNECTION.value:
                    update_connection(context_client, device_client, self.service_handler_factory, grpc_obj)
                    context_client.SetConnection(grpc_obj)
                elif obj_type == ObjectType.SERVICE.value:
                    update_service(context_client, device_client, self.service_handler_factory, grpc_obj)
                    context_client.SetService(grpc_obj)
                else:
                    MSG_EXTRA_DETAILS = 'obj_type={:s} obj_key={:s} grpc_objid={:s} grpc_obj={:s}'
                    str_grpc_obj = 'None' if grpc_obj is None else grpc_message_to_json_string(grpc_obj)
                    str_grpc_objid = 'None' if grpc_objid is None else grpc_message_to_json_string(grpc_objid)
                    msg_extra_details = MSG_EXTRA_DETAILS.format(obj_type, obj_key, str_grpc_objid, str_grpc_obj)
                    raise NotImplementedException('Specified Dependency', extra_details=msg_extra_details)

        return request.service_id

    @safe_and_metered_rpc_method(METRICS, LOGGER)
    def DeleteService(self, request : ServiceId, context : grpc.ServicerContext) -> Empty:
        LOGGER.info('[DeleteService] begin ; request = {:s}'.format(grpc_message_to_json_string(request)))

        pce = PathComputationElement()
        pce.load_topology(self.context_client)
        pce.load_connectivity(self.context_client, request)
        #pce.dump_topology_to_file('../data/topo.dot')
        #pce.dump_connectivity_to_file('../data/conn-before.txt')
        connectivity = pce.get_connectivity_from_service_id(request)
        if connectivity is None: return Empty()
        #pce.dump_connectivity_to_file('../data/conn-after.txt')

        LOGGER.info('[DeleteService] connectivity = {:s}'.format(str(dump_connectivity(connectivity))))

        for connection in connectivity.get('connections'):
            delete_service(
                self.database, self.context_client, self.device_client, self.service_handler_factory,
                request, connection)

        for sub_service, sub_connections in connectivity.get('requirements', []):
            for sub_connection in sub_connections:
                delete_service(
                    self.database, self.context_client, self.device_client, self.service_handler_factory,
                    sub_service.service_id, sub_connection)

        return Empty()