Add capability for L2 VNIs

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

Introduce Layer-2 EVPN Features

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

Author: fzakfeld

osism/tasks/conductor/netbox.py

@@ -160,14 +160,19 @@ 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 = {}
+    l2vni_vlans = {}
+    # Map of NetBox VLAN object id -> vid, collected while iterating interfaces
+    vlan_obj_ids = {}
 
     try:
         # Use cached interfaces instead of separate query
@@ -204,6 +209,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 +229,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:
@@ -250,9 +257,7 @@ def get_device_vlans(device):
                     for ip_addr in ip_addresses:
                         if ip_addr.address:
                             role = getattr(ip_addr, "role", None)
-                            role_value = (
-                                getattr(role, "value", None) if role else None
-                            )
+                            role_value = getattr(role, "value", None) if role else None
                             if role_value == "anycast":
                                 anycast_addresses.append(ip_addr.address)
                             else:
@@ -264,18 +269,67 @@ def get_device_vlans(device):
                         if addresses:
                             vlan_interfaces[vid]["addresses"] = addresses
                         if anycast_addresses:
-                            vlan_interfaces[vid]["anycast_addresses"] = anycast_addresses
+                            vlan_interfaces[vid][
+                                "anycast_addresses"
+                            ] = anycast_addresses
+                        if hasattr(interface, "vrf") and interface.vrf:
+                            vlan_interfaces[vid]["vrf_name"] = interface.vrf.name
+                        # Extract default route nexthops from sonic_parameters
+
+                        sonic_params = (
+                            interface.custom_fields.get("sonic_parameters")
+                            if hasattr(interface, "custom_fields")
+                            and interface.custom_fields
+                            else None
+                        )
+
+                        if sonic_params:
+                            if (
+                                "default_route_ipv4" in sonic_params
+                                and sonic_params["default_route_ipv4"]
+                            ):
+                                vlan_interfaces[vid]["default_route_ipv4"] = (
+                                    sonic_params["default_route_ipv4"]
+                                )
+                            if (
+                                "default_route_ipv6" in sonic_params
+                                and sonic_params["default_route_ipv6"]
+                            ):
+                                vlan_interfaces[vid]["default_route_ipv6"] = (
+                                    sonic_params["default_route_ipv6"]
+                                )
+
                 except (ValueError, IndexError):
                     # 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

@@ -34,7 +34,12 @@
     get_connected_interface_ipv4_address,
 )
 from .cache import get_cached_device_interfaces
-from .constants import BGP_AF_L2VPN_EVPN_TAG, DEFAULT_SONIC_ROLES
+from .constants import (
+    BGP_AF_L2VPN_EVPN_TAG,
+    DEFAULT_SONIC_ROLES,
+    DEFAULT_EVPN_SYSTEM_MAC,
+    DEFAULT_SAG_MAC,
+)
 
 # Global cache for NTP servers to avoid multiple queries
 _ntp_servers_cache = None
