# 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. # Convert the Explicit Route Object (ERO)-like paths produced by the PathComp component (response['path']) into # explicit hops with ingress and egress endpoints per device (path_hops). # # response['path'] = [{ # 'path-capacity': {'total-size': {'value': 200, 'unit': 0}}, # 'path-latency': {'fixed-latency-characteristic': '12.000000'}, # 'path-cost': {'cost-name': '', 'cost-value': '6.000000', 'cost-algorithm': '0.000000'}, # 'devices': [ # {'device_id': 'DC1-GW', 'endpoint_uuid': 'int'}, # {'device_id': 'DC1-GW', 'endpoint_uuid': 'eth1'}, # {'device_id': 'CS1-GW1', 'endpoint_uuid': '1/2'}, # {'device_id': 'TN-R2', 'endpoint_uuid': '2/1'}, # {'device_id': 'TN-OLS', 'endpoint_uuid': 'ca46812e8ad7'}, # {'device_id': 'TN-R3', 'endpoint_uuid': '1/1'}, # {'device_id': 'CS2-GW1', 'endpoint_uuid': '10/1'}, # {'device_id': 'DC2-GW', 'endpoint_uuid': 'int'} # ] # }] # # path_hops = [ # {'device': 'DC1-GW', 'ingress_ep': 'int', 'egress_ep': 'eth1'}, # {'device': 'CS1-GW1', 'ingress_ep': '10/1', 'egress_ep': '1/2'}, # {'device': 'TN-R2', 'ingress_ep': '1/2', 'egress_ep': '2/1'}, # {'device': 'TN-OLS', 'ingress_ep': '951f2f57e4a4', 'egress_ep': 'ca46812e8ad7'}, # {'device': 'TN-R3', 'ingress_ep': '2/1', 'egress_ep': '1/1'}, # {'device': 'CS2-GW1', 'ingress_ep': '1/1', 'egress_ep': '10/1'}, # {'device': 'DC2-GW', 'ingress_ep': 'eth1', 'egress_ep': 'int'} # ] # import logging from typing import Dict, List, Tuple from common.proto.context_pb2 import Link LOGGER = logging.getLogger(__name__) def eropath_to_hops( ero_path : List[Dict], endpoint_to_link_dict : Dict[Tuple[str, str, str], Tuple[Dict, Link]] ) -> List[Dict]: try: path_hops = [] num_ero_hops = len(ero_path) for endpoint in ero_path: device_uuid = endpoint['device_id'] endpoint_uuid = endpoint['endpoint_uuid'] if len(path_hops) == 0: path_hops.append({'device': device_uuid, 'ingress_ep': endpoint_uuid}) continue last_hop = path_hops[-1] if last_hop['device'] != device_uuid: raise Exception('Malformed path') last_hop['egress_ep'] = endpoint_uuid if num_ero_hops - 1 == len(path_hops): break link_tuple = endpoint_to_link_dict[(device_uuid, endpoint_uuid, 'src')] if link_tuple is None: raise Exception('Malformed path') ingress = link_tuple[0]['link_endpoint_ids'][-1] path_hops.append({ 'device': ingress['endpoint_id']['device_id'], 'ingress_ep': ingress['endpoint_id']['endpoint_uuid'] }) return path_hops except: LOGGER.exception('Unhandled exception: ero_path={:s} endpoint_to_link_dict={:s}'.format( str(ero_path), str(endpoint_to_link_dict))) raise