CMK version: 2.5.0p3.ultimate
OS version: Ubuntu 24.04
I am trying to create a checkmk plugin to monitor WAN IN RX/TX DROP on Mikrotik routers.
so far i have this plugin:
#!/usr/bin/env python3
"""
CheckMK SNMP Check Plugin: MikroTik WAN Interface RX/TX Drop
=============================================================
Monitors sfp-sfpplus* drop counters dynamically — works regardless of
which row index the interface occupies in the MikroTik SNMP table.
Walks the full mtxrInterfaceStats table and keys by interface name.
Only sfp-sfpplus* interfaces are discovered by default.
SNMP Table : 1.3.6.1.4.1.14988.1.1.14.1.1
col 2 = mtxrInterfaceStatsName
col 14 = mtxrInterfaceStatsRxDrop (cumulative counter)
col 19 = mtxrInterfaceStatsTxDrop (cumulative counter)
Deploy to:
~/local/lib/python3/cmk_addons/plugins/mikrotik/agent_based/wan_in_rx_tx_drop.py
Then run:
cmk-validate-plugins
cmk -vII --detect-plugins=mikrotik_wan_drop <hostname>
cmk -v --detect-plugins=mikrotik_wan_drop <hostname>
cmk -R
"""
import time
from typing import Dict
from cmk.agent_based.v2 import (
CheckPlugin,
CheckResult,
DiscoveryResult,
Metric,
OIDEnd,
Result,
Service,
SNMPSection,
SNMPTree,
State,
StringTable,
contains,
get_rate,
get_value_store,
)
# ---------------------------------------------------------------------------
# Section type: { "sfp-sfpplus1": {"rx_drop": int, "tx_drop": int}, ... }
# ---------------------------------------------------------------------------
Section = Dict[str, Dict[str, int]]
# ---------------------------------------------------------------------------
# Parse function — walks full table, keys by interface name
# ---------------------------------------------------------------------------
def parse_mikrotik_wan_drop(string_table: list[StringTable]) -> Section:
section: Section = {}
if not string_table or not string_table[0]:
return section
for row in string_table[0]:
# row: [oid_end, name, rx_drop, tx_drop]
if len(row) < 4:
continue
name = row[1]
try:
rx = int(row[2])
tx = int(row[3])
except (ValueError, IndexError):
continue
if name:
section[name] = {"rx_drop": rx, "tx_drop": tx}
return section
# ---------------------------------------------------------------------------
# SNMP Section — full table walk, dynamic row resolution
# ---------------------------------------------------------------------------
snmp_section_mikrotik_wan_drop = SNMPSection(
name="mikrotik_wan_drop",
parse_function=parse_mikrotik_wan_drop,
detect=contains(".1.3.6.1.2.1.1.2.0", ".1.3.6.1.4.1.14988"),
fetch=[
SNMPTree(
base=".1.3.6.1.4.1.14988.1.1.14.1.1",
oids=[
OIDEnd(), # row index (for reference)
"2", # mtxrInterfaceStatsName
"55", # mtxrInterfaceStatsRxDrop
"85", # mtxrInterfaceStatsTxDrop
],
)
],
)
# ---------------------------------------------------------------------------
# Discovery — yield a service for every sfp-sfpplus* interface found
# ---------------------------------------------------------------------------
def discover_mikrotik_wan_drop(section: Section) -> DiscoveryResult:
for iface_name in section:
if iface_name.startswith("sfp-sfpplus"):
yield Service(item=iface_name)
# ---------------------------------------------------------------------------
# Check function
# ---------------------------------------------------------------------------
def check_mikrotik_wan_drop(item: str, params: dict, section: Section) -> CheckResult:
if item not in section:
yield Result(state=State.UNKNOWN, summary=f"Interface '{item}' not found in SNMP data")
return
data = section[item]
value_store = get_value_store()
now = time.time()
try:
rx_rate = get_rate(
value_store,
f"mikrotik_wan_drop.rx.{item}",
now,
data["rx_drop"],
raise_overflow=True,
)
tx_rate = get_rate(
value_store,
f"mikrotik_wan_drop.tx.{item}",
now,
data["tx_drop"],
raise_overflow=True,
)
except Exception:
yield Result(
state=State.OK,
summary="Initializing counters — rate will appear on next check cycle",
)
return
warn_rx = params.get("warn_rx", 10.0)
crit_rx = params.get("crit_rx", 50.0)
warn_tx = params.get("warn_tx", 10.0)
crit_tx = params.get("crit_tx", 50.0)
if rx_rate >= crit_rx or tx_rate >= crit_tx:
state = State.CRIT
elif rx_rate >= warn_rx or tx_rate >= warn_tx:
state = State.WARN
else:
state = State.OK
yield Result(
state=state,
summary=f"RX Drop: {rx_rate:.2f} /s | TX Drop: {tx_rate:.2f} /s",
)
yield Metric(
name="rx_drop_rate",
value=rx_rate,
levels=(warn_rx, crit_rx),
boundaries=(0.0, None),
)
yield Metric(
name="tx_drop_rate",
value=tx_rate,
levels=(warn_tx, crit_tx),
boundaries=(0.0, None),
)
# ---------------------------------------------------------------------------
# Plugin registration
# ---------------------------------------------------------------------------
check_plugin_mikrotik_wan_drop = CheckPlugin(
name="mikrotik_wan_drop",
sections=["mikrotik_wan_drop"],
service_name="WAN RX/TX Drop %s",
discovery_function=discover_mikrotik_wan_drop,
check_function=check_mikrotik_wan_drop,
check_default_parameters={
"warn_rx": 10.0,
"crit_rx": 50.0,
"warn_tx": 10.0,
"crit_tx": 50.0,
},
)
but the plugin does not seem to work properly, it is either not detecting or on the host it detect, it place the host in warning when they should not be…