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 <noreply@anthropic.com>
2026-06-07 00:43:38 +02:00
parent cbe73cab64
commit 2a9ddac02d
1 changed files with 53 additions and 0 deletions
@@ -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)}