diff options
Diffstat (limited to 'collectors')
| -rw-r--r-- | collectors/smart-selftest.py | 330 | ||||
| -rw-r--r-- | collectors/zfs-metrics.py | 610 |
2 files changed, 940 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()) diff --git a/collectors/zfs-metrics.py b/collectors/zfs-metrics.py new file mode 100644 index 0000000..b313743 --- /dev/null +++ b/collectors/zfs-metrics.py @@ -0,0 +1,610 @@ +#!/usr/bin/env python3 +"""Collect ZFS metrics for the Prometheus node_exporter textfile collector. + +node_exporter's built-in zfs collector reads /proc/spl/kstat/zfs, which covers ARC +statistics, per-dataset objset counters and pool state -- but nothing about capacity, +fragmentation, vdev errors, scrub history, compression, encryption or snapshots. All of +that has to come from the zpool/zfs CLIs, which is what this script does. + +Design notes: + +* Each collection stage is independent. A stage that fails logs to stderr and simply + contributes no samples, rather than taking the whole file down with it -- a broken + snapshot listing should not cost you your pool capacity graphs. +* Output is written to a temporary file and renamed into place, so node_exporter never + reads a half-written file. +* Samples are grouped into metric families by a registry rather than printed inline, so + the exposition format is correct no matter what order stages emit things in. + +Run it with --output PATH. Optionally pass --retention label=count (repeatable) to +publish the configured auto-snapshot retention targets alongside the observed counts. +""" + +from __future__ import annotations + +import argparse +import os +import re +import subprocess +import sys +import time +from collections import defaultdict + +# zpool/zfs report state as a string; dashboards want something orderable. +STATE_CODES = { + "ONLINE": 0, + "DEGRADED": 1, + "FAULTED": 2, + "OFFLINE": 3, + "UNAVAIL": 4, + "REMOVED": 5, + "SUSPENDED": 6, + "AVAIL": 7, + "INUSE": 7, +} +STATE_UNKNOWN = 8 + +# zfs-auto-snapshot (zfstools) names snapshots "zfs-auto-snap_<label>-<Y-m-d-HhM>". +AUTOSNAP_RE = re.compile(r"^zfs-auto-snap_([A-Za-z0-9]+)-") + +# How stale a snapshot of each label is allowed to get before it counts as overdue. +# Mirrors the systemd timers NixOS creates for services.zfs.autoSnapshot. +LABEL_INTERVALS = { + "frequent": 15 * 60, + "hourly": 60 * 60, + "daily": 24 * 60 * 60, + "weekly": 7 * 24 * 60 * 60, + "monthly": 31 * 24 * 60 * 60, +} + + +def warn(msg: str) -> None: + print(f"zfs-metrics: {msg}", file=sys.stderr) + + +class Registry: + """Collects samples and renders them in Prometheus text exposition format.""" + + 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.get(name) + if fam is None: + fam = {"help": help, "type": type, "samples": []} + self._families[name] = fam + 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"]: + out.append(f"{name}{render_labels(labels)} {render_value(value)}") + return "\n".join(out) + "\n" + + +def render_labels(labels: dict) -> str: + if not labels: + return "" + # Pool, dataset and vdev names cannot contain quotes or backslashes, so stripping + # them is a no-op in practice while guaranteeing we never emit a malformed label. + parts = [f'{k}="{str(v).translate(_LABEL_STRIP)}"' for k, v in sorted(labels.items())] + return "{" + ",".join(parts) + "}" + + +_LABEL_STRIP = str.maketrans("", "", '"\\\n') + + +def render_value(value) -> str: + if isinstance(value, float): + if value != value or value in (float("inf"), float("-inf")): + return "NaN" + # Avoid scientific notation, which Prometheus accepts but is unpleasant to read, + # while still keeping enough precision for byte counts. + return f"{value:.6f}".rstrip("0").rstrip(".") or "0" + return str(value) + + +def run(args: list[str]) -> str | None: + """Run a command, returning stdout, or None if it failed.""" + try: + proc = subprocess.run( + args, capture_output=True, text=True, timeout=120, check=False + ) + except (OSError, subprocess.TimeoutExpired) as exc: + warn(f"{args[0]} {args[1] if len(args) > 1 else ''}: {exc}") + return None + if proc.returncode != 0: + err = proc.stderr.strip().splitlines() + warn(f"{' '.join(args)} exited {proc.returncode}: {err[0] if err else ''}") + return None + return proc.stdout + + +def num(value: str): + """Parse a zfs/zpool numeric field, tolerating '-' and human-formatted suffixes.""" + if value in ("", "-", "none"): + return None + value = value.rstrip("%x") + try: + return float(value) if "." in value else int(value) + except ValueError: + return None + + +_SIZE_SUFFIXES = {"B": 1} +for _i, _s in enumerate("KMGTPEZ", start=1): + _SIZE_SUFFIXES[_s] = 1024**_i + + +def nicebytes(value: str): + """Parse zfs_nicebytes output, e.g. '0B', '512B', '1.05G'.""" + if not value or value == "-": + return None + match = re.fullmatch(r"([0-9]+(?:\.[0-9]+)?)\s*([BKMGTPEZ])?", value.strip()) + if not match: + return None + return float(match.group(1)) * _SIZE_SUFFIXES.get(match.group(2) or "B", 1) + + +def parse_ctime(value: str): + """Parse a ctime(3) timestamp into a Unix epoch. + + zpool status prints scan timestamps with ctime(), whose format is fixed and + locale-independent ("Sun Aug 10 05:36:35 2025"), so this is safe to parse. The day + is space-padded, hence the whitespace normalisation. + """ + value = " ".join(value.split()) + try: + return int(time.mktime(time.strptime(value, "%a %b %d %H:%M:%S %Y"))) + except (ValueError, OverflowError): + return None + + +def parse_dhms(value: str): + """Parse secs_to_dhms output: 'HH:MM:SS' or 'N days HH:MM:SS'.""" + match = re.fullmatch(r"(?:(\d+) days )?(\d+):(\d{2}):(\d{2})", value.strip()) + if not match: + return None + days, hours, mins, secs = match.groups() + return int(days or 0) * 86400 + int(hours) * 3600 + int(mins) * 60 + int(secs) + + +def split_dataset(name: str) -> tuple[str, str]: + return name.split("/", 1)[0], name + + +# -------------------------------------------------------------------------------------- +# Stage: pool capacity and health +# -------------------------------------------------------------------------------------- + +POOL_COLUMNS = [ + "name", "size", "allocated", "free", "checkpoint", "fragmentation", + "capacity", "dedupratio", "health", "freeing", "leaked", +] + + +def collect_pools(reg: Registry) -> None: + out = run(["zpool", "list", "-Hp", "-o", ",".join(POOL_COLUMNS)]) + if out is None: + return + for line in out.splitlines(): + fields = line.split("\t") + if len(fields) != len(POOL_COLUMNS): + continue + row = dict(zip(POOL_COLUMNS, fields)) + pool = row["name"] + lbl = {"pool": pool} + + reg.add("zfs_pool_size_bytes", num(row["size"]), lbl, + "Total size of the pool.") + reg.add("zfs_pool_allocated_bytes", num(row["allocated"]), lbl, + "Space allocated in the pool.") + reg.add("zfs_pool_free_bytes", num(row["free"]), lbl, + "Space free in the pool.") + reg.add("zfs_pool_checkpoint_bytes", num(row["checkpoint"]), lbl, + "Space consumed by a pool checkpoint.") + reg.add("zfs_pool_freeing_bytes", num(row["freeing"]), lbl, + "Space still to be reclaimed in the background.") + reg.add("zfs_pool_leaked_bytes", num(row["leaked"]), lbl, + "Space leaked, which should normally be zero.") + reg.add("zfs_pool_dedup_ratio", num(row["dedupratio"]), lbl, + "Deduplication ratio.") + + frag = num(row["fragmentation"]) + if frag is not None: + reg.add("zfs_pool_fragmentation_ratio", frag / 100, lbl, + "Free space fragmentation, 0-1.") + cap = num(row["capacity"]) + if cap is not None: + reg.add("zfs_pool_capacity_ratio", cap / 100, lbl, + "Allocated space as a fraction of pool size, 0-1.") + + health = row["health"] + reg.add("zfs_pool_health", 1, {"pool": pool, "state": health}, + "Pool health, 1 for the currently reported state.") + reg.add("zfs_pool_health_code", STATE_CODES.get(health, STATE_UNKNOWN), lbl, + "Pool health as a code: 0 ONLINE, 1 DEGRADED, 2 FAULTED, 3 OFFLINE, " + "4 UNAVAIL, 5 REMOVED, 6 SUSPENDED, 8 unknown.") + + +# -------------------------------------------------------------------------------------- +# Stage: zpool status -- vdev topology, errors, scrub/resilver history +# -------------------------------------------------------------------------------------- + +CONFIG_HEADER_RE = re.compile(r"^\s*NAME\s+STATE\s+READ\s+WRITE\s+CKSUM") +POOL_RE = re.compile(r"^\s*pool:\s*(\S+)") +SCAN_RE = re.compile(r"^\s*scan:\s*(.*)$") +ERRORS_RE = re.compile(r"^\s*errors:\s*(.*)$") + +SCAN_PATTERNS = [ + ("scrub", "finished", + re.compile(r"^scrub repaired (\S+) in (.+?) with (\d+) errors on (.+)$")), + ("resilver", "finished", + re.compile(r"^resilvered (\S+) in (.+?) with (\d+) errors on (.+)$")), + ("scrub", "canceled", re.compile(r"^scrub canceled on (.+)$")), + ("resilver", "canceled", re.compile(r"^resilver canceled on (.+)$")), + ("scrub", "scanning", re.compile(r"^scrub in progress since (.+)$")), + ("scrub", "paused", re.compile(r"^scrub paused since (.+)$")), + ("resilver", "scanning", re.compile(r"^resilver in progress since (.+)$")), +] + + +def collect_status(reg: Registry) -> None: + out = run(["zpool", "status", "-p"]) + if out is None: + return + + pool = None + in_config = False + seen_scan: set[str] = set() + pools: list[str] = [] + + for line in out.splitlines(): + match = POOL_RE.match(line) + if match: + pool = match.group(1) + pools.append(pool) + in_config = False + continue + + if pool is None: + continue + + match = SCAN_RE.match(line) + if match and pool not in seen_scan: + seen_scan.add(pool) + parse_scan(reg, pool, match.group(1).strip()) + continue + + match = ERRORS_RE.match(line) + if match: + in_config = False + reg.add("zfs_pool_data_errors", parse_data_errors(match.group(1)), + {"pool": pool}, + "Data errors reported by zpool status; 0 means no known errors.") + continue + + if CONFIG_HEADER_RE.match(line): + in_config = True + continue + + if in_config: + if not line.strip(): + in_config = False + continue + parse_vdev(reg, pool, line) + + # A pool with no scan line at all has never been scrubbed; make that explicit + # rather than leaving the series absent. + for p in pools: + if p not in seen_scan: + set_scan_flags(reg, p, scrub=0, resilver=0) + reg.add("zfs_pool_scan_state", 1, {"pool": p, "state": "none"}, + "Current scan state, 1 for the active state.") + + +def parse_data_errors(text: str): + text = text.strip() + if not text: + return None + if "No known data errors" in text: + return 0 + match = re.search(r"(\d+) data errors", text) + if match: + return int(match.group(1)) + # e.g. "Permanent errors have been detected in the following files:" -- we know + # something is wrong but not how many, so report a non-zero count. + return 1 + + +def set_scan_flags(reg: Registry, pool: str, scrub: int, resilver: int) -> None: + reg.add("zfs_pool_scrub_in_progress", scrub, {"pool": pool}, + "Whether a scrub is currently running.") + reg.add("zfs_pool_resilver_in_progress", resilver, {"pool": pool}, + "Whether a resilver is currently running.") + + +def parse_scan(reg: Registry, pool: str, text: str) -> None: + lbl = {"pool": pool} + + for kind, state, pattern in SCAN_PATTERNS: + match = pattern.match(text) + if not match: + continue + + scanning = state in ("scanning", "paused") + set_scan_flags( + reg, pool, + scrub=int(scanning and kind == "scrub"), + resilver=int(scanning and kind == "resilver"), + ) + reg.add("zfs_pool_scan_state", 1, {"pool": pool, "state": f"{kind}_{state}"}, + "Current scan state, 1 for the active state.") + + if state == "finished": + repaired, duration, errors, when = match.groups() + reg.add(f"zfs_pool_last_{kind}_end_timestamp_seconds", parse_ctime(when), + lbl, f"Unix time the last {kind} completed.") + reg.add(f"zfs_pool_last_{kind}_duration_seconds", parse_dhms(duration), + lbl, f"How long the last completed {kind} took.") + reg.add(f"zfs_pool_last_{kind}_errors", int(errors), lbl, + f"Unrepairable errors found by the last completed {kind}.") + reg.add(f"zfs_pool_last_{kind}_repaired_bytes", nicebytes(repaired), lbl, + f"Bytes repaired by the last completed {kind}.") + elif state == "scanning": + reg.add(f"zfs_pool_{kind}_started_timestamp_seconds", + parse_ctime(match.group(1)), lbl, + f"Unix time the running {kind} started.") + return + + # "none requested", or a format we do not recognise. + set_scan_flags(reg, pool, scrub=0, resilver=0) + state = "none" if "none requested" in text else "unknown" + reg.add("zfs_pool_scan_state", 1, {"pool": pool, "state": state}, + "Current scan state, 1 for the active state.") + + +def parse_vdev(reg: Registry, pool: str, line: str) -> None: + fields = line.split() + # vdev rows are "NAME STATE READ WRITE CKSUM" plus an optional trailing note. + # Section headers (logs/cache/spares) and spare rows have fewer fields. + if len(fields) < 5: + return + name, state, read, write, cksum = fields[:5] + if not (read.isdigit() and write.isdigit() and cksum.isdigit()): + return + + reg.add("zfs_pool_vdev_state_code", STATE_CODES.get(state, STATE_UNKNOWN), + {"pool": pool, "vdev": name}, + "vdev state as a code, same encoding as zfs_pool_health_code.") + for kind, value in (("read", read), ("write", write), ("checksum", cksum)): + reg.add("zfs_pool_vdev_errors_total", int(value), + {"pool": pool, "vdev": name, "type": kind}, + "Cumulative vdev errors as reported by zpool status. " + "Reset by zpool clear.", type="counter") + + +# -------------------------------------------------------------------------------------- +# Stage: datasets -- space, compression, encryption +# -------------------------------------------------------------------------------------- + +DATASET_COLUMNS = [ + "name", "type", "used", "available", "referenced", "logicalused", + "usedbysnapshots", "usedbydataset", "compressratio", "compression", + "encryption", "keystatus", "quota", "refquota", "recordsize", "mounted", + "readonly", "com.sun:auto-snapshot", +] + +ONOFF = {"on": 1, "off": 0, "yes": 1, "no": 0, "true": 1, "false": 0} + +# `zfs get keystatus` reports "none" for datasets that are not encrypted at all. +KEYSTATUS_CODES = {"none": 0, "-": 0, "available": 1, "unavailable": 2} + + +def collect_datasets(reg: Registry) -> None: + out = run([ + "zfs", "list", "-Hp", "-t", "filesystem,volume", + "-o", ",".join(DATASET_COLUMNS), + ]) + if out is None: + return + + for line in out.splitlines(): + fields = line.split("\t") + if len(fields) != len(DATASET_COLUMNS): + continue + row = dict(zip(DATASET_COLUMNS, fields)) + pool, dataset = split_dataset(row["name"]) + lbl = {"pool": pool, "dataset": dataset} + + reg.add("zfs_dataset_used_bytes", num(row["used"]), lbl, + "Space consumed by the dataset and its descendants.") + reg.add("zfs_dataset_available_bytes", num(row["available"]), lbl, + "Space available to the dataset.") + reg.add("zfs_dataset_referenced_bytes", num(row["referenced"]), lbl, + "Space referenced by the dataset itself.") + reg.add("zfs_dataset_logical_used_bytes", num(row["logicalused"]), lbl, + "Logical (pre-compression) space consumed.") + reg.add("zfs_dataset_used_by_snapshots_bytes", num(row["usedbysnapshots"]), lbl, + "Space consumed by this dataset's snapshots.") + reg.add("zfs_dataset_used_by_dataset_bytes", num(row["usedbydataset"]), lbl, + "Space consumed by the dataset's own data.") + reg.add("zfs_dataset_compression_ratio", num(row["compressratio"]), lbl, + "Achieved compression ratio; 1 means no saving.") + reg.add("zfs_dataset_quota_bytes", num(row["quota"]), lbl, + "Configured quota in bytes; 0 means unlimited.") + reg.add("zfs_dataset_refquota_bytes", num(row["refquota"]), lbl, + "Configured refquota in bytes; 0 means unlimited.") + reg.add("zfs_dataset_recordsize_bytes", num(row["recordsize"]), lbl, + "Configured record size.") + reg.add("zfs_dataset_mounted", ONOFF.get(row["mounted"]), lbl, + "Whether the dataset is currently mounted.") + reg.add("zfs_dataset_readonly", ONOFF.get(row["readonly"]), lbl, + "Whether the dataset is read-only.") + + autosnap = ONOFF.get(row["com.sun:auto-snapshot"]) + reg.add("zfs_dataset_autosnapshot_enabled", autosnap, lbl, + "Whether com.sun:auto-snapshot is set true on the dataset.") + + # Info-style metrics: the value is always 1, the detail lives in the labels. + # The label is named after the property rather than a generic "algorithm" so + # that dashboards can attach it to a numeric series with group_left() without + # two of these colliding on the same label name. + reg.add("zfs_dataset_compression", 1, + {**lbl, "compression": row["compression"]}, + "Configured compression algorithm, 1 for the active setting.") + reg.add("zfs_dataset_encryption", 1, + {**lbl, "encryption": row["encryption"]}, + "Configured encryption suite, 1 for the active setting.") + reg.add("zfs_dataset_encrypted", int(row["encryption"] not in ("off", "-")), + lbl, "Whether the dataset is encrypted.") + reg.add("zfs_dataset_key_available", int(row["keystatus"] == "available"), + lbl, "Whether the encryption key is loaded (unencrypted datasets " + "report 0).") + # keystatus is a three-state property -- "none" on an unencrypted dataset is + # not the same thing as "unavailable" on an encrypted one. Collapsing it to a + # boolean loses that distinction and makes every unencrypted dataset look + # locked, so publish the three states as a code too. + reg.add("zfs_dataset_keystatus_code", KEYSTATUS_CODES.get(row["keystatus"], 3), + lbl, "Encryption key status: 0 not encrypted, 1 key loaded, " + "2 key not loaded, 3 unknown.") + reg.add("zfs_dataset_type", 1, {**lbl, "type": row["type"]}, + "Dataset type, 1 for the actual type.") + + +# -------------------------------------------------------------------------------------- +# Stage: snapshots +# -------------------------------------------------------------------------------------- + +def collect_snapshots(reg: Registry) -> None: + out = run(["zfs", "list", "-Hp", "-t", "snapshot", "-o", "name,creation,used"]) + if out is None: + return + + # (dataset, label) -> [count, newest, oldest, used] + per_dataset: dict[tuple[str, str], list] = defaultdict( + lambda: [0, None, None, 0.0] + ) + per_pool: dict[tuple[str, str], list] = defaultdict(lambda: [0, None, None, 0.0]) + + for line in out.splitlines(): + fields = line.split("\t") + if len(fields) != 3: + continue + full, creation, used = fields + if "@" not in full: + continue + dataset, snapname = full.split("@", 1) + pool, _ = split_dataset(dataset) + + created = num(creation) + if created is None: + continue + size = num(used) or 0 + + match = AUTOSNAP_RE.match(snapname) + label = match.group(1) if match else "manual" + + for bucket, key in ( + (per_dataset, (dataset, label)), + (per_pool, (pool, label)), + ): + entry = bucket[key] + entry[0] += 1 + entry[1] = created if entry[1] is None else max(entry[1], created) + entry[2] = created if entry[2] is None else min(entry[2], created) + entry[3] += size + + for (dataset, label), (count, newest, oldest, used) in per_dataset.items(): + pool, _ = split_dataset(dataset) + lbl = {"pool": pool, "dataset": dataset, "label": label} + reg.add("zfs_snapshot_count", count, lbl, + "Number of snapshots of this dataset with this label.") + reg.add("zfs_snapshot_newest_timestamp_seconds", newest, lbl, + "Creation time of the most recent snapshot.") + reg.add("zfs_snapshot_oldest_timestamp_seconds", oldest, lbl, + "Creation time of the oldest snapshot; with the newest one this is " + "the retention window actually on disk.") + reg.add("zfs_snapshot_used_bytes", used, lbl, + "Space consumed by these snapshots.") + + for (pool, label), (count, newest, oldest, used) in per_pool.items(): + lbl = {"pool": pool, "label": label} + reg.add("zfs_pool_snapshot_count", count, lbl, + "Number of snapshots in the pool with this label.") + reg.add("zfs_pool_snapshot_newest_timestamp_seconds", newest, lbl, + "Creation time of the pool's most recent snapshot with this label.") + reg.add("zfs_pool_snapshot_oldest_timestamp_seconds", oldest, lbl, + "Creation time of the pool's oldest snapshot with this label.") + reg.add("zfs_pool_snapshot_used_bytes", used, lbl, + "Space consumed by the pool's snapshots with this label.") + + +def collect_retention(reg: Registry, retention: dict[str, int]) -> None: + """Publish the configured auto-snapshot policy so dashboards can compare + observed snapshot counts and ages against what is supposed to exist.""" + for label, keep in sorted(retention.items()): + reg.add("zfs_snapshot_retention_target", keep, {"label": label}, + "Configured number of snapshots to keep for this label.") + for label, seconds in sorted(LABEL_INTERVALS.items()): + if label in retention: + reg.add("zfs_snapshot_interval_seconds", seconds, {"label": label}, + "Expected interval between snapshots of this label.") + + +# -------------------------------------------------------------------------------------- + + +def parse_retention(values: list[str]) -> dict[str, int]: + out = {} + for item in values: + label, _, count = item.partition("=") + try: + out[label] = int(count) + except ValueError: + warn(f"ignoring malformed --retention {item!r}") + return out + + +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("--retention", action="append", default=[], metavar="LABEL=N", + help="configured auto-snapshot retention, repeatable") + args = parser.parse_args() + + reg = Registry() + started = time.time() + + for stage in (collect_pools, collect_status, collect_datasets, collect_snapshots): + try: + stage(reg) + except Exception as exc: # noqa: BLE001 - one bad stage must not lose the rest + warn(f"{stage.__name__} failed: {exc!r}") + + try: + collect_retention(reg, parse_retention(args.retention)) + except Exception as exc: # noqa: BLE001 + warn(f"collect_retention failed: {exc!r}") + + reg.add("zfs_metrics_collection_duration_seconds", time.time() - started, {}, + "Time this collector took to gather ZFS metrics.") + + 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()) |