NAPALM/napalm-device-types
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add get_health_metrics() interface and UCD-MIB shared implementation

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
christianmanivong
2026-06-07 00:43:26 +02:00
co-authored by Claude Sonnet 4.6
parent 197e0b4ab7
commit 879419d5ba
7 changed files with 313 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
napalm_device_types/_ucd_metrics.py
+165
View File
@@ -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
napalm_device_types/access_point.py
+13
View File
@@ -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 {
napalm_device_types/firewall.py
+14
View File
@@ -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.
napalm_device_types/hypervisor.py
+13
View File
@@ -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
# ------------------------------------------------------------------
napalm_device_types/models.py
+44
View File
@@ -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]]
napalm_device_types/os.py
+54 -1
View File
@@ -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
# ------------------------------------------------------------------
napalm_device_types/switch.py
+10
View File
@@ -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.