# Copyright 2022-2025 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.
from typing import Dict, Tuple
import grpc, logging, uuid
from common.Constants import DEFAULT_CONTEXT_NAME, DEFAULT_TOPOLOGY_NAME, INTERDOMAIN_TOPOLOGY_NAME, ServiceNameEnum
from common.Settings import (
ENVVAR_SUFIX_SERVICE_HOST, ENVVAR_SUFIX_SERVICE_PORT_GRPC, find_environment_variables, get_env_var_name)
from common.proto.context_pb2 import (
AuthenticationResult, Empty, EndPointId, Slice, SliceId, SliceStatusEnum, TeraFlowController, TopologyId)
from common.proto.interdomain_pb2_grpc import InterdomainServiceServicer
from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method
from common.tools.context_queries.CheckType import endpoint_type_is_border
from common.tools.context_queries.Context import create_context
from common.tools.context_queries.Device import get_device
from common.tools.context_queries.InterDomain import (
compute_interdomain_sub_slices, get_local_device_uuids, is_inter_domain)
from common.tools.context_queries.Slice import get_slice_by_id
from common.tools.context_queries.Topology import create_topology, get_topology
from common.tools.grpc.Tools import grpc_message_to_json, grpc_message_to_json_string
from common.tools.object_factory.Context import json_context_id
from common.tools.object_factory.Device import json_device_id
from common.tools.object_factory.EndPoint import json_endpoint_id
from common.tools.object_factory.Topology import json_topology_id
from context.client.ContextClient import ContextClient
from dlt.connector.client.DltConnectorClientAsync import DltConnectorClientAsync
from pathcomp.frontend.client.PathCompClient import PathCompClient
from service.client.ServiceClient import ServiceClient
from slice.client.SliceClient import SliceClient
from .topology_abstractor.DltRecordSender import DltRecordSender
from .RemoteDomainClients import RemoteDomainClients
from .Tools import compose_slice, compute_slice_owner #, map_abstract_endpoints_to_real
LOGGER = logging.getLogger(__name__)
METRICS_POOL = MetricsPool('Interdomain', 'RPC')
class InterdomainServiceServicerImpl(InterdomainServiceServicer):
def __init__(self, remote_domain_clients : RemoteDomainClients):
LOGGER.debug('Creating Servicer...')
self.remote_domain_clients = remote_domain_clients
LOGGER.debug('Servicer Created')
@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
def RequestSlice(self, request : Slice, context : grpc.ServicerContext) -> SliceId:
context_client = ContextClient()
pathcomp_client = PathCompClient()
slice_client = SliceClient()
local_device_uuids = get_local_device_uuids(context_client)
slice_owner_uuid = request.slice_owner.owner_uuid.uuid
not_inter_domain = not is_inter_domain(context_client, request.slice_endpoint_ids)
no_slice_owner = len(slice_owner_uuid) == 0
is_local_slice_owner = slice_owner_uuid in local_device_uuids
if not_inter_domain and (no_slice_owner or is_local_slice_owner):
str_slice = grpc_message_to_json_string(request)
raise Exception('InterDomain can only handle inter-domain slice requests: {:s}'.format(str_slice))
local_slices, remote_slices = compute_interdomain_sub_slices(
context_client, pathcomp_client, request)
traversed_domain_uuids = set()
traversed_domain_uuids.update(local_slices.keys())
traversed_domain_uuids.update(remote_slices.keys())
LOGGER.debug('traversed_domain_uuids={:s}'.format(str(traversed_domain_uuids)))
slice_owner_uuid = compute_slice_owner(context_client, traversed_domain_uuids)
LOGGER.debug('slice_owner_uuid={:s}'.format(str(slice_owner_uuid)))
if slice_owner_uuid is None:
raise Exception('Unable to identify slice owner')
reply = Slice()
reply.CopyFrom(request)
env_vars = find_environment_variables([
get_env_var_name(ServiceNameEnum.DLT, ENVVAR_SUFIX_SERVICE_HOST ),
get_env_var_name(ServiceNameEnum.DLT, ENVVAR_SUFIX_SERVICE_PORT_GRPC),
])
if len(env_vars) == 2:
# DLT available
dlt_connector_client = DltConnectorClientAsync()
dlt_connector_client.connect()
else:
dlt_connector_client = None
dlt_record_sender = DltRecordSender(context_client, dlt_connector_client)
for domain_uuid, endpoint_id_groups in local_slices.items():
domain_topology = get_topology(context_client, domain_uuid)
if domain_topology is None: raise Exception('Topology({:s}) not found'.format(str(domain_uuid)))
domain_name = domain_topology.name
for endpoint_ids in endpoint_id_groups:
slice_uuid = str(uuid.uuid4())
MSG = '[loop] [local] domain_uuid={:s} slice_uuid={:s} endpoint_ids={:s}'
LOGGER.debug(MSG.format(str(domain_uuid), str(slice_uuid), str([
grpc_message_to_json(ep_id) for ep_id in endpoint_ids
])))
# local slices always in DEFAULT_CONTEXT_NAME
#context_uuid = request.slice_id.context_id.context_uuid.uuid
context_uuid = DEFAULT_CONTEXT_NAME
#endpoint_ids = map_abstract_endpoints_to_real(context_client, domain_uuid, endpoint_ids)
slice_name = '{:s}:local:{:s}'.format(request.name, domain_name)
sub_slice = compose_slice(
context_uuid, slice_uuid, endpoint_ids, slice_name=slice_name, constraints=request.slice_constraints,
config_rules=request.slice_config.config_rules)
LOGGER.debug('[loop] [local] sub_slice={:s}'.format(grpc_message_to_json_string(sub_slice)))
sub_slice_id = slice_client.CreateSlice(sub_slice)
LOGGER.debug('[loop] adding sub-slice')
reply.slice_subslice_ids.add().CopyFrom(sub_slice_id) # pylint: disable=no-member
for domain_uuid, endpoint_id_groups in remote_slices.items():
domain_topology = get_device(context_client, domain_uuid)
if domain_topology is None: raise Exception('Device({:s}) not found'.format(str(domain_uuid)))
domain_name = domain_topology.name
domain_endpoint_ids_to_names = {
endpoint.endpoint_id.endpoint_uuid.uuid : endpoint.name
for endpoint in domain_topology.device_endpoints
if endpoint_type_is_border(endpoint.endpoint_type)
}
for endpoint_ids in endpoint_id_groups:
slice_uuid = str(uuid.uuid4())
MSG = '[loop] [remote] domain_uuid={:s} slice_uuid={:s} endpoint_ids={:s}'
LOGGER.debug(MSG.format(str(domain_uuid), str(slice_uuid), str([
grpc_message_to_json(ep_id) for ep_id in endpoint_ids
])))
# create context/topology for the remote domains where we are creating slices
create_context(context_client, domain_uuid, name=domain_name)
create_topology(context_client, domain_uuid, DEFAULT_TOPOLOGY_NAME)
create_topology(context_client, domain_uuid, INTERDOMAIN_TOPOLOGY_NAME)
slice_name = '{:s}:remote:{:s}'.format(request.name, domain_name)
# convert endpoint ids to names to enable conversion to uuids on the remote domain
endpoint_ids = [
EndPointId(**json_endpoint_id(
json_device_id(domain_name),
domain_endpoint_ids_to_names[endpoint_id.endpoint_uuid.uuid],
topology_id=json_topology_id(
INTERDOMAIN_TOPOLOGY_NAME,
context_id=json_context_id(DEFAULT_CONTEXT_NAME)
)
))
for endpoint_id in endpoint_ids
]
sub_slice = compose_slice(
DEFAULT_CONTEXT_NAME, slice_uuid, endpoint_ids, slice_name=slice_name,
constraints=request.slice_constraints, config_rules=request.slice_config.config_rules,
owner_uuid=slice_owner_uuid, owner_string=domain_uuid)
LOGGER.debug('[loop] [remote] sub_slice={:s}'.format(grpc_message_to_json_string(sub_slice)))
sub_slice_id = context_client.SetSlice(sub_slice)
if dlt_connector_client is not None:
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(domain_uuid)
if interdomain_client is None:
raise Exception('InterDomain Client not found for Domain({:s})'.format(str(domain_uuid)))
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)
sub_slice.slice_status.slice_status = remote_sub_slice.slice_status.slice_status
context_client.SetSlice(sub_slice)
if remote_sub_slice.slice_status.slice_status != SliceStatusEnum.SLICESTATUS_ACTIVE:
raise Exception('Remote Slice creation failed. Wrong Slice status returned')
LOGGER.debug('[loop] adding sub-slice')
reply.slice_subslice_ids.add().CopyFrom(sub_slice_id) # pylint: disable=no-member
if dlt_connector_client is not None:
LOGGER.debug('Recording Remote Slice requests to DLT')
dlt_record_sender.commit()
LOGGER.debug('Activating interdomain slice')
reply.slice_status.slice_status = SliceStatusEnum.SLICESTATUS_ACTIVE # pylint: disable=no-member
LOGGER.debug('Updating interdomain slice')
slice_id = context_client.SetSlice(reply)
return slice_id
@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
def Authenticate(self, request : TeraFlowController, context : grpc.ServicerContext) -> AuthenticationResult:
auth_result = AuthenticationResult()
auth_result.context_id.CopyFrom(request.context_id) # pylint: disable=no-member
auth_result.authenticated = True
return auth_result
@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
def LookUpSlice(self, request : Slice, context : grpc.ServicerContext) -> SliceId:
try:
context_client = ContextClient()
slice_id = SliceId()
slice_id.CopyFrom(request.slice_id)
slice_id.context_id.context_uuid.uuid = DEFAULT_CONTEXT_NAME
slice_ = context_client.GetSlice(slice_id)
return slice_.slice_id
except grpc.RpcError:
#LOGGER.exception('Unable to get slice({:s})'.format(grpc_message_to_json_string(request.slice_id)))
return SliceId()
@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
def CreateSliceAndAddToCatalog(self, request : Slice, context : grpc.ServicerContext) -> Slice:
context_client = ContextClient()
slice_client = SliceClient()
_request = Slice()
_request.CopyFrom(request)
_request.slice_id.context_id.context_uuid.uuid = DEFAULT_CONTEXT_NAME
#admin_context = context_client.GetContext(ContextId(**json_context_id(DEFAULT_CONTEXT_NAME)))
#admin_context_uuid = admin_context.context_id.context_uuid.uuid
#admin_context_name = admin_context.name
#interdomain_topology = context_client.GetTopology(TopologyId(**json_topology_id(
# DEFAULT_TOPOLOGY_NAME, context_id=json_context_id(DEFAULT_CONTEXT_NAME)
#)))
#interdomain_topology_uuid = interdomain_topology.topology_id.topology_uuid.uuid
#interdomain_topology_name = interdomain_topology.name
devices = context_client.ListDevices(Empty())
interdomain_endpoint_map : Dict[str, Tuple[str, str, str, str]] = dict()
for device in devices.devices:
device_uuid = device.device_id.device_uuid.uuid
device_name = device.name
for endpoint in device.device_endpoints:
if not endpoint_type_is_border(endpoint.endpoint_type): continue
#endpoint_context_uuid = endpoint.endpoint_id.topology_id.context_id.context_uuid.uuid
#if endpoint_context_uuid not in {admin_context_uuid, admin_context_name}: continue
#endpoint_topology_uuid = endpoint.endpoint_id.topology_id.topology_uuid.uuid
#if endpoint_topology_uuid not in {interdomain_topology_uuid, interdomain_topology_name}: continue
endpoint_uuid = endpoint.endpoint_id.endpoint_uuid.uuid
endpoint_name = endpoint.name
interdomain_endpoint_map[endpoint_name] = (device_uuid, device_name, endpoint_uuid, endpoint_name)
LOGGER.debug('interdomain_endpoint_map={:s}'.format(str(interdomain_endpoint_map)))
# Map endpoints to local real counterparts
del _request.slice_endpoint_ids[:]
for endpoint_id in request.slice_endpoint_ids:
#endpoint_context_uuid = endpoint_id.topology_id.context_id.context_uuid.uuid
#if endpoint_context_uuid not in {admin_context_uuid, admin_context_name}:
# MSG = 'Unexpected ContextId in EndPointId({:s})'
# raise Exception(MSG.format(grpc_message_to_json_string(endpoint_id)))
#endpoint_topology_uuid = endpoint_id.topology_id.topology_uuid.uuid
#if endpoint_topology_uuid not in {admin_topology_uuid, admin_topology_name}:
# MSG = 'Unexpected TopologyId in EndPointId({:s})'
# raise Exception(MSG.format(grpc_message_to_json_string(endpoint_id)))
endpoint_uuid = endpoint_id.endpoint_uuid.uuid
real_endpoint = interdomain_endpoint_map.get(endpoint_uuid)
if real_endpoint is None:
MSG = 'Unable to map EndPointId({:s}) to real endpoint. interdomain_endpoint_map={:s}'
raise Exception(MSG.format(grpc_message_to_json_string(endpoint_id), str(interdomain_endpoint_map)))
real_device_uuid, _, real_endpoint_uuid, _ = real_endpoint
real_endpoint_id = _request.slice_endpoint_ids.add()
real_endpoint_id.topology_id.context_id.context_uuid.uuid = DEFAULT_CONTEXT_NAME
real_endpoint_id.topology_id.topology_uuid.uuid = DEFAULT_TOPOLOGY_NAME
real_endpoint_id.device_id.device_uuid.uuid = real_device_uuid
real_endpoint_id.endpoint_uuid.uuid = real_endpoint_uuid
slice_id = slice_client.CreateSlice(_request)
return context_client.GetSlice(slice_id)
@safe_and_metered_rpc_method(METRICS_POOL, LOGGER)
def DeleteSlice(self, request : SliceId, context : grpc.ServicerContext) -> Empty:
context_client = ContextClient()
try:
_slice = context_client.GetSlice(request)
except: # pylint: disable=bare-except
context_client.close()
return Empty()
_slice_rw = Slice()
_slice_rw.CopyFrom(_slice)
_slice_rw.slice_status.slice_status = SliceStatusEnum.SLICESTATUS_DEINIT # pylint: disable=no-member
context_client.SetSlice(_slice_rw)
local_device_uuids = get_local_device_uuids(context_client)
slice_owner_uuid = _slice.slice_owner.owner_uuid.uuid
not_inter_domain = not is_inter_domain(context_client, _slice.slice_endpoint_ids)
no_slice_owner = len(slice_owner_uuid) == 0
is_local_slice_owner = slice_owner_uuid in local_device_uuids
if not_inter_domain and (no_slice_owner or is_local_slice_owner):
str_slice = grpc_message_to_json_string(_slice)
raise Exception('InterDomain can only handle inter-domain slice requests: {:s}'.format(str_slice))
slice_client = SliceClient()
for subslice_id in _slice_rw.slice_subslice_ids:
sub_slice = get_slice_by_id(context_client, subslice_id, rw_copy=True)
if ':remote:' in sub_slice.name:
domain_uuid = sub_slice.slice_owner.owner_string
interdomain_client = self.remote_domain_clients.get_peer(domain_uuid)
if interdomain_client is None:
raise Exception('InterDomain Client not found for Domain({:s})'.format(str(domain_uuid)))
interdomain_client.DeleteSlice(subslice_id)
tmp_slice = Slice()
tmp_slice.slice_id.CopyFrom(_slice_rw.slice_id) # pylint: disable=no-member
slice_subslice_id = tmp_slice.slice_subslice_ids.add() # pylint: disable=no-member
slice_subslice_id.CopyFrom(subslice_id)
context_client.UnsetSlice(tmp_slice)
if ':remote:' in sub_slice.name:
context_client.RemoveSlice(subslice_id)
else:
slice_client.DeleteSlice(subslice_id)
service_client = ServiceClient()
for service_id in _slice_rw.slice_service_ids:
tmp_slice = Slice()
tmp_slice.slice_id.CopyFrom(_slice_rw.slice_id) # pylint: disable=no-member
slice_service_id = tmp_slice.slice_service_ids.add() # pylint: disable=no-member
slice_service_id.CopyFrom(service_id)
context_client.UnsetSlice(tmp_slice)
service_client.DeleteService(service_id)
context_client.RemoveSlice(request)
slice_client.close()
service_client.close()
context_client.close()
return Empty()