From 2a9ddac02d04308ff7569d132ad540caf8453841 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Sun, 7 Jun 2026 00:43:38 +0200 Subject: [PATCH] feat: add get_health_metrics() via TP-Link proprietary OIDs SNMP health metrics using TP-Link enterprise MIB (OID 11863) for uptime, per-unit CPU utilization, and per-unit memory utilization. Co-Authored-By: Claude Sonnet 4.6 --- napalm_tplink_jetstream/tplink_jetstream.py | 53 +++++++++++++++++++++ 1 file changed, 53 insertions(+) diff --git a/napalm_tplink_jetstream/tplink_jetstream.py b/napalm_tplink_jetstream/tplink_jetstream.py index 9f4af01..a838d5f 100644 --- a/napalm_tplink_jetstream/tplink_jetstream.py +++ b/napalm_tplink_jetstream/tplink_jetstream.py @@ -1848,4 +1848,57 @@ class TPLinkJetstreamDriver(SwitchDriver): else: lines.append(f"[warn] Verification failed: {out[:100]}") + # ------------------------------------------------------------------ + # Health metrics (SNMP) + # ------------------------------------------------------------------ + + @classmethod + async def get_health_metrics(cls, snmp_get, snmp_walk) -> dict: + import asyncio + from napalm_device_types._ucd_metrics import build_if_metrics, ticks_to_seconds + + _TPLINK_SYS_UPTIME = "1.3.6.1.4.1.11863.6.1.1.9.0" + _TPLINK_CPU_1MIN = "1.3.6.1.4.1.11863.6.4.1.1.1.1.3" + _TPLINK_MEM_UTIL = "1.3.6.1.4.1.11863.6.4.1.2.1.1.2" + _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" + + (tp_uptime, cpu_rows, mem_rows, + descr, speed, in_oct, out_oct, in_err, out_err) = await asyncio.gather( + snmp_get(_TPLINK_SYS_UPTIME), + snmp_walk(_TPLINK_CPU_1MIN), + snmp_walk(_TPLINK_MEM_UTIL), + 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(tp_uptime) + if secs is not None: + metrics["uptime_seconds"] = secs + + if cpu_rows: + try: + metrics["cpu_percent"] = float(next(iter(cpu_rows.values()))) + except (ValueError, StopIteration): + pass + + if mem_rows: + try: + metrics["memory_percent"] = float(next(iter(mem_rows.values()))) + except (ValueError, StopIteration): + pass + + build_if_metrics(metrics, descr, speed, in_oct, out_oct, in_err, out_err) + return metrics + return {"success": success, "output": "\n".join(lines)}