Merge branch 'feat/interdomain' into 'develop'

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

coverage==6.3

grpcio==1.47.*

grpcio-health-checking==1.47.*

import grpc, logging

from typing import Dict, List, Set, Tuple

from common.Checkers import chk_string

from common.exceptions.ServiceException import ServiceException

from service.proto.context_pb2 import Constraint

def check_constraint(

    logger : logging.Logger, constraint_number : int, parent_name : str, constraint : Constraint,

    add_constraints : Dict[str, Dict[str, Set[str]]]) -> Tuple[str, str]:

    try:

        constraint_type  = chk_string('constraint[#{}].constraint_type'.format(constraint_number),

                                      constraint.constraint_type,

                                      allow_empty=False)

        constraint_value = chk_string('constraint[#{}].constraint_value'.format(constraint_number),

                                      constraint.constraint_value,

                                      allow_empty=False)

    except Exception as e:

        logger.exception('Invalid arguments:')

        raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, str(e))

    if constraint_type in add_constraints:

        msg = 'Duplicated ConstraintType({}) in {}.'

        msg = msg.format(constraint_type, parent_name)

        raise ServiceException(grpc.StatusCode.INVALID_ARGUMENT, msg)

    add_constraints[constraint_type] = constraint_value

    return constraint_type, constraint_value

def check_constraints(logger : logging.Logger, parent_name : str, constraints):

    add_constraints : Dict[str, str] = {}

    constraint_tuples : List[Tuple[str, str]] = []

    for constraint_number,constraint in enumerate(constraints):

        _parent_name = 'Constraint(#{}) of {}'.format(constraint_number, parent_name)

        constraint_type, constraint_value = check_constraint(

            logger, constraint_number, _parent_name, constraint, add_constraints)

        constraint_tuples.append((constraint_type, constraint_value))

    return constraint_tuples

import grpc

from common.database.api.Database import Database

from common.database.api.context.slice.Slice import Slice

from common.exceptions.ServiceException import ServiceException

def check_slice_exists(database : Database, context_id : str, slice_id : str) -> Slice:

    db_context = database.context(context_id).create()

    if db_context.slices.contains(slice_id): return db_context.slice(slice_id)

    msg = 'Context({})/Slice({}) does not exist in the database.'

    msg = msg.format(context_id, slice_id)

    raise ServiceException(grpc.StatusCode.NOT_FOUND, msg)

def check_slice_not_exists(database : Database, context_id : str, slice_id : str):

    db_context = database.context(context_id).create()

    if not db_context.slices.contains(slice_id): return

    msg = 'Context({})/Slice({}) already exists in the database.'

    msg = msg.format(context_id, slice_id)

    raise ServiceException(grpc.StatusCode.ALREADY_EXISTS, msg)

from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method

from common.tools.context_queries.Context import create_context

from common.tools.context_queries.InterDomain import (

    compute_interdomain_path, compute_traversed_domains, get_local_device_uuids, is_inter_domain, is_multi_domain)

    compute_interdomain_path, compute_traversed_domains, get_local_device_uuids, is_inter_domain)

from common.tools.context_queries.Topology import create_topology

from common.tools.grpc.Tools import grpc_message_to_json_string

from common.tools.object_factory.Topology import json_topology_id

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

USE_DLT = True

class InterdomainServiceServicerImpl(InterdomainServiceServicer):

    def __init__(self, remote_domain_clients : RemoteDomainClients):

        LOGGER.debug('Creating Servicer...')

                    config_rules=request.slice_config.config_rules)

                LOGGER.info('[loop] [local] sub_slice={:s}'.format(grpc_message_to_json_string(sub_slice)))

                sub_slice_id = slice_client.CreateSlice(sub_slice)

                if sub_slice_id != sub_slice.slice_id: # pylint: disable=no-member

                    raise Exception('Local Slice creation failed. Wrong Slice Id was returned')

            else:

                slice_uuid = request.slice_id.slice_uuid.uuid

                LOGGER.info('[loop] [remote] domain_uuid={:s} is_local_domain={:s} slice_uuid={:s}'.format(

                # create context/topology for the remote domains where we are creating slices

                create_context(context_client, domain_uuid)

                create_topology(context_client, domain_uuid, DEFAULT_TOPOLOGY_UUID)

                sub_slice = compose_slice(

                    domain_uuid, slice_uuid, endpoint_ids, constraints=request.slice_constraints,

                    config_rules=request.slice_config.config_rules, owner_uuid=slice_owner_uuid)

                LOGGER.info('[loop] [remote] sub_slice={:s}'.format(grpc_message_to_json_string(sub_slice)))

                sub_slice_id = context_client.SetSlice(sub_slice)

                if USE_DLT:

                    topology_id = TopologyId(**json_topology_id(domain_uuid))

                    dlt_record_sender.add_slice(topology_id, sub_slice)

                else:

                    interdomain_client = self.remote_domain_clients.get_peer('remote-teraflow')

                    sub_slice_reply = interdomain_client.LookUpSlice(sub_slice)

                    if sub_slice_reply == sub_slice.slice_id: # pylint: disable=no-member

                        # successful case

                        remote_sub_slice = interdomain_client.OrderSliceFromCatalog(sub_slice)

                    else:

                        # not in catalog

                        remote_sub_slice = interdomain_client.CreateSliceAndAddToCatalog(sub_slice)

                    if remote_sub_slice.slice_status.slice_status != SliceStatusEnum.SLICESTATUS_ACTIVE:

                        raise Exception('Remote Slice creation failed. Wrong Slice status returned')

            LOGGER.info('[loop] adding sub-slice')

            reply.slice_subslice_ids.add().CopyFrom(sub_slice_id)   # pylint: disable=no-member

        context_client = ContextClient()

        slice_client = SliceClient()

        reply = slice_client.CreateSlice(request)

        if reply != request.slice_id: # pylint: disable=no-member

        if reply != request.slice_id:

            raise Exception('Slice creation failed. Wrong Slice Id was returned')

        return context_client.GetSlice(request.slice_id)

# OECC/PSC'22 Paper - Interdomain slices

This functional test reproduces the experiment in paper "... paper title ..." presented at OECC/PSC'22 conference

[OECC/PSC'22](... demo link ...).

## Functional test folder

This functional test can be found in folder `./src/tests/oeccpsc22/`. A convenience alias `./oeccpsc22/` pointing to that folder has been defined.

# TO BE WRITTEN