Storage management: Add SMART status to disk infos

- Also rewrite API using sdbus-python

Author: christophehenry

debian/control

@@ -17,7 +17,9 @@ Depends: python3-all (>= 3.11),
  , python3-ldap, python3-zeroconf (>= 0.47), python3-lexicon,
  , python3-cryptography, python3-jwt, python3-passlib, python3-magic
  , python-is-python3, python3-pydantic, python3-email-validator
+ , python3-sortedcollections, python3-sdbus
  , udisks2, udisks2-bcache, udisks2-btrfs, udisks2-lvm2, udisks2-zram
+ , smartmontools
  , python3-zmq
  , nginx, nginx-extras (>=1.22)
  , apt, apt-transport-https, apt-utils, aptitude, dirmngr

share/actionsmap.yml

@@ -2117,27 +2117,27 @@ storage:
     subcategories:
         disk:
             subcategory_help: Manage et get infos about hard-drives
+            storage_disks_common_args: &storage_disks_common_args
+                -H:
+                    full: --human-readable
+                    help: Print informations in a human-readable format
+                    action: store_true
+                --human-readable-size:
+                    help: Print sizes in a human-readable format
+                    action: store_true
             actions:
                 # storage_disks_list
                 list:
                     action_help: List hard-drives currently attached to this system optionnaly with infos
                     api: GET /storage/disk/list
                     arguments:
+                        <<: *storage_disks_common_args
                         -i:
                             full: --with-info
                             help: Get all informations for each archive
                             action: store_true
-                        -H:
-                            full: --human-readable
-                            help: Print sizes in human readable format
-                            action: store_true
+                # storage_disks_info
                 info:
                     action_help: Get hard-drive information
                     api: GET /storage/disk/info/<name>
-                    arguments:
-                        name:
-                            help: Name of the hard drive as returned by 'list' command
-                        -H:
-                            full: --human-readable
-                            help: Print sizes in human readable format
-                            action: store_true
+                    arguments: *storage_disks_common_args

src/disk.py

@@ -1,40 +1,54 @@
-import dbus
+import enum
+
+from sdbus import sd_bus_open_system
 from yunohost.utils.system import binary_to_human
+from yunohost.utils.udisks2_interfaces import Udisks2Manager
 
 UDISK_DRIVE_PATH = "/org/freedesktop/UDisks2/drives/"
 UDISK_DRIVE_IFC = "org.freedesktop.UDisks2.Drive"
-
-
-def _query_udisks() -> list[tuple[str, dict]]:
-    bus = dbus.SystemBus()
-    manager = dbus.Interface(
-        bus.get_object("org.freedesktop.UDisks2", "/org/freedesktop/UDisks2"),
-        "org.freedesktop.DBus.ObjectManager",
-    )
-
-    return sorted(
-        (
-            (name.removeprefix(UDISK_DRIVE_PATH), dev[UDISK_DRIVE_IFC])
-            for name, dev in manager.GetManagedObjects().items()
-            if name.startswith(UDISK_DRIVE_PATH)
-        ),
-        key=lambda item: item[1]["SortKey"],
-    )
-
-
-def _disk_infos(name: str, drive: dict, human_readable=False):
+UDISK_DRIVE_ATA_IFC = "org.freedesktop.UDisks2.Drive.Ata"
+UDISK_DRIVE_NVME_IFC = "org.freedesktop.UDisks2.Manager.NVMe"
+
+
+class DiskState(enum.StrEnum):
+    """https://github.com/storaged-project/udisks/issues/1352#issuecomment-2678874537"""
+    @staticmethod
+    def _generate_next_value_(name, start, count, last_values):
+        return name.upper()
+
+    SANE = enum.auto()
+    CRITICAL = enum.auto()
+    UNKNOWN = enum.auto()
+
+    @staticmethod
+    def parse(drive: dict) -> "DiskState":
+        match drive:
+            case {"smart_failing": failing}:
+                # ATA disk, see https://storaged.org/doc/udisks2-api/latest/gdbus-org.freedesktop.UDisks2.Drive.Ata.html#gdbus-property-org-freedesktop-UDisks2-Drive-Ata.SmartFailing
+                return DiskState.SANE if not failing else DiskState.CRITICAL
+            case {"smart_critical_warning": failing}:
+                # NVME; see https://storaged.org/doc/udisks2-api/latest/gdbus-org.freedesktop.UDisks2.NVMe.Controller.html#gdbus-property-org-freedesktop-UDisks2-NVMe-Controller.SmartCriticalWarning
+                return DiskState.SANE if not failing else DiskState.CRITICAL
+            case _:
+                return DiskState.UNKNOWN
+
+
+def _disk_infos(name: str, drive: dict, **kwargs):
+    human_readable = kwargs.get("human_readable", False)
+    human_readable_size = kwargs.get("human_readable_size", human_readable)
     result = {
         "name": name,
-        "model": drive["Model"],
-        "serial": drive["Serial"],
-        "removable": bool(drive["MediaRemovable"]),
-        "size": binary_to_human(drive["Size"]) if human_readable else drive["Size"],
+        "model": drive["model"],
+        "serial": drive["serial"],
+        "removable": bool(drive["media_removable"]),
+        "size": binary_to_human(drive["size"]) if human_readable_size else drive["size"],
+        "smartStatus": DiskState.parse(drive),
     }
 
