Commit 5bc275e7 authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Service component: 

- Implemented connection_to_string helper method
- Extended TaskScheduler with method compose_service_connection_update
- Implemented basic logic for gRPC method RecomputeConnections
parent 14b361d0

src/service/service/ServiceServiceServicerImpl.py

+129 −2
Original line number Diff line number Diff line
@@ -15,7 +15,8 @@ 
import grpc, json, logging

from typing import Optional

from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method

from common.method_wrappers.ServiceExceptions import AlreadyExistsException, InvalidArgumentException

from common.method_wrappers.ServiceExceptions import (

    AlreadyExistsException, InvalidArgumentException, NotFoundException, NotImplementedException)

from common.proto.context_pb2 import Empty, Service, ServiceId, ServiceStatusEnum, ServiceTypeEnum

from common.proto.pathcomp_pb2 import PathCompRequest

from common.proto.service_pb2_grpc import ServiceServiceServicer
@@ -23,6 +24,7 @@ from common.tools.context_queries.Service import get_service_by_id 
from common.tools.grpc.Tools import grpc_message_to_json, grpc_message_to_json_string

from context.client.ContextClient import ContextClient

from pathcomp.frontend.client.PathCompClient import PathCompClient

from service.service.tools.ConnectionToString import connection_to_string

from .service_handler_api.ServiceHandlerFactory import ServiceHandlerFactory

from .task_scheduler.TaskScheduler import TasksScheduler
@@ -171,5 +173,130 @@ class ServiceServiceServicerImpl(ServiceServiceServicer): 


    @safe_and_metered_rpc_method(METRICS_POOL, LOGGER)

    def RecomputeConnections(self, request : Service, context : grpc.ServicerContext) -> Empty:

        raise NotImplementedError()

        if len(request.service_endpoint_ids) > 0:

            raise NotImplementedException('update-endpoints')



        if len(request.service_constraints) > 0:

            raise NotImplementedException('update-constraints')



        if len(request.service_config.config_rules) > 0:

            raise NotImplementedException('update-config-rules')



        context_client = ContextClient()



        updated_service : Optional[Service] = get_service_by_id(

            context_client, request.service_id, rw_copy=True,

            include_config_rules=False, include_constraints=False, include_endpoint_ids=False)



        if updated_service is None:

            raise NotFoundException('service', request.service_id.service_uuid.uuid)



        # pylint: disable=no-member

        if updated_service.service_type == ServiceTypeEnum.SERVICETYPE_UNKNOWN:

            raise InvalidArgumentException(

                'request.service_type', ServiceTypeEnum.Name(updated_service.service_type)

            )



        # Set service status to "SERVICESTATUS_UPDATING" to ensure rest of components are aware the service is

        # being modified.

        # pylint: disable=no-member

        updated_service.service_status.service_status = ServiceStatusEnum.SERVICESTATUS_UPDATING



        # Update endpoints

        # pylint: disable=no-member

        #del updated_service.service_endpoint_ids[:]

        #updated_service.service_endpoint_ids.extend(request.service_endpoint_ids)



        # Update constraints

        # pylint: disable=no-member

        #del updated_service.service_constraints[:]

        #updated_service.service_constraints.extend(request.service_constraints)



        # Update config rules

        # pylint: disable=no-member

        #del updated_service.service_config.config_rules[:]

        #updated_service.service_config.config_rules.extend(request.service_config.config_rules)



        updated_service_id_with_uuids = context_client.SetService(updated_service)



        # PathComp requires endpoints, constraints and config rules

        updated_service_with_uuids = get_service_by_id(

            context_client, updated_service_id_with_uuids, rw_copy=True,

            include_config_rules=True, include_constraints=True, include_endpoint_ids=True)



        # Get active connection

        connections = context_client.ListConnections(updated_service_id_with_uuids)

        if len(connections.connections) == 0:

            MSG = 'Service({:s}) has no connections'

            str_service_id = grpc_message_to_json_string(updated_service_id_with_uuids)

            str_extra_details = MSG.format(str_service_id)

            raise NotImplementedException('service-with-no-connections', extra_details=str_extra_details)

        if len(connections.connections) > 1:

            MSG = 'Service({:s}) has multiple ({:d}) connections({:s})'

            str_service_id = grpc_message_to_json_string(updated_service_id_with_uuids)

            num_connections = len(connections.connections)

            str_connections = grpc_message_to_json_string(connections)

            str_extra_details = MSG.format(str_service_id, num_connections, str_connections)

            raise NotImplementedException('service-with-multiple-connections', extra_details=str_extra_details)



        old_connection = connections.connections[0]

        if len(old_connection.sub_service_ids) > 0:

            MSG = 'Service({:s})/Connection({:s}) has sub-services: {:s}'

            str_service_id = grpc_message_to_json_string(updated_service_id_with_uuids)

            str_connection_id = grpc_message_to_json_string(old_connection.connection_id)

            str_connection = grpc_message_to_json_string(old_connection)

            str_extra_details = MSG.format(str_service_id, str_connection_id, str_connection)

            raise NotImplementedException('service-connection-with-subservices', extra_details=str_extra_details)



        # Find alternative connections

        # pylint: disable=no-member

        pathcomp_request = PathCompRequest()

        pathcomp_request.services.append(updated_service_with_uuids)

        pathcomp_request.k_disjoint_path.num_disjoint = 100

        LOGGER.debug('pathcomp_request={:s}'.format(grpc_message_to_json_string(pathcomp_request)))

        pathcomp = PathCompClient()

        pathcomp_reply = pathcomp.Compute(pathcomp_request)

        pathcomp.close()

        LOGGER.debug('pathcomp_reply={:s}'.format(grpc_message_to_json_string(pathcomp_reply)))



        if len(pathcomp_reply.services) == 0:

            MSG = 'KDisjointPath reported no services for Service({:s}): {:s}'

            str_service_id = grpc_message_to_json_string(updated_service_id_with_uuids)

            str_pathcomp_reply = grpc_message_to_json_string(pathcomp_reply)

            str_extra_details = MSG.format(str_service_id, str_pathcomp_reply)

            raise NotImplementedException('kdisjointpath-no-services', extra_details=str_extra_details)



        if len(pathcomp_reply.services) > 1:

            MSG = 'KDisjointPath reported subservices for Service({:s}): {:s}'

            str_service_id = grpc_message_to_json_string(updated_service_id_with_uuids)

            str_pathcomp_reply = grpc_message_to_json_string(pathcomp_reply)

            str_extra_details = MSG.format(str_service_id, str_pathcomp_reply)

            raise NotImplementedException('kdisjointpath-subservices', extra_details=str_extra_details)



        if len(pathcomp_reply.connections) == 0:

            MSG = 'KDisjointPath reported no connections for Service({:s}): {:s}'

            str_service_id = grpc_message_to_json_string(updated_service_id_with_uuids)

            str_pathcomp_reply = grpc_message_to_json_string(pathcomp_reply)

            str_extra_details = MSG.format(str_service_id, str_pathcomp_reply)

            raise NotImplementedException('kdisjointpath-no-connections', extra_details=str_extra_details)



        # compute a string representing the old connection

        str_old_connection = connection_to_string(old_connection)



        new_connection = None

        for candidate_new_connection in pathcomp_reply.connections:

            str_candidate_new_connection = connection_to_string(candidate_new_connection)

            if str_candidate_new_connection != str_old_connection:

                new_connection = candidate_new_connection

                break



        # Feed TaskScheduler with the service to update, the old connection to

        # deconfigure and the new connection to configure. It will produce a

        # schedule of tasks (an ordered list of tasks to be executed) to

        # implement the requested changes.

        tasks_scheduler = TasksScheduler(self.service_handler_factory)

        tasks_scheduler.compose_service_connection_update(

            updated_service_with_uuids, old_connection, new_connection)

        tasks_scheduler.execute_all()



        return Empty()

