#!/usr/bin/env python3 """Synology SSD cache / SSD wear collector for Zabbix (external check). Reads the DSM Web API (SYNO.Storage.CGI.Storage / load_info, plus a SMART call for each SSD) and prints ONE JSON document that Zabbix dependent items parse with JSONPath. Zabbix external check (args come from the item key): synology_ssd.py [port] Debug / exploration: synology_ssd.py [port] --dump raw load_info answer synology_ssd.py [port] --api SYNO.Storage.CGI.Smart --method get_health_info \ --param device=/dev/nvme1n1 any raw DSM API call (prints JSON) synology_ssd.py --from-file dump.json normalize a saved --dump Credentials come from a config file, not from arguments. Default /etc/zabbix/synology_api.conf, override with the SYNO_CONF environment variable: [default] # or [] to override per device user = zbx_monitor password = ... # no quotes https = 1 verify = 0 timeout = 20 Output (exit 0): {"cache": [{"id","status","mode","ok","members","designed","failed", "hit_rate","hit_rate_write"}], "ssd": [{"name","model","device","slot","status","temp","remain_life", "wear","spare","written_tb","media_errors","critical_warning", "poweron_hours"}], "ssd_count": N, "cache_count": N} wear = Percentage Used from NVMe SMART (0-100), -1 if it could not be read. SMART comes from SYNO.Storage.CGI.Smart/get_health_info -> healthInfo.smartInfo. NOTE: DSM's own "remain_life" is NOT the same thing (on a Samsung 980 PRO it can report 100 while SMART Percentage Used is 99), so wear never falls back to it. On any error: message on stderr, exit 1 (Zabbix item becomes unsupported). """ import argparse import configparser import contextlib import json import os import ssl import sys import urllib.parse import urllib.request CONF = os.environ.get("SYNO_CONF", "/etc/zabbix/synology_api.conf") # (api, version, method, extra params) tried in order to read SMART of an SSD. # Field names vary by DSM version; use --api/--method/--param to explore. SMART_CALLS = [ ("SYNO.Storage.CGI.Smart", "1", "get_health_info"), ("SYNO.Storage.CGI.Smart", "1", "get_smart_info"), ("SYNO.Core.Storage.Disk", "1", "get_smart_info"), ] def load_conf(host): cp = configparser.ConfigParser(interpolation=None) # passwords may contain % if not cp.read(CONF): raise RuntimeError(f"cannot read config {CONF}") sec = host if cp.has_section(host) else "default" if not cp.has_section(sec): raise RuntimeError(f"no [{host}] or [default] section in {CONF}") g = cp[sec] return { "user": g.get("user"), "password": g.get("password"), "https": g.getboolean("https", True), "verify": g.getboolean("verify", False), "timeout": g.getint("timeout", 20), } @contextlib.contextmanager def session(host, port, conf): """Yield api(api_name, version, method, **params) -> dict, logging out at the end.""" scheme = "https" if conf["https"] else "http" base = f"{scheme}://{host}:{port}/webapi/" ctx = ssl.create_default_context() if not conf["verify"]: ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE timeout = conf["timeout"] def post(path, params): req = urllib.request.Request(base + path, data=urllib.parse.urlencode(params).encode()) with urllib.request.urlopen(req, context=ctx, timeout=timeout) as r: # POST keeps secrets out of URLs return json.load(r) login = post("auth.cgi", { "api": "SYNO.API.Auth", "version": "6", "method": "login", "account": conf["user"], "passwd": conf["password"], "session": "zbx_ssd", "format": "sid", }) if not login.get("success"): raise RuntimeError(f"DSM login failed: {login.get('error')}") sid = login["data"]["sid"] def api(name, version, method, **params): p = {"api": name, "version": version, "method": method, "_sid": sid} p.update(params) return post("entry.cgi", p) try: yield api finally: try: post("auth.cgi", {"api": "SYNO.API.Auth", "version": "6", "method": "logout", "session": "zbx_ssd", "_sid": sid}) except Exception: pass def num(v, default=-1): """float(v); also accepts strings like '99%' or '29 C'.""" if isinstance(v, str): v = v.strip().split(" ")[0].rstrip("%") try: return float(v) except (TypeError, ValueError): return default def collect_attrs(obj, out): """Collect SMART rows shaped like {"name": "...", "raw": "..."} anywhere in the answer.""" if isinstance(obj, dict): if "name" in obj and "raw" in obj: out[str(obj["name"]).strip().lower()] = obj["raw"] for v in obj.values(): collect_attrs(v, out) elif isinstance(obj, list): for v in obj: collect_attrs(v, out) return out def read_smart(api, disk): """SMART attributes of one disk, {} if no call worked.""" for name, ver, method in SMART_CALLS: try: ans = api(name, ver, method, device=disk.get("device", "")) except Exception: continue if ans.get("success"): attrs = collect_attrs(ans.get("data", ans), {}) if attrs: return attrs return {} def is_ssd(disk): dev = str(disk.get("device", "")).lower() typ = str(disk.get("diskType", "")).lower() return bool(disk.get("isSsd") or "nvme" in dev or "ssd" in typ) def normalize(info, api=None): d = info.get("data", info) out = {"cache": [], "ssd": []} for i, c in enumerate(d.get("ssdCaches") or []): status = str(c.get("status", "unknown")) members = c.get("disks") out["cache"].append({ "id": str(c.get("id") or i), "status": status, "mode": str(c.get("mode", "")), "ok": 1 if status.lower() == "normal" else 0, "members": len(members) if isinstance(members, list) else -1, "designed": int(num(c.get("dev_count"), -1)), "failed": int(num(c.get("disk_failure_number"), 0)), "hit_rate": num(c.get("hit_rate")), "hit_rate_write": num(c.get("hit_rate_write")), }) for dk in d.get("disks") or []: if not is_ssd(dk): continue sm = read_smart(api, dk) if api else {} written = num(sm.get("data units written")) # 1 unit = 512000 bytes out["ssd"].append({ "name": str(dk.get("name") or dk.get("id")), "model": str(dk.get("model", "")).strip(), "device": str(dk.get("device", "")), "slot": str((dk.get("container") or {}).get("str", "")), "status": str(dk.get("status", "unknown")), "temp": num(dk.get("temp")), "remain_life": num(dk.get("remain_life")), "wear": num(sm.get("percentage used")), "spare": num(sm.get("available spare")), "written_tb": round(written * 512000 / 1e12, 1) if written >= 0 else -1, "media_errors": int(num(sm.get("media errors"), -1)), "critical_warning": int(num(sm.get("critical warning"), -1)), "poweron_hours": int(num(sm.get("power on hours"), -1)), }) out["ssd_count"] = len(out["ssd"]) out["cache_count"] = len(out["cache"]) return out def main(): ap = argparse.ArgumentParser() ap.add_argument("host", nargs="?") ap.add_argument("port", nargs="?", default="5001") ap.add_argument("--dump", action="store_true", help="print raw load_info") ap.add_argument("--from-file") ap.add_argument("--api", help="raw API call: API name, e.g. SYNO.Storage.CGI.Smart") ap.add_argument("--version", default="1") ap.add_argument("--method") ap.add_argument("--param", action="append", default=[], help="key=value (repeatable)") a = ap.parse_args() try: if a.from_file: info = json.load(open(a.from_file, encoding="utf-8")) print(json.dumps(normalize(info), ensure_ascii=False)) return if not a.host: ap.error("host required") with session(a.host, a.port, load_conf(a.host)) as api: if a.api: if not a.method: ap.error("--method required with --api") params = dict(p.split("=", 1) for p in a.param) print(json.dumps(api(a.api, a.version, a.method, **params), ensure_ascii=False)) return info = api("SYNO.Storage.CGI.Storage", "1", "load_info") if not info.get("success"): raise RuntimeError(f"load_info failed: {info.get('error')}") print(json.dumps(info if a.dump else normalize(info, api), ensure_ascii=False)) except Exception as e: # short message, never print credentials print(f"ERROR: {e}", file=sys.stderr) sys.exit(1) if __name__ == "__main__": main()