-    if connection_bus := drive["ConnectionBus"]:
-        result["connection_bus"] = connection_bus
+    if "connection_bus" in drive:
+        result["connectionBus"] = drive["connection_bus"]
 
-    if (rotation_rate := drive["RotationRate"]) == -1:
+    if (rotation_rate := drive["rotation_rate"]) == -1:
         result.update(
             {
                 "type": "HDD",
@@ -54,25 +68,29 @@ def _disk_infos(name: str, drive: dict, human_readable=False):
     return result
 
 
-def disk_list(with_info=False, human_readable=False):
-    if not with_info:
-        return [name for name, _ in _query_udisks()]
+def disk_list(**kwargs):
+    bus = sd_bus_open_system()
+    disks = Udisks2Manager(bus).get_disks()
+
+    with_info = kwargs.get("with_info", False)
 
-    result = []
+    if not with_info:
+        return list(disks.keys())
 
-    for name, drive in _query_udisks():
-        result.append(_disk_infos(name, drive, human_readable))
+    result = [
+        _disk_infos(name, disk.props, **kwargs) for name, disk in disks.items()
+    ]
 
     return {"disks": result}
 
 
-def disk_info(name, human_readable=False):
-    bus = dbus.SystemBus()
-    drive = dbus.Interface(
-        bus.get_object(
-            "org.freedesktop.UDisks2", f"/org/freedesktop/UDisks2/drives/{name}"
-        ),
-        dbus.PROPERTIES_IFACE,
-    ).GetAll("org.freedesktop.UDisks2.Drive")
+def disk_info(name, **kwargs):
+    bus = sd_bus_open_system()
+    disk = Udisks2Manager(bus).get_disks().get(name)
+
+    human_readable = kwargs.get("human_readable", False)
+
+    if not disk:
+        return f"Unknown disk with name {name}" if human_readable else None
 
-    return _disk_infos(name, drive, human_readable)
+    return _disk_infos(name, disk.props, **kwargs)

src/smart.py

@@ -1,10 +1,10 @@
-def storage_disk_list(with_info=False, human_readable=False):
+def storage_disk_list(**kargs):
     from yunohost.disk import disk_list
 
-    return disk_list(with_info=with_info, human_readable=human_readable)
+    return disk_list(**kargs)
 
 
-def storage_disk_info(name, human_readable=False):
+def storage_disk_info(name, **kargs):
     from yunohost.disk import disk_info
 
-    return disk_info(name, human_readable=human_readable)
+    return disk_info(name, **kargs)