Add capability for L2 VNIs

AI-assisted: Claude Code
Signed-off-by: Freerk-Ole Zakfeld <fzakfeld@scaleuptech.com>

Author: fzakfeld

osism/tasks/conductor/netbox.py

@@ -160,14 +160,18 @@ def get_device_vlans(device):
               {
                   'vlans': {vid: {'name': name, 'description': desc}},
                   'vlan_members': {vid: {'port_name': 'tagging_mode'}},
-                  'vlan_interfaces': {vid: {'addresses': [ip_with_prefix, ...]}}
+                  'vlan_interfaces': {vid: {'addresses': [ip_with_prefix, ...]}},
+                  'l2vni_vlans': {vid: vni}  -- VLANs tagged evpn-l2vni (VNI == VID)
               }
     """
     from .sonic.cache import get_cached_device_interfaces
+    from .sonic.constants import EVPN_L2VNI_TAG
 
     vlans = {}
     vlan_members = {}
     vlan_interfaces = {}
+    # Map of NetBox VLAN object id -> vid, collected while iterating interfaces
+    vlan_obj_ids = {}
 
     try:
         # Use cached interfaces instead of separate query
@@ -204,6 +208,7 @@ def get_device_vlans(device):
                         "name": vlan.name or f"Vlan{vid}",
                         "description": vlan.description or "",
                     }
+                vlan_obj_ids[vlan.id] = vid
 
                 # Add interface to VLAN members as untagged
                 if vid not in vlan_members:
@@ -223,6 +228,7 @@ def get_device_vlans(device):
                             "name": vlan.name or f"Vlan{vid}",
                             "description": vlan.description or "",
                         }
+                    vlan_obj_ids[vlan.id] = vid
 
                     # Add interface to VLAN members as tagged
                     if vid not in vlan_members:
@@ -259,13 +265,33 @@ def get_device_vlans(device):
                     # Skip if interface name doesn't follow Vlan<number> pattern
                     pass
 
+        # Determine which VLANs are tagged evpn-l2vni by fetching full VLAN objects
+        l2vni_vlans = {}
+        if vlan_obj_ids:
+            try:
+                full_vlans = list(
+                    utils.nb.ipam.vlans.filter(id=list(vlan_obj_ids.keys()))
+                )
+                for v in full_vlans:
+                    if any(
+                        getattr(t, "slug", None) == EVPN_L2VNI_TAG
+                        for t in getattr(v, "tags", [])
+                    ):
+                        l2vni_vlans[v.vid] = v.vid  # VNI equals VID
+                        logger.debug(
+                            f"VLAN {v.vid} tagged {EVPN_L2VNI_TAG}, will add L2 VXLAN_TUNNEL_MAP entry"
+                        )
+            except Exception as e:
+                logger.warning(f"Could not fetch VLAN tags for L2 VNI check: {e}")
+
     except Exception as e:
         logger.warning(f"Could not get VLANs for device {device.name}: {e}")
 
     return {
         "vlans": vlans,
         "vlan_members": vlan_members,
         "vlan_interfaces": vlan_interfaces,
+        "l2vni_vlans": l2vni_vlans,
     }
 
 

osism/tasks/conductor/sonic/config_generator.py

@@ -278,7 +278,7 @@ def generate_sonic_config(device, hwsku, device_as_mapping=None, config_version=
     _add_portchannel_configuration(config, portchannel_info, evpn_system_mac)
 
     # Add VRF configuration
-    _add_vrf_configuration(config, vrf_info, netbox_interfaces)
+    _add_vrf_configuration(config, vrf_info, vlan_info, netbox_interfaces)
 
     # Set DATABASE VERSION from config_version parameter or default
     if "VERSIONS" not in config:
@@ -1927,12 +1927,13 @@ def _get_vrf_info(device):
     return vrf_info
 
 
-def _add_vrf_configuration(config, vrf_info, netbox_interfaces):
+def _add_vrf_configuration(config, vrf_info, vlan_info, netbox_interfaces):
     """Add VRF configuration to config.
 
     Args:
         config: Configuration dictionary to update
         vrf_info: VRF information dictionary from _get_vrf_info()
+        vlan_info: VLAN information dictionary from get_device_vlans()
         netbox_interfaces: Dict mapping SONiC names to NetBox interface info
     """
     # Track VRFs with VNI for VXLAN configuration
@@ -2019,8 +2020,11 @@ def _add_vrf_configuration(config, vrf_info, netbox_interfaces):
             config["BGP_GLOBALS"][vrf_name] = copy.deepcopy(default_bgp)
             logger.info(f"Added BGP_GLOBALS for VRF {vrf_name}")
 
-    # Add VXLAN configuration if there are VRFs with VNI
-    if vrfs_with_vni:
+    # Collect L2 VNI VLANs (tagged evpn-l2vni in NetBox, VNI == VID)
+    l2vni_vlans = vlan_info.get("l2vni_vlans", {})
+
+    # Add VXLAN configuration if there are VRFs with VNI or L2 VNI VLANs
+    if vrfs_with_vni or l2vni_vlans:
         # Get source IP from BGP_GLOBALS default router_id
         src_ip = config.get("BGP_GLOBALS", {}).get("default", {}).get("router_id", "")
 
@@ -2039,7 +2043,7 @@ def _add_vrf_configuration(config, vrf_info, netbox_interfaces):
         }
         logger.info(f"Added VXLAN_EVPN_NVO nvo1 with source_vtep {VXLAN_VTEP_NAME}")
 
-        # Add VXLAN_TUNNEL_MAP for each VRF with VNI
+        # Add VXLAN_TUNNEL_MAP for each VRF with VNI (L3 / IRB)
         for vrf_entry in vrfs_with_vni:
             vni = vrf_entry["vni"]
             vlan_name = f"Vlan{vni}"
@@ -2050,6 +2054,22 @@ def _add_vrf_configuration(config, vrf_info, netbox_interfaces):
             }
             logger.info(f"Added VXLAN_TUNNEL_MAP {map_key}")
 
+        # Add VXLAN_TUNNEL_MAP for each L2 VNI VLAN (pure L2, no VRF assignment)
+        vrf_vnis = {entry["vni"] for entry in vrfs_with_vni}
+        for vid, vni in l2vni_vlans.items():
+            if vni in vrf_vnis:
+                logger.debug(
+                    f"Skipping L2 VNI {vni} for Vlan{vid}: already covered by VRF tunnel map"
+                )
+                continue
+            vlan_name = f"Vlan{vid}"
+            map_key = f"{VXLAN_VTEP_NAME}|map_{vni}_{vlan_name}"
+            config["VXLAN_TUNNEL_MAP"][map_key] = {
+                "vlan": vlan_name,
+                "vni": str(vni),
+            }
+            logger.info(f"Added L2 VXLAN_TUNNEL_MAP {map_key}")
+
     # Add VRF assignments to interfaces
     for sonic_interface, vrf_name in vrf_info["interface_vrf_mapping"].items():
         # Check if this is a regular interface

osism/tasks/conductor/sonic/constants.py

@@ -8,6 +8,9 @@
 # Tag to enable EVPN Multihoming (evpn-lag mode) on a port channel
 EVPN_LAG_TAG = "evpn-lag"
 
+# Tag to enable L2 VxLAN (EVPN L2 VNI) for a VLAN — VNI equals VLAN ID
+EVPN_L2VNI_TAG = "evpn-l2vni"
+
 # Default AS prefix for local ASN calculation
 DEFAULT_LOCAL_AS_PREFIX = 4200