From 879419d5bac15a8d09bc62150f4473f278adb3c1 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Sun, 7 Jun 2026 00:43:26 +0200 Subject: [PATCH] feat: add get_health_metrics() interface and UCD-MIB shared implementation Defines the driver-level health metrics interface across all base classes. OSDriver/FirewallDriver/HypervisorDriver/AccessPointDriver get a default UCD-MIB + IF-MIB implementation via shared _ucd_metrics.py; SwitchDriver raises NotImplementedError (vendor-proprietary OIDs). Adds HealthMetricsDict and HealthMetricsIfaceDict TypedDicts to models.py. Co-Authored-By: Claude Sonnet 4.6 --- napalm_device_types/_ucd_metrics.py | 165 ++++++++++++++++++++++++++++ napalm_device_types/access_point.py | 13 +++ napalm_device_types/firewall.py | 14 +++ napalm_device_types/hypervisor.py | 13 +++ napalm_device_types/models.py | 44 ++++++++ napalm_device_types/os.py | 55 +++++++++- napalm_device_types/switch.py | 10 ++ 7 files changed, 313 insertions(+), 1 deletion(-) create mode 100644 napalm_device_types/_ucd_metrics.py diff --git a/napalm_device_types/_ucd_metrics.py b/napalm_device_types/_ucd_metrics.py new file mode 100644 index 0000000..e67f0a8 --- /dev/null +++ b/napalm_device_types/_ucd_metrics.py @@ -0,0 +1,165 @@ +"""Shared SNMP metric collection helpers for UCD-MIB and IF-MIB.""" +from __future__ import annotations + +import asyncio +import re +from typing import Awaitable, Callable, Dict, Optional + +# ── UCD-MIB OIDs ────────────────────────────────────────────────────────────── +OID_SYS_UPTIME = "1.3.6.1.2.1.25.1.1.0" # hrSystemUptime (centiseconds) +OID_CPU_IDLE = "1.3.6.1.4.1.2021.11.11.0" # UCD ssCpuIdle (%) +OID_MEM_TOTAL = "1.3.6.1.4.1.2021.4.5.0" # memTotalReal (kB) +OID_MEM_FREE = "1.3.6.1.4.1.2021.4.6.0" # memAvailReal (kB) +OID_MEM_BUFFER = "1.3.6.1.4.1.2021.4.14.0" # memBuffer (kB) +OID_MEM_CACHED = "1.3.6.1.4.1.2021.4.15.0" # memCached (kB) +OID_SWAP_TOTAL = "1.3.6.1.4.1.2021.4.3.0" # memTotalSwap (kB) +OID_SWAP_AVAIL = "1.3.6.1.4.1.2021.4.4.0" # memAvailSwap (kB) +OID_LOAD_1 = "1.3.6.1.4.1.2021.10.1.3.1" # laLoad 1-min +OID_LOAD_5 = "1.3.6.1.4.1.2021.10.1.3.2" # laLoad 5-min +OID_LOAD_15 = "1.3.6.1.4.1.2021.10.1.3.3" # laLoad 15-min + +# ── IF-MIB OIDs ─────────────────────────────────────────────────────────────── +OID_IF_DESCR = "1.3.6.1.2.1.2.2.1.2" +OID_IF_SPEED = "1.3.6.1.2.1.2.2.1.5" +OID_IF_IN_OCT = "1.3.6.1.2.1.2.2.1.10" +OID_IF_OUT_OCT = "1.3.6.1.2.1.2.2.1.16" +OID_IF_IN_ERR = "1.3.6.1.2.1.2.2.1.14" +OID_IF_OUT_ERR = "1.3.6.1.2.1.2.2.1.20" + +IF_SKIP_DEFAULT = re.compile(r'^(lo|sit\d|tun\d|docker|veth|br-|virbr)', re.IGNORECASE) + +SnmpGetFn = Callable[[str], Awaitable[Optional[str]]] +SnmpWalkFn = Callable[[str], Awaitable[Dict[str, str]]] + + +def ticks_to_seconds(raw: str | None) -> int | None: + if raw is None: + return None + try: + return int(raw) // 100 + except (ValueError, TypeError): + return None + + +def build_if_metrics( + metrics: dict, + descr: Dict[str, str], + speed: Dict[str, str], + in_oct: Dict[str, str], + out_oct: Dict[str, str], + in_err: Dict[str, str], + out_err: Dict[str, str], + *, + if_skip: re.Pattern | None = None, + tx_err_is_drop: bool = False, +) -> None: + if if_skip is None: + if_skip = IF_SKIP_DEFAULT + interfaces: dict = {} + for idx, name in descr.items(): + if if_skip.match(name): + continue + iface: dict = {"name": name} + try: + iface["rx_bytes"] = int(in_oct.get(idx, 0) or 0) + iface["tx_bytes"] = int(out_oct.get(idx, 0) or 0) + iface["rx_errors"] = int(in_err.get(idx, 0) or 0) + tx_val = int(out_err.get(idx, 0) or 0) + if tx_err_is_drop: + iface["tx_errors"] = 0 + iface["tx_drops"] = tx_val + else: + iface["tx_errors"] = tx_val + spd = speed.get(idx) + iface["speed_mbps"] = int(spd) // 1_000_000 if spd and int(spd) > 0 else 0 + except (ValueError, TypeError): + pass + interfaces[name] = iface + if interfaces: + metrics["interfaces"] = interfaces + + +async def collect_ucd_metrics( + snmp_get: SnmpGetFn, + snmp_walk: SnmpWalkFn, + *, + tx_err_is_drop: bool = False, + if_skip: re.Pattern | None = None, +) -> dict: + """Collect UCD-MIB system metrics + IF-MIB interface counters in parallel.""" + ( + uptime_raw, cpu_idle_raw, + mem_total_raw, mem_free_raw, mem_buf_raw, mem_cache_raw, + swap_total_raw, swap_avail_raw, + load1_raw, load5_raw, load15_raw, + ), (descr, speed, in_oct, out_oct, in_err, out_err) = await asyncio.gather( + asyncio.gather( + snmp_get(OID_SYS_UPTIME), + snmp_get(OID_CPU_IDLE), + snmp_get(OID_MEM_TOTAL), + snmp_get(OID_MEM_FREE), + snmp_get(OID_MEM_BUFFER), + snmp_get(OID_MEM_CACHED), + snmp_get(OID_SWAP_TOTAL), + snmp_get(OID_SWAP_AVAIL), + snmp_get(OID_LOAD_1), + snmp_get(OID_LOAD_5), + snmp_get(OID_LOAD_15), + ), + asyncio.gather( + snmp_walk(OID_IF_DESCR), + snmp_walk(OID_IF_SPEED), + snmp_walk(OID_IF_IN_OCT), + snmp_walk(OID_IF_OUT_OCT), + snmp_walk(OID_IF_IN_ERR), + snmp_walk(OID_IF_OUT_ERR), + ), + ) + + metrics: dict = {} + + secs = ticks_to_seconds(uptime_raw) + if secs is not None: + metrics["uptime_seconds"] = secs + + if cpu_idle_raw is not None: + try: + metrics["cpu_percent"] = round(100.0 - float(cpu_idle_raw), 1) + except ValueError: + pass + + if mem_total_raw and mem_free_raw: + try: + total_kb = int(mem_total_raw) + free_kb = int(mem_free_raw) + buf_kb = int(mem_buf_raw) if mem_buf_raw else 0 + cache_kb = int(mem_cache_raw) if mem_cache_raw else 0 + used_kb = max(0, total_kb - free_kb - buf_kb - cache_kb) + metrics["memory_total_bytes"] = total_kb * 1024 + metrics["memory_used_bytes"] = used_kb * 1024 + metrics["memory_percent"] = round(used_kb / total_kb * 100, 1) if total_kb else 0.0 + metrics["memory_free_bytes"] = free_kb * 1024 + except ValueError: + pass + + if swap_total_raw and swap_avail_raw: + try: + stotal = int(swap_total_raw) + savail = int(swap_avail_raw) + sused = stotal - savail + metrics["swap_total_bytes"] = stotal * 1024 + metrics["swap_used_bytes"] = sused * 1024 + metrics["swap_percent"] = round(sused / stotal * 100, 1) if stotal else 0.0 + except ValueError: + pass + + for key, raw in [("load_1", load1_raw), ("load_5", load5_raw), ("load_15", load15_raw)]: + if raw is not None: + try: + metrics[key] = float(raw) + except ValueError: + pass + + build_if_metrics(metrics, descr, speed, in_oct, out_oct, in_err, out_err, + if_skip=if_skip, tx_err_is_drop=tx_err_is_drop) + return metrics diff --git a/napalm_device_types/access_point.py b/napalm_device_types/access_point.py index 52148cf..99097f1 100644 --- a/napalm_device_types/access_point.py +++ b/napalm_device_types/access_point.py @@ -12,9 +12,11 @@ Usage:: from typing import Any, Dict, List from napalm.base import NetworkDriver +from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics from napalm_device_types.models import ( Dot1XConfigDict, FastTransitionConfigDict, + HealthMetricsDict, MACACLDict, MeshConfigDict, MeshPeerDict, @@ -45,6 +47,17 @@ class AccessPointDriver(NetworkDriver): # devices and have no IP/Ethernet significance at the AP level). _EXCLUDED_INTERFACE_PREFIXES: tuple = ("phy",) + _SNMP_SKIP_IF = IF_SKIP_DEFAULT + _SNMP_TX_ERR_IS_DROP: bool = False + + @classmethod + async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict: + return await collect_ucd_metrics( + snmp_get, snmp_walk, + tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP, + if_skip=cls._SNMP_SKIP_IF, + ) + def _filter_interfaces(self, interfaces: Dict[str, Any]) -> Dict[str, Any]: """Remove loopback and radio-device (phy*) interfaces from an interface dict.""" return { diff --git a/napalm_device_types/firewall.py b/napalm_device_types/firewall.py index cdb4a5f..d143f30 100644 --- a/napalm_device_types/firewall.py +++ b/napalm_device_types/firewall.py @@ -12,7 +12,9 @@ Usage:: from typing import Any, Dict, List from napalm.base import NetworkDriver +from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics from napalm_device_types.models import ( + HealthMetricsDict, NATTranslationDict, PackageDict, SecurityZoneDict, @@ -30,6 +32,18 @@ class FirewallDriver(NetworkDriver): that concrete drivers must implement. """ + _SNMP_SKIP_IF = IF_SKIP_DEFAULT + # OPNsense reports drops in the out-error counter. + _SNMP_TX_ERR_IS_DROP: bool = True + + @classmethod + async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict: + return await collect_ucd_metrics( + snmp_get, snmp_walk, + tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP, + if_skip=cls._SNMP_SKIP_IF, + ) + def get_nat_translations(self) -> List[NATTranslationDict]: """ Returns a list of active NAT translation entries. diff --git a/napalm_device_types/hypervisor.py b/napalm_device_types/hypervisor.py index 11ffd86..b2c4f6d 100644 --- a/napalm_device_types/hypervisor.py +++ b/napalm_device_types/hypervisor.py @@ -12,7 +12,9 @@ Usage:: from typing import Any, Dict, List from napalm.base import NetworkDriver +from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics from napalm_device_types.models import ( + HealthMetricsDict, PackageDict, SnapshotDict, StorageVolumeDict, @@ -31,6 +33,17 @@ class HypervisorDriver(NetworkDriver): hypervisor-specific operations that concrete drivers must implement. """ + _SNMP_SKIP_IF = IF_SKIP_DEFAULT + _SNMP_TX_ERR_IS_DROP: bool = False + + @classmethod + async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict: + return await collect_ucd_metrics( + snmp_get, snmp_walk, + tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP, + if_skip=cls._SNMP_SKIP_IF, + ) + # ------------------------------------------------------------------ # Virtual machines – read # ------------------------------------------------------------------ diff --git a/napalm_device_types/models.py b/napalm_device_types/models.py index 7e077f8..2996e1b 100644 --- a/napalm_device_types/models.py +++ b/napalm_device_types/models.py @@ -611,3 +611,47 @@ class DeviceActionResultDict(TypedDict): success: bool # True if the action completed without error output: str # human-readable output or status message + + +class SNMPConfigDict(TypedDict): + """SNMP agent configuration detected on the device.""" + + running: bool # True if the SNMP daemon is active + community: str # read community string (e.g. "public") + port: int # listening port (default 161) + version: str # highest supported version: "1", "2c", or "3" + + +# --------------------------------------------------------------------------- +# Health Metrics (returned by get_health_metrics()) +# --------------------------------------------------------------------------- + + +class HealthMetricsIfaceDict(TypedDict): + """Per-interface counters inside a HealthMetricsDict.""" + + name: str + rx_bytes: NotRequired[int] + tx_bytes: NotRequired[int] + rx_errors: NotRequired[int] + tx_errors: NotRequired[int] + tx_drops: NotRequired[int] # populated instead of tx_errors when the driver reports drops + speed_mbps: NotRequired[int] + + +class HealthMetricsDict(TypedDict): + """Return value of ``get_health_metrics()``.""" + + uptime_seconds: NotRequired[int] + cpu_percent: NotRequired[float] + memory_total_bytes: NotRequired[int] + memory_used_bytes: NotRequired[int] + memory_free_bytes: NotRequired[int] + memory_percent: NotRequired[float] + swap_total_bytes: NotRequired[int] + swap_used_bytes: NotRequired[int] + swap_percent: NotRequired[float] + load_1: NotRequired[float] + load_5: NotRequired[float] + load_15: NotRequired[float] + interfaces: NotRequired[Dict[str, HealthMetricsIfaceDict]] diff --git a/napalm_device_types/os.py b/napalm_device_types/os.py index 2514bd5..6f526f7 100644 --- a/napalm_device_types/os.py +++ b/napalm_device_types/os.py @@ -10,16 +10,20 @@ Usage:: ... """ -from typing import List +import re +from typing import List, Optional from napalm.base import NetworkDriver +from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics from napalm_device_types.models import ( ApplyUpdatesResultDict, CronJobDict, DeviceActionResultDict, DockerInfoDict, + HealthMetricsDict, PackageDict, ProcessDict, ServiceDict, + SNMPConfigDict, UpdateDict, UserDict, ) @@ -34,6 +38,24 @@ class OSDriver(NetworkDriver): operations that concrete drivers must implement. """ + # Interfaces matching this pattern are excluded from health-metric collection. + _SNMP_SKIP_IF: re.Pattern = IF_SKIP_DEFAULT + # Set True for drivers where the out-error counter actually reports drops (e.g. OPNsense). + _SNMP_TX_ERR_IS_DROP: bool = False + + @classmethod + async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict: + """Collect SNMP health metrics (CPU, memory, uptime, load, interfaces). + + :param snmp_get: async callable ``(oid: str) -> Optional[str]`` + :param snmp_walk: async callable ``(oid: str) -> Dict[str, str]`` + """ + return await collect_ucd_metrics( + snmp_get, snmp_walk, + tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP, + if_skip=cls._SNMP_SKIP_IF, + ) + # ------------------------------------------------------------------ # Package management # ------------------------------------------------------------------ @@ -254,6 +276,37 @@ class OSDriver(NetworkDriver): """ raise NotImplementedError + # ------------------------------------------------------------------ + # SNMP + # ------------------------------------------------------------------ + + def get_snmp_config(self) -> Optional[SNMPConfigDict]: + """ + Returns the SNMP agent configuration currently active on the device, + or ``None`` if no SNMP daemon is running or detectable. + + The returned dictionary contains: + + * running (bool) - whether the SNMP daemon is currently active + * community (string) - the read community string (e.g. ``"public"``) + * port (int) - the UDP port the agent listens on (default ``161``) + * version (string) - highest supported SNMP version: ``"1"``, ``"2c"``, or ``"3"`` + + Example:: + + # snmpd running with community "public": + { + "running": True, + "community": "public", + "port": 161, + "version": "2c", + } + + # snmpd not installed / not running: + None + """ + raise NotImplementedError + # ------------------------------------------------------------------ # Docker # ------------------------------------------------------------------ diff --git a/napalm_device_types/switch.py b/napalm_device_types/switch.py index bac11ef..3d123fc 100644 --- a/napalm_device_types/switch.py +++ b/napalm_device_types/switch.py @@ -14,6 +14,7 @@ from typing import Dict from napalm.base import NetworkDriver from napalm_device_types.models import ( Dot1XPortDict, + HealthMetricsDict, InterfaceConfigDict, MACACLDict, PoESummaryDict, @@ -32,6 +33,15 @@ class SwitchDriver(NetworkDriver): operations that concrete drivers must implement. """ + @classmethod + async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict: + """Collect SNMP health metrics for this switch type. + + Switch vendors use proprietary OIDs — each concrete driver must + override this classmethod. + """ + raise NotImplementedError + def get_spanning_tree(self) -> Dict[str, SpanningTreeDict]: """ Returns spanning tree status for each STP instance.