Loading src/context/service/database/models/enums/DeviceDriver.py +0 −1 Original line number Diff line number Diff line Loading @@ -35,7 +35,6 @@ class ORM_DeviceDriverEnum(enum.Enum): IETF_ACTN = DeviceDriverEnum.DEVICEDRIVER_IETF_ACTN OC = DeviceDriverEnum.DEVICEDRIVER_OC QKD = DeviceDriverEnum.DEVICEDRIVER_QKD IETF_L3VPN = DeviceDriverEnum.DEVICEDRIVER_IETF_L3VPN grpc_to_enum__device_driver = functools.partial( grpc_to_enum, DeviceDriverEnum, ORM_DeviceDriverEnum) src/service/service/service_handlers/l3nm_ietfl3vpn/ConfigRules.py 0 → 100644 +291 −0 Original line number Diff line number Diff line # Copyright 2022-2024 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, List, Tuple, TypedDict from common.proto.context_pb2 import Link from common.tools.object_factory.ConfigRule import json_config_rule_delete from context.client.ContextClient import ContextClient class LANPrefixesDict(TypedDict): lan: str lan_tag: str SITE_NETWORK_ACCESS_TYPE = "ietf-l3vpn-svc:multipoint" def setup_config_rules(service_uuid: str, json_settings: Dict) -> List[Dict]: src_device_uuid: str = json_settings["src_device_name"] src_endpoint_uuid: str = json_settings["src_endpoint_name"] src_site_location: str = json_settings["src_site_location"] src_ipv4_lan_prefixes: list[LANPrefixesDict] = json_settings.get( "src_ipv4_lan_prefixes" ) src_site_id: str = json_settings.get("src_site_id", f"site_{src_site_location}") src_management_type: str = json_settings.get( "src_management_type", "ietf-l3vpn-svc:provider-managed" ) if src_management_type != "ietf-l3vpn-svc:provider-managed": raise Exception("management type %s not supported", src_management_type) src_role: str = "ietf-l3vpn-svc:hub-role" src_ce_address: str = json_settings["src_ce_address"] src_pe_address: str = json_settings["src_pe_address"] src_ce_pe_network_prefix: int = json_settings["src_ce_pe_network_prefix"] src_mtu: int = json_settings["src_mtu"] src_input_bw: int = json_settings.get("src_input_bw", 1000000000) src_output_bw: int = json_settings.get("src_input_bw", 1000000000) src_qos_profile_id = "qos-realtime" src_qos_profile_direction = "ietf-l3vpn-svc:both" src_qos_profile_latency: int = json_settings.get("src_qos_profile_latency", 10) src_qos_profile_bw_guarantee: int = json_settings.get( "src_qos_profile_bw_guarantee", 100 ) dst_device_uuid = json_settings["dst_device_name"] dst_endpoint_uuid = json_settings["dst_endpoint_name"] dst_site_location: str = json_settings["dst_site_location"] dst_ipv4_lan_prefixes: list[LANPrefixesDict] = json_settings[ "dst_ipv4_lan_prefixes" ] dst_site_id: str = json_settings.get("dst_site_id", f"site_{dst_site_location}") dst_management_type: str = json_settings.get( "dst_management_type", "ietf-l3vpn-svc:provider-managed" ) if dst_management_type != "ietf-l3vpn-svc:provider-managed": raise Exception("management type %s not supported", dst_management_type) dst_role: str = "ietf-l3vpn-svc:spoke-role" dst_ce_address: str = json_settings["dst_ce_address"] dst_pe_address: str = json_settings["dst_pe_address"] dst_ce_pe_network_prefix: int = json_settings["dst_ce_pe_network_prefix"] dst_mtu: int = json_settings["dst_mtu"] dst_input_bw: int = json_settings.get("dst_input_bw", 1000000000) dst_output_bw: int = json_settings.get("dst_output_bw", 1000000000) dst_qos_profile_id = "qos-realtime" dst_qos_profile_direction = "ietf-l3vpn-svc:both" dst_qos_profile_latency: int = json_settings.get("dst_qos_profile_latency", 10) dst_qos_profile_bw_guarantee: int = json_settings.get( "dst_qos_profile_bw_guarantee", 100 ) # Create source site information src_management = {"type": src_management_type} src_locations = {"location": [{"location-id": src_site_location}]} src_devices = { "device": [{"device-id": src_device_uuid, "location": src_site_location}] } src_site_lan_prefixes = [ {"lan": lp["lan"], "lan-tag": lp["lan_tag"], "next-hop": src_ce_address} for lp in src_ipv4_lan_prefixes ] src_site_routing_protocols = { "routing-protocol": [ { "type": "ietf-l3vpn-svc:static", "static": { "cascaded-lan-prefixes": { "ipv4-lan-prefixes": src_site_lan_prefixes } }, } ] } src_site_network_accesses = { "site-network-access": [ { "site-network-access-id": src_endpoint_uuid, "site-network-access-type": SITE_NETWORK_ACCESS_TYPE, "device-reference": src_device_uuid, "vpn-attachment": {"vpn-id": service_uuid, "site-role": src_role}, "ip-connection": { "ipv4": { "address-allocation-type": "ietf-l3vpn-svc:static-address", "addresses": { "provider-address": src_pe_address, "customer-address": src_ce_address, "prefix-length": src_ce_pe_network_prefix, }, } }, "service": { "svc-mtu": src_mtu, "svc-input-bandwidth": src_input_bw, "svc-output-bandwidth": src_output_bw, "qos": { "qos-profile": { "classes": { "class": [ { "class-id": src_qos_profile_id, "direction": src_qos_profile_direction, "latency": { "latency-boundary": src_qos_profile_latency }, "bandwidth": { "guaranteed-bw-percent": src_qos_profile_bw_guarantee }, } ] } } }, }, } ] } # Create destination site information dst_management = {"type": src_management_type} dst_locations = {"location": [{"location-id": dst_site_location}]} dst_devices = { "device": [{"device-id": dst_device_uuid, "location": dst_site_location}] } dst_site_lan_prefixes = [ {"lan": lp["lan"], "lan-tag": lp["lan_tag"], "next-hop": dst_ce_address} for lp in dst_ipv4_lan_prefixes ] dst_site_routing_protocols = { "routing-protocol": [ { "type": "ietf-l3vpn-svc:static", "static": { "cascaded-lan-prefixes": { "ipv4-lan-prefixes": dst_site_lan_prefixes } }, } ] } dst_site_network_accesses = { "site-network-access": [ { "site-network-access-id": dst_endpoint_uuid, "site-network-access-type": SITE_NETWORK_ACCESS_TYPE, "device-reference": dst_device_uuid, "vpn-attachment": {"vpn-id": service_uuid, "site-role": dst_role}, "ip-connection": { "ipv4": { "address-allocation-type": "ietf-l3vpn-svc:static-address", "addresses": { "provider-address": dst_pe_address, "customer-address": dst_ce_address, "prefix-length": dst_ce_pe_network_prefix, }, } }, "service": { "svc-mtu": dst_mtu, "svc-input-bandwidth": dst_input_bw, "svc-output-bandwidth": dst_output_bw, "qos": { "qos-profile": { "classes": { "class": [ { "class-id": dst_qos_profile_id, "direction": dst_qos_profile_direction, "latency": { "latency-boundary": dst_qos_profile_latency }, "bandwidth": { "guaranteed-bw-percent": dst_qos_profile_bw_guarantee }, } ] } } }, }, } ] } sites = { "site": [ { "site-id": src_site_id, "management": src_management, "locations": src_locations, "devices": src_devices, "routing-protocols": src_site_routing_protocols, "site-network-accesses": src_site_network_accesses, }, { "site-id": dst_site_id, "management": dst_management, "locations": dst_locations, "devices": dst_devices, "routing-protocols": dst_site_routing_protocols, "site-network-accesses": dst_site_network_accesses, }, ] } l3_vpn_data_model = { "ietf-l3vpn-svc:l3vpn-svc": { "vpn-services": {"vpn-service": [{"vpn-id": service_uuid}]}, "sites": sites, } } return l3_vpn_data_model def teardown_config_rules(service_uuid: str, json_settings: Dict) -> List[Dict]: json_config_rules = [ json_config_rule_delete( "/service[{:s}]/IETFSlice".format(service_uuid), {}, ), json_config_rule_delete( "/service[{:s}]/IETFSlice/operation".format(service_uuid), {}, ), ] return json_config_rules def get_link_ep_device_names( link: Link, context_client: ContextClient ) -> Tuple[str, str, str, str]: ep_ids = link.link_endpoint_ids ep_device_id_1 = ep_ids[0].device_id ep_uuid_1 = ep_ids[0].endpoint_uuid.uuid device_obj_1 = context_client.GetDevice(ep_device_id_1) for d_ep in device_obj_1.device_endpoints: if d_ep.endpoint_id.endpoint_uuid.uuid == ep_uuid_1: ep_name_1 = d_ep.name break else: raise Exception("endpoint not found") device_obj_name_1 = device_obj_1.name ep_device_id_2 = ep_ids[1].device_id ep_uuid_2 = ep_ids[1].endpoint_uuid.uuid device_obj_2 = context_client.GetDevice(ep_device_id_2) for d_ep in device_obj_2.device_endpoints: if d_ep.endpoint_id.endpoint_uuid.uuid == ep_uuid_2: ep_name_2 = d_ep.name break else: raise Exception("endpoint not found") device_obj_name_2 = device_obj_2.name return ( device_obj_name_1, ep_name_1, device_obj_1, device_obj_name_2, ep_name_2, device_obj_2, ) Loading
src/context/service/database/models/enums/DeviceDriver.py +0 −1 Original line number Diff line number Diff line Loading @@ -35,7 +35,6 @@ class ORM_DeviceDriverEnum(enum.Enum): IETF_ACTN = DeviceDriverEnum.DEVICEDRIVER_IETF_ACTN OC = DeviceDriverEnum.DEVICEDRIVER_OC QKD = DeviceDriverEnum.DEVICEDRIVER_QKD IETF_L3VPN = DeviceDriverEnum.DEVICEDRIVER_IETF_L3VPN grpc_to_enum__device_driver = functools.partial( grpc_to_enum, DeviceDriverEnum, ORM_DeviceDriverEnum)
src/service/service/service_handlers/l3nm_ietfl3vpn/ConfigRules.py 0 → 100644 +291 −0 Original line number Diff line number Diff line # Copyright 2022-2024 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, List, Tuple, TypedDict from common.proto.context_pb2 import Link from common.tools.object_factory.ConfigRule import json_config_rule_delete from context.client.ContextClient import ContextClient class LANPrefixesDict(TypedDict): lan: str lan_tag: str SITE_NETWORK_ACCESS_TYPE = "ietf-l3vpn-svc:multipoint" def setup_config_rules(service_uuid: str, json_settings: Dict) -> List[Dict]: src_device_uuid: str = json_settings["src_device_name"] src_endpoint_uuid: str = json_settings["src_endpoint_name"] src_site_location: str = json_settings["src_site_location"] src_ipv4_lan_prefixes: list[LANPrefixesDict] = json_settings.get( "src_ipv4_lan_prefixes" ) src_site_id: str = json_settings.get("src_site_id", f"site_{src_site_location}") src_management_type: str = json_settings.get( "src_management_type", "ietf-l3vpn-svc:provider-managed" ) if src_management_type != "ietf-l3vpn-svc:provider-managed": raise Exception("management type %s not supported", src_management_type) src_role: str = "ietf-l3vpn-svc:hub-role" src_ce_address: str = json_settings["src_ce_address"] src_pe_address: str = json_settings["src_pe_address"] src_ce_pe_network_prefix: int = json_settings["src_ce_pe_network_prefix"] src_mtu: int = json_settings["src_mtu"] src_input_bw: int = json_settings.get("src_input_bw", 1000000000) src_output_bw: int = json_settings.get("src_input_bw", 1000000000) src_qos_profile_id = "qos-realtime" src_qos_profile_direction = "ietf-l3vpn-svc:both" src_qos_profile_latency: int = json_settings.get("src_qos_profile_latency", 10) src_qos_profile_bw_guarantee: int = json_settings.get( "src_qos_profile_bw_guarantee", 100 ) dst_device_uuid = json_settings["dst_device_name"] dst_endpoint_uuid = json_settings["dst_endpoint_name"] dst_site_location: str = json_settings["dst_site_location"] dst_ipv4_lan_prefixes: list[LANPrefixesDict] = json_settings[ "dst_ipv4_lan_prefixes" ] dst_site_id: str = json_settings.get("dst_site_id", f"site_{dst_site_location}") dst_management_type: str = json_settings.get( "dst_management_type", "ietf-l3vpn-svc:provider-managed" ) if dst_management_type != "ietf-l3vpn-svc:provider-managed": raise Exception("management type %s not supported", dst_management_type) dst_role: str = "ietf-l3vpn-svc:spoke-role" dst_ce_address: str = json_settings["dst_ce_address"] dst_pe_address: str = json_settings["dst_pe_address"] dst_ce_pe_network_prefix: int = json_settings["dst_ce_pe_network_prefix"] dst_mtu: int = json_settings["dst_mtu"] dst_input_bw: int = json_settings.get("dst_input_bw", 1000000000) dst_output_bw: int = json_settings.get("dst_output_bw", 1000000000) dst_qos_profile_id = "qos-realtime" dst_qos_profile_direction = "ietf-l3vpn-svc:both" dst_qos_profile_latency: int = json_settings.get("dst_qos_profile_latency", 10) dst_qos_profile_bw_guarantee: int = json_settings.get( "dst_qos_profile_bw_guarantee", 100 ) # Create source site information src_management = {"type": src_management_type} src_locations = {"location": [{"location-id": src_site_location}]} src_devices = { "device": [{"device-id": src_device_uuid, "location": src_site_location}] } src_site_lan_prefixes = [ {"lan": lp["lan"], "lan-tag": lp["lan_tag"], "next-hop": src_ce_address} for lp in src_ipv4_lan_prefixes ] src_site_routing_protocols = { "routing-protocol": [ { "type": "ietf-l3vpn-svc:static", "static": { "cascaded-lan-prefixes": { "ipv4-lan-prefixes": src_site_lan_prefixes } }, } ] } src_site_network_accesses = { "site-network-access": [ { "site-network-access-id": src_endpoint_uuid, "site-network-access-type": SITE_NETWORK_ACCESS_TYPE, "device-reference": src_device_uuid, "vpn-attachment": {"vpn-id": service_uuid, "site-role": src_role}, "ip-connection": { "ipv4": { "address-allocation-type": "ietf-l3vpn-svc:static-address", "addresses": { "provider-address": src_pe_address, "customer-address": src_ce_address, "prefix-length": src_ce_pe_network_prefix, }, } }, "service": { "svc-mtu": src_mtu, "svc-input-bandwidth": src_input_bw, "svc-output-bandwidth": src_output_bw, "qos": { "qos-profile": { "classes": { "class": [ { "class-id": src_qos_profile_id, "direction": src_qos_profile_direction, "latency": { "latency-boundary": src_qos_profile_latency }, "bandwidth": { "guaranteed-bw-percent": src_qos_profile_bw_guarantee }, } ] } } }, }, } ] } # Create destination site information dst_management = {"type": src_management_type} dst_locations = {"location": [{"location-id": dst_site_location}]} dst_devices = { "device": [{"device-id": dst_device_uuid, "location": dst_site_location}] } dst_site_lan_prefixes = [ {"lan": lp["lan"], "lan-tag": lp["lan_tag"], "next-hop": dst_ce_address} for lp in dst_ipv4_lan_prefixes ] dst_site_routing_protocols = { "routing-protocol": [ { "type": "ietf-l3vpn-svc:static", "static": { "cascaded-lan-prefixes": { "ipv4-lan-prefixes": dst_site_lan_prefixes } }, } ] } dst_site_network_accesses = { "site-network-access": [ { "site-network-access-id": dst_endpoint_uuid, "site-network-access-type": SITE_NETWORK_ACCESS_TYPE, "device-reference": dst_device_uuid, "vpn-attachment": {"vpn-id": service_uuid, "site-role": dst_role}, "ip-connection": { "ipv4": { "address-allocation-type": "ietf-l3vpn-svc:static-address", "addresses": { "provider-address": dst_pe_address, "customer-address": dst_ce_address, "prefix-length": dst_ce_pe_network_prefix, }, } }, "service": { "svc-mtu": dst_mtu, "svc-input-bandwidth": dst_input_bw, "svc-output-bandwidth": dst_output_bw, "qos": { "qos-profile": { "classes": { "class": [ { "class-id": dst_qos_profile_id, "direction": dst_qos_profile_direction, "latency": { "latency-boundary": dst_qos_profile_latency }, "bandwidth": { "guaranteed-bw-percent": dst_qos_profile_bw_guarantee }, } ] } } }, }, } ] } sites = { "site": [ { "site-id": src_site_id, "management": src_management, "locations": src_locations, "devices": src_devices, "routing-protocols": src_site_routing_protocols, "site-network-accesses": src_site_network_accesses, }, { "site-id": dst_site_id, "management": dst_management, "locations": dst_locations, "devices": dst_devices, "routing-protocols": dst_site_routing_protocols, "site-network-accesses": dst_site_network_accesses, }, ] } l3_vpn_data_model = { "ietf-l3vpn-svc:l3vpn-svc": { "vpn-services": {"vpn-service": [{"vpn-id": service_uuid}]}, "sites": sites, } } return l3_vpn_data_model def teardown_config_rules(service_uuid: str, json_settings: Dict) -> List[Dict]: json_config_rules = [ json_config_rule_delete( "/service[{:s}]/IETFSlice".format(service_uuid), {}, ), json_config_rule_delete( "/service[{:s}]/IETFSlice/operation".format(service_uuid), {}, ), ] return json_config_rules def get_link_ep_device_names( link: Link, context_client: ContextClient ) -> Tuple[str, str, str, str]: ep_ids = link.link_endpoint_ids ep_device_id_1 = ep_ids[0].device_id ep_uuid_1 = ep_ids[0].endpoint_uuid.uuid device_obj_1 = context_client.GetDevice(ep_device_id_1) for d_ep in device_obj_1.device_endpoints: if d_ep.endpoint_id.endpoint_uuid.uuid == ep_uuid_1: ep_name_1 = d_ep.name break else: raise Exception("endpoint not found") device_obj_name_1 = device_obj_1.name ep_device_id_2 = ep_ids[1].device_id ep_uuid_2 = ep_ids[1].endpoint_uuid.uuid device_obj_2 = context_client.GetDevice(ep_device_id_2) for d_ep in device_obj_2.device_endpoints: if d_ep.endpoint_id.endpoint_uuid.uuid == ep_uuid_2: ep_name_2 = d_ep.name break else: raise Exception("endpoint not found") device_obj_name_2 = device_obj_2.name return ( device_obj_name_1, ep_name_1, device_obj_1, device_obj_name_2, ep_name_2, device_obj_2, )
