From f335d1fc33380ed933e569f1763a66b92cd9d7f7 Mon Sep 17 00:00:00 2001 From: Damien Arnodo Date: Fri, 26 Dec 2025 13:42:31 +0000 Subject: [PATCH] feat: add YANG path constants module Python module with validated gNMI YANG paths for: - Interfaces, Loopbacks, VLANs (OpenConfig) - BGP with neighbor and AFI-SAFI helpers (OpenConfig) - VXLAN VNI mappings (Arista experimental) - MLAG and EVPN config (Arista experimental) - Port-Channel/LAG paths - Subscription helpers for fabric monitoring Part of #3 --- src/yang/paths.py | 345 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 345 insertions(+) create mode 100644 src/yang/paths.py diff --git a/src/yang/paths.py b/src/yang/paths.py new file mode 100644 index 0000000..3719bba --- /dev/null +++ b/src/yang/paths.py @@ -0,0 +1,345 @@ +""" +YANG Path Constants for Arista EOS 4.35.0F + +This module provides validated gNMI YANG paths for managing +Arista EVPN-VXLAN fabrics via gNMI. + +Usage: + from yang.paths import Interfaces, BGP, VXLAN + + # Get interface state path + path = Interfaces.state("Ethernet1") + + # Get all BGP neighbors + path = BGP.NEIGHBORS +""" + +from dataclasses import dataclass +from typing import Optional + + +# ============================================================================= +# Interfaces (OpenConfig) +# ============================================================================= + +class Interfaces: + """OpenConfig interface paths.""" + + # Base paths + ROOT = "/interfaces/interface" + + @staticmethod + def interface(name: str) -> str: + """Get path for specific interface.""" + return f"/interfaces/interface[name={name}]" + + @staticmethod + def config(name: str) -> str: + """Get interface config path.""" + return f"/interfaces/interface[name={name}]/config" + + @staticmethod + def state(name: str) -> str: + """Get interface state path.""" + return f"/interfaces/interface[name={name}]/state" + + @staticmethod + def oper_status(name: str) -> str: + """Get interface operational status path.""" + return f"/interfaces/interface[name={name}]/state/oper-status" + + @staticmethod + def admin_status(name: str) -> str: + """Get interface admin status path.""" + return f"/interfaces/interface[name={name}]/state/admin-status" + + @staticmethod + def counters(name: str) -> str: + """Get interface counters path.""" + return f"/interfaces/interface[name={name}]/state/counters" + + +# ============================================================================= +# Loopbacks (OpenConfig) +# ============================================================================= + +class Loopbacks: + """Loopback interface paths.""" + + @staticmethod + def interface(index: int) -> str: + """Get loopback interface path.""" + return f"/interfaces/interface[name=Loopback{index}]" + + @staticmethod + def config(index: int) -> str: + """Get loopback config path.""" + return f"/interfaces/interface[name=Loopback{index}]/config" + + # Common loopbacks + LOOPBACK0 = "/interfaces/interface[name=Loopback0]" + LOOPBACK1 = "/interfaces/interface[name=Loopback1]" # VTEP source + + +# ============================================================================= +# VLANs (OpenConfig) +# ============================================================================= + +class VLANs: + """OpenConfig VLAN paths.""" + + # Base path + ROOT = "/network-instances/network-instance[name=default]/vlans/vlan" + + @staticmethod + def vlan(vlan_id: int) -> str: + """Get specific VLAN path.""" + return f"/network-instances/network-instance[name=default]/vlans/vlan[vlan-id={vlan_id}]" + + @staticmethod + def config(vlan_id: int) -> str: + """Get VLAN config path.""" + return f"/network-instances/network-instance[name=default]/vlans/vlan[vlan-id={vlan_id}]/config" + + @staticmethod + def svi(vlan_id: int) -> str: + """Get SVI interface path.""" + return f"/interfaces/interface[name=Vlan{vlan_id}]" + + +# ============================================================================= +# BGP (OpenConfig) +# ============================================================================= + +class BGP: + """OpenConfig BGP paths.""" + + # Base path for BGP + _BASE = "/network-instances/network-instance[name=default]/protocols/protocol[identifier=BGP][name=BGP]/bgp" + + # Global paths + GLOBAL = f"{_BASE}/global" + GLOBAL_CONFIG = f"{_BASE}/global/config" + GLOBAL_STATE = f"{_BASE}/global/state" + + # Neighbors + NEIGHBORS = f"{_BASE}/neighbors/neighbor" + + @staticmethod + def neighbor(address: str) -> str: + """Get specific neighbor path.""" + return f"{BGP._BASE}/neighbors/neighbor[neighbor-address={address}]" + + @staticmethod + def neighbor_config(address: str) -> str: + """Get neighbor config path.""" + return f"{BGP._BASE}/neighbors/neighbor[neighbor-address={address}]/config" + + @staticmethod + def neighbor_state(address: str) -> str: + """Get neighbor state path.""" + return f"{BGP._BASE}/neighbors/neighbor[neighbor-address={address}]/state" + + @staticmethod + def neighbor_session_state(address: str) -> str: + """Get neighbor session state path.""" + return f"{BGP._BASE}/neighbors/neighbor[neighbor-address={address}]/state/session-state" + + @staticmethod + def neighbor_afi_safi(address: str, afi_safi: str) -> str: + """ + Get neighbor AFI-SAFI path. + + Args: + address: Neighbor IP address + afi_safi: AFI-SAFI name (e.g., 'IPV4_UNICAST', 'L2VPN_EVPN') + """ + return f"{BGP._BASE}/neighbors/neighbor[neighbor-address={address}]/afi-safis/afi-safi[afi-safi-name={afi_safi}]" + + +# AFI-SAFI constants +class AfiSafi: + """BGP AFI-SAFI identifiers.""" + + IPV4_UNICAST = "IPV4_UNICAST" + IPV6_UNICAST = "IPV6_UNICAST" + L2VPN_EVPN = "L2VPN_EVPN" + + +# ============================================================================= +# VXLAN (Arista Experimental) +# ============================================================================= + +class VXLAN: + """ + Arista VXLAN paths. + + Model: arista-exp-eos-vxlan (augments openconfig-interfaces) + """ + + # Base paths + INTERFACE = "/interfaces/interface[name=Vxlan1]" + ROOT = "/interfaces/interface[name=Vxlan1]/arista-vxlan" + CONFIG = "/interfaces/interface[name=Vxlan1]/arista-vxlan/config" + STATE = "/interfaces/interface[name=Vxlan1]/arista-vxlan/state" + + # Specific config leaves + SOURCE_INTERFACE = "/interfaces/interface[name=Vxlan1]/arista-vxlan/config/src-ip-intf" + UDP_PORT = "/interfaces/interface[name=Vxlan1]/arista-vxlan/config/udp-port" + + # VNI mappings + VLAN_TO_VNIS = "/interfaces/interface[name=Vxlan1]/arista-vxlan/vlan-to-vnis" + VRF_TO_VNIS = "/interfaces/interface[name=Vxlan1]/arista-vxlan/vrf-to-vnis" + + @staticmethod + def vlan_to_vni(vlan_id: int) -> str: + """Get specific VLAN-to-VNI mapping path.""" + return f"/interfaces/interface[name=Vxlan1]/arista-vxlan/vlan-to-vnis/vlan-to-vni[vlan={vlan_id}]" + + @staticmethod + def vrf_to_vni(vrf_name: str) -> str: + """Get specific VRF-to-VNI mapping path.""" + return f"/interfaces/interface[name=Vxlan1]/arista-vxlan/vrf-to-vnis/vrf-to-vni[vrf={vrf_name}]" + + +# ============================================================================= +# MLAG (Arista Experimental) +# ============================================================================= + +class MLAG: + """ + Arista MLAG paths. + + Model: arista-exp-eos-mlag + + WARNING: Only config is exposed via gNMI. + State (peer status, role) requires eAPI. + """ + + ROOT = "/arista/eos/mlag" + CONFIG = "/arista/eos/mlag/config" + + # Config fields (for reference) + # - domain-id + # - local-intf + # - peer-address + # - peer-link-intf + # - dual-primary-action + # - dual-primary-detection-delay + # - heartbeat-peer-address + + +# ============================================================================= +# EVPN (Arista Experimental) +# ============================================================================= + +class EVPN: + """ + Arista EVPN paths. + + Model: arista-exp-eos-evpn + + WARNING: Only config is exposed via gNMI. + Learned routes/MACs require eAPI. + """ + + ROOT = "/arista/eos/evpn" + INSTANCES = "/arista/eos/evpn/evpn-instances" + + @staticmethod + def instance(name: str) -> str: + """Get specific EVPN instance path.""" + return f"/arista/eos/evpn/evpn-instances/evpn-instance[name={name}]" + + @staticmethod + def instance_config(name: str) -> str: + """Get EVPN instance config path.""" + return f"/arista/eos/evpn/evpn-instances/evpn-instance[name={name}]/config" + + @staticmethod + def route_target(name: str) -> str: + """Get EVPN instance route-target path.""" + return f"/arista/eos/evpn/evpn-instances/evpn-instance[name={name}]/route-target/config" + + +# ============================================================================= +# Port-Channel / LAG (OpenConfig) +# ============================================================================= + +class PortChannel: + """OpenConfig LAG/Port-Channel paths.""" + + @staticmethod + def interface(channel_id: int) -> str: + """Get port-channel interface path.""" + return f"/interfaces/interface[name=Port-Channel{channel_id}]" + + @staticmethod + def config(channel_id: int) -> str: + """Get port-channel config path.""" + return f"/interfaces/interface[name=Port-Channel{channel_id}]/config" + + @staticmethod + def aggregation(channel_id: int) -> str: + """Get LAG aggregation config path.""" + return f"/interfaces/interface[name=Port-Channel{channel_id}]/aggregation/config" + + @staticmethod + def aggregation_state(channel_id: int) -> str: + """Get LAG aggregation state path.""" + return f"/interfaces/interface[name=Port-Channel{channel_id}]/aggregation/state" + + +# ============================================================================= +# System (OpenConfig) +# ============================================================================= + +class System: + """OpenConfig system paths.""" + + ROOT = "/system" + CONFIG = "/system/config" + STATE = "/system/state" + HOSTNAME = "/system/config/hostname" + + +# ============================================================================= +# Subscription Helpers +# ============================================================================= + +@dataclass +class SubscriptionPath: + """Helper for building subscription paths.""" + + path: str + mode: str = "on-change" # on-change, sample, target-defined + + def __str__(self) -> str: + return self.path + + +# Common subscription paths for fabric monitoring +class FabricSubscriptions: + """Pre-defined subscription paths for fabric monitoring.""" + + # Interface state changes + INTERFACE_STATE = SubscriptionPath("/interfaces/interface/state") + + # BGP session state changes + BGP_SESSIONS = SubscriptionPath( + "/network-instances/network-instance[name=default]/protocols/protocol[identifier=BGP][name=BGP]/bgp/neighbors/neighbor/state/session-state" + ) + + # VXLAN VNI changes + VXLAN_VNIS = SubscriptionPath( + "/interfaces/interface[name=Vxlan1]/arista-vxlan/vlan-to-vnis" + ) + + @classmethod + def all(cls) -> list[SubscriptionPath]: + """Get all fabric subscription paths.""" + return [ + cls.INTERFACE_STATE, + cls.BGP_SESSIONS, + cls.VXLAN_VNIS, + ]