initial commit

1 files changed, 330 insertions, 0 deletions

diff --git a/collectors/smart-selftest.py b/collectors/smart-selftest.py
new file mode 100644
index 0000000..fa9d68f
--- /dev/null
+++ b/collectors/smart-selftest.py
@@ -0,0 +1,330 @@
+#!/usr/bin/env python3
+"""Collect SMART self-test history and drive identity for node_exporter's textfile
+collector.
+
+Two things smartctl_exporter cannot give us:
+
+* Self-test history. The exporter reads only `count` and `error_count_total` out of
+  smartctl's self-test log and discards the log table itself, so it cannot tell you
+  when a drive last ran a short or an extended test, or whether that test passed.
+
+* A ready-to-display drive identity string. Grafana can only put a *label* in a row
+  title, and PromQL has no way to turn a number (capacity, rotation rate) into one.
+  So the formatting happens here, in `smart_drive_label`.
+
+Device names are emitted without the /dev/ prefix so they join against
+smartctl_exporter's `device` label, which is the bare name ("sda").
+
+Run with --output PATH. Reading the self-test log does not spin an idle disk up, but
+there is no reason to poll it often -- a 15 minute timer is plenty.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import subprocess
+import sys
+import time
+from collections import defaultdict
+
+# smartctl spells these differently per transport and firmware; normalise to the names
+# people actually use.
+TEST_TYPES = [
+    (re.compile(r"\bshort\b", re.I), "short"),
+    (re.compile(r"\b(extended|long)\b", re.I), "extended"),
+    (re.compile(r"\bconveyance\b", re.I), "conveyance"),
+    (re.compile(r"\bselective\b", re.I), "selective"),
+    (re.compile(r"\boffline\b", re.I), "offline"),
+]
+
+
+def warn(msg: str) -> None:
+    print(f"smart-selftest: {msg}", file=sys.stderr)
+
+
+def render_labels(labels: dict) -> str:
+    if not labels:
+        return ""
+    parts = [f'{k}="{_clean(str(v))}"' for k, v in sorted(labels.items())]
+    return "{" + ",".join(parts) + "}"
+
+
+def _clean(value: str) -> str:
+    return value.translate(str.maketrans("", "", '"\\\n'))
+
+
+class Registry:
+    def __init__(self) -> None:
+        self._families: dict[str, dict] = {}
+
+    def add(self, name, value, labels=None, help="", type="gauge") -> None:
+        if value is None:
+            return
+        fam = self._families.setdefault(name, {"help": help, "type": type,
+                                               "samples": []})
+        fam["samples"].append((labels or {}, value))
+
+    def render(self) -> str:
+        out = []
+        for name in sorted(self._families):
+            fam = self._families[name]
+            if not fam["samples"]:
+                continue
+            out.append(f"# HELP {name} {fam['help']}")
+            out.append(f"# TYPE {name} {fam['type']}")
+            for labels, value in fam["samples"]:
+                if isinstance(value, float):
+                    rendered = f"{value:.6f}".rstrip("0").rstrip(".") or "0"
+                else:
+                    rendered = str(value)
+                out.append(f"{name}{render_labels(labels)} {rendered}")
+        return "\n".join(out) + "\n"
+
+
+def run_json(args: list[str]) -> dict | None:
+    """Run smartctl with JSON output.
+
+    smartctl uses its exit status as a bitfield rather than a simple success flag --
+    bit 3 means "a self-test failed", which is data we specifically want, not an error.
+    So parse stdout regardless of exit status and only give up if there is no JSON.
+    """
+    try:
+        proc = subprocess.run(args, capture_output=True, text=True, timeout=120,
+                              check=False)
+    except (OSError, subprocess.TimeoutExpired) as exc:
+        warn(f"{' '.join(args)}: {exc}")
+        return None
+    if not proc.stdout.strip():
+        err = proc.stderr.strip().splitlines()
+        warn(f"{' '.join(args)} produced no output: {err[0] if err else ''}")
+        return None
+    try:
+        return json.loads(proc.stdout)
+    except json.JSONDecodeError as exc:
+        warn(f"{' '.join(args)}: could not parse JSON: {exc}")
+        return None
+
+
+def scan_devices() -> list[str]:
+    data = run_json(["smartctl", "--scan", "-j"])
+    if not data:
+        return []
+    return [d["name"] for d in data.get("devices", []) if d.get("name")]
+
+
+def normalise_type(text: str) -> str:
+    for pattern, name in TEST_TYPES:
+        if pattern.search(text):
+            return name
+    return "other"
+
+
+def iter_entries(data: dict):
+    """Yield self-test log entries, whatever shape smartctl used.
+
+    ATA groups them into tables:
+        ata_smart_self_test_log: {standard: {table: [...]}, extended: {table: [...]}}
+
+    SCSI does not have a table at all -- smartctl emits one flat top-level key per
+    entry (scsiprint.cpp: `snprintf(st, "scsi_self_test_%d", k)`):
+        scsi_self_test_0: {code: {...}, result: {...}, power_on_time: {hours: N}}
+
+    Handling only the ATA shape would silently produce nothing for SAS drives.
+    """
+    for key, value in sorted(data.items()):
+        if "self_test" not in key or not isinstance(value, dict):
+            continue
+        # SCSI: the value *is* one entry.
+        if "code" in value or "result" in value:
+            yield value
+            continue
+        # ATA: a table here, or one level down under standard/extended.
+        if isinstance(value.get("table"), list):
+            yield from (e for e in value["table"] if isinstance(e, dict))
+        for sub in value.values():
+            if isinstance(sub, dict) and isinstance(sub.get("table"), list):
+                yield from (e for e in sub["table"] if isinstance(e, dict))
+
+
+def _field(entry: dict, *names):
+    for name in names:
+        if name in entry:
+            return entry[name]
+    return {}
+
+
+def entry_fields(entry: dict) -> tuple[str, str, bool | None, float | None]:
+    """Pull (test type, status text, passed, lifetime hours) out of a log entry.
+
+    All three transports name these differently:
+        ATA   type / status{passed} / lifetime_hours
+        SCSI  code / result        / power_on_time{hours}
+        NVMe  self_test_code / self_test_result / power_on_hours
+    """
+    type_field = _field(entry, "type", "code", "self_test_code")
+    if isinstance(type_field, dict):
+        type_text = type_field.get("string") or str(type_field.get("value", ""))
+    else:
+        type_text = str(type_field)
+
+    status = _field(entry, "status", "result", "self_test_result")
+    if isinstance(status, dict):
+        status_text = status.get("string") or str(status.get("value", ""))
+        passed = status.get("passed")
+    else:
+        status_text = str(status)
+        passed = None
+
+    if passed is None and status_text:
+        # SCSI logs carry no "passed" boolean, only a result string. Check for failure
+        # before checking for "Completed", because SCSI reports a failed long test as
+        # "Completed, segment failed" -- a naive startswith("completed") reads that as
+        # a pass.
+        low = status_text.lower()
+        if "in progress" in low or low.startswith("aborted"):
+            passed = None
+        elif "fail" in low or "error" in low:
+            passed = "without error" in low
+        elif low.startswith("completed"):
+            passed = True
+
+    hours = entry.get("lifetime_hours")
+    if hours is None and isinstance(entry.get("power_on_time"), dict):
+        hours = entry["power_on_time"].get("hours")
+    if hours is None and isinstance(entry.get("power_on_hours"), (int, float)):
+        hours = entry["power_on_hours"]
+
+    return normalise_type(type_text), status_text, passed, hours
+
+
+def human_capacity(nbytes) -> str:
+    """Format a capacity the way the label on the drive does: decimal, not binary."""
+    if not isinstance(nbytes, (int, float)) or nbytes <= 0:
+        return ""
+    for unit, scale in (("TB", 10 ** 12), ("GB", 10 ** 9), ("MB", 10 ** 6)):
+        if nbytes >= scale:
+            value = nbytes / scale
+            return f"{value:.1f} {unit}" if value < 100 else f"{value:.0f} {unit}"
+    return f"{int(nbytes)} B"
+
+
+def technology(data: dict) -> str:
+    """A short "what kind of drive is this" string: bus plus spindle speed."""
+    protocol = (data.get("device") or {}).get("protocol", "")
+    bus = {"ATA": "SATA", "SCSI": "SAS", "NVMe": "NVMe"}.get(protocol, protocol)
+
+    rpm = data.get("rotation_rate")
+    if rpm is None or rpm == 0:
+        # smartctl reports 0 (or omits it) for anything without platters.
+        return f"{bus} SSD".strip()
+    # 7200 -> "7.2K"; the round numbers people actually quote.
+    spin = f"{rpm / 1000:.1f}K".replace(".0K", "K")
+    return f"{bus} {spin}".strip()
+
+
+def collect_identity(reg: Registry, name: str, data: dict) -> None:
+    """Emit one info metric whose label is a preformatted identity line.
+
+    This exists to be read by a human in a Grafana row title, so it is deliberately a
+    single rendered string rather than separate labels -- Grafana can interpolate a
+    variable's display text into a title, but cannot assemble one from several labels.
+    """
+    model = (data.get("model_name") or data.get("scsi_model_name")
+             or (data.get("scsi_product") or "") or "unknown model")
+    serial = data.get("serial_number") or "no serial"
+    capacity = human_capacity((data.get("user_capacity") or {}).get("bytes"))
+
+    parts = [p for p in (capacity, technology(data), model, serial) if p]
+    reg.add("smart_drive_label", 1, {"device": name, "label": " · ".join(parts)},
+            "Drive identity rendered for display; the value is always 1.")
+
+
+def collect_device(reg: Registry, dev: str) -> bool:
+    # -i rides along on the same invocation, so identity costs no extra smartctl run
+    # and no extra chance of waking an idle disk.
+    data = run_json(["smartctl", "-j", "-i", "-l", "selftest", dev])
+    if data is None:
+        return False
+
+    name = os.path.basename(dev)
+    collect_identity(reg, name, data)
+    counts: dict[tuple[str, str], int] = defaultdict(int)
+    totals: dict[str, int] = defaultdict(int)
+    # smartctl lists newest first, so the first entry of each type is the latest.
+    latest: dict[str, tuple[str, bool | None, float | None]] = {}
+
+    entries = 0
+    for entry in iter_entries(data):
+        entries += 1
+        ttype, status_text, passed, hours = entry_fields(entry)
+        counts[(ttype, status_text)] += 1
+        totals[ttype] += 1
+        latest.setdefault(ttype, (status_text, passed, hours))
+
+    for (ttype, status_text), n in sorted(counts.items()):
+        reg.add("smart_selftest_result_count", n,
+                {"device": name, "test_type": ttype, "result": status_text},
+                "Self-test log entries by type and reported result.")
+
+    for ttype, n in sorted(totals.items()):
+        reg.add("smart_selftest_count", n, {"device": name, "test_type": ttype},
+                "Self-test log entries of this type.")
+
+    for ttype, (status_text, passed, hours) in sorted(latest.items()):
+        lbl = {"device": name, "test_type": ttype}
+        if passed is not None:
+            reg.add("smart_selftest_last_passed", int(bool(passed)), lbl,
+                    "Whether the most recent self-test of this type passed.")
+        reg.add("smart_selftest_last_lifetime_hours", hours, lbl,
+                "Drive powered-on hours at which the most recent self-test of this "
+                "type ran. Subtract from the drive's current power-on hours to get "
+                "how long ago that was.")
+        reg.add("smart_selftest_last_status", 1, {**lbl, "status": status_text},
+                "Reported status text of the most recent self-test, 1 for the "
+                "current value.")
+
+    reg.add("smart_selftest_log_entries", entries, {"device": name},
+            "Total entries found in the drive's self-test log.")
+    return True
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--output", required=True,
+                        help="path of the .prom file to write")
+    parser.add_argument("--device", action="append", default=[],
+                        help="device to inspect; repeatable. Default: smartctl --scan")
+    args = parser.parse_args()
+
+    reg = Registry()
+    started = time.time()
+
+    devices = args.device or scan_devices()
+    if not devices:
+        warn("no devices found")
+
+    seen = 0
+    for dev in devices:
+        try:
+            if collect_device(reg, dev):
+                seen += 1
+        except Exception as exc:  # noqa: BLE001 - one bad disk must not lose the rest
+            warn(f"{dev}: {exc!r}")
+
+    reg.add("smart_selftest_devices", seen, {},
+            "Devices successfully queried for self-test history.")
+    reg.add("smart_selftest_collection_duration_seconds", time.time() - started, {},
+            "Time this collector took to gather self-test history.")
+
+    tmp = f"{args.output}.tmp"
+    with open(tmp, "w") as handle:
+        handle.write(reg.render())
+    os.replace(tmp, args.output)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())