src/service/service/task_scheduler/TaskScheduler.py

+39 −0
Original line number Diff line number Diff line
@@ -198,6 +198,45 @@ class TasksScheduler: 
        t1 = time.time()

        LOGGER.debug('[compose_from_service] elapsed_time: {:f} sec'.format(t1-t0))



    def compose_service_connection_update(

        self, service : Service, old_connection : Connection, new_connection : Connection

    ) -> None:

        t0 = time.time()



        self._add_service_to_executor_cache(service)

        self._add_connection_to_executor_cache(old_connection)

        self._add_connection_to_executor_cache(new_connection)



        service_updating_key = self._add_task_if_not_exists(Task_ServiceSetStatus(

            self._executor, service.service_id, ServiceStatusEnum.SERVICESTATUS_UPDATING))



        old_connection_deconfigure_key = self._add_task_if_not_exists(Task_ConnectionDeconfigure(

            self._executor, old_connection.connection_id))



        new_connection_configure_key = self._add_task_if_not_exists(Task_ConnectionConfigure(

            self._executor, new_connection.connection_id))



        service_active_key = self._add_task_if_not_exists(Task_ServiceSetStatus(

            self._executor, service.service_id, ServiceStatusEnum.SERVICESTATUS_ACTIVE))



        # the old connection deconfiguration depends on service being in updating state

        self._dag.add(old_connection_deconfigure_key, service_updating_key)



        # the new connection configuration depends on service being in updating state

        self._dag.add(new_connection_configure_key, service_updating_key)



        # the new connection configuration depends on the old connection having been deconfigured

        self._dag.add(new_connection_configure_key, old_connection_deconfigure_key)



        # re-activating the service depends on the service being in updating state before

        self._dag.add(service_active_key, service_updating_key)



        # re-activating the service depends on the new connection having been configured

        self._dag.add(service_active_key, service_updating_key)



        t1 = time.time()

        LOGGER.debug('[compose_service_connection_update] elapsed_time: {:f} sec'.format(t1-t0))



    def execute_all(self, dry_run : bool = False) -> None:

        ordered_task_keys = list(self._dag.static_order())

        LOGGER.debug('[execute_all] ordered_task_keys={:s}'.format(str(ordered_task_keys)))

src/service/service/tools/ConnectionToString.py

0 → 100644
+25 −0
Original line number Diff line number Diff line 
# 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 List

from common.proto.context_pb2 import Connection



def connection_to_string(connection : Connection) -> str:

    str_device_endpoint_uuids : List[str] = list()

    for endpoint_id in connection.path_hops_endpoint_ids:

        device_uuid = endpoint_id.device_id.device_uuid.uuid

        endpoint_uuid = endpoint_id.endpoint_uuid.uuid

        device_endpoint_uuid = '{:s}:{:s}'.format(device_uuid, endpoint_uuid)

        str_device_endpoint_uuids.append(device_endpoint_uuid)

    return ','.join(str_device_endpoint_uuids)