@@ -73,16 +78,7 @@ def generate_sonic_config(device, hwsku, device_as_mapping=None, config_version=
     # Get port channel configuration from NetBox first (needed by get_connected_interfaces)
     portchannel_info = detect_port_channels(device)
 
-    # Resolve evpn_system_mac early so it is validated once and passed explicitly later
-    _raw_evpn_mac = device.config_context.get("_evpn_system_mac")
-    evpn_system_mac = (
-        _raw_evpn_mac if isinstance(_raw_evpn_mac, str) and _raw_evpn_mac else None
-    )
-    if _raw_evpn_mac and not evpn_system_mac:
-        logger.warning(
-            f"Device {device.name}: '_evpn_system_mac' in config_context is not a valid string"
-            f" (got {type(_raw_evpn_mac).__name__!r}), ignoring"
-        )
+    evpn_system_mac = DEFAULT_EVPN_SYSTEM_MAC
 
     # Get connected interfaces to determine admin_status
     connected_interfaces, connected_portchannels = get_connected_interfaces(
@@ -270,7 +266,8 @@ def generate_sonic_config(device, hwsku, device_as_mapping=None, config_version=
         config["MGMT_INTERFACE"]["eth0"] = {"admin_status": "up"}
         config["MGMT_INTERFACE"][f"eth0|{oob_ip}/{prefix_len}"] = {}
         metalbox_ip = _get_metalbox_ip_for_device(device)
-        config["STATIC_ROUTE"] = {}
+        if "STATIC_ROUTE" not in config:
+            config["STATIC_ROUTE"] = {}
         config["STATIC_ROUTE"]["mgmt|0.0.0.0/0"] = {"nexthop": metalbox_ip}
     else:
         oob_ip = None
@@ -288,7 +285,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:
@@ -1756,7 +1753,11 @@ def _add_vlan_configuration(config, vlan_info, netbox_interfaces, device):
 
         if addresses or anycast_addresses:
             # Add the VLAN interface base entry
-            config["VLAN_INTERFACE"][vlan_name] = {"admin_status": "up"}
+            vlan_iface_entry = {"admin_status": "up"}
+            vrf_name = interface_data.get("vrf_name")
+            if vrf_name:
+                vlan_iface_entry["vrf_name"] = vrf_name
+            config["VLAN_INTERFACE"][vlan_name] = vlan_iface_entry
 
         # Add regular IP configuration for each address (IPv4 and IPv6)
         for address in addresses:
@@ -1786,20 +1787,40 @@ def _add_vlan_configuration(config, vlan_info, netbox_interfaces, device):
                 config["SAG"][f"{vlan_name}|IPv6"] = {"gwip": ipv6_anycast}
 
     if sag_enabled:
-        gwmac = device.config_context.get("_sag_gwmac")
-        if not gwmac:
-            raise ValueError(
-                f"Device {device.name} has SAG anycast addresses but no '_sag_gwmac' "
-                "defined in its config context"
-            )
         if "SAG_GLOBAL" not in config:
             config["SAG_GLOBAL"] = {}
         config["SAG_GLOBAL"]["IP"] = {
             "IPv4": "enable",
             "IPv6": "enable",
-            "gwmac": gwmac,
+            "gwmac": DEFAULT_SAG_MAC,
         }
 
+    # Add static default routes per VRF from sonic_parameters on VLAN interfaces
+    for vid, interface_data in vlan_info["vlan_interfaces"].items():
+        vrf_name = interface_data.get("vrf_name")
+        if not vrf_name:
+            continue
+        logger.debug(f"Adding static default routes for VRF {vrf_name} (Vlan{vid})")
+
+        default_route_ipv4 = interface_data.get("default_route_ipv4")
+        default_route_ipv6 = interface_data.get("default_route_ipv6")
+        if not default_route_ipv4 and not default_route_ipv6:
+            continue
+        if "STATIC_ROUTE" not in config:
+            config["STATIC_ROUTE"] = {}
+        if default_route_ipv4:
+            config["STATIC_ROUTE"][f"{vrf_name}|0.0.0.0/0"] = {
+                "nexthop": default_route_ipv4
+            }
+            logger.debug(
+                f"Added static IPv4 default route for VRF {vrf_name} via {default_route_ipv4} (Vlan{vid})"
+            )
+        if default_route_ipv6:
+            config["STATIC_ROUTE"][f"{vrf_name}|::/0"] = {"nexthop": default_route_ipv6}
+            logger.debug(
+                f"Added static IPv6 default route for VRF {vrf_name} via {default_route_ipv6} (Vlan{vid})"
+            )
+
 
 def _add_loopback_configuration(config, loopback_info):
     """Add Loopback configuration from NetBox."""
@@ -1978,12 +1999,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
@@ -2070,8 +2092,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", "")
 
@@ -2090,7 +2115,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}"
@@ -2101,6 +2126,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
@@ -2119,7 +2160,7 @@ def _add_vrf_configuration(config, vrf_info, netbox_interfaces):
             )
 
 
-def _add_portchannel_configuration(config, portchannel_info, evpn_system_mac=None):
+def _add_portchannel_configuration(config, portchannel_info, evpn_system_mac):
     """Add port channel configuration from NetBox."""
     if portchannel_info["portchannels"]:
         for pc_name, pc_data in portchannel_info["portchannels"].items():
@@ -2134,6 +2175,18 @@ def _add_portchannel_configuration(config, portchannel_info, evpn_system_mac=Non
                 pc_config["system_mac"] = evpn_system_mac
             config["PORTCHANNEL"][pc_name] = pc_config
 
+            # Add EVPN_ETHERNET_SEGMENT configuration for EVPN multihoming LAGs
+            if pc_data.get("evpn_lag"):
+                if "EVPN_ETHERNET_SEGMENT" not in config:
+                    config["EVPN_ETHERNET_SEGMENT"] = {}
+                config["EVPN_ETHERNET_SEGMENT"][pc_name] = {
+                    "esi": "AUTO",
+                    "esi_type": "TYPE_3_MAC_BASED",
+                    "ifname": pc_name,
+                }
+                if "EVPN_MH_GLOBAL" not in config:
+                    config["EVPN_MH_GLOBAL"] = {"default": {"startup_delay": "300"}}
+
             # Add PORTCHANNEL_INTERFACE configuration to enable IPv6 link-local
             config["PORTCHANNEL_INTERFACE"][pc_name] = {
                 "ipv6_use_link_local_only": "enable"

osism/tasks/conductor/sonic/constants.py

@@ -8,9 +8,20 @@
 # 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
 
+# Default Base MAC for EVPN PortChannels (Calculated with each PortChannel Index)
+# using a locally administered MAC adress range
+DEFAULT_EVPN_SYSTEM_MAC = "02:00:00:00:00:00"
+
+# Default MAC for Static Anycast Gateway (L2 Anycast Gateway)
+# using a locally administered MAC adress range
+DEFAULT_SAG_MAC = "02:00:10:00:00:00"
+
 # Default SONiC device roles
 DEFAULT_SONIC_ROLES = [
     "accessleaf",