Loading src/service/service/service_handlers/l3nm_gnmi_openconfig/ConfigRuleComposer.py +101 −10 Original line number Diff line number Diff line Loading @@ -30,6 +30,27 @@ RE_IF = re.compile(r'^\/interface\[([^\]]+)\]$') RE_SUBIF = re.compile(r'^\/interface\[([^\]]+)\]\/subinterface\[([^\]]+)\]$') RE_SR = re.compile(r'^\/network_instance\[([^\]]+)\]\/protocols\[STATIC\]/route\[([^\:]+)\:([^\]]+)\]$') def _safe_int(value: Optional[object]) -> Optional[int]: try: return int(value) if value is not None else None except (TypeError, ValueError): return None def _safe_bool(value: Optional[object]) -> Optional[bool]: if value is None: return None if isinstance(value, bool): return value if isinstance(value, (int, float)): return bool(value) if isinstance(value, str): lowered = value.strip().lower() if lowered in {'true', '1', 'yes', 'y', 'on', 'tagged'}: return True if lowered in {'false', '0', 'no', 'n', 'off', 'untagged'}: return False return None def _interface( interface : str, if_type : Optional[str] = 'l3ipvlan', index : int = 0, vlan_id : Optional[int] = None, address_ip : Optional[str] = None, address_prefix : Optional[int] = None, mtu : Optional[int] = None, Loading Loading @@ -82,12 +103,28 @@ class EndpointComposer: self.sub_interface_index = 0 self.ipv4_address = None self.ipv4_prefix_len = None self.explicit_vlan_ids : Set[int] = set() self.force_trunk = False def _add_vlan_id(self, vlan_id : Optional[int]) -> None: if vlan_id is not None: self.explicit_vlan_ids.add(vlan_id) def _configure_from_settings(self, json_settings : Dict) -> None: if not isinstance(json_settings, dict): return vlan_id = _safe_int(json_settings.get('vlan_id', json_settings.get('vlan-id'))) self._add_vlan_id(vlan_id) def configure(self, endpoint_obj : Optional[EndPoint], settings : Optional[TreeNode]) -> None: if endpoint_obj is not None: self.objekt = endpoint_obj if settings is None: return json_settings : Dict = settings.value json_settings : Dict = settings.value or dict() self._configure_from_settings(json_settings) for child in settings.children: if isinstance(child.value, dict): self._configure_from_settings(child.value) if 'address_ip' in json_settings: self.ipv4_address = json_settings['address_ip'] Loading @@ -107,7 +144,20 @@ class EndpointComposer: self.sub_interface_index = json_settings.get('index', 0) def get_config_rules(self, network_instance_name : str, delete : bool = False) -> List[Dict]: def set_force_trunk(self, enable : bool = True) -> None: self.force_trunk = enable def _select_vlan_id(self, service_vlan_id : Optional[int]) -> Optional[int]: if service_vlan_id is not None and service_vlan_id in self.explicit_vlan_ids: return service_vlan_id if len(self.explicit_vlan_ids) > 0: return sorted(self.explicit_vlan_ids)[0] return service_vlan_id def get_config_rules( self, network_instance_name : str, service_vlan_id : Optional[int] = None, access_vlan_tagged : bool = False, delete : bool = False ) -> List[Dict]: if self.ipv4_address is None: return [] if self.ipv4_prefix_len is None: return [] json_config_rule = json_config_rule_delete if delete else json_config_rule_set Loading @@ -118,18 +168,24 @@ class EndpointComposer: network_instance_name, self.objekt.name, self.sub_interface_index ))) vlan_id = None if self.force_trunk or access_vlan_tagged or len(self.explicit_vlan_ids) > 0: vlan_id = self._select_vlan_id(service_vlan_id) if vlan_id is None: LOGGER.warning('VLAN tagging requested but no vlan_id provided for endpoint={:s}'.format(self.uuid)) if delete: config_rules.extend([ json_config_rule(*_interface( self.objekt.name, index=self.sub_interface_index, address_ip=None, address_prefix=None, enabled=False address_prefix=None, enabled=False, vlan_id=vlan_id )), ]) else: config_rules.extend([ json_config_rule(*_interface( self.objekt.name, index=self.sub_interface_index, address_ip=self.ipv4_address, address_prefix=self.ipv4_prefix_len, enabled=True address_prefix=self.ipv4_prefix_len, enabled=True, vlan_id=vlan_id )), ]) return config_rules Loading @@ -139,6 +195,8 @@ class EndpointComposer: 'index' : self.sub_interface_index, 'address_ip' : self.ipv4_address, 'address_prefix': self.ipv4_prefix_len, 'explicit_vlan_ids': list(self.explicit_vlan_ids), 'force_trunk' : self.force_trunk, } def __str__(self): Loading @@ -155,6 +213,8 @@ class DeviceComposer: self.endpoints : Dict[str, EndpointComposer] = dict() # endpoint_uuid => EndpointComposer self.connected : Set[str] = set() self.static_routes : Dict[str, Dict[int, str]] = dict() # {prefix => {metric => next_hop}} self.service_vlan_id : Optional[int] = None self.access_vlan_tagged = False def set_endpoint_alias(self, endpoint_name : str, endpoint_uuid : str) -> None: self.aliases[endpoint_name] = endpoint_uuid Loading Loading @@ -225,7 +285,12 @@ class DeviceComposer: metric = static_route.get('metric', 0) self.static_routes.setdefault(prefix, dict())[metric] = next_hop def get_config_rules(self, network_instance_name : str, delete : bool = False) -> List[Dict]: def get_config_rules( self, network_instance_name : str, service_vlan_id : Optional[int] = None, access_vlan_tagged : bool = False, delete : bool = False ) -> List[Dict]: self.service_vlan_id = service_vlan_id self.access_vlan_tagged = access_vlan_tagged SELECTED_DEVICES = { DeviceTypeEnum.PACKET_POP.value, DeviceTypeEnum.PACKET_ROUTER.value, Loading @@ -238,7 +303,10 @@ class DeviceComposer: if network_instance_name != DEFAULT_NETWORK_INSTANCE: json_config_rule(*_network_instance(network_instance_name, 'L3VRF')) for endpoint in self.endpoints.values(): config_rules.extend(endpoint.get_config_rules(network_instance_name, delete=delete)) config_rules.extend(endpoint.get_config_rules( network_instance_name, self.service_vlan_id, access_vlan_tagged=self.access_vlan_tagged, delete=delete )) if len(self.static_routes) > 0: config_rules.append( json_config_rule(*_network_instance_protocol_static(network_instance_name)) Loading Loading @@ -274,6 +342,8 @@ class ConfigRuleComposer: self.objekt : Optional[Service] = None self.aliases : Dict[str, str] = dict() # device_name => device_uuid self.devices : Dict[str, DeviceComposer] = dict() # device_uuid => DeviceComposer self.vlan_id : Optional[int] = None self.access_vlan_tagged = False def set_device_alias(self, device_name : str, device_uuid : str) -> None: self.aliases[device_name] = device_uuid Loading @@ -286,15 +356,34 @@ class ConfigRuleComposer: def configure(self, service_obj : Service, settings : Optional[TreeNode]) -> None: self.objekt = service_obj self.vlan_id = None self.access_vlan_tagged = False if settings is None: return #json_settings : Dict = settings.value # For future use json_settings : Dict = settings.value or dict() if 'vlan_id' in json_settings: self.vlan_id = _safe_int(json_settings['vlan_id']) elif 'vlan-id' in json_settings: self.vlan_id = _safe_int(json_settings['vlan-id']) if 'access_vlan_tagged' in json_settings or 'access-vlan-tagged' in json_settings: access_vlan_tagged = json_settings.get('access_vlan_tagged', json_settings.get('access-vlan-tagged')) parsed = _safe_bool(access_vlan_tagged) if parsed is None: MSG = 'Invalid access_vlan_tagged value in service settings: {:s}' LOGGER.warning(MSG.format(str(access_vlan_tagged))) self.access_vlan_tagged = False else: self.access_vlan_tagged = parsed def get_config_rules( self, network_instance_name : str = NETWORK_INSTANCE, delete : bool = False ) -> Dict[str, List[Dict]]: return { device_uuid : device.get_config_rules(network_instance_name, delete=delete) device_uuid : device.get_config_rules( network_instance_name, self.vlan_id, access_vlan_tagged=self.access_vlan_tagged, delete=delete ) for device_uuid, device in self.devices.items() } Loading @@ -303,5 +392,7 @@ class ConfigRuleComposer: 'devices' : { device_uuid : device.dump() for device_uuid, device in self.devices.items() } }, 'vlan_id': self.vlan_id, 'access_vlan_tagged': self.access_vlan_tagged, } src/service/service/service_handlers/l3nm_gnmi_openconfig/L3NMGnmiOpenConfigServiceHandler.py +4 −0 Original line number Diff line number Diff line Loading @@ -26,6 +26,7 @@ from service.service.task_scheduler.TaskExecutor import TaskExecutor from service.service.tools.EndpointIdFormatters import endpointids_to_raw from .ConfigRuleComposer import ConfigRuleComposer from .StaticRouteGenerator import StaticRouteGenerator from .VlanIdPropagator import VlanIdPropagator LOGGER = logging.getLogger(__name__) Loading @@ -39,6 +40,7 @@ class L3NMGnmiOpenConfigServiceHandler(_ServiceHandler): self.__task_executor = task_executor self.__settings_handler = SettingsHandler(service.service_config, **settings) self.__config_rule_composer = ConfigRuleComposer() self.__vlan_id_propagator = VlanIdPropagator(self.__config_rule_composer) self.__static_route_generator = StaticRouteGenerator(self.__config_rule_composer) self.__endpoint_map : Dict[Tuple[str, str], Tuple[str, str]] = dict() Loading Loading @@ -66,6 +68,8 @@ class L3NMGnmiOpenConfigServiceHandler(_ServiceHandler): self.__endpoint_map[(device_uuid, endpoint_uuid)] = (device_obj.name, endpoint_obj.name) LOGGER.debug('[pre] config_rule_composer = {:s}'.format(json.dumps(self.__config_rule_composer.dump()))) self.__vlan_id_propagator.compose(endpoints) LOGGER.debug('[post-vlan] config_rule_composer = {:s}'.format(json.dumps(self.__config_rule_composer.dump()))) self.__static_route_generator.compose(endpoints) LOGGER.debug('[post] config_rule_composer = {:s}'.format(json.dumps(self.__config_rule_composer.dump()))) Loading src/service/service/service_handlers/l3nm_gnmi_openconfig/VlanIdPropagator.py 0 → 100644 +87 −0 Original line number Diff line number Diff line # 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. import json, logging from typing import List, Optional, Tuple from common.DeviceTypes import DeviceTypeEnum from .ConfigRuleComposer import ConfigRuleComposer LOGGER = logging.getLogger(__name__) class VlanIdPropagator: def __init__(self, config_rule_composer : ConfigRuleComposer) -> None: self._config_rule_composer = config_rule_composer self._router_types = { DeviceTypeEnum.PACKET_ROUTER.value, DeviceTypeEnum.EMULATED_PACKET_ROUTER.value, DeviceTypeEnum.PACKET_POP.value, DeviceTypeEnum.PACKET_RADIO_ROUTER.value, DeviceTypeEnum.EMULATED_PACKET_RADIO_ROUTER.value, } def _is_router_device(self, device) -> bool: return device.objekt is not None and device.objekt.device_type in self._router_types def compose(self, connection_hop_list : List[Tuple[str, str, Optional[str]]]) -> None: link_endpoints = self._compute_link_endpoints(connection_hop_list) LOGGER.debug('link_endpoints = {:s}'.format(str(link_endpoints))) self._propagate_vlan_id(link_endpoints) LOGGER.debug('config_rule_composer = {:s}'.format(json.dumps(self._config_rule_composer.dump()))) def _compute_link_endpoints( self, connection_hop_list : List[Tuple[str, str, Optional[str]]] ) -> List[Tuple[Tuple[str, str, Optional[str]], Tuple[str, str, Optional[str]]]]: # In some cases connection_hop_list might contain repeated endpoints, remove them here. added_connection_hops = set() filtered_connection_hop_list = list() for connection_hop in connection_hop_list: if connection_hop in added_connection_hops: continue filtered_connection_hop_list.append(connection_hop) added_connection_hops.add(connection_hop) connection_hop_list = filtered_connection_hop_list # In some cases connection_hop_list first and last items might be internal endpoints of # devices instead of link endpoints. Filter those endpoints not reaching a new device. if len(connection_hop_list) > 2 and connection_hop_list[0][0] == connection_hop_list[1][0]: # same device on first 2 endpoints connection_hop_list = connection_hop_list[1:] if len(connection_hop_list) > 2 and connection_hop_list[-1][0] == connection_hop_list[-2][0]: # same device on last 2 endpoints connection_hop_list = connection_hop_list[:-1] num_connection_hops = len(connection_hop_list) if num_connection_hops % 2 != 0: raise Exception('Number of connection hops must be even') if num_connection_hops < 4: raise Exception('Number of connection hops must be >= 4') it_connection_hops = iter(connection_hop_list) return list(zip(it_connection_hops, it_connection_hops)) def _propagate_vlan_id( self, link_endpoints_list : List[Tuple[Tuple[str, str, Optional[str]], Tuple[str, str, Optional[str]]]] ) -> None: for link_endpoints in link_endpoints_list: device_endpoint_a, device_endpoint_b = link_endpoints device_uuid_a, endpoint_uuid_a = device_endpoint_a[0:2] device_a = self._config_rule_composer.get_device(device_uuid_a) endpoint_a = device_a.get_endpoint(endpoint_uuid_a) device_uuid_b, endpoint_uuid_b = device_endpoint_b[0:2] device_b = self._config_rule_composer.get_device(device_uuid_b) endpoint_b = device_b.get_endpoint(endpoint_uuid_b) if self._is_router_device(device_a) and self._is_router_device(device_b): endpoint_a.set_force_trunk() endpoint_b.set_force_trunk() src/service/tests/test_l3nm_gnmi_static_rule_gen/MockServiceHandler.py +5 −1 Original line number Diff line number Diff line Loading @@ -26,6 +26,7 @@ from .MockTaskExecutor import MockTaskExecutor from service.service.tools.EndpointIdFormatters import endpointids_to_raw from service.service.service_handlers.l3nm_gnmi_openconfig.ConfigRuleComposer import ConfigRuleComposer from service.service.service_handlers.l3nm_gnmi_openconfig.StaticRouteGenerator import StaticRouteGenerator from service.service.service_handlers.l3nm_gnmi_openconfig.VlanIdPropagator import VlanIdPropagator LOGGER = logging.getLogger(__name__) Loading @@ -37,6 +38,7 @@ class MockServiceHandler(_ServiceHandler): self.__task_executor = task_executor self.__settings_handler = SettingsHandler(service.service_config, **settings) self.__config_rule_composer = ConfigRuleComposer() self.__vlan_id_propagator = VlanIdPropagator(self.__config_rule_composer) self.__static_route_generator = StaticRouteGenerator(self.__config_rule_composer) self.__endpoint_map : Dict[Tuple[str, str], Tuple[str, str]] = dict() Loading Loading @@ -94,8 +96,10 @@ class MockServiceHandler(_ServiceHandler): #prev_endpoint = _endpoint #prev_endpoint_obj = endpoint_obj self.__vlan_id_propagator.compose(endpoints) LOGGER.debug('[post-vlan] config_rule_composer = {:s}'.format(json.dumps(self.__config_rule_composer.dump()))) self.__static_route_generator.compose(endpoints) LOGGER.debug('config_rule_composer = {:s}'.format(json.dumps(self.__config_rule_composer.dump()))) LOGGER.debug('[post] config_rule_composer = {:s}'.format(json.dumps(self.__config_rule_composer.dump()))) def _do_configurations( self, config_rules_per_device : Dict[str, List[Dict]], endpoints : List[Tuple[str, str, Optional[str]]], Loading Loading
src/service/service/service_handlers/l3nm_gnmi_openconfig/ConfigRuleComposer.py +101 −10 Original line number Diff line number Diff line Loading @@ -30,6 +30,27 @@ RE_IF = re.compile(r'^\/interface\[([^\]]+)\]$') RE_SUBIF = re.compile(r'^\/interface\[([^\]]+)\]\/subinterface\[([^\]]+)\]$') RE_SR = re.compile(r'^\/network_instance\[([^\]]+)\]\/protocols\[STATIC\]/route\[([^\:]+)\:([^\]]+)\]$') def _safe_int(value: Optional[object]) -> Optional[int]: try: return int(value) if value is not None else None except (TypeError, ValueError): return None def _safe_bool(value: Optional[object]) -> Optional[bool]: if value is None: return None if isinstance(value, bool): return value if isinstance(value, (int, float)): return bool(value) if isinstance(value, str): lowered = value.strip().lower() if lowered in {'true', '1', 'yes', 'y', 'on', 'tagged'}: return True if lowered in {'false', '0', 'no', 'n', 'off', 'untagged'}: return False return None def _interface( interface : str, if_type : Optional[str] = 'l3ipvlan', index : int = 0, vlan_id : Optional[int] = None, address_ip : Optional[str] = None, address_prefix : Optional[int] = None, mtu : Optional[int] = None, Loading Loading @@ -82,12 +103,28 @@ class EndpointComposer: self.sub_interface_index = 0 self.ipv4_address = None self.ipv4_prefix_len = None self.explicit_vlan_ids : Set[int] = set() self.force_trunk = False def _add_vlan_id(self, vlan_id : Optional[int]) -> None: if vlan_id is not None: self.explicit_vlan_ids.add(vlan_id) def _configure_from_settings(self, json_settings : Dict) -> None: if not isinstance(json_settings, dict): return vlan_id = _safe_int(json_settings.get('vlan_id', json_settings.get('vlan-id'))) self._add_vlan_id(vlan_id) def configure(self, endpoint_obj : Optional[EndPoint], settings : Optional[TreeNode]) -> None: if endpoint_obj is not None: self.objekt = endpoint_obj if settings is None: return json_settings : Dict = settings.value json_settings : Dict = settings.value or dict() self._configure_from_settings(json_settings) for child in settings.children: if isinstance(child.value, dict): self._configure_from_settings(child.value) if 'address_ip' in json_settings: self.ipv4_address = json_settings['address_ip'] Loading @@ -107,7 +144,20 @@ class EndpointComposer: self.sub_interface_index = json_settings.get('index', 0) def get_config_rules(self, network_instance_name : str, delete : bool = False) -> List[Dict]: def set_force_trunk(self, enable : bool = True) -> None: self.force_trunk = enable def _select_vlan_id(self, service_vlan_id : Optional[int]) -> Optional[int]: if service_vlan_id is not None and service_vlan_id in self.explicit_vlan_ids: return service_vlan_id if len(self.explicit_vlan_ids) > 0: return sorted(self.explicit_vlan_ids)[0] return service_vlan_id def get_config_rules( self, network_instance_name : str, service_vlan_id : Optional[int] = None, access_vlan_tagged : bool = False, delete : bool = False ) -> List[Dict]: if self.ipv4_address is None: return [] if self.ipv4_prefix_len is None: return [] json_config_rule = json_config_rule_delete if delete else json_config_rule_set Loading @@ -118,18 +168,24 @@ class EndpointComposer: network_instance_name, self.objekt.name, self.sub_interface_index ))) vlan_id = None if self.force_trunk or access_vlan_tagged or len(self.explicit_vlan_ids) > 0: vlan_id = self._select_vlan_id(service_vlan_id) if vlan_id is None: LOGGER.warning('VLAN tagging requested but no vlan_id provided for endpoint={:s}'.format(self.uuid)) if delete: config_rules.extend([ json_config_rule(*_interface( self.objekt.name, index=self.sub_interface_index, address_ip=None, address_prefix=None, enabled=False address_prefix=None, enabled=False, vlan_id=vlan_id )), ]) else: config_rules.extend([ json_config_rule(*_interface( self.objekt.name, index=self.sub_interface_index, address_ip=self.ipv4_address, address_prefix=self.ipv4_prefix_len, enabled=True address_prefix=self.ipv4_prefix_len, enabled=True, vlan_id=vlan_id )), ]) return config_rules Loading @@ -139,6 +195,8 @@ class EndpointComposer: 'index' : self.sub_interface_index, 'address_ip' : self.ipv4_address, 'address_prefix': self.ipv4_prefix_len, 'explicit_vlan_ids': list(self.explicit_vlan_ids), 'force_trunk' : self.force_trunk, } def __str__(self): Loading @@ -155,6 +213,8 @@ class DeviceComposer: self.endpoints : Dict[str, EndpointComposer] = dict() # endpoint_uuid => EndpointComposer self.connected : Set[str] = set() self.static_routes : Dict[str, Dict[int, str]] = dict() # {prefix => {metric => next_hop}} self.service_vlan_id : Optional[int] = None self.access_vlan_tagged = False def set_endpoint_alias(self, endpoint_name : str, endpoint_uuid : str) -> None: self.aliases[endpoint_name] = endpoint_uuid Loading Loading @@ -225,7 +285,12 @@ class DeviceComposer: metric = static_route.get('metric', 0) self.static_routes.setdefault(prefix, dict())[metric] = next_hop def get_config_rules(self, network_instance_name : str, delete : bool = False) -> List[Dict]: def get_config_rules( self, network_instance_name : str, service_vlan_id : Optional[int] = None, access_vlan_tagged : bool = False, delete : bool = False ) -> List[Dict]: self.service_vlan_id = service_vlan_id self.access_vlan_tagged = access_vlan_tagged SELECTED_DEVICES = { DeviceTypeEnum.PACKET_POP.value, DeviceTypeEnum.PACKET_ROUTER.value, Loading @@ -238,7 +303,10 @@ class DeviceComposer: if network_instance_name != DEFAULT_NETWORK_INSTANCE: json_config_rule(*_network_instance(network_instance_name, 'L3VRF')) for endpoint in self.endpoints.values(): config_rules.extend(endpoint.get_config_rules(network_instance_name, delete=delete)) config_rules.extend(endpoint.get_config_rules( network_instance_name, self.service_vlan_id, access_vlan_tagged=self.access_vlan_tagged, delete=delete )) if len(self.static_routes) > 0: config_rules.append( json_config_rule(*_network_instance_protocol_static(network_instance_name)) Loading Loading @@ -274,6 +342,8 @@ class ConfigRuleComposer: self.objekt : Optional[Service] = None self.aliases : Dict[str, str] = dict() # device_name => device_uuid self.devices : Dict[str, DeviceComposer] = dict() # device_uuid => DeviceComposer self.vlan_id : Optional[int] = None self.access_vlan_tagged = False def set_device_alias(self, device_name : str, device_uuid : str) -> None: self.aliases[device_name] = device_uuid Loading @@ -286,15 +356,34 @@ class ConfigRuleComposer: def configure(self, service_obj : Service, settings : Optional[TreeNode]) -> None: self.objekt = service_obj self.vlan_id = None self.access_vlan_tagged = False if settings is None: return #json_settings : Dict = settings.value # For future use json_settings : Dict = settings.value or dict() if 'vlan_id' in json_settings: self.vlan_id = _safe_int(json_settings['vlan_id']) elif 'vlan-id' in json_settings: self.vlan_id = _safe_int(json_settings['vlan-id']) if 'access_vlan_tagged' in json_settings or 'access-vlan-tagged' in json_settings: access_vlan_tagged = json_settings.get('access_vlan_tagged', json_settings.get('access-vlan-tagged')) parsed = _safe_bool(access_vlan_tagged) if parsed is None: MSG = 'Invalid access_vlan_tagged value in service settings: {:s}' LOGGER.warning(MSG.format(str(access_vlan_tagged))) self.access_vlan_tagged = False else: self.access_vlan_tagged = parsed def get_config_rules( self, network_instance_name : str = NETWORK_INSTANCE, delete : bool = False ) -> Dict[str, List[Dict]]: return { device_uuid : device.get_config_rules(network_instance_name, delete=delete) device_uuid : device.get_config_rules( network_instance_name, self.vlan_id, access_vlan_tagged=self.access_vlan_tagged, delete=delete ) for device_uuid, device in self.devices.items() } Loading @@ -303,5 +392,7 @@ class ConfigRuleComposer: 'devices' : { device_uuid : device.dump() for device_uuid, device in self.devices.items() } }, 'vlan_id': self.vlan_id, 'access_vlan_tagged': self.access_vlan_tagged, }
src/service/service/service_handlers/l3nm_gnmi_openconfig/L3NMGnmiOpenConfigServiceHandler.py +4 −0 Original line number Diff line number Diff line Loading @@ -26,6 +26,7 @@ from service.service.task_scheduler.TaskExecutor import TaskExecutor from service.service.tools.EndpointIdFormatters import endpointids_to_raw from .ConfigRuleComposer import ConfigRuleComposer from .StaticRouteGenerator import StaticRouteGenerator from .VlanIdPropagator import VlanIdPropagator LOGGER = logging.getLogger(__name__) Loading @@ -39,6 +40,7 @@ class L3NMGnmiOpenConfigServiceHandler(_ServiceHandler): self.__task_executor = task_executor self.__settings_handler = SettingsHandler(service.service_config, **settings) self.__config_rule_composer = ConfigRuleComposer() self.__vlan_id_propagator = VlanIdPropagator(self.__config_rule_composer) self.__static_route_generator = StaticRouteGenerator(self.__config_rule_composer) self.__endpoint_map : Dict[Tuple[str, str], Tuple[str, str]] = dict() Loading Loading @@ -66,6 +68,8 @@ class L3NMGnmiOpenConfigServiceHandler(_ServiceHandler): self.__endpoint_map[(device_uuid, endpoint_uuid)] = (device_obj.name, endpoint_obj.name) LOGGER.debug('[pre] config_rule_composer = {:s}'.format(json.dumps(self.__config_rule_composer.dump()))) self.__vlan_id_propagator.compose(endpoints) LOGGER.debug('[post-vlan] config_rule_composer = {:s}'.format(json.dumps(self.__config_rule_composer.dump()))) self.__static_route_generator.compose(endpoints) LOGGER.debug('[post] config_rule_composer = {:s}'.format(json.dumps(self.__config_rule_composer.dump()))) Loading
src/service/service/service_handlers/l3nm_gnmi_openconfig/VlanIdPropagator.py 0 → 100644 +87 −0 Original line number Diff line number Diff line # 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. import json, logging from typing import List, Optional, Tuple from common.DeviceTypes import DeviceTypeEnum from .ConfigRuleComposer import ConfigRuleComposer LOGGER = logging.getLogger(__name__) class VlanIdPropagator: def __init__(self, config_rule_composer : ConfigRuleComposer) -> None: self._config_rule_composer = config_rule_composer self._router_types = { DeviceTypeEnum.PACKET_ROUTER.value, DeviceTypeEnum.EMULATED_PACKET_ROUTER.value, DeviceTypeEnum.PACKET_POP.value, DeviceTypeEnum.PACKET_RADIO_ROUTER.value, DeviceTypeEnum.EMULATED_PACKET_RADIO_ROUTER.value, } def _is_router_device(self, device) -> bool: return device.objekt is not None and device.objekt.device_type in self._router_types def compose(self, connection_hop_list : List[Tuple[str, str, Optional[str]]]) -> None: link_endpoints = self._compute_link_endpoints(connection_hop_list) LOGGER.debug('link_endpoints = {:s}'.format(str(link_endpoints))) self._propagate_vlan_id(link_endpoints) LOGGER.debug('config_rule_composer = {:s}'.format(json.dumps(self._config_rule_composer.dump()))) def _compute_link_endpoints( self, connection_hop_list : List[Tuple[str, str, Optional[str]]] ) -> List[Tuple[Tuple[str, str, Optional[str]], Tuple[str, str, Optional[str]]]]: # In some cases connection_hop_list might contain repeated endpoints, remove them here. added_connection_hops = set() filtered_connection_hop_list = list() for connection_hop in connection_hop_list: if connection_hop in added_connection_hops: continue filtered_connection_hop_list.append(connection_hop) added_connection_hops.add(connection_hop) connection_hop_list = filtered_connection_hop_list # In some cases connection_hop_list first and last items might be internal endpoints of # devices instead of link endpoints. Filter those endpoints not reaching a new device. if len(connection_hop_list) > 2 and connection_hop_list[0][0] == connection_hop_list[1][0]: # same device on first 2 endpoints connection_hop_list = connection_hop_list[1:] if len(connection_hop_list) > 2 and connection_hop_list[-1][0] == connection_hop_list[-2][0]: # same device on last 2 endpoints connection_hop_list = connection_hop_list[:-1] num_connection_hops = len(connection_hop_list) if num_connection_hops % 2 != 0: raise Exception('Number of connection hops must be even') if num_connection_hops < 4: raise Exception('Number of connection hops must be >= 4') it_connection_hops = iter(connection_hop_list) return list(zip(it_connection_hops, it_connection_hops)) def _propagate_vlan_id( self, link_endpoints_list : List[Tuple[Tuple[str, str, Optional[str]], Tuple[str, str, Optional[str]]]] ) -> None: for link_endpoints in link_endpoints_list: device_endpoint_a, device_endpoint_b = link_endpoints device_uuid_a, endpoint_uuid_a = device_endpoint_a[0:2] device_a = self._config_rule_composer.get_device(device_uuid_a) endpoint_a = device_a.get_endpoint(endpoint_uuid_a) device_uuid_b, endpoint_uuid_b = device_endpoint_b[0:2] device_b = self._config_rule_composer.get_device(device_uuid_b) endpoint_b = device_b.get_endpoint(endpoint_uuid_b) if self._is_router_device(device_a) and self._is_router_device(device_b): endpoint_a.set_force_trunk() endpoint_b.set_force_trunk()
src/service/tests/test_l3nm_gnmi_static_rule_gen/MockServiceHandler.py +5 −1 Original line number Diff line number Diff line Loading @@ -26,6 +26,7 @@ from .MockTaskExecutor import MockTaskExecutor from service.service.tools.EndpointIdFormatters import endpointids_to_raw from service.service.service_handlers.l3nm_gnmi_openconfig.ConfigRuleComposer import ConfigRuleComposer from service.service.service_handlers.l3nm_gnmi_openconfig.StaticRouteGenerator import StaticRouteGenerator from service.service.service_handlers.l3nm_gnmi_openconfig.VlanIdPropagator import VlanIdPropagator LOGGER = logging.getLogger(__name__) Loading @@ -37,6 +38,7 @@ class MockServiceHandler(_ServiceHandler): self.__task_executor = task_executor self.__settings_handler = SettingsHandler(service.service_config, **settings) self.__config_rule_composer = ConfigRuleComposer() self.__vlan_id_propagator = VlanIdPropagator(self.__config_rule_composer) self.__static_route_generator = StaticRouteGenerator(self.__config_rule_composer) self.__endpoint_map : Dict[Tuple[str, str], Tuple[str, str]] = dict() Loading Loading @@ -94,8 +96,10 @@ class MockServiceHandler(_ServiceHandler): #prev_endpoint = _endpoint #prev_endpoint_obj = endpoint_obj self.__vlan_id_propagator.compose(endpoints) LOGGER.debug('[post-vlan] config_rule_composer = {:s}'.format(json.dumps(self.__config_rule_composer.dump()))) self.__static_route_generator.compose(endpoints) LOGGER.debug('config_rule_composer = {:s}'.format(json.dumps(self.__config_rule_composer.dump()))) LOGGER.debug('[post] config_rule_composer = {:s}'.format(json.dumps(self.__config_rule_composer.dump()))) def _do_configurations( self, config_rules_per_device : Dict[str, List[Dict]], endpoints : List[Tuple[str, str, Optional[str]]], Loading
