Skip to content
Snippets Groups Projects
Commit 9a71c35a authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

InterDomain component: 

- Updated to use only names instead of UUIDs.
- Other Minor fixes
parent 77c8b389
No related branches found
No related tags found
1 merge request!148Draft: Resolve "Adapt InterDomain and DLT components for Release 3.0 to automate NFV-SDN'22 experiment"
Expand all Show whitespace changes
Inline Side-by-side
src/interdomain/service/topology_abstractor/AbstractDevice.py
+ 89
− 89
View file @ 9a71c35a
......@@ -16,78 +16,81 @@ import copy, logging
from typing import Dict, Optional
from common.Constants import DEFAULT_CONTEXT_NAME, INTERDOMAIN_TOPOLOGY_NAME
from common.DeviceTypes import DeviceTypeEnum
from common.proto.context_pb2 import (
ContextId, Device, DeviceDriverEnum, DeviceId, DeviceOperationalStatusEnum, EndPoint)
from common.proto.context_pb2 import Device, DeviceDriverEnum, DeviceOperationalStatusEnum, EndPoint
from common.tools.context_queries.CheckType import (
device_type_is_datacenter, device_type_is_network, endpoint_type_is_border)
from common.tools.context_queries.Device import add_device_to_topology, get_existing_device_uuids
from common.tools.object_factory.Context import json_context_id
from common.tools.object_factory.Device import json_device, json_device_id
from common.tools.context_queries.Device import get_device
from common.tools.grpc.Tools import grpc_message_to_json
from common.tools.object_factory.Device import json_device
from context.client.ContextClient import ContextClient
from context.service.database.uuids.EndPoint import endpoint_get_uuid
from .Tools import replace_device_uuids_by_names
# Remark on UUIDs:
# TopologyAbstractor, AbstractDevice and AbstractLink are used
# to compose network reporesentations to be forwarded to remote
# instances. Constraining it to use UUIDs is pointless given
# these UUIDs, to be unique, need to be bound to the local
# context/topology UUIDs, which might be different than that for
# the remote TeraFlowSDN instances. For this very reason, we use
# the device/link/endpoint/topology/context names as UUIDs, to
# prevent UUID-related issues.
LOGGER = logging.getLogger(__name__)
class AbstractDevice:
def __init__(self, device_uuid : str, device_name : str, device_type : DeviceTypeEnum):
def __init__(self, device_name : str, device_type : DeviceTypeEnum):
self.__context_client = ContextClient()
self.__device_uuid : str = device_uuid
self.__device_name : str = device_name
self.__device_type : DeviceTypeEnum = device_type
self.__device : Optional[Device] = None
self.__device_id : Optional[DeviceId] = None
# Dict[device_uuid, Dict[endpoint_uuid, abstract EndPoint]]
# Dict[device_uuid, Dict[endpoint_name, abstract EndPoint]]
self.__device_endpoint_to_abstract : Dict[str, Dict[str, EndPoint]] = dict()
# Dict[endpoint_uuid, device_uuid]
# Dict[endpoint_name, device_name]
self.__abstract_endpoint_to_device : Dict[str, str] = dict()
def to_json(self) -> Dict:
return {
'device_uuid' : self.__device_uuid,
'device_name' : self.__device_name,
'device_type' : self.__device_type,
'device' : self.__device,
'device_id' : self.__device_id,
'device' : grpc_message_to_json(self.__device),
'device_endpoint_to_abstract' : self.__device_endpoint_to_abstract,
'abstract_endpoint_to_device' : self.__abstract_endpoint_to_device,
}
@property
def uuid(self) -> str: return self.__device_uuid
@property
def name(self) -> str: return self.__device_name
@property
def device_id(self) -> Optional[DeviceId]: return self.__device_id
@property
def device(self) -> Optional[Device]: return self.__device
def get_endpoint(self, device_uuid : str, endpoint_uuid : str) -> Optional[EndPoint]:
return self.__device_endpoint_to_abstract.get(device_uuid, {}).get(endpoint_uuid)
def get_endpoint(self, device_name : str, endpoint_name : str) -> Optional[EndPoint]:
return self.__device_endpoint_to_abstract.get(device_name, {}).get(endpoint_name)
def initialize(self) -> bool:
if self.__device is not None: return False
existing_device_uuids = get_existing_device_uuids(self.__context_client)
create_abstract_device = self.__device_uuid not in existing_device_uuids
local_interdomain_device = get_device(
self.__context_client, self.__device_name, rw_copy=True, include_endpoints=True,
include_config_rules=False, include_components=False)
create_abstract_device = local_interdomain_device is None
if create_abstract_device:
self._create_empty()
else:
self._load_existing()
is_datacenter = device_type_is_datacenter(self.__device_type)
is_network = device_type_is_network(self.__device_type)
if is_datacenter or is_network:
# Add abstract device to topologies [INTERDOMAIN_TOPOLOGY_NAME]
context_id = ContextId(**json_context_id(DEFAULT_CONTEXT_NAME))
topology_uuids = [INTERDOMAIN_TOPOLOGY_NAME]
for topology_uuid in topology_uuids:
add_device_to_topology(self.__context_client, context_id, topology_uuid, self.__device_uuid)
self._load_existing(local_interdomain_device)
#is_datacenter = device_type_is_datacenter(self.__device_type)
#is_network = device_type_is_network(self.__device_type)
#if is_datacenter or is_network:
# # Add abstract device to topologies [INTERDOMAIN_TOPOLOGY_NAME]
# admin_context_id = ContextId(**json_context_id(DEFAULT_CONTEXT_NAME))
# topology_uuids = [INTERDOMAIN_TOPOLOGY_NAME]
# for topology_uuid in topology_uuids:
# # This action is done automatically; commented out by now.
# add_device_to_topology(
# self.__context_client, admin_context_id, topology_uuid, self.__device_name)
# seems not needed; to be removed in future releases
#if is_datacenter and create_abstract_device:
......@@ -101,27 +104,25 @@ class AbstractDevice:
# if device_type_is_datacenter(device.device_type): continue
# self.update_endpoints(device)
return True
return create_abstract_device
def _create_empty(self) -> None:
device_uuid = self.__device_uuid
device_name = self.__device_name
device = Device(**json_device(
device_uuid, self.__device_type.value, DeviceOperationalStatusEnum.DEVICEOPERATIONALSTATUS_ENABLED,
name=self.__device_name, endpoints=[], config_rules=[], drivers=[DeviceDriverEnum.DEVICEDRIVER_UNDEFINED]
device_name, self.__device_type.value, DeviceOperationalStatusEnum.DEVICEOPERATIONALSTATUS_ENABLED,
name=device_name, endpoints=[], config_rules=[], drivers=[DeviceDriverEnum.DEVICEDRIVER_UNDEFINED]
))
self.__context_client.SetDevice(device)
self.__device = device
self.__device_id = self.__device.device_id
self.__device = device # Store copy with names as UUIDs
def _load_existing(self) -> None:
def _load_existing(self, local_interdomain_device : Device) -> None:
self.__device_endpoint_to_abstract = dict()
self.__abstract_endpoint_to_device = dict()
self.__device_id = DeviceId(**json_device_id(self.__device_uuid))
self.__device = self.__context_client.GetDevice(self.__device_id)
replace_device_uuids_by_names(self.__context_client, local_interdomain_device)
self.__device = local_interdomain_device
self.__device_type = self.__device.device_type
device_uuid = self.__device_id.device_uuid.uuid
device_type = self.__device_type
is_datacenter = device_type_is_datacenter(device_type)
......@@ -132,67 +133,63 @@ class AbstractDevice:
# for each endpoint in abstract device, populate internal data structures and mappings
for interdomain_endpoint in self.__device.device_endpoints:
endpoint_uuid : str = interdomain_endpoint.endpoint_id.endpoint_uuid.uuid
endpoint_name : str = interdomain_endpoint.name
if is_network:
endpoint_uuid,device_uuid = endpoint_uuid.split('@', maxsplit=1)
endpoint_name,device_name = endpoint_name.split('@', maxsplit=1)
else:
device_name = self.__device_name
self.__device_endpoint_to_abstract\
.setdefault(device_uuid, {}).setdefault(endpoint_uuid, interdomain_endpoint)
.setdefault(device_name, {}).setdefault(endpoint_name, interdomain_endpoint)
self.__abstract_endpoint_to_device\
.setdefault(endpoint_uuid, device_uuid)
def _update_endpoint_name(self, device_uuid : str, endpoint_uuid : str, endpoint_name : str) -> bool:
device_endpoint_to_abstract = self.__device_endpoint_to_abstract.get(device_uuid, {})
interdomain_endpoint = device_endpoint_to_abstract.get(endpoint_uuid)
interdomain_endpoint_name = interdomain_endpoint.name
if endpoint_name == interdomain_endpoint_name: return False
interdomain_endpoint.name = endpoint_name
return True
def _update_endpoint_type(self, device_uuid : str, endpoint_uuid : str, endpoint_type : str) -> bool:
device_endpoint_to_abstract = self.__device_endpoint_to_abstract.get(device_uuid, {})
interdomain_endpoint = device_endpoint_to_abstract.get(endpoint_uuid)
.setdefault(endpoint_name, device_name)
# This method becomes useless; the endpoint name is considered a primary key; cannot be updated
#def _update_endpoint_name(self, device_uuid : str, endpoint_uuid : str, endpoint_name : str) -> bool:
# device_endpoint_to_abstract = self.__device_endpoint_to_abstract.get(device_uuid, {})
# interdomain_endpoint = device_endpoint_to_abstract.get(endpoint_uuid)
# interdomain_endpoint_name = interdomain_endpoint.name
# if endpoint_name == interdomain_endpoint_name: return False
# interdomain_endpoint.name = endpoint_name
# return True
def _update_endpoint_type(self, device_name : str, endpoint_name : str, endpoint_type : str) -> bool:
device_endpoint_to_abstract = self.__device_endpoint_to_abstract.get(device_name, {})
interdomain_endpoint = device_endpoint_to_abstract.get(endpoint_name)
interdomain_endpoint_type = interdomain_endpoint.endpoint_type
if endpoint_type == interdomain_endpoint_type: return False
interdomain_endpoint.endpoint_type = endpoint_type
return True
def _add_endpoint(
self, device_uuid : str, endpoint_uuid : str, endpoint_name : str, endpoint_type : str
) -> EndPoint:
def _add_endpoint(self, device_name : str, endpoint_name : str, endpoint_type : str) -> EndPoint:
interdomain_endpoint = self.__device.device_endpoints.add()
interdomain_endpoint.endpoint_id.topology_id.topology_uuid.uuid = INTERDOMAIN_TOPOLOGY_NAME
interdomain_endpoint.endpoint_id.topology_id.context_id.context_uuid.uuid = DEFAULT_CONTEXT_NAME
interdomain_endpoint.endpoint_id.device_id.CopyFrom(self.__device_id)
interdomain_endpoint.endpoint_id.device_id.device_uuid.uuid = self.__device.name
interdomain_endpoint.endpoint_id.endpoint_uuid.uuid = endpoint_name
interdomain_endpoint.name = endpoint_name
interdomain_endpoint.endpoint_type = endpoint_type
uuids = endpoint_get_uuid(interdomain_endpoint.endpoint_id, endpoint_name=endpoint_name, allow_random=False)
_, _, interdomain_endpoint_uuid = uuids
self.__device_endpoint_to_abstract\
.setdefault(device_uuid, {}).setdefault(endpoint_uuid, interdomain_endpoint)
.setdefault(device_name, {}).setdefault(endpoint_name, interdomain_endpoint)
self.__abstract_endpoint_to_device\
.setdefault(interdomain_endpoint_uuid, device_uuid)
.setdefault(endpoint_name, device_name)
return interdomain_endpoint
def _remove_endpoint(
self, device_uuid : str, endpoint_uuid : str, interdomain_endpoint : EndPoint
) -> None:
self.__abstract_endpoint_to_device.pop(endpoint_uuid, None)
device_endpoint_to_abstract = self.__device_endpoint_to_abstract.get(device_uuid, {})
device_endpoint_to_abstract.pop(endpoint_uuid, None)
def _remove_endpoint(self, device_name : str, endpoint_name : str, interdomain_endpoint : EndPoint) -> None:
self.__abstract_endpoint_to_device.pop(endpoint_name, None)
device_endpoint_to_abstract = self.__device_endpoint_to_abstract.get(device_name, {})
device_endpoint_to_abstract.pop(endpoint_name, None)
self.__device.device_endpoints.remove(interdomain_endpoint)
def update_endpoints(self, device : Device) -> bool:
if device_type_is_datacenter(self.__device.device_type): return False
device_uuid = device.device_id.device_uuid.uuid
device_border_endpoint_uuids = {
endpoint.endpoint_id.endpoint_uuid.uuid : (endpoint.name, endpoint.endpoint_type)
device_name = device.name
device_border_endpoint_names = {
endpoint.name : endpoint.endpoint_type
for endpoint in device.device_endpoints
if endpoint_type_is_border(endpoint.endpoint_type)
}
......@@ -200,24 +197,27 @@ class AbstractDevice:
updated = False
# for each border endpoint in abstract device that is not in device; remove from abstract device
device_endpoint_to_abstract = self.__device_endpoint_to_abstract.get(device_uuid, {})
device_endpoint_to_abstract = self.__device_endpoint_to_abstract.get(device_name, {})
_device_endpoint_to_abstract = copy.deepcopy(device_endpoint_to_abstract)
for endpoint_uuid, interdomain_endpoint in _device_endpoint_to_abstract.items():
if endpoint_uuid in device_border_endpoint_uuids: continue
for endpoint_name, interdomain_endpoint in _device_endpoint_to_abstract.items():
if endpoint_name in device_border_endpoint_names: continue
# remove interdomain endpoint that is not in device
self._remove_endpoint(device_uuid, endpoint_uuid, interdomain_endpoint)
self._remove_endpoint(device_name, endpoint_name, interdomain_endpoint)
updated = True
# for each border endpoint in device that is not in abstract device; add to abstract device
for endpoint_uuid,(endpoint_name, endpoint_type) in device_border_endpoint_uuids.items():
# if already added; just check endpoint name and type are not modified
if endpoint_uuid in self.__abstract_endpoint_to_device:
updated = updated or self._update_endpoint_name(device_uuid, endpoint_uuid, endpoint_name)
updated = updated or self._update_endpoint_type(device_uuid, endpoint_uuid, endpoint_type)
for endpoint_name,endpoint_type in device_border_endpoint_names.items():
abstract_endpoint = self.__device_endpoint_to_abstract.get(device_name, {}).get(endpoint_name)
abstract_endpoint_name = None if abstract_endpoint is None else abstract_endpoint.name
# if already added; just check endpoint type is not modified
if abstract_endpoint_name in self.__abstract_endpoint_to_device:
#updated = updated or self._update_endpoint_name(device_name, endpoint_name, endpoint_name)
updated = updated or self._update_endpoint_type(device_name, endpoint_name, endpoint_type)
continue
# otherwise, add it to the abstract device
self._add_endpoint(device_uuid, endpoint_uuid, endpoint_name, endpoint_type)
self._add_endpoint(device_name, endpoint_name, endpoint_type)
updated = True
return updated
src/interdomain/service/topology_abstractor/AbstractLink.py
+ 64
− 57
View file @ 9a71c35a
......@@ -15,122 +15,129 @@
import copy, logging
from typing import Dict, List, Optional, Tuple
from common.Constants import DEFAULT_CONTEXT_NAME, INTERDOMAIN_TOPOLOGY_NAME
from common.proto.context_pb2 import ContextId, EndPointId, Link, LinkId
from common.tools.context_queries.Link import add_link_to_topology, get_existing_link_uuids
from common.tools.object_factory.Context import json_context_id
from common.tools.object_factory.Link import json_link, json_link_id
from common.proto.context_pb2 import EndPointId, Link
from common.tools.context_queries.Link import get_link
from common.tools.object_factory.Link import json_link
from context.client.ContextClient import ContextClient
from .Tools import replace_link_uuids_by_names
# Remark on UUIDs:
# TopologyAbstractor, AbstractDevice and AbstractLink are used
# to compose network reporesentations to be forwarded to remote
# instances. Constraining it to use UUIDs is pointless given
# these UUIDs, to be unique, need to be bound to the local
# context/topology UUIDs, which might be different than that for
# the remote TeraFlowSDN instances. For this very reason, we use
# the device/link/endpoint/topology/context names as UUIDs, to
# prevent UUID-related issues.
LOGGER = logging.getLogger(__name__)
class AbstractLink:
def __init__(self, link_uuid : str):
def __init__(self, link_name : str):
self.__context_client = ContextClient()
self.__link_uuid : str = link_uuid
self.__link_name : str = link_name
self.__link : Optional[Link] = None
self.__link_id : Optional[LinkId] = None
# Dict[(device_uuid, endpoint_uuid), abstract EndPointId]
# Dict[(device_name, endpoint_name), abstract EndPointId]
self.__device_endpoint_to_abstract : Dict[Tuple[str, str], EndPointId] = dict()
def to_json(self) -> Dict:
return {
'link_uuid' : self.__link_uuid,
'link_name' : self.__link_name,
'link' : self.__link,
'link_id' : self.__link_id,
'device_endpoint_to_abstract' : self.__device_endpoint_to_abstract,
}
@property
def uuid(self) -> str: return self.__link_uuid
@property
def link_id(self) -> Optional[LinkId]: return self.__link_id
def name(self) -> str: return self.__link_name
@property
def link(self) -> Optional[Link]: return self.__link
@staticmethod
def compose_uuid(
device_uuid_a : str, endpoint_uuid_a : str, device_uuid_z : str, endpoint_uuid_z : str
def compose_name(
device_name_a : str, endpoint_name_a : str, device_name_z : str, endpoint_name_z : str
) -> str:
# sort endpoints lexicographically to prevent duplicities
link_endpoint_uuids = sorted([
(device_uuid_a, endpoint_uuid_a),
(device_uuid_z, endpoint_uuid_z)
link_endpoint_names = sorted([
(device_name_a, endpoint_name_a),
(device_name_z, endpoint_name_z)
])
link_uuid = '{:s}/{:s}=={:s}/{:s}'.format(
link_endpoint_uuids[0][0], link_endpoint_uuids[0][1],
link_endpoint_uuids[1][0], link_endpoint_uuids[1][1])
return link_uuid
link_name = '{:s}/{:s}=={:s}/{:s}'.format(
link_endpoint_names[0][0], link_endpoint_names[0][1],
link_endpoint_names[1][0], link_endpoint_names[1][1])
return link_name
def initialize(self) -> bool:
if self.__link is not None: return False
existing_link_uuids = get_existing_link_uuids(self.__context_client)
local_interdomain_link = get_link(self.__context_client, self.__link_name, rw_copy=False)
create_abstract_link = local_interdomain_link is None
create = self.__link_uuid not in existing_link_uuids
if create:
if create_abstract_link:
self._create_empty()
else:
self._load_existing()
self._load_existing(local_interdomain_link)
# Add abstract link to topologies [INTERDOMAIN_TOPOLOGY_NAME]
context_id = ContextId(**json_context_id(DEFAULT_CONTEXT_NAME))
topology_uuids = [INTERDOMAIN_TOPOLOGY_NAME]
for topology_uuid in topology_uuids:
add_link_to_topology(self.__context_client, context_id, topology_uuid, self.__link_uuid)
## Add abstract link to topologies [INTERDOMAIN_TOPOLOGY_NAME]
#admin_context_id = ContextId(**json_context_id(DEFAULT_CONTEXT_NAME))
#topology_uuids = [INTERDOMAIN_TOPOLOGY_NAME]
#for topology_uuid in topology_uuids:
# # This action is done automatically; commented out by now.
# add_link_to_topology(
# self.__context_client, admin_context_id, topology_uuid, self.__link_name)
return create
return create_abstract_link
def _create_empty(self) -> None:
link = Link(**json_link(self.__link_uuid, endpoint_ids=[]))
link_name = self.__link_name
link = Link(**json_link(link_name, name=link_name, endpoint_ids=[]))
self.__context_client.SetLink(link)
self.__link = link
self.__link_id = self.__link.link_id
def _load_existing(self) -> None:
self.__link_id = LinkId(**json_link_id(self.__link_uuid))
self.__link = self.__context_client.GetLink(self.__link_id)
self.__link = link # Store copy with names as UUIDs
def _load_existing(self, local_interdomain_link : Link) -> None:
self.__device_endpoint_to_abstract = dict()
self.__link = local_interdomain_link
replace_link_uuids_by_names(self.__context_client, local_interdomain_link)
# for each endpoint in abstract link, populate internal data structures and mappings
for endpoint_id in self.__link.link_endpoint_ids:
device_uuid : str = endpoint_id.device_id.device_uuid.uuid
endpoint_uuid : str = endpoint_id.endpoint_uuid.uuid
self.__device_endpoint_to_abstract.setdefault((device_uuid, endpoint_uuid), endpoint_id)
device_name = endpoint_id.device_id.device_uuid.uuid
endpoint_name = endpoint_id.endpoint_uuid.uuid
self.__device_endpoint_to_abstract.setdefault((device_name, endpoint_name), endpoint_id)
def _add_endpoint(self, device_uuid : str, endpoint_uuid : str) -> None:
def _add_endpoint(self, device_name : str, endpoint_name : str) -> None:
endpoint_id = self.__link.link_endpoint_ids.add()
endpoint_id.topology_id.topology_uuid.uuid = INTERDOMAIN_TOPOLOGY_NAME
endpoint_id.topology_id.context_id.context_uuid.uuid = DEFAULT_CONTEXT_NAME
endpoint_id.device_id.device_uuid.uuid = device_uuid
endpoint_id.endpoint_uuid.uuid = endpoint_uuid
self.__device_endpoint_to_abstract.setdefault((device_uuid, endpoint_uuid), endpoint_id)
endpoint_id.device_id.device_uuid.uuid = device_name
endpoint_id.endpoint_uuid.uuid = endpoint_name
self.__device_endpoint_to_abstract.setdefault((device_name, endpoint_name), endpoint_id)
def _remove_endpoint(self, device_uuid : str, endpoint_uuid : str) -> None:
device_endpoint_to_abstract = self.__device_endpoint_to_abstract.get(device_uuid, {})
endpoint_id = device_endpoint_to_abstract.pop(endpoint_uuid, None)
def _remove_endpoint(self, device_name : str, endpoint_name : str) -> None:
device_endpoint_to_abstract = self.__device_endpoint_to_abstract.get(device_name, {})
endpoint_id = device_endpoint_to_abstract.pop(endpoint_name, None)
if endpoint_id is not None: self.__link.link_endpoint_ids.remove(endpoint_id)
def update_endpoints(self, link_endpoint_uuids : List[Tuple[str, str]] = []) -> bool:
def update_endpoints(self, link_endpoint_names : List[Tuple[str, str]] = []) -> bool:
updated = False
# for each endpoint in abstract link that is not in link; remove from abstract link
device_endpoint_to_abstract = copy.deepcopy(self.__device_endpoint_to_abstract)
for device_uuid, endpoint_uuid in device_endpoint_to_abstract.keys():
if (device_uuid, endpoint_uuid) in link_endpoint_uuids: continue
for device_name, endpoint_name in device_endpoint_to_abstract.keys():
if (device_name, endpoint_name) in link_endpoint_names: continue
# remove endpoint_id that is not in link
self._remove_endpoint(device_uuid, endpoint_uuid)
self._remove_endpoint(device_name, endpoint_name)
updated = True
# for each endpoint in link that is not in abstract link; add to abstract link
for device_uuid, endpoint_uuid in link_endpoint_uuids:
for device_name, endpoint_name in link_endpoint_names:
# if already added; just check endpoint type is not modified
if (device_uuid, endpoint_uuid) in self.__device_endpoint_to_abstract: continue
if (device_name, endpoint_name) in self.__device_endpoint_to_abstract: continue
# otherwise, add it to the abstract device
self._add_endpoint(device_uuid, endpoint_uuid)
self._add_endpoint(device_name, endpoint_name)
updated = True
return updated
src/interdomain/service/topology_abstractor/Tools.py 0 → 100644
+ 86
− 0
View file @ 9a71c35a
# 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 Dict, Tuple
from common.proto.context_pb2 import Device, Link
from common.tools.context_queries.Context import get_context
from common.tools.context_queries.EndPoint import get_endpoint_names
from common.tools.context_queries.Topology import get_topology
from context.client.ContextClient import ContextClient
def get_context_name(
context_client : ContextClient, context_names_cache : Dict[str, str], context_uuid : str
) -> str:
if context_uuid not in context_names_cache:
context = get_context(context_client, context_uuid)
context_name = context.name
context_names_cache[context_uuid] = context_name
context_names_cache[context_name] = context_name
else:
context_name = context_names_cache[context_uuid]
return context_name
def get_topology_name(
context_client : ContextClient, topology_names_cache : Dict[Tuple[str, str], str],
context_name : str, topology_uuid : str
) -> str:
topology_key = (context_name, topology_uuid)
if topology_key not in topology_names_cache:
topology = get_topology(context_client, topology_uuid, context_uuid=context_name)
topology_name = topology.name
topology_names_cache[(context_name, topology_uuid)] = topology_name
topology_names_cache[(context_name, topology_name)] = topology_name
else:
topology_name = topology_names_cache[topology_key]
return topology_name
def replace_device_uuids_by_names(
context_client : ContextClient, device : Device, context_names_cache : Dict[str, str] = dict(),
topology_names_cache : Dict[Tuple[str, str], str] = dict()
) -> None:
device_name = device.name
device.device_id.device_uuid.uuid = device_name
for endpoint in device.device_endpoints:
endpoint.endpoint_id.endpoint_uuid.uuid = endpoint.name
endpoint.endpoint_id.device_id.device_uuid.uuid = device_name
context_uuid = endpoint.endpoint_id.topology_id.context_id.context_uuid.uuid
context_name = get_context_name(context_client, context_names_cache, context_uuid)
endpoint.endpoint_id.topology_id.context_id.context_uuid.uuid = context_name
topology_uuid = endpoint.endpoint_id.topology_id.topology_uuid.uuid
topology_name = get_topology_name(context_client, topology_names_cache, context_name, topology_uuid)
endpoint.endpoint_id.topology_id.topology_uuid.uuid = topology_name
def replace_link_uuids_by_names(
context_client : ContextClient, link : Link, context_names_cache : Dict[str, str] = dict(),
topology_names_cache : Dict[Tuple[str, str], str] = dict()
) -> None:
link.link_id.link_uuid.uuid = link.name
device_names, endpoints_data = get_endpoint_names(context_client, link.link_endpoint_ids)
# for each endpoint in abstract link, populate internal data structures and mappings
for endpoint_id in link.link_endpoint_ids:
endpoint_id.device_id.device_uuid.uuid = device_names[endpoint_id.device_id.device_uuid.uuid]
endpoint_id.endpoint_uuid.uuid = endpoints_data[endpoint_id.endpoint_uuid.uuid][0]
context_uuid = endpoint_id.topology_id.context_id.context_uuid.uuid
context_name = get_context_name(context_client, context_names_cache, context_uuid)
endpoint_id.topology_id.context_id.context_uuid.uuid = context_name
topology_uuid = endpoint_id.topology_id.topology_uuid.uuid
topology_name = get_topology_name(context_client, topology_names_cache, context_name, topology_uuid)
endpoint_id.topology_id.topology_uuid.uuid = topology_name
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment