Loading src/service/service/service_handlers/l2nm_gnmi_openconfig/ConfigRuleComposer.py +91 −94 Original line number Diff line number Diff line Loading @@ -12,10 +12,10 @@ # See the License for the specific language governing permissions and # limitations under the License. import json, logging, netaddr, re import json, logging from typing import Dict, List, Optional, Set, Tuple from common.DeviceTypes import DeviceTypeEnum from common.proto.context_pb2 import ConfigActionEnum, Device, EndPoint, Service from common.proto.context_pb2 import Device, EndPoint, Service from common.tools.grpc.Tools import grpc_message_to_json_string from common.tools.object_factory.ConfigRule import json_config_rule_delete, json_config_rule_set from service.service.service_handler_api.AnyTreeTools import TreeNode Loading @@ -26,19 +26,26 @@ LOGGER = logging.getLogger(__name__) NETWORK_INSTANCE = 'default' DEFAULT_NETWORK_INSTANCE = 'default' RE_IF = re.compile(r'^\/interface\[([^\]]+)\]$') RE_SUBIF = re.compile(r'^\/interface\[([^\]]+)\]\/subinterface\[([^\]]+)\]$') 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 _interface( interface : str, if_type : Optional[str] = 'l3ipvlan', index : int = 0, vlan_id : Optional[int] = None, mtu : Optional[int] = None, enabled : bool = True def _interface_switched_vlan( interface : str, interface_mode : str, access_vlan_id : Optional[int] = None, trunk_vlan_id : Optional[int] = None, native_vlan : int = 1 ) -> Tuple[str, Dict]: path = '/interface[{:s}]/subinterface[{:d}]'.format(interface, index) data = {'name': interface, 'type': if_type, 'index': index, 'enabled': enabled} if if_type is not None: data['type'] = if_type if vlan_id is not None: data['vlan_id'] = vlan_id if mtu is not None: data['mtu'] = mtu return path, data path = '/interface[{:s}]/ethernet/switched-vlan'.format(interface) config : Dict[str, object] = {'interface-mode': interface_mode} if interface_mode == 'ACCESS': if access_vlan_id is not None: config['access-vlan'] = access_vlan_id elif interface_mode == 'TRUNK': config['native-vlan'] = native_vlan if trunk_vlan_id is not None: config['trunk-vlans'] = [trunk_vlan_id] return path, {'config': config} def _network_instance(ni_name : str, ni_type : str) -> Tuple[str, Dict]: path = '/network_instance[{:s}]'.format(ni_name) Loading @@ -50,63 +57,73 @@ def _network_instance_vlan(ni_name : str, vlan_id : int, vlan_name : str = None) data = {'name': ni_name, 'vlan_id': vlan_id, 'vlan_name': vlan_name} return path, data def _network_instance_interface(ni_name : str, interface : str, sub_interface_index : int) -> Tuple[str, Dict]: sub_interface_name = '{:s}.{:d}'.format(interface, sub_interface_index) path = '/network_instance[{:s}]/interface[{:s}]'.format(ni_name, sub_interface_name) data = {'name': ni_name, 'id': sub_interface_name, 'interface': interface, 'subinterface': sub_interface_index} return path, data class EndpointComposer: def __init__(self, endpoint_uuid : str) -> None: self.uuid = endpoint_uuid self.objekt : Optional[EndPoint] = None self.vlan_id = 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 if 'vlan_id' in json_settings: self.vlan_id = json_settings['vlan_id'] elif 'vlan-id' in json_settings: self.vlan_id = json_settings['vlan-id'] else: MSG = 'VLAN ID not found. Tried: vlan_id and vlan-id. endpoint_obj={:s} settings={:s}' LOGGER.warning(MSG.format(grpc_message_to_json_string(endpoint_obj), str(settings))) def get_config_rules(self, network_instance_name : str, delete : bool = False) -> List[Dict]: 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) def set_force_trunk(self, enable : bool = True) -> None: self.force_trunk = enable def _select_trunk_vlan_id(self, service_vlan_id : int) -> int: if 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_vlan_ids(self) -> Set[int]: return set(self.explicit_vlan_ids) def has_vlan(self, vlan_id : int) -> bool: return vlan_id in self.get_vlan_ids() def get_config_rules(self, service_vlan_id : int, delete : bool = False) -> List[Dict]: if self.objekt is None: MSG = 'Endpoint object not defined for uuid={:s}' LOGGER.warning(MSG.format(self.uuid)) return [] config_rules : List[Dict] = list() if self.vlan_id is None: MSG = 'VLAN ID not defined for endpoint_obj={:s}' LOGGER.warning(MSG.format(grpc_message_to_json_string(self.objekt))) return config_rules json_config_rule = json_config_rule_delete if delete else json_config_rule_set if network_instance_name != DEFAULT_NETWORK_INSTANCE: config_rules.append(json_config_rule(*_network_instance_interface( network_instance_name, self.objekt.name, self.vlan_id if self.force_trunk or len(self.explicit_vlan_ids) > 0: trunk_vlan_id = self._select_trunk_vlan_id(service_vlan_id) config_rules.append(json_config_rule(*_interface_switched_vlan( self.objekt.name, 'TRUNK', trunk_vlan_id=trunk_vlan_id ))) if delete: config_rules.extend([ json_config_rule(*_interface( self.objekt.name, index=self.vlan_id, vlan_id=self.vlan_id, enabled=False )), ]) else: config_rules.extend([ json_config_rule(*_interface( self.objekt.name, index=self.vlan_id, vlan_id=self.vlan_id, enabled=True )), ]) config_rules.append(json_config_rule(*_interface_switched_vlan( self.objekt.name, 'ACCESS', access_vlan_id=service_vlan_id ))) return config_rules def dump(self) -> Dict: return { 'vlan_id' : self.vlan_id, 'explicit_vlan_ids' : list(self.explicit_vlan_ids), 'force_trunk' : self.force_trunk, } def __str__(self): Loading @@ -132,6 +149,10 @@ class DeviceComposer: self.endpoints[endpoint_uuid] = EndpointComposer(endpoint_uuid) return self.endpoints[endpoint_uuid] def _refresh_vlan_ids(self, service_vlan_id : int) -> None: # Only keep the service VLAN; others are ignored for composition self.vlan_ids = {service_vlan_id} def configure(self, device_obj : Device, settings : Optional[TreeNode]) -> None: self.objekt = device_obj for endpoint_obj in device_obj.device_endpoints: Loading @@ -139,37 +160,9 @@ class DeviceComposer: self.set_endpoint_alias(endpoint_obj.name, endpoint_uuid) self.get_endpoint(endpoint_obj.name).configure(endpoint_obj, None) # Find management interfaces mgmt_ifaces = set() for config_rule in device_obj.device_config.config_rules: if config_rule.action != ConfigActionEnum.CONFIGACTION_SET: continue if config_rule.WhichOneof('config_rule') != 'custom': continue config_rule_custom = config_rule.custom match = RE_IF.match(config_rule_custom.resource_key) if match is None: continue if_name = match.groups()[0] resource_value = json.loads(config_rule_custom.resource_value) management = resource_value.get('management', False) if management: mgmt_ifaces.add(if_name) # Find data plane interfaces for config_rule in device_obj.device_config.config_rules: if config_rule.action != ConfigActionEnum.CONFIGACTION_SET: continue if config_rule.WhichOneof('config_rule') != 'custom': continue config_rule_custom = config_rule.custom match = RE_SUBIF.match(config_rule_custom.resource_key) if match is not None: if_name, _ = match.groups() if if_name in mgmt_ifaces: continue resource_value = json.loads(config_rule_custom.resource_value) if 'vlan_id' not in resource_value: continue vlan_id = int(resource_value['vlan_id']) self.vlan_ids.add(vlan_id) endpoint = self.get_endpoint(if_name) endpoint.vlan_id = vlan_id 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 : int, delete : bool = False ) -> List[Dict]: SELECTED_DEVICES = { DeviceTypeEnum.PACKET_POP.value, DeviceTypeEnum.PACKET_ROUTER.value, Loading @@ -179,11 +172,12 @@ class DeviceComposer: json_config_rule = json_config_rule_delete if delete else json_config_rule_set config_rules : List[Dict] = list() self._refresh_vlan_ids(service_vlan_id) 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)) for vlan_id in self.vlan_ids: config_rules.extend(endpoint.get_config_rules(service_vlan_id, delete=delete)) for vlan_id in sorted(self.vlan_ids): vlan_name = 'tfs-vlan-{:s}'.format(str(vlan_id)) config_rules.append(json_config_rule(*_network_instance_vlan( network_instance_name, vlan_id, vlan_name=vlan_name Loading Loading @@ -224,27 +218,30 @@ class ConfigRuleComposer: def configure(self, service_obj : Service, settings : Optional[TreeNode]) -> None: self.objekt = service_obj if settings is None: return json_settings : Dict = settings.value if settings is None: raise Exception('Service settings are required to extract vlan_id') json_settings : Dict = settings.value or dict() if 'vlan_id' in json_settings: self.vlan_id = json_settings['vlan_id'] self.vlan_id = _safe_int(json_settings['vlan_id']) elif 'vlan-id' in json_settings: self.vlan_id = json_settings['vlan-id'] self.vlan_id = _safe_int(json_settings['vlan-id']) else: MSG = 'VLAN ID not found. Tried: vlan_id and vlan-id. service_obj={:s} settings={:s}' LOGGER.warning(MSG.format(grpc_message_to_json_string(service_obj), str(settings))) raise Exception(MSG.format(grpc_message_to_json_string(service_obj), str(settings))) if self.vlan_id is None: MSG = 'Invalid VLAN ID value in service settings: {:s}' raise Exception(MSG.format(str(json_settings))) def get_config_rules( self, network_instance_name : str = NETWORK_INSTANCE, delete : bool = False ) -> Dict[str, List[Dict]]: if self.vlan_id is None: MSG = 'VLAN ID not defined for service_obj={:s}' LOGGER.warning(MSG.format(grpc_message_to_json_string(self.objekt))) return dict() raise Exception('VLAN ID must be configured at service level before composing rules') return { device_uuid : device.get_config_rules(network_instance_name, delete=delete) device_uuid : device.get_config_rules(network_instance_name, self.vlan_id, delete=delete) for device_uuid, device in self.devices.items() } Loading src/service/service/service_handlers/l2nm_gnmi_openconfig/VlanIdPropagator.py +14 −25 Original line number Diff line number Diff line Loading @@ -14,6 +14,7 @@ import json, logging from typing import List, Optional, Tuple from common.DeviceTypes import DeviceTypeEnum from .ConfigRuleComposer import ConfigRuleComposer LOGGER = logging.getLogger(__name__) Loading @@ -21,6 +22,16 @@ 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) Loading Loading @@ -71,28 +82,6 @@ class VlanIdPropagator: device_b = self._config_rule_composer.get_device(device_uuid_b) endpoint_b = device_b.get_endpoint(endpoint_uuid_b) svc_vlan_id = self._config_rule_composer.vlan_id ep_a_vlan_id = endpoint_a.vlan_id ep_b_vlan_id = endpoint_b.vlan_id if ep_a_vlan_id is None and ep_b_vlan_id is None: device_a.vlan_ids.add(svc_vlan_id) endpoint_a.vlan_id = svc_vlan_id device_b.vlan_ids.add(svc_vlan_id) endpoint_b.vlan_id = svc_vlan_id elif ep_a_vlan_id is not None and ep_b_vlan_id is None: if ep_a_vlan_id != svc_vlan_id: MSG = 'Incompatible VLAN-IDs: endpoint_a({:s}), service({:s})' raise Exception(MSG.format(str(endpoint_a), str(svc_vlan_id))) device_b.vlan_ids.add(svc_vlan_id) endpoint_b.vlan_id = svc_vlan_id elif ep_a_vlan_id is None and ep_b_vlan_id is not None: if ep_b_vlan_id != svc_vlan_id: MSG = 'Incompatible VLAN-IDs: endpoint_b({:s}), service({:s})' raise Exception(MSG.format(str(endpoint_b), str(svc_vlan_id))) device_a.vlan_ids.add(svc_vlan_id) endpoint_a.vlan_id = svc_vlan_id elif ep_a_vlan_id is not None and ep_b_vlan_id is not None: if ep_a_vlan_id != svc_vlan_id or ep_b_vlan_id != svc_vlan_id: MSG = 'Incompatible VLAN-IDs: endpoint_a({:s}), endpoint_a({:s}), service({:s})' raise Exception(MSG.format(str(endpoint_a), str(endpoint_b), str(svc_vlan_id))) if self._is_router_device(device_a) and self._is_router_device(device_b): endpoint_a.set_force_trunk() endpoint_b.set_force_trunk() Loading
src/service/service/service_handlers/l2nm_gnmi_openconfig/ConfigRuleComposer.py +91 −94 Original line number Diff line number Diff line Loading @@ -12,10 +12,10 @@ # See the License for the specific language governing permissions and # limitations under the License. import json, logging, netaddr, re import json, logging from typing import Dict, List, Optional, Set, Tuple from common.DeviceTypes import DeviceTypeEnum from common.proto.context_pb2 import ConfigActionEnum, Device, EndPoint, Service from common.proto.context_pb2 import Device, EndPoint, Service from common.tools.grpc.Tools import grpc_message_to_json_string from common.tools.object_factory.ConfigRule import json_config_rule_delete, json_config_rule_set from service.service.service_handler_api.AnyTreeTools import TreeNode Loading @@ -26,19 +26,26 @@ LOGGER = logging.getLogger(__name__) NETWORK_INSTANCE = 'default' DEFAULT_NETWORK_INSTANCE = 'default' RE_IF = re.compile(r'^\/interface\[([^\]]+)\]$') RE_SUBIF = re.compile(r'^\/interface\[([^\]]+)\]\/subinterface\[([^\]]+)\]$') 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 _interface( interface : str, if_type : Optional[str] = 'l3ipvlan', index : int = 0, vlan_id : Optional[int] = None, mtu : Optional[int] = None, enabled : bool = True def _interface_switched_vlan( interface : str, interface_mode : str, access_vlan_id : Optional[int] = None, trunk_vlan_id : Optional[int] = None, native_vlan : int = 1 ) -> Tuple[str, Dict]: path = '/interface[{:s}]/subinterface[{:d}]'.format(interface, index) data = {'name': interface, 'type': if_type, 'index': index, 'enabled': enabled} if if_type is not None: data['type'] = if_type if vlan_id is not None: data['vlan_id'] = vlan_id if mtu is not None: data['mtu'] = mtu return path, data path = '/interface[{:s}]/ethernet/switched-vlan'.format(interface) config : Dict[str, object] = {'interface-mode': interface_mode} if interface_mode == 'ACCESS': if access_vlan_id is not None: config['access-vlan'] = access_vlan_id elif interface_mode == 'TRUNK': config['native-vlan'] = native_vlan if trunk_vlan_id is not None: config['trunk-vlans'] = [trunk_vlan_id] return path, {'config': config} def _network_instance(ni_name : str, ni_type : str) -> Tuple[str, Dict]: path = '/network_instance[{:s}]'.format(ni_name) Loading @@ -50,63 +57,73 @@ def _network_instance_vlan(ni_name : str, vlan_id : int, vlan_name : str = None) data = {'name': ni_name, 'vlan_id': vlan_id, 'vlan_name': vlan_name} return path, data def _network_instance_interface(ni_name : str, interface : str, sub_interface_index : int) -> Tuple[str, Dict]: sub_interface_name = '{:s}.{:d}'.format(interface, sub_interface_index) path = '/network_instance[{:s}]/interface[{:s}]'.format(ni_name, sub_interface_name) data = {'name': ni_name, 'id': sub_interface_name, 'interface': interface, 'subinterface': sub_interface_index} return path, data class EndpointComposer: def __init__(self, endpoint_uuid : str) -> None: self.uuid = endpoint_uuid self.objekt : Optional[EndPoint] = None self.vlan_id = 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 if 'vlan_id' in json_settings: self.vlan_id = json_settings['vlan_id'] elif 'vlan-id' in json_settings: self.vlan_id = json_settings['vlan-id'] else: MSG = 'VLAN ID not found. Tried: vlan_id and vlan-id. endpoint_obj={:s} settings={:s}' LOGGER.warning(MSG.format(grpc_message_to_json_string(endpoint_obj), str(settings))) def get_config_rules(self, network_instance_name : str, delete : bool = False) -> List[Dict]: 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) def set_force_trunk(self, enable : bool = True) -> None: self.force_trunk = enable def _select_trunk_vlan_id(self, service_vlan_id : int) -> int: if 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_vlan_ids(self) -> Set[int]: return set(self.explicit_vlan_ids) def has_vlan(self, vlan_id : int) -> bool: return vlan_id in self.get_vlan_ids() def get_config_rules(self, service_vlan_id : int, delete : bool = False) -> List[Dict]: if self.objekt is None: MSG = 'Endpoint object not defined for uuid={:s}' LOGGER.warning(MSG.format(self.uuid)) return [] config_rules : List[Dict] = list() if self.vlan_id is None: MSG = 'VLAN ID not defined for endpoint_obj={:s}' LOGGER.warning(MSG.format(grpc_message_to_json_string(self.objekt))) return config_rules json_config_rule = json_config_rule_delete if delete else json_config_rule_set if network_instance_name != DEFAULT_NETWORK_INSTANCE: config_rules.append(json_config_rule(*_network_instance_interface( network_instance_name, self.objekt.name, self.vlan_id if self.force_trunk or len(self.explicit_vlan_ids) > 0: trunk_vlan_id = self._select_trunk_vlan_id(service_vlan_id) config_rules.append(json_config_rule(*_interface_switched_vlan( self.objekt.name, 'TRUNK', trunk_vlan_id=trunk_vlan_id ))) if delete: config_rules.extend([ json_config_rule(*_interface( self.objekt.name, index=self.vlan_id, vlan_id=self.vlan_id, enabled=False )), ]) else: config_rules.extend([ json_config_rule(*_interface( self.objekt.name, index=self.vlan_id, vlan_id=self.vlan_id, enabled=True )), ]) config_rules.append(json_config_rule(*_interface_switched_vlan( self.objekt.name, 'ACCESS', access_vlan_id=service_vlan_id ))) return config_rules def dump(self) -> Dict: return { 'vlan_id' : self.vlan_id, 'explicit_vlan_ids' : list(self.explicit_vlan_ids), 'force_trunk' : self.force_trunk, } def __str__(self): Loading @@ -132,6 +149,10 @@ class DeviceComposer: self.endpoints[endpoint_uuid] = EndpointComposer(endpoint_uuid) return self.endpoints[endpoint_uuid] def _refresh_vlan_ids(self, service_vlan_id : int) -> None: # Only keep the service VLAN; others are ignored for composition self.vlan_ids = {service_vlan_id} def configure(self, device_obj : Device, settings : Optional[TreeNode]) -> None: self.objekt = device_obj for endpoint_obj in device_obj.device_endpoints: Loading @@ -139,37 +160,9 @@ class DeviceComposer: self.set_endpoint_alias(endpoint_obj.name, endpoint_uuid) self.get_endpoint(endpoint_obj.name).configure(endpoint_obj, None) # Find management interfaces mgmt_ifaces = set() for config_rule in device_obj.device_config.config_rules: if config_rule.action != ConfigActionEnum.CONFIGACTION_SET: continue if config_rule.WhichOneof('config_rule') != 'custom': continue config_rule_custom = config_rule.custom match = RE_IF.match(config_rule_custom.resource_key) if match is None: continue if_name = match.groups()[0] resource_value = json.loads(config_rule_custom.resource_value) management = resource_value.get('management', False) if management: mgmt_ifaces.add(if_name) # Find data plane interfaces for config_rule in device_obj.device_config.config_rules: if config_rule.action != ConfigActionEnum.CONFIGACTION_SET: continue if config_rule.WhichOneof('config_rule') != 'custom': continue config_rule_custom = config_rule.custom match = RE_SUBIF.match(config_rule_custom.resource_key) if match is not None: if_name, _ = match.groups() if if_name in mgmt_ifaces: continue resource_value = json.loads(config_rule_custom.resource_value) if 'vlan_id' not in resource_value: continue vlan_id = int(resource_value['vlan_id']) self.vlan_ids.add(vlan_id) endpoint = self.get_endpoint(if_name) endpoint.vlan_id = vlan_id 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 : int, delete : bool = False ) -> List[Dict]: SELECTED_DEVICES = { DeviceTypeEnum.PACKET_POP.value, DeviceTypeEnum.PACKET_ROUTER.value, Loading @@ -179,11 +172,12 @@ class DeviceComposer: json_config_rule = json_config_rule_delete if delete else json_config_rule_set config_rules : List[Dict] = list() self._refresh_vlan_ids(service_vlan_id) 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)) for vlan_id in self.vlan_ids: config_rules.extend(endpoint.get_config_rules(service_vlan_id, delete=delete)) for vlan_id in sorted(self.vlan_ids): vlan_name = 'tfs-vlan-{:s}'.format(str(vlan_id)) config_rules.append(json_config_rule(*_network_instance_vlan( network_instance_name, vlan_id, vlan_name=vlan_name Loading Loading @@ -224,27 +218,30 @@ class ConfigRuleComposer: def configure(self, service_obj : Service, settings : Optional[TreeNode]) -> None: self.objekt = service_obj if settings is None: return json_settings : Dict = settings.value if settings is None: raise Exception('Service settings are required to extract vlan_id') json_settings : Dict = settings.value or dict() if 'vlan_id' in json_settings: self.vlan_id = json_settings['vlan_id'] self.vlan_id = _safe_int(json_settings['vlan_id']) elif 'vlan-id' in json_settings: self.vlan_id = json_settings['vlan-id'] self.vlan_id = _safe_int(json_settings['vlan-id']) else: MSG = 'VLAN ID not found. Tried: vlan_id and vlan-id. service_obj={:s} settings={:s}' LOGGER.warning(MSG.format(grpc_message_to_json_string(service_obj), str(settings))) raise Exception(MSG.format(grpc_message_to_json_string(service_obj), str(settings))) if self.vlan_id is None: MSG = 'Invalid VLAN ID value in service settings: {:s}' raise Exception(MSG.format(str(json_settings))) def get_config_rules( self, network_instance_name : str = NETWORK_INSTANCE, delete : bool = False ) -> Dict[str, List[Dict]]: if self.vlan_id is None: MSG = 'VLAN ID not defined for service_obj={:s}' LOGGER.warning(MSG.format(grpc_message_to_json_string(self.objekt))) return dict() raise Exception('VLAN ID must be configured at service level before composing rules') return { device_uuid : device.get_config_rules(network_instance_name, delete=delete) device_uuid : device.get_config_rules(network_instance_name, self.vlan_id, delete=delete) for device_uuid, device in self.devices.items() } Loading
src/service/service/service_handlers/l2nm_gnmi_openconfig/VlanIdPropagator.py +14 −25 Original line number Diff line number Diff line Loading @@ -14,6 +14,7 @@ import json, logging from typing import List, Optional, Tuple from common.DeviceTypes import DeviceTypeEnum from .ConfigRuleComposer import ConfigRuleComposer LOGGER = logging.getLogger(__name__) Loading @@ -21,6 +22,16 @@ 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) Loading Loading @@ -71,28 +82,6 @@ class VlanIdPropagator: device_b = self._config_rule_composer.get_device(device_uuid_b) endpoint_b = device_b.get_endpoint(endpoint_uuid_b) svc_vlan_id = self._config_rule_composer.vlan_id ep_a_vlan_id = endpoint_a.vlan_id ep_b_vlan_id = endpoint_b.vlan_id if ep_a_vlan_id is None and ep_b_vlan_id is None: device_a.vlan_ids.add(svc_vlan_id) endpoint_a.vlan_id = svc_vlan_id device_b.vlan_ids.add(svc_vlan_id) endpoint_b.vlan_id = svc_vlan_id elif ep_a_vlan_id is not None and ep_b_vlan_id is None: if ep_a_vlan_id != svc_vlan_id: MSG = 'Incompatible VLAN-IDs: endpoint_a({:s}), service({:s})' raise Exception(MSG.format(str(endpoint_a), str(svc_vlan_id))) device_b.vlan_ids.add(svc_vlan_id) endpoint_b.vlan_id = svc_vlan_id elif ep_a_vlan_id is None and ep_b_vlan_id is not None: if ep_b_vlan_id != svc_vlan_id: MSG = 'Incompatible VLAN-IDs: endpoint_b({:s}), service({:s})' raise Exception(MSG.format(str(endpoint_b), str(svc_vlan_id))) device_a.vlan_ids.add(svc_vlan_id) endpoint_a.vlan_id = svc_vlan_id elif ep_a_vlan_id is not None and ep_b_vlan_id is not None: if ep_a_vlan_id != svc_vlan_id or ep_b_vlan_id != svc_vlan_id: MSG = 'Incompatible VLAN-IDs: endpoint_a({:s}), endpoint_a({:s}), service({:s})' raise Exception(MSG.format(str(endpoint_a), str(endpoint_b), str(svc_vlan_id))) if self._is_router_device(device_a) and self._is_router_device(device_b): endpoint_a.set_force_trunk() endpoint_b.set_force_trunk()
