FabledSteward/plugins/snmp/readings.py

# plugins/snmp/readings.py
"""What an SNMP device is asked for, and how each answer reads.

A device's readings come from the presets ticked on its form plus any custom
lines. Each reading has a key (the metric name it is stored and alerted on),
a name a person reads, and a unit. A counter (an interface's byte count, any
"total so far") is stored as a rate: the poller turns two samples into a
per-second figure, so a chart and an alert see traffic, not a lifetime total.
"""
from __future__ import annotations

import re
from dataclasses import dataclass

from steward.core.formatting import human_bytes, human_duration


@dataclass(frozen=True)
class Reading:
    oid: str
    key: str
    name: str
    unit: str = "number"
    scale: float = 1.0
    offset: float = 0.0

    @property
    def counter(self) -> bool:
        return self.unit in COUNTER_UNITS


# A counter's unit names what it counts; it is stored as the rate unit it maps
# to, multiplied by the factor (a byte counter is stored as bits per second).
COUNTER_UNITS = {"octets": ("bits_per_second", 8.0), "count": ("per_second", 1.0)}

STATUS_WORDS = {
    "if_status": {1: "up", 2: "down", 3: "testing", 4: "unknown", 5: "dormant",
                  6: "not present", 7: "lower layer down"},
    "ups_battery": {1: "unknown", 2: "normal", 3: "low", 4: "depleted"},
}

# The units a custom line may name, with how each one reads, for the form's hint.
UNITS = {
    "number": "a plain number",
    "percent": "0 to 100",
    "celsius": "degrees Celsius",
    "volts": "volts",
    "decivolts": "tenths of a volt",
    "seconds": "a span in seconds",
    "minutes": "a span in minutes",
    "centiseconds": "a span in hundredths of a second (sysUpTime)",
    "bytes": "a size in bytes",
    "kilobytes": "a size in KiB",
    "octets": "a byte counter, shown as traffic (bit/s)",
    "count": "any other counter, shown per second",
}

_UPTIME = Reading("1.3.6.1.2.1.1.3.0", "uptime", "Uptime", "centiseconds")

# id: (label, what it is for, readings). Interface traffic is per interface
# and built by interface_readings from the numbers on the form.
PRESETS: dict[str, tuple[str, str, tuple[Reading, ...]]] = {
    "system": ("System", "How long the device has been up. Nearly every SNMP device answers it.",
               (_UPTIME,)),
    "ups": ("UPS", "Battery state, charge, runtime left, load and input voltage (UPS-MIB, RFC 1628; "
                   "APC, Eaton and CyberPower management cards).",
            (Reading("1.3.6.1.2.1.33.1.2.1.0", "battery_status", "Battery", "ups_battery"),
             Reading("1.3.6.1.2.1.33.1.2.4.0", "battery_charge", "Charge", "percent"),
             Reading("1.3.6.1.2.1.33.1.2.3.0", "runtime_left", "Runtime left", "minutes"),
             Reading("1.3.6.1.2.1.33.1.4.4.1.5.1", "load", "Load", "percent"),
             Reading("1.3.6.1.2.1.33.1.3.3.1.3.1", "input_voltage", "Input", "volts"),
             Reading("1.3.6.1.2.1.33.1.2.2.0", "on_battery_for", "On battery", "battery_seconds"))),
    "linux": ("Linux server", "CPU, memory and load from net-snmp (snmpd's UCD-SNMP-MIB).",
              (Reading("1.3.6.1.4.1.2021.11.11.0", "cpu_used", "CPU used", "percent", scale=-1.0, offset=100.0),
               Reading("1.3.6.1.4.1.2021.10.1.5.1", "load_1m", "Load (1m)", "number", scale=0.01),
               Reading("1.3.6.1.4.1.2021.4.6.0", "memory_free", "Memory free", "kilobytes"),
               Reading("1.3.6.1.4.1.2021.4.5.0", "memory_total", "Memory total", "kilobytes"))),
}

PRESET_ORDER = ("system", "interfaces", "ups", "linux")
INTERFACES_LABEL = ("Interface traffic", "Traffic in and out and link state for the interfaces numbered "
                    "below. Test the device to see its interfaces and their numbers.")


def parse_interfaces(text: str) -> list[int]:
    """'1, 2, 5-7' → [1, 2, 5, 6, 7]. Raises ValueError naming what it can't read."""
    out: list[int] = []
    for part in re.split(r"[,\s]+", (text or "").strip()):
        if not part:
            continue
        m = re.fullmatch(r"(\d+)(?:-(\d+))?", part)
        if not m:
            raise ValueError(f"“{part}” is not an interface number or range")
        lo, hi = int(m.group(1)), int(m.group(2) or m.group(1))
        if hi < lo or hi - lo > 64:
            raise ValueError(f"“{part}” is not a range of up to 64 interfaces")
        out.extend(n for n in range(lo, hi + 1) if n not in out)
    return out


def interface_readings(numbers: list[int], version: str) -> list[Reading]:
    """In, out and link state per interface. v2c and v3 read the 64-bit
    counters (ifHCInOctets), which don't wrap on a fast link; v1 has only the
    32-bit ones (ifInOctets)."""
    hc = version != "1"
    out: list[Reading] = []
    for n in numbers:
        rx = f"1.3.6.1.2.1.31.1.1.1.6.{n}" if hc else f"1.3.6.1.2.1.2.2.1.10.{n}"
        tx = f"1.3.6.1.2.1.31.1.1.1.10.{n}" if hc else f"1.3.6.1.2.1.2.2.1.16.{n}"
        out += [Reading(f"1.3.6.1.2.1.2.2.1.8.{n}", f"if{n}_status", f"Interface {n}", "if_status"),
                Reading(rx, f"if{n}_in", f"Interface {n} in", "octets"),
                Reading(tx, f"if{n}_out", f"Interface {n} out", "octets")]
    return out


_OID = re.compile(r"\.?\d+(\.\d+)+")
_KEY = re.compile(r"[a-z][a-z0-9_]{0,63}")


def parse_custom(text: str) -> list[Reading]:
    """One reading a line: `OID name [unit]`. The name is a single word
    (letters, digits, underscores) and becomes the metric name; the unit is
    one of UNITS and defaults to number. Raises ValueError naming the line."""
    out: list[Reading] = []
    for i, raw in enumerate((text or "").splitlines(), start=1):
        line = raw.split("#", 1)[0].strip()
        if not line:
            continue
        parts = line.split()
        if len(parts) not in (2, 3):
            raise ValueError(f"Line {i}: write it as “OID name” or “OID name unit”")
        oid, key = parts[0].lstrip("."), parts[1].lower()
        unit = parts[2].lower() if len(parts) == 3 else "number"
        if not _OID.fullmatch(oid):
            raise ValueError(f"Line {i}: “{parts[0]}” is not an OID (numbers separated by dots)")
        if not _KEY.fullmatch(key):
            raise ValueError(f"Line {i}: the name “{parts[1]}” must start with a letter and use "
                             "only letters, digits and underscores")
        if unit not in UNITS:
            raise ValueError(f"Line {i}: “{parts[2]}” is not a unit; use one of {', '.join(UNITS)}")
        out.append(Reading(oid, key, key.replace("_", " ").capitalize(), unit))
    return out


def device_readings(presets: list[str], interfaces: list[int], custom: str, version: str) -> list[Reading]:
    """Every reading a device is asked for, presets in their order then custom
    lines. Raises ValueError when two readings share a name."""
    out: list[Reading] = []
    for pid in PRESET_ORDER:
        if pid == "interfaces":
            if "interfaces" in presets:
                out += interface_readings(interfaces, version)
        elif pid in presets:
            out += PRESETS[pid][2]
    out += parse_custom(custom)
    seen: set[str] = set()
    for r in out:
        if r.key in seen:
            raise ValueError(f"Two readings are named “{r.key}”; give the custom one another name")
        seen.add(r.key)
    return out


def stored_unit(reading: Reading) -> str:
    """The unit a reading is stored in: a counter's rate unit, else its own."""
    return COUNTER_UNITS[reading.unit][0] if reading.counter else reading.unit


def _bits(bps: float) -> str:
    for unit, size in (("Gbit/s", 1e9), ("Mbit/s", 1e6), ("kbit/s", 1e3)):
        if abs(bps) >= size:
            return f"{bps / size:.3g} {unit}"
    return f"{bps:.0f} bit/s"


def format_value(unit: str, value: float | None) -> str:
    """A stored value as a person reads it, in its unit."""
    if value is None:
        return "—"
    if unit in STATUS_WORDS:
        return STATUS_WORDS[unit].get(int(value), f"code {value:.0f}")
    if unit == "centiseconds":
        return human_duration(value / 100)
    if unit == "seconds":
        return human_duration(value)
    if unit == "battery_seconds":
        # UPS-MIB upsSecondsOnBattery: zero whenever the UPS is on mains power.
        return f"for {human_duration(value)}" if value > 0 else "No, on mains"
    if unit == "minutes":
        return human_duration(value * 60)
    if unit == "percent":
        return f"{value:.0f}%"
    if unit == "celsius":
        return f"{value:.1f} °C"
    if unit == "volts":
        return f"{value:.0f} V"
    if unit == "decivolts":
        return f"{value / 10:.1f} V"
    if unit == "bytes":
        return human_bytes(value)
    if unit == "kilobytes":
        return human_bytes(value * 1024)
    if unit == "bits_per_second":
        return _bits(value)
    if unit == "per_second":
        return f"{value:.3g}/s"
    return f"{value:,.0f}" if float(value).is_integer() else f"{value:.4g}"


# What a chart's axis says for a stored unit, and how its values scale for it.
def chart_axis(unit: str) -> tuple[str, float]:
    return {"centiseconds": ("days", 1 / 8_640_000), "seconds": ("seconds", 1.0),
            "battery_seconds": ("seconds on battery", 1.0),
            "minutes": ("minutes", 1.0), "percent": ("%", 1.0), "celsius": ("°C", 1.0),
            "volts": ("V", 1.0), "decivolts": ("V", 0.1), "bytes": ("MiB", 1 / 1_048_576),
            "kilobytes": ("MiB", 1 / 1024), "bits_per_second": ("Mbit/s", 1e-6),
            "per_second": ("per second", 1.0)}.get(unit, ("", 1.0))