e041dab724
- Replace placeholder deps with click, pygnmi, and rich for CLI functionality - Add fabric-orch CLI entry point via project.scripts - Configure hatchling as build backend with wheel/sdist targets - Add ruff linter configuration for code quality - Add dev dependency group with ruff - Alphabetize yang module imports
344 lines
11 KiB
Python
344 lines
11 KiB
Python
"""
|
|
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
|
|
|
|
# =============================================================================
|
|
# 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,
|
|
]
|