diff --git a/plugins/snmp/migrations/versions/snmp_002_table_rows.py b/plugins/snmp/migrations/versions/snmp_002_table_rows.py new file mode 100644 index 0000000..45e7cf5 --- /dev/null +++ b/plugins/snmp/migrations/versions/snmp_002_table_rows.py @@ -0,0 +1,31 @@ +"""SNMP devices keep the rows of their vendor tables, and states in words + +A NAS preset reads a row per disk and volume, found by walking the device's +tables; the rows found, when, and each state reading's newest word are kept +on the device (milestone 530). + +Revision ID: snmp_002_table_rows +Revises: snmp_001_devices +Create Date: 2026-10-17 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "snmp_002_table_rows" +down_revision: Union[str, None] = "snmp_001_devices" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column("snmp_devices", sa.Column("tables", sa.JSON, nullable=False, server_default="{}")) + op.add_column("snmp_devices", sa.Column("tables_at", sa.DateTime(timezone=True), nullable=True)) + op.add_column("snmp_devices", sa.Column("texts", sa.JSON, nullable=False, server_default="{}")) + + +def downgrade() -> None: + op.drop_column("snmp_devices", "texts") + op.drop_column("snmp_devices", "tables_at") + op.drop_column("snmp_devices", "tables") diff --git a/plugins/snmp/models.py b/plugins/snmp/models.py index 214d6e6..3da6895 100644 --- a/plugins/snmp/models.py +++ b/plugins/snmp/models.py @@ -16,6 +16,9 @@ class SnmpDevice(Base): Secrets (the community, the v3 keys) are stored encrypted. The last_* columns are the newest poll's outcome, written by the poller, so a page can say why a device isn't answering without asking the device itself. + `tables` is the rows the poller found in the presets' vendor tables (a + disk or volume each), as of `tables_at`; `texts` is each state reading's + newest answer in the device's own word. """ __tablename__ = "snmp_devices" @@ -40,3 +43,6 @@ class SnmpDevice(Base): last_ok_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) last_error: Mapped[str | None] = mapped_column(Text, nullable=True) missing: Mapped[list] = mapped_column(JSON, nullable=False, default=list) + tables: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict) + tables_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + texts: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict) diff --git a/plugins/snmp/poller.py b/plugins/snmp/poller.py index 0b24713..30a74d6 100644 --- a/plugins/snmp/poller.py +++ b/plugins/snmp/poller.py @@ -48,6 +48,13 @@ class PollResult: readings: dict[str, float] = field(default_factory=dict) error: str | None = None # the device didn't answer, and why missing: list[str] = field(default_factory=list) # keys it answered without + texts: dict[str, str] = field(default_factory=dict) # keys answered in words, not a number + + +@dataclass +class TableScan: + rows: dict[str, list[dict]] = field(default_factory=dict) # table id -> [{"index", "label", "columns"}] + error: str | None = None def _pysnmp_available() -> bool: @@ -149,9 +156,10 @@ async def poll_device(address: str, port: int, creds: Credentials, oids: list[tuple[str, str]]) -> PollResult: """GET every (oid, key) of a device, a few dozen to a request. - Returns the numeric answers by key, the keys the device answered without - (no such object, or not a number), and an error in words when it didn't - answer at all. Never raises: a broken device must not stop the others. + Returns the numeric answers by key, the answers in words (a vendor's + status), the keys the device answered without, and an error in words when + it didn't answer at all. Never raises: a broken device must not stop the + others. """ result = PollResult() if not oids: @@ -198,8 +206,12 @@ async def poll_device(address: str, port: int, creds: Credentials, try: result.readings[key] = float(val) except (TypeError, ValueError): - result.missing.append(key) - if not result.readings and result.missing: + text = str(val).strip() + if text: + result.texts[key] = text + else: + result.missing.append(key) + if not result.readings and not result.texts and result.missing: result.error = "The device answered but has none of these readings." return result except Exception as exc: @@ -210,6 +222,87 @@ async def poll_device(address: str, port: int, creds: Credentials, _close_engine(engine) +# Rows read of one table: enough for any NAS's disks and volumes, and a bound +# on a device that answers a walk forever. +ROWS_PER_TABLE = 64 + + +async def _walk_column(api, engine, auth, transport, column: str) -> dict[str, object] | str: + """{row index: value} for one column, or the error in words when the + device didn't answer. A column the device lacks is an empty dict.""" + found: dict[str, object] = {} + async for ei, es, _, rows in api.walk(engine, auth, transport, api.hl.ContextData(), + api.hl.ObjectType(api.hl.ObjectIdentity(column)), + lexicographicMode=False, lookupMib=False): + if ei: + return str(ei) + if es: + break + for name, val in rows: + oid = str(name) + if type(val).__name__ in _ABSENT or not oid.startswith(column + "."): + continue + found[oid[len(column) + 1:]] = val + if len(found) >= ROWS_PER_TABLE: + break + return found + + +async def walk_tables(address: str, port: int, creds: Credentials, tables: list) -> TableScan: + """Which rows each vendor table has: each row's index, its label, and the + columns it answered. A walk per column, so it is asked for by the + scheduled poll when a device's rows are unknown or an hour old, never on a + page load. Never raises.""" + scan = TableScan() + if not tables: + return scan + if not _pysnmp_available(): + scan.error = "pysnmp isn't installed in this image, so SNMP polling is off." + return scan + api = _api() + engine = api.hl.SnmpEngine() + try: + try: + transport = await _transport(api, address, port) + except Exception as exc: + scan.error = f"Can't reach {address}:{port}: {exc}" + return scan + auth = _auth_data(api, creds) + for table in tables: + labels: dict[str, object] = {} + if table.label: + got = await _walk_column(api, engine, auth, transport, table.label) + if isinstance(got, str): + scan.error = explain(got, address, port, creds) + return scan + labels = got + answered: dict[str, set[str]] = {} + for column in table.columns: + got = await _walk_column(api, engine, auth, transport, column.oid) + if isinstance(got, str): + scan.error = explain(got, address, port, creds) + return scan + for index in got: + answered.setdefault(index, set()).add(column.key) + indexes = labels if table.label else answered + rows = [] + for index in sorted(indexes, key=lambda i: [int(p) for p in i.split(".") if p.isdigit()]): + label = str(labels.get(index, "")).strip() + if table.label and not table.keeps(label): + continue + columns = [c.key for c in table.columns if c.key in answered.get(index, ())] + if columns: + rows.append({"index": index, "label": label, "columns": columns}) + scan.rows[table.id] = rows[:ROWS_PER_TABLE] + return scan + except Exception as exc: + logger.warning("SNMP table walk of %s:%s failed: %s", address, port, exc) + scan.error = f"SNMP table walk failed: {exc}" + return scan + finally: + _close_engine(engine) + + SYS_DESCR, SYS_UPTIME, SYS_NAME = "1.3.6.1.2.1.1.1.0", "1.3.6.1.2.1.1.3.0", "1.3.6.1.2.1.1.5.0" IF_DESCR, IF_NAME = "1.3.6.1.2.1.2.2.1.2", "1.3.6.1.2.1.31.1.1.1.1" INTERFACES_LISTED = 128 diff --git a/plugins/snmp/readings.py b/plugins/snmp/readings.py index 9232c57..9f2cdc8 100644 --- a/plugins/snmp/readings.py +++ b/plugins/snmp/readings.py @@ -6,6 +6,14 @@ 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. + +A preset may also read a vendor's tables, a row per disk or volume (Table). +Which rows a device has is discovered from the device by the scheduled poll +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. """ from __future__ import annotations @@ -23,6 +31,7 @@ class Reading: unit: str = "number" scale: float = 1.0 offset: float = 0.0 + states: str = "" # a state reading's codes: a STATE_CODES name, or "words" @property def counter(self) -> bool: @@ -39,6 +48,49 @@ STATUS_WORDS = { "ups_battery": {1: "unknown", 2: "normal", 3: "low", 4: "depleted"}, } +# A state reading is stored as one of these levels; an alert rule on "level +# at least 2" catches any vendor's degraded or failed. +LEVELS = {0: "OK", 1: "Busy", 2: "Warning", 3: "Failed"} + +# A vendor's status codes: code -> (the vendor's word, level). Synology's are +# from its MIB guide (SYNOLOGY-SYSTEM-, -DISK- and -RAID-MIB). +STATE_CODES: dict[str, dict[int, tuple[str, int]]] = { + "synology_ok": {1: ("Normal", 0), 2: ("Failed", 3)}, + "synology_disk": {1: ("Normal", 0), 2: ("Initialized", 0), 3: ("Not initialized", 2), + 4: ("System partition failed", 3), 5: ("Crashed", 3)}, + "synology_health": {1: ("Normal", 0), 2: ("Warning", 2), 3: ("Critical", 3), 4: ("Failing", 3)}, + "synology_raid": {1: ("Normal", 0), 2: ("Repairing", 1), 3: ("Migrating", 1), 4: ("Expanding", 1), + 5: ("Deleting", 1), 6: ("Creating", 1), 7: ("Syncing", 1), 8: ("Parity checking", 1), + 9: ("Assembling", 1), 10: ("Canceling", 1), 11: ("Degraded", 2), 12: ("Crashed", 3), + 13: ("Data scrubbing", 1), 14: ("Deploying", 1), 15: ("Undeploying", 1), + 16: ("Mounting cache", 1), 17: ("Unmounting cache", 1), 18: ("Expanding", 1), + 19: ("Converting to a storage pool", 1), 20: ("Migrating SHR1 to SHR2", 1), + 21: ("Unknown", 2)}, +} + +# A vendor that answers in words (Asustor) is read by the words' beginnings, +# worst first: "Unhealthy" is failed before "healthy" can call it OK. +_WORD_LEVELS = ( + (3, re.compile(r"\b(fail|crash|critical|bad|error|abnormal|unhealthy|broken|removed|missing|offline)", re.I)), + (2, re.compile(r"\b(degrad|warn|unknown)", re.I)), + (1, re.compile(r"\b(sync|resync|rebuild|repair|recover|scrub|check|migrat|expand|creat|initiali|build)", + re.I)), + (0, re.compile(r"\b(normal|good|healthy|ok|ready|active|online|clean|pass)", re.I)), +) + + +def state_of(reading: Reading, raw: float | str) -> tuple[str, int | None]: + """A state reading's answer as (the device's word, level). The level is + None for a code or word Steward doesn't know: it is shown, not stored.""" + if reading.states == "words": + word = str(raw).strip() + return word, next((level for level, rx in _WORD_LEVELS if rx.search(word)), None) + try: + code = int(float(raw)) + except (TypeError, ValueError): + return str(raw).strip(), None + return STATE_CODES.get(reading.states, {}).get(code, (f"code {code}", None)) + # The units a custom line may name, with how each one reads, for the form's hint. UNITS = { "number": "a plain number", @@ -51,12 +103,82 @@ UNITS = { "centiseconds": "a span in hundredths of a second (sysUpTime)", "bytes": "a size in bytes", "kilobytes": "a size in KiB", + "rpm": "revolutions per minute (a fan)", "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") + +@dataclass(frozen=True) +class Column: + """One column of a vendor table: a row's reading is this OID plus the + row's index.""" + oid: str + key: str + name: str + unit: str = "number" + scale: float = 1.0 + states: str = "" + + +@dataclass(frozen=True) +class Table: + """A vendor table with a row per disk, volume, fan or sensor. + + `label` is the column naming each row ("Disk 1", "Volume 1"); a row's + readings are keyed by prefix, index and column ("disk0_status") and named + by its label and column ("Disk 1 status"). A table with no label column + (one temperature sensor, one memory) is keyed and named by its columns + alone while it has one row. `rows_like` keeps only rows whose label + matches it (physical disks, not partitions and loop devices).""" + id: str + prefix: str + label: str + noun: str + columns: tuple[Column, ...] + rows_like: str = "" + + def column_oids(self) -> list[str]: + return ([self.label] if self.label else []) + [c.oid for c in self.columns] + + def keeps(self, label: str) -> bool: + return not self.rows_like or re.fullmatch(self.rows_like, label) is not None + + +def table_readings(table: Table, rows: list[dict]) -> list[Reading]: + """The readings of a table's discovered rows: [{"index", "label", + "columns"}], each asked only for the columns it answered.""" + out: list[Reading] = [] + alone = not table.label and len(rows) == 1 + for row in rows: + index, label = str(row["index"]), str(row.get("label") or "") + if label.isdigit() or (table.label and not label): + label = f"{table.noun} {label or index}" + for c in table.columns: + if c.key not in row.get("columns", ()): + continue + if table.label: + key, name = f"{table.prefix}{index.replace('.', '_')}_{c.key}", f"{label} {c.name}" + elif alone: + key, name = c.key, c.name + else: + key, name = f"{c.key}{index.replace('.', '_')}", f"{c.name} ({table.noun.lower()} {index})" + out.append(Reading(f"{c.oid}.{index}", key, name, c.unit, scale=c.scale, states=c.states)) + return out + + +def _syn(oid: str) -> str: + return f"1.3.6.1.4.1.6574.{oid}" + + +def _asus(oid: str) -> str: + return f"1.3.6.1.4.1.44738.{oid}" + + +_GIB = float(2 ** 30) + # 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, ...]]] = { @@ -70,18 +192,72 @@ 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).", + "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"))), + "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.", + (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"), + Reading(_syn("1.4.1.0"), "system_fan", "System fan", "state", states="synology_ok"), + Reading(_syn("1.4.2.0"), "cpu_fan", "CPU fan", "state", states="synology_ok"))), + "asustor": ("Asustor NAS", "Each disk's status, SMART verdict and temperature; each volume's status and " + "space; CPU and system temperature, free memory and fan speed (ADM's " + "ASUSTOR-DISK- and -SYSTEM-MIB). Tick Linux server too for CPU, load and disk load.", + ()), } -PRESET_ORDER = ("system", "interfaces", "ups", "linux") +# 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", + (Column("1.3.6.1.4.1.2021.13.15.1.1.9", "busy", "busy", "percent"),), + rows_like=r"(sd[a-z]+|hd[a-z]+|vd[a-z]+|xvd[a-z]+|nvme\d+n\d+|sata\d+)"),), + "synology": ( + Table("synology_disks", "disk", _syn("2.1.1.2"), "Disk", + (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"))), + 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"))), + ), + "asustor": ( + Table("asustor_disks", "disk", _asus("4.1.1.2"), "Disk", + (Column(_asus("4.1.1.5"), "status", "status", "state", states="words"), + Column(_asus("4.1.1.6"), "smart", "SMART", "state", states="words"), + Column(_asus("4.1.1.7"), "temp", "temperature", "celsius"))), + Table("asustor_volumes", "volume", _asus("5.1.1.2"), "Volume", + (Column(_asus("5.1.1.4"), "status", "status", "state", states="words"), + Column(_asus("5.1.1.7"), "free", "free", "bytes", scale=_GIB), + Column(_asus("5.1.1.6"), "size", "size", "bytes", scale=_GIB))), + Table("asustor_temperatures", "", "", "Sensor", + (Column(_asus("2.5.1.2"), "cpu_temp", "CPU temperature", "celsius"), + Column(_asus("2.5.1.3"), "system_temp", "System temperature", "celsius"))), + Table("asustor_memory", "", "", "Memory", + (Column(_asus("2.6.1.4"), "memory_free_pct", "Memory free", "percent"),)), + Table("asustor_fans", "", "", "Fan", + (Column(_asus("2.4.1.2"), "fan", "Fan speed", "rpm"),)), + ), +} + +PRESET_ORDER = ("system", "interfaces", "ups", "linux", "synology", "asustor") 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 tables_for(presets: list[str]) -> list[Table]: + """The vendor tables the ticked presets read, in preset order.""" + return [t for pid in PRESET_ORDER if pid in (presets or []) for t in TABLES.get(pid, ())] + + 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] = [] @@ -142,9 +318,13 @@ def parse_custom(text: str) -> list[Reading]: 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.""" +def device_readings(presets: list[str], interfaces: list[int], custom: str, version: str, + tables: dict | None = None) -> list[Reading]: + """Every reading a device is asked for: presets in their order, the rows + discovered in their tables (`tables`, {table id: rows}), then custom + lines. Raises ValueError when a custom line repeats a preset's name; a + table row a custom line already names is left to the custom line, since + rows appear after the form was saved.""" out: list[Reading] = [] for pid in PRESET_ORDER: if pid == "interfaces": @@ -152,7 +332,14 @@ def device_readings(presets: list[str], interfaces: list[int], custom: str, vers out += interface_readings(interfaces, version) elif pid in presets: out += PRESETS[pid][2] - out += parse_custom(custom) + custom_readings = parse_custom(custom) + taken = {r.key for r in out} | {r.key for r in custom_readings} + for table in tables_for(presets): + for r in table_readings(table, (tables or {}).get(table.id, [])): + if r.key not in taken: + out.append(r) + taken.add(r.key) + out += custom_readings seen: set[str] = set() for r in out: if r.key in seen: @@ -177,6 +364,8 @@ 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 == "state": + return LEVELS.get(int(value), f"level {value:.0f}") if unit in STATUS_WORDS: return STATUS_WORDS[unit].get(int(value), f"code {value:.0f}") if unit == "centiseconds": @@ -204,6 +393,8 @@ def format_value(unit: str, value: float | None) -> str: return _bits(value) if unit == "per_second": return f"{value:.3g}/s" + if unit == "rpm": + return f"{value:,.0f} rpm" return f"{value:,.0f}" if float(value).is_integer() else f"{value:.4g}" @@ -214,4 +405,13 @@ 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), - "per_second": ("per second", 1.0)}.get(unit, ("", 1.0)) + "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)) + + +def reading_text(reading: Reading, value: float | None, texts: dict | None) -> str: + """What a page shows for a reading: a state in the device's own word, as + its newest poll answered it; anything else in its unit.""" + if reading.unit == "state" and (texts or {}).get(reading.key): + return texts[reading.key] + return format_value(stored_unit(reading), value) diff --git a/plugins/snmp/routes.py b/plugins/snmp/routes.py index 8b223a8..d8232e5 100644 --- a/plugins/snmp/routes.py +++ b/plugins/snmp/routes.py @@ -17,8 +17,7 @@ from steward.models.metrics import PluginMetric, PluginMetricLatest from steward.models.users import UserRole from .models import SnmpDevice -from .readings import (INTERFACES_LABEL, PRESETS, PRESET_ORDER, UNITS, chart_axis, format_value, - stored_unit) +from .readings import INTERFACES_LABEL, PRESETS, PRESET_ORDER, UNITS, chart_axis, reading_text, stored_unit from .service import (FormError, device_state, form_credentials, form_values, parse_form, readings_for) @@ -40,10 +39,10 @@ async def _latest(db, names: list[str]) -> dict[str, dict]: def shown_readings(device: SnmpDevice, values: dict) -> list[dict]: - """The device's readings in its own order, each named and in its unit; - "—" for one with no value yet.""" + """The device's readings in its own order, each named and in its unit (a + state in the device's own word); "—" for one with no value yet.""" return [{"key": r.key, "name": r.name, "counter": r.counter, - "text": format_value(stored_unit(r), values.get(r.key))} + "text": reading_text(r, values.get(r.key), device.texts)} for r in readings_for(device)] @@ -168,9 +167,10 @@ async def save_device(device_id: str): for column, value in values.items(): setattr(device, column, value) # What it is asked, or how, may have changed: the last outcome no - # longer describes it, so the page says "not polled yet" until the next poll. - device.last_polled_at = device.last_ok_at = device.last_error = None - device.missing = [] + # longer describes it, so the page says "not polled yet" until the + # next poll, which also walks its vendor tables again. + device.last_polled_at = device.last_ok_at = device.last_error = device.tables_at = None + device.missing, device.texts = [], {} if old_name != device.name: # Its history is stored under its name; move it with the rename. for model in (PluginMetric, PluginMetricLatest): @@ -263,7 +263,7 @@ async def device_detail(device_id: str): unit = stored_unit(r) axis, factor = chart_axis(unit) charts.append({"key": r.key, "name": r.name, "axis": axis, - "now": format_value(unit, latest.get(device.name, {}).get(r.key)), + "now": reading_text(r, latest.get(device.name, {}).get(r.key), device.texts), "points": [(t, v * factor) for t, v in hist.get(r.key, [])]}) return await render_template("snmp/device.html", device=view(device, latest.get(device.name, {}), now), charts=charts, charts_json=json.dumps(charts), hours=hours, ranges=RANGES) diff --git a/plugins/snmp/scheduler.py b/plugins/snmp/scheduler.py index 70a8f7a..6291e9e 100644 --- a/plugins/snmp/scheduler.py +++ b/plugins/snmp/scheduler.py @@ -3,7 +3,7 @@ from __future__ import annotations import asyncio import logging -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from typing import TYPE_CHECKING from sqlalchemy import select @@ -18,6 +18,20 @@ logger = logging.getLogger(__name__) # Counters' previous samples, for their rates (service.RateKeeper). RATES = None +# How long a device's discovered table rows stand before they are walked +# again: a disk swapped or a volume added shows within the hour, and a NAS is +# walked 24 times a day, not 1,440. +TABLES_REFRESH = timedelta(hours=1) + + +def tables_due(device, now: datetime) -> bool: + """Whether this poll walks the device's vendor tables: it reads some, it + answered its last poll (a device that isn't answering is only asked its + GET), and its rows are unknown, reset by a save, or an hour old.""" + from .readings import tables_for + return (bool(tables_for(device.presets or [])) and not device.last_error + and (device.tables_at is None or now - device.tables_at >= TABLES_REFRESH)) + def make_poll_task(app: "Quart") -> ScheduledTask: interval = app.config["PLUGINS"]["snmp"].get("poll_interval_seconds", 60) @@ -33,18 +47,21 @@ def make_poll_task(app: "Quart") -> ScheduledTask: ) -async def _do_poll(app: "Quart", poll=None) -> None: +async def _do_poll(app: "Quart", poll=None, walk=None) -> None: """Poll every device at once, then write each one's outcome to its row - and its answers to the metric tables. `poll` stands in for - poller.poll_device in tests.""" + and its answers to the metric tables. A device whose vendor tables are + due is walked first, so a new disk is read in the same poll. `poll` and + `walk` stand in for poller.poll_device and poller.walk_tables in tests.""" global RATES from steward.core.alerts import MetricBatch from .models import SnmpDevice - from .poller import poll_device + from .poller import poll_device, walk_tables + from .readings import state_of, tables_for from .service import RateKeeper, credentials, readings_for poll = poll or poll_device + walk = walk or walk_tables if RATES is None: RATES = RateKeeper() @@ -53,10 +70,18 @@ async def _do_poll(app: "Quart", poll=None) -> None: if not devices: return - asked = {d.id: readings_for(d) for d in devices} - results = await asyncio.gather(*( - poll(d.address, d.port, credentials(d), [(r.oid, r.key) for r in asked[d.id]]) - for d in devices), return_exceptions=True) + started = datetime.now(timezone.utc) + scans, asked = {}, {} + + async def one(d): + creds = credentials(d) + if tables_due(d, started): + 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]]) + + results = await asyncio.gather(*(one(d) for d in devices), return_exceptions=True) now = datetime.now(timezone.utc) async with app.db_sessionmaker() as session: @@ -71,20 +96,35 @@ async def _do_poll(app: "Quart", poll=None) -> None: logger.warning("SNMP poll of %s failed: %s", device.name, result) row.last_error = f"SNMP poll failed: {result}" continue + scan = scans.get(device.id) + if scan is not None and scan.error is None: + row.tables, row.tables_at = scan.rows, now row.last_error = result.error - row.missing = sorted(set(result.missing)) - if result.readings: + missing = set(result.missing) + if result.readings or result.texts: row.last_ok_at = now + texts = dict(row.texts or {}) if result.error else {} for reading in asked[device.id]: value = result.readings.get(reading.key) - if value is None: - continue - if reading.counter: - value = RATES.rate(device.id, reading, value, now) - if value is None: + 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 + texts[reading.key], value = state_of(reading, raw) + elif 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) else: value = value * reading.scale + reading.offset + if value is None: + continue batch.add(source_module="snmp", resource_name=device.name, metric_name=reading.key, value=value, recorded_at=now) + row.missing = sorted(missing) + row.texts = texts await batch.record(session) diff --git a/plugins/snmp/service.py b/plugins/snmp/service.py index 634fc4e..631f689 100644 --- a/plugins/snmp/service.py +++ b/plugins/snmp/service.py @@ -11,7 +11,7 @@ from steward.core.crypto import decrypt_secret, encrypt_secret from .models import SnmpDevice from .poller import AUTH_PROTOCOLS, PRIV_PROTOCOLS, Credentials -from .readings import PRESETS, Reading, device_readings, parse_interfaces +from .readings import PRESETS, Reading, device_readings, parse_interfaces, tables_for VERSIONS = ("1", "2c", "3") PRESET_IDS = set(PRESETS) | {"interfaces"} @@ -22,13 +22,15 @@ class FormError(ValueError): """What is wrong with a submitted device form, said to the person.""" -def readings_for(device: SnmpDevice) -> list[Reading]: - """What the device is asked for. A row saved through the form always - parses; one that somehow doesn't is asked for nothing rather than raising - in the poll loop.""" +def readings_for(device: SnmpDevice, tables: dict | None = None) -> list[Reading]: + """What the device is asked for, its vendor tables' rows included (the + stored ones, or `tables` when the poll has just found them). A row saved + through the form always parses; one that somehow doesn't is asked for + nothing rather than raising in the poll loop.""" try: return device_readings(device.presets or [], parse_interfaces(device.interfaces), - device.custom, device.version) + device.custom, device.version, + device.tables if tables is None else tables) except ValueError: return [] @@ -84,7 +86,9 @@ def parse_form(form, existing: SnmpDevice | None = None) -> dict: numbers = parse_interfaces(interfaces) if "interfaces" in presets and not numbers: raise FormError("Interface traffic needs the interface numbers to read, e.g. 1, 2 or 1-4.") - if not device_readings(presets, numbers, custom, version): + # A NAS preset's rows are found by the first poll, so it reads + # nothing until then and still counts as a choice. + if not device_readings(presets, numbers, custom, version) and not tables_for(presets): raise FormError("Choose at least one preset or add a custom reading.") except ValueError as exc: raise FormError(str(exc)) from None diff --git a/tests/integration/test_snmp_poll.py b/tests/integration/test_snmp_poll.py index d33d941..229f519 100644 --- a/tests/integration/test_snmp_poll.py +++ b/tests/integration/test_snmp_poll.py @@ -85,3 +85,52 @@ def test_a_poll_records_why_a_device_is_silent_and_what_another_answered(app): # 60,000 bytes over the 1.1 s or more between the polls, stored as bits # per second: at most 436 kbit/s, and above 150 unless a poll took seconds. assert 150_000 < second["if1_in"] <= 436_364 + + +def test_a_nas_is_walked_once_then_read_by_row_its_states_stored_as_levels_and_shown_in_its_words(app): + from sqlalchemy import select + + from steward.models.metrics import PluginMetricLatest + + scheduler = _snmp("scheduler") + SnmpDevice = _snmp("models").SnmpDevice + poller = _snmp("poller") + + name = f"nas-{uuid.uuid4().hex[:8]}" + walks = [] + + async def fake_walk(address, port, creds, tables): + walks.append([t.id for t in tables]) + return poller.TableScan(rows={ + "asustor_disks": [{"index": "1", "label": "Disk1", "columns": ["status", "temp"]}], + "asustor_volumes": [{"index": "1", "label": "Volume 1", "columns": ["status", "free"]}]}) + + async def fake_poll(address, port, creds, oids): + keys = {k for _, k in oids} + assert {"disk1_status", "disk1_temp", "volume1_status", "volume1_free"} <= keys + return poller.PollResult(readings={"disk1_temp": 41.0, "volume1_free": 100.0}, + texts={"disk1_status": "Normal", "volume1_status": "Degraded"}) + + async def go(): + from steward.core.crypto import encrypt_secret + async with app.db_sessionmaker() as s: + async with s.begin(): + s.add(SnmpDevice(name=name, address="10.9.9.5", presets=["asustor"], + community=encrypt_secret("public"))) + scheduler.RATES = None + await scheduler._do_poll(app, poll=fake_poll, walk=fake_walk) + await scheduler._do_poll(app, poll=fake_poll, walk=fake_walk) + async with app.db_sessionmaker() as s: + async with s.begin(): + row = (await s.execute(select(SnmpDevice).where(SnmpDevice.name == name))).scalar_one() + latest = {r.metric_name: r.value for r in (await s.execute(select(PluginMetricLatest).where( + PluginMetricLatest.source_module == "snmp", PluginMetricLatest.resource_name == name))).scalars()} + await s.delete(row) + return row, latest + + row, latest = asyncio.run(go()) + assert len(walks) == 1 and walks[0][:2] == ["asustor_disks", "asustor_volumes"] # not again within the hour + assert row.tables_at is not None and row.tables["asustor_volumes"][0]["label"] == "Volume 1" + assert row.texts == {"disk1_status": "Normal", "volume1_status": "Degraded"} and row.last_error is None + assert latest["disk1_status"] == 0.0 and latest["volume1_status"] == 2.0 and latest["disk1_temp"] == 41.0 + assert latest["volume1_free"] == 100.0 * 2 ** 30 # ADM's GB, stored in bytes diff --git a/tests/plugins/snmp/test_mcp_tools.py b/tests/plugins/snmp/test_mcp_tools.py index ca0bcfd..2525d88 100644 --- a/tests/plugins/snmp/test_mcp_tools.py +++ b/tests/plugins/snmp/test_mcp_tools.py @@ -18,7 +18,7 @@ def _device(name, *, polled=timedelta(seconds=30), error=None): community="s3cret", v3_user="", v3_auth_key="authkey", v3_priv_key="privkey", presets=[], interfaces="", custom=None, host_id=None, last_polled_at=NOW - polled if polled else None, last_ok_at=NOW - timedelta(hours=1), - last_error=error, missing=[]) + last_error=error, missing=[], tables={}, texts={}) def test_only_a_device_whose_poll_failed_is_an_overview_finding(): diff --git a/tests/plugins/snmp/test_poller.py b/tests/plugins/snmp/test_poller.py index 626fdf5..75e6332 100644 --- a/tests/plugins/snmp/test_poller.py +++ b/tests/plugins/snmp/test_poller.py @@ -65,7 +65,8 @@ def test_one_get_carries_every_oid_and_answers_come_back_by_key(monkeypatch): [("1.1", "uptime"), ("1.2", "fan"), ("1.3", "label")]) assert len(calls) == 1 and calls[0][1] == ["1.1", "1.2", "1.3"] and calls[0][2] == {"lookupMib": False} assert calls[0][0] == ("community", "public", 1) - assert result.readings == {"uptime": 4200.0} and result.missing == ["fan", "label"] and result.error is None + assert result.readings == {"uptime": 4200.0} and result.missing == ["fan"] and result.error is None + assert result.texts == {"label": "eth0"} # a word, for a state reading def test_v1_drops_the_oid_it_does_not_know_and_asks_again(monkeypatch): @@ -151,3 +152,42 @@ def test_close_engine_never_raises(): poller._close_engine(Engine()) poller._close_engine(object()) + + +def _walk(monkeypatch, columns, error=None): + """Walk tables against a device whose columns are {column oid: {index: value}}.""" + walked = [] + + async def walk(engine, auth, transport, ctx, obj, **opts): + walked.append(obj) + if error: + yield error, None, None, [] + return + for index, value in columns.get(obj, {}).items(): + yield None, None, None, [(f"{obj}.{index}", value)] + + monkeypatch.setattr(poller, "_pysnmp_available", lambda: True) + api = _fake_api(lambda oids: (None, None, None, []), []) + api.walk = walk + monkeypatch.setattr(poller, "_api", lambda: api) + return walked + + +def test_a_table_walk_finds_each_row_its_label_and_the_columns_it_answered(monkeypatch): + from plugins.snmp.readings import Column, Table + disks = Table("t", "disk", "9.2", "Disk", (Column("9.5", "status", "status"), Column("9.6", "temp", "temp")), + rows_like=r"Disk \d+") + walked = _walk(monkeypatch, {"9.2": {"1": "Disk 1", "0": "Disk 0", "7": "Cache"}, + "9.5": {"0": 1, "1": 1, "7": 1}, "9.6": {"1": 38}}) + scan = asyncio.run(poller.walk_tables("10.0.0.2", 161, Credentials(community="public"), [disks])) + assert walked == ["9.2", "9.5", "9.6"] and scan.error is None + assert scan.rows == {"t": [{"index": "0", "label": "Disk 0", "columns": ["status"]}, + {"index": "1", "label": "Disk 1", "columns": ["status", "temp"]}]} + + +def test_a_table_walk_with_no_reply_says_what_to_check(monkeypatch): + from plugins.snmp.readings import Column, Table + _walk(monkeypatch, {}, error="No SNMP response received before timeout") + scan = asyncio.run(poller.walk_tables("10.0.0.2", 161, Credentials(community="public"), + [Table("t", "", "", "Fan", (Column("9.2", "fan", "Fan speed"),))])) + assert scan.rows == {} and scan.error.startswith("No reply from 10.0.0.2:161") diff --git a/tests/plugins/snmp/test_readings.py b/tests/plugins/snmp/test_readings.py index 6efb576..f507365 100644 --- a/tests/plugins/snmp/test_readings.py +++ b/tests/plugins/snmp/test_readings.py @@ -1,8 +1,9 @@ """What an SNMP device is asked for and how each answer reads (milestone 491).""" import pytest -from plugins.snmp.readings import (device_readings, format_value, interface_readings, parse_custom, - parse_interfaces, stored_unit) +from plugins.snmp.readings import (TABLES, Column, Reading, Table, device_readings, format_value, + interface_readings, parse_custom, parse_interfaces, reading_text, state_of, + stored_unit, table_readings, tables_for) def test_interface_numbers_read_as_a_list(): @@ -64,6 +65,72 @@ def test_presets_come_in_order_and_names_may_not_repeat(): ("if_status", 99, "code 99"), ("number", 41.25, "41.25"), ("number", None, "—"), + ("state", 2, "Warning"), + ("rpm", 1450, "1,450 rpm"), ]) def test_a_value_reads_in_its_unit(unit, value, text): assert format_value(unit, value) == text + + +_DISKS = Table("d", "disk", "9.2", "Disk", (Column("9.5", "status", "status", "state", states="words"), + Column("9.6", "temp", "temperature", "celsius"))) + + +def test_a_table_row_is_keyed_by_its_index_and_named_by_its_label(): + got = table_readings(_DISKS, [{"index": "0", "label": "Disk 1", "columns": ["status", "temp"]}, + {"index": "1", "label": "2", "columns": ["temp"]}]) + assert [(r.oid, r.key, r.name, r.unit) for r in got] == [ + ("9.5.0", "disk0_status", "Disk 1 status", "state"), + ("9.6.0", "disk0_temp", "Disk 1 temperature", "celsius"), + ("9.6.1", "disk1_temp", "Disk 2 temperature", "celsius")] # a bare bay number gets its noun + assert got[0].states == "words" + + +def test_a_table_without_labels_is_named_by_its_columns(): + fans = Table("f", "", "", "Fan", (Column("9.2", "fan", "Fan speed", "rpm"),)) + [one] = table_readings(fans, [{"index": "1", "label": "", "columns": ["fan"]}]) + assert (one.key, one.name) == ("fan", "Fan speed") + two = table_readings(fans, [{"index": "1", "label": "", "columns": ["fan"]}, + {"index": "2", "label": "", "columns": ["fan"]}]) + assert [(r.key, r.name) for r in two] == [("fan1", "Fan speed (fan 1)"), ("fan2", "Fan speed (fan 2)")] + + +def test_rows_come_after_the_presets_only_for_ticked_presets_and_a_custom_name_wins(): + rows = {"synology_volumes": [{"index": "0", "label": "Volume 1", "columns": ["status", "free"]}], + "asustor_fans": [{"index": "1", "label": "", "columns": ["fan"]}]} + keys = [r.key for r in device_readings(["system", "synology"], [], "1.3.6.1.9.9 raid0_free bytes", "2c", rows)] + assert keys[0] == "uptime" and "system_status" in keys and "fan" not in keys + assert keys.index("raid0_status") > keys.index("cpu_fan") and keys.count("raid0_free") == 1 + assert keys[-1] == "raid0_free" # the custom line, not the row + assert [t.id for t in tables_for(["asustor", "linux"])] == [ + "ucd_diskio", "asustor_disks", "asustor_volumes", "asustor_temperatures", "asustor_memory", "asustor_fans"] + + +def test_only_physical_disks_are_read_for_disk_load(): + [diskio] = TABLES["linux"] + assert diskio.keeps("sda") and diskio.keeps("nvme0n1") and diskio.keeps("sata2") + assert not diskio.keeps("sda1") and not diskio.keeps("loop0") and not diskio.keeps("md2") + + +@pytest.mark.parametrize("states, raw, want", [ + ("synology_raid", 13, ("Data scrubbing", 1)), + ("synology_raid", 11.0, ("Degraded", 2)), + ("synology_raid", 1, ("Normal", 0)), + ("synology_disk", 5, ("Crashed", 3)), + ("synology_raid", 99, ("code 99", None)), + ("words", " Normal ", ("Normal", 0)), + ("words", "Healthy", ("Healthy", 0)), + ("words", "Unhealthy", ("Unhealthy", 3)), + ("words", "Degraded", ("Degraded", 2)), + ("words", "Resyncing", ("Resyncing", 1)), + ("words", "Broken", ("Broken", 3)), + ("words", "Frobnicated", ("Frobnicated", None)), +]) +def test_a_vendor_state_reads_as_its_word_and_a_level(states, raw, want): + assert state_of(Reading("1.1", "s", "S", "state", states=states), raw) == want + + +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) == "—" diff --git a/tests/plugins/snmp/test_service.py b/tests/plugins/snmp/test_service.py index e83a46e..e98dabd 100644 --- a/tests/plugins/snmp/test_service.py +++ b/tests/plugins/snmp/test_service.py @@ -87,6 +87,21 @@ def test_test_uses_what_is_typed_and_falls_back_to_the_stored_secret(): assert form_credentials(_form(), None)[2].community == "public" +def test_a_nas_preset_alone_is_a_choice_though_its_rows_come_with_the_first_poll(): + assert parse_form(_form(presets=["asustor"]))["presets"] == ["asustor"] + with pytest.raises(FormError, match="Choose at least one preset"): + parse_form(_form(presets=[])) + + +def test_a_device_is_asked_for_its_stored_rows_or_the_ones_just_found(): + row = NS(presets=["synology"], interfaces="", custom="", version="2c", + tables={"synology_volumes": [{"index": "0", "label": "Volume 1", "columns": ["status"]}]}) + assert "raid0_status" in [r.key for r in readings_for(row)] + found = {"synology_disks": [{"index": "0", "label": "Disk 1", "columns": ["temp"]}]} + keys = [r.key for r in readings_for(row, found)] + assert "disk0_temp" in keys and "raid0_status" not in keys + + def test_a_row_that_no_longer_parses_is_asked_for_nothing(): row = NS(presets=["system"], interfaces="eth0", custom="", version="2c") assert readings_for(row) == []