diff --git a/plugins/snmp/readings.py b/plugins/snmp/readings.py index 94ed183..9da373c 100644 --- a/plugins/snmp/readings.py +++ b/plugins/snmp/readings.py @@ -14,11 +14,17 @@ and stored on its row, so every reader builds the same readings from it. A vendor's status (a code, or a word) is a "state" reading: stored as a level (0 OK, 1 busy, 2 warning, 3 failed), so one alert rule reads any vendor's, and shown in the device's own word. + +A reading may be worked out from others (`inputs`, combined by `combine`): +the device is asked for its inputs, each read in its own unit (a counter as +its rate), and only the result is stored and shown. CPU waiting on disk is +one counter's share of the CPU's. """ from __future__ import annotations import re -from dataclasses import dataclass +from dataclasses import dataclass, field +from typing import Callable from steward.core.formatting import human_bytes, human_duration @@ -32,6 +38,8 @@ class Reading: scale: float = 1.0 offset: float = 0.0 states: str = "" # a state reading's codes: a STATE_CODES name, or "words" + inputs: tuple[Reading, ...] = () + combine: Callable[[dict], float | None] | None = field(default=None, compare=False) @property def counter(self) -> bool: @@ -40,7 +48,8 @@ class Reading: # 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)} +COUNTER_UNITS = {"octets": ("bits_per_second", 8.0), "byte_count": ("bytes_per_second", 1.0), + "count": ("per_second", 1.0)} STATUS_WORDS = { "if_status": {1: "up", 2: "down", 3: "testing", 4: "unknown", 5: "dormant", @@ -105,6 +114,7 @@ UNITS = { "kilobytes": "a size in KiB", "rpm": "revolutions per minute (a fan)", "octets": "a byte counter, shown as traffic (bit/s)", + "byte_count": "a byte counter, shown as bytes a second (disk I/O)", "count": "any other counter, shown per second", } @@ -184,6 +194,32 @@ def _asus(oid: str) -> str: _GIB = float(2 ** 30) + +def _ucd(oid: str) -> str: + return f"1.3.6.1.4.1.2021.{oid}" + + +# UCD-SNMP-MIB's CPU tick counters. On Linux ssCpuRawSystem already holds the +# interrupt time, so these five are the whole. +_CPU_TICKS = tuple(Reading(_ucd(f"11.{n}.0"), f"cpu_ticks_{part}", f"CPU ticks {part}", "count") + for n, part in ((50, "user"), (51, "nice"), (52, "system"), (53, "idle"), (54, "wait"))) + + +def _wait_share(v: dict) -> float | None: + total = sum(v.values()) + return 100.0 * v["cpu_ticks_wait"] / total if total > 0 else None + + +_CPU_WAIT = Reading("", "cpu_iowait", "CPU waiting on disk", "percent", inputs=_CPU_TICKS, + combine=_wait_share) +# What the kernel could hand a program without swapping: free, plus the +# buffers and page cache it gives back on demand. +_MEMORY_AVAILABLE = Reading( + "", "memory_available", "Memory available", "kilobytes", + inputs=tuple(Reading(_ucd(f"4.{n}.0"), f"memory_{part}_kb", f"Memory {part}", "kilobytes") + for n, part in ((6, "free"), (14, "buffers"), (15, "cached"))), + combine=lambda v: sum(v.values())) + # 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, ...]]] = { @@ -197,16 +233,19 @@ PRESETS: dict[str, tuple[str, str, tuple[Reading, ...]]] = { 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), and how busy " - "each disk has been over the last minute (UCD-DISKIO-MIB). Most NAS boxes " - "answer it too.", - (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"))), + "linux": ("Linux server", "CPU, how much of it waits on disk, memory and load from net-snmp (snmpd's " + "UCD-SNMP-MIB), and how busy each disk has been over the last minute " + "(UCD-DISKIO-MIB). Most NAS boxes answer it too.", + (Reading(_ucd("11.11.0"), "cpu_used", "CPU used", "percent", scale=-1.0, offset=100.0), + _CPU_WAIT, + Reading(_ucd("10.1.5.1"), "load_1m", "Load (1m)", "number", scale=0.01), + Reading(_ucd("4.6.0"), "memory_free", "Memory free", "kilobytes"), + _MEMORY_AVAILABLE, + Reading(_ucd("4.5.0"), "memory_total", "Memory total", "kilobytes"))), "synology": ("Synology NAS", "System, power and fan status and temperature; each disk's status, health, " - "temperature and bad sectors; each volume's state (scrubbing, rebuilding, " - "degraded) and space. Tick Linux server too for CPU, memory and disk load.", + "temperature, bad sectors and reconnects, and what it reads and writes; " + "each volume's state (scrubbing, rebuilding, degraded), space, reads and " + "writes. Tick Linux server too for CPU, memory and disk load.", (Reading(_syn("1.1.0"), "system_status", "System", "state", states="synology_ok"), Reading(_syn("1.2.0"), "temperature", "Temperature", "celsius"), Reading(_syn("1.3.0"), "power_status", "Power", "state", states="synology_ok"), @@ -218,6 +257,14 @@ PRESETS: dict[str, tuple[str, str, tuple[Reading, ...]]] = { ()), } +def _io_columns(mib: int) -> tuple[Column, ...]: + return (Column(_syn(f"{mib}.1.1.12"), "read", "read", "byte_count"), + Column(_syn(f"{mib}.1.1.13"), "written", "written", "byte_count"), + Column(_syn(f"{mib}.1.1.5"), "reads", "reads", "count"), + Column(_syn(f"{mib}.1.1.6"), "writes", "writes", "count"), + Column(_syn(f"{mib}.1.1.8"), "load", "load", "percent")) + + # A preset's vendor tables, read as one row per disk, volume, fan or sensor. TABLES: dict[str, tuple[Table, ...]] = { "linux": (Table("ucd_diskio", "diskio", "1.3.6.1.4.1.2021.13.15.1.1.2", "Disk", @@ -228,11 +275,21 @@ TABLES: dict[str, tuple[Table, ...]] = { (Column(_syn("2.1.1.5"), "status", "status", "state", states="synology_disk"), Column(_syn("2.1.1.13"), "health", "health", "state", states="synology_health"), Column(_syn("2.1.1.6"), "temp", "temperature", "celsius"), - Column(_syn("2.1.1.9"), "bad_sectors", "bad sectors"))), + Column(_syn("2.1.1.9"), "bad_sectors", "bad sectors"), + # How often DSM has had to reconnect to the disk, and how often + # the disk couldn't find a sector it was asked for (IDNF). + Column(_syn("2.1.1.8"), "retries", "reconnects"), + Column(_syn("2.1.1.10"), "identify_fails", "sectors not found"))), Table("synology_volumes", "raid", _syn("3.1.1.2"), "Volume", (Column(_syn("3.1.1.3"), "status", "status", "state", states="synology_raid"), Column(_syn("3.1.1.4"), "free", "free", "bytes"), Column(_syn("3.1.1.5"), "size", "size", "bytes"))), + # SYNOLOGY-STORAGEIO-MIB, a row per disk, and -SPACEIO-MIB, a row per + # volume: bytes from the 64-bit counters, requests a second, and the + # load DSM gives it (%). A volume slow to answer NFS shows here first. + Table("synology_disk_io", "storageio", _syn("101.1.1.2"), "Disk", _io_columns(101), + rows_like=r"(sd[a-z]+|sata\d+|nvme\d+n\d+)"), + Table("synology_volume_io", "spaceio", _syn("102.1.1.2"), "Volume", _io_columns(102)), ), "asustor": ( Table("asustor_disks", "disk", _asus("4.1.1.2"), "Disk", @@ -280,9 +337,10 @@ def parse_interfaces(text: str) -> list[int]: 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).""" + """In, out and link state per interface, and the packets it got wrong or + had to throw away. 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: @@ -290,7 +348,12 @@ def interface_readings(numbers: list[int], version: str) -> list[Reading]: 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")] + Reading(tx, f"if{n}_out", f"Interface {n} out", "octets"), + Reading(f"1.3.6.1.2.1.2.2.1.14.{n}", f"if{n}_in_errors", f"Interface {n} errors in", "count"), + Reading(f"1.3.6.1.2.1.2.2.1.20.{n}", f"if{n}_out_errors", f"Interface {n} errors out", "count"), + Reading(f"1.3.6.1.2.1.2.2.1.13.{n}", f"if{n}_in_discards", f"Interface {n} discards in", "count"), + Reading(f"1.3.6.1.2.1.2.2.1.19.{n}", f"if{n}_out_discards", f"Interface {n} discards out", + "count")] return out @@ -396,6 +459,8 @@ def format_value(unit: str, value: float | None) -> str: return human_bytes(value * 1024) if unit == "bits_per_second": return _bits(value) + if unit == "bytes_per_second": + return f"{human_bytes(value)}/s" if unit == "per_second": return f"{value:.3g}/s" if unit == "rpm": @@ -410,6 +475,7 @@ def chart_axis(unit: str) -> tuple[str, float]: "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), + "bytes_per_second": ("MiB/s", 1 / 1_048_576), "per_second": ("per second", 1.0), "rpm": ("rpm", 1.0), "state": ("0 OK · 1 busy · 2 warning · 3 failed", 1.0)}.get(unit, ("", 1.0)) diff --git a/plugins/snmp/scheduler.py b/plugins/snmp/scheduler.py index fe349e1..15de0ec 100644 --- a/plugins/snmp/scheduler.py +++ b/plugins/snmp/scheduler.py @@ -139,7 +139,7 @@ async def _do_poll(app: "Quart", poll=None, walk=None) -> None: scans[d.id] = await walk(d.address, d.port, creds, tables_for(d.presets or [])) found = scans.get(d.id) asked[d.id] = readings_for(d, found.rows if found is not None and found.error is None else None) - return await poll(d.address, d.port, creds, [(r.oid, r.key) for r in asked[d.id]]) + return await poll(d.address, d.port, creds, [(r.oid, r.key) for r in _asks(asked[d.id])]) results = await asyncio.gather(*(one(d) for d in devices), return_exceptions=True) @@ -158,23 +158,35 @@ async def _do_poll(app: "Quart", poll=None, walk=None) -> None: device.tables, device.tables_at = p.scan.rows, now device.last_error = result.error missing = set(result.missing) - for reading in asked[device.id]: + + def read(reading): value = result.readings.get(reading.key) if reading.unit == "state": # Its word is shown as answered; its level is stored, # unless Steward doesn't know the word (level None). raw = value if value is not None else result.texts.get(reading.key) if raw is None: - continue + return None p.texts[reading.key], value = state_of(reading, raw) - elif value is None: + return value + if value is None: if reading.key in result.texts: # a word where a number was asked for missing.add(reading.key) - continue - elif reading.counter: - value = RATES.rate(device.id, reading, value, now) + return None + if reading.counter: + return RATES.rate(device.id, reading, value, now) + return value * reading.scale + reading.offset + + for reading in asked[device.id]: + if reading.inputs: + # Every input is read, so each counter keeps its last sample. + got = {i.key: read(i) for i in reading.inputs} + if any(i.key in missing for i in reading.inputs): + missing -= {i.key for i in reading.inputs} + missing.add(reading.key) + value = None if None in got.values() else reading.combine(got) else: - value = value * reading.scale + reading.offset + value = read(reading) if value is not None: p.values.append((reading.key, value)) p.missing = sorted(missing) @@ -183,6 +195,12 @@ async def _do_poll(app: "Quart", poll=None, walk=None) -> None: await held.flush(STORE_WAIT_S) +def _asks(readings: list) -> list: + """What the device is asked for: each reading, or the inputs of one + worked out from others.""" + return [i for r in readings for i in (r.inputs or (r,))] + + async def _store(app: "Quart", rounds: list[list[Polled]]) -> None: """Write held polls, oldest first, in one transaction.""" from steward.core.alerts import MetricBatch diff --git a/tests/plugins/snmp/test_readings.py b/tests/plugins/snmp/test_readings.py index 4015931..abc87e2 100644 --- a/tests/plugins/snmp/test_readings.py +++ b/tests/plugins/snmp/test_readings.py @@ -17,10 +17,12 @@ def test_interface_numbers_read_as_a_list(): def test_interfaces_read_64_bit_counters_except_on_v1(): v2 = {r.key: r for r in interface_readings([3], "2c")} v1 = {r.key: r for r in interface_readings([3], "1")} - assert set(v2) == {"if3_status", "if3_in", "if3_out"} + assert set(v2) == {"if3_status", "if3_in", "if3_out", "if3_in_errors", "if3_out_errors", + "if3_in_discards", "if3_out_discards"} assert v2["if3_in"].oid == "1.3.6.1.2.1.31.1.1.1.6.3" and v2["if3_in"].counter assert v1["if3_in"].oid == "1.3.6.1.2.1.2.2.1.10.3" assert stored_unit(v2["if3_out"]) == "bits_per_second" and not v2["if3_status"].counter + assert v2["if3_in_errors"].oid == "1.3.6.1.2.1.2.2.1.14.3" and stored_unit(v2["if3_in_errors"]) == "per_second" def test_a_custom_line_is_an_oid_a_name_and_a_unit(): @@ -60,6 +62,7 @@ def test_presets_come_in_order_and_names_may_not_repeat(): ("kilobytes", 2048, "2.0 MiB"), ("bits_per_second", 12_345_678, "12.3 Mbit/s"), ("bits_per_second", 950, "950 bit/s"), + ("bytes_per_second", 3 * 1024 * 1024, "3.0 MiB/s"), ("if_status", 2, "down"), ("ups_battery", 3, "low"), ("if_status", 99, "code 99"), @@ -140,3 +143,32 @@ def test_a_state_shows_the_device_word_and_falls_back_to_its_level(): r = Reading("1.1", "raid0_status", "Volume 1 status", "state", states="synology_raid") assert reading_text(r, 1.0, {"raid0_status": "Data scrubbing"}) == "Data scrubbing" assert reading_text(r, 1.0, {}) == "Busy" and reading_text(r, None, None) == "—" + + +def test_cpu_waiting_on_disk_is_the_wait_counter_s_share_of_the_cpu_s(): + """#5997: asked for as its five tick counters, kept as one percentage.""" + linux = {r.key: r for r in device_readings(["linux"], [], "", "2c")} + wait = linux["cpu_iowait"] + assert wait.oid == "" and stored_unit(wait) == "percent" + assert [i.oid for i in wait.inputs] == [f"1.3.6.1.4.1.2021.11.{n}.0" for n in (50, 51, 52, 53, 54)] + assert all(i.counter for i in wait.inputs) + ticks = {"cpu_ticks_user": 100.0, "cpu_ticks_nice": 0.0, "cpu_ticks_system": 50.0, + "cpu_ticks_idle": 800.0, "cpu_ticks_wait": 50.0} + assert wait.combine(ticks) == 5.0 + assert wait.combine(dict.fromkeys(ticks, 0.0)) is None # no ticks between two polls + assert linux["memory_available"].combine( + {"memory_free_kb": 1000.0, "memory_buffers_kb": 200.0, "memory_cached_kb": 3000.0}) == 4200.0 + + +def test_a_synology_disk_and_volume_read_their_io_and_the_disk_its_reconnects(): + tables = {t.id: t for t in TABLES["synology"]} + io = table_readings(tables["synology_disk_io"], [{"index": "1", "label": "sata1", + "columns": ["read", "writes", "load"]}]) + assert [(r.oid, r.key, r.name, stored_unit(r)) for r in io] == [ + ("1.3.6.1.4.1.6574.101.1.1.12.1", "storageio1_read", "sata1 read", "bytes_per_second"), + ("1.3.6.1.4.1.6574.101.1.1.6.1", "storageio1_writes", "sata1 writes", "per_second"), + ("1.3.6.1.4.1.6574.101.1.1.8.1", "storageio1_load", "sata1 load", "percent")] + assert tables["synology_disk_io"].keeps("sata2") and not tables["synology_disk_io"].keeps("md2") + assert tables["synology_volume_io"].columns[0].oid == "1.3.6.1.4.1.6574.102.1.1.12" + disks = {c.key: c.oid for c in tables["synology_disks"].columns} + assert disks["retries"] == "1.3.6.1.4.1.6574.2.1.1.8" and disks["identify_fails"] == "1.3.6.1.4.1.6574.2.1.1.10" diff --git a/tests/plugins/snmp/test_scheduler.py b/tests/plugins/snmp/test_scheduler.py index b51e4ec..590576a 100644 --- a/tests/plugins/snmp/test_scheduler.py +++ b/tests/plugins/snmp/test_scheduler.py @@ -4,16 +4,24 @@ import asyncio from contextlib import asynccontextmanager from types import SimpleNamespace as NS +import pytest + import plugins.snmp.scheduler as scheduler import plugins.snmp.service as service from plugins.snmp.poller import Credentials, PollResult -def _device(): - return NS(id="d1", name="nas", address="10.0.0.5", port=161, version="2c", presets=["system"], +def _device(presets=("system",)): + return NS(id="d1", name="nas", address="10.0.0.5", port=161, version="2c", presets=list(presets), interfaces="", custom=None, tables={}, tables_at=None, last_error=None, texts={}) +@asynccontextmanager +async def _hung_session(): + await asyncio.Event().wait() # never answers: the device list comes from memory + yield None + + def test_polls_go_on_while_the_database_hangs_and_are_stored_in_order_after(monkeypatch): async def main(): stalled = asyncio.Event() @@ -53,3 +61,50 @@ def test_polls_go_on_while_the_database_hangs_and_are_stored_in_order_after(monk assert [[(p.name, p.values) for p in r] for r in rounds] == [ [("nas", [("uptime", 100.0)])], [("nas", [("uptime", 200.0)])]] assert rounds[0][0].at < rounds[1][0].at # each stored at the time it was collected + + +def test_a_reading_worked_out_from_others_is_asked_for_by_its_inputs_and_kept_alone(monkeypatch): + """#5997: CPU waiting on disk from the tick counters' rates, and available + memory from three gauges; the inputs are neither stored nor named missing.""" + mem = {"memory_free_kb": 1000.0, "memory_buffers_kb": 200.0, "memory_cached_kb": 3000.0} + answers = iter([ + {"cpu_ticks_user": 1000.0, "cpu_ticks_nice": 0.0, "cpu_ticks_system": 500.0, + "cpu_ticks_idle": 8000.0, "cpu_ticks_wait": 500.0} | mem, + {"cpu_ticks_user": 1100.0, "cpu_ticks_nice": 0.0, "cpu_ticks_system": 550.0, + "cpu_ticks_idle": 8800.0, "cpu_ticks_wait": 550.0} | mem, + {"cpu_ticks_user": 1200.0, "cpu_ticks_nice": 0.0, "cpu_ticks_system": 600.0, + "cpu_ticks_idle": 9600.0} | mem]) # this time it doesn't answer the wait counter + asked, stored = [], [] + + async def poll(address, port, creds, oids): + asked.append(dict((k, o) for o, k in oids)) + got = next(answers) + return PollResult(readings=got, missing=[k for _, k in oids if k not in got]) + + async def store(app_, rounds): + stored.extend(rounds) + + async def main(): + monkeypatch.setattr(scheduler, "DEVICES", [_device(["linux"])]) + monkeypatch.setattr(scheduler, "DEVICES_READ_S", 0.05) + monkeypatch.setattr(scheduler, "_store", store) + monkeypatch.setattr(service, "credentials", lambda d: Credentials(community="public")) + scheduler.RATES = None + app = NS(db_sessionmaker=_hung_session) + for _ in range(3): + await scheduler._do_poll(app, poll=poll, walk=_no_tables) + await asyncio.sleep(0.01) + + asyncio.run(main()) + assert asked[0]["cpu_ticks_wait"] == "1.3.6.1.4.1.2021.11.54.0" and "cpu_iowait" not in asked[0] + assert "" not in asked[0].values() + first, second, third = (dict(r[0].values) for r in stored) + assert first == {"memory_available": 4200.0} # a counter needs two samples + assert second == {"cpu_iowait": pytest.approx(5.0), "memory_available": 4200.0} # 50 of 1,000 ticks waited + assert third == {"memory_available": 4200.0} + assert "cpu_iowait" in stored[2][0].missing and not any(k.startswith("cpu_ticks") for k in stored[2][0].missing) + + +async def _no_tables(address, port, creds, tables): + from plugins.snmp.poller import TableScan + return TableScan(rows={})