Add eMMC life-time estimate support (#2413)

* Add eMMC life-time estimate support

Expose life time estimate as Host Info property "ssd_life_time"

* Fix pytest

* Fix path to helper

* Allow protected access in tests

* Apply suggestions from code review

Rename SSD to disk.

Co-authored-by: Pascal Vizeli <pascal.vizeli@syshack.ch>

* Rename functions as well

* Update host.py

Co-authored-by: Pascal Vizeli <pascal.vizeli@syshack.ch>

supervisor/api/host.py

@@ -11,6 +11,7 @@ from ..const import (
     ATTR_DEPLOYMENT,
     ATTR_DESCRIPTON,
     ATTR_DISK_FREE,
+    ATTR_DISK_LIFE_TIME,
     ATTR_DISK_TOTAL,
     ATTR_DISK_USED,
     ATTR_FEATURES,
@@ -43,6 +44,7 @@ class APIHost(CoreSysAttributes):
             ATTR_DISK_FREE: self.sys_host.info.free_space,
             ATTR_DISK_TOTAL: self.sys_host.info.total_space,
             ATTR_DISK_USED: self.sys_host.info.used_space,
+            ATTR_DISK_LIFE_TIME: self.sys_host.info.disk_life_time,
             ATTR_FEATURES: self.sys_host.features,
             ATTR_HOSTNAME: self.sys_host.info.hostname,
             ATTR_KERNEL: self.sys_host.info.kernel,

supervisor/const.py

@@ -236,6 +236,7 @@ ATTR_SNAPSHOTS = "snapshots"
 ATTR_SOURCE = "source"
 ATTR_SQUASH = "squash"
 ATTR_SSID = "ssid"
+ATTR_DISK_LIFE_TIME = "disk_life_time"
 ATTR_SSL = "ssl"
 ATTR_STAGE = "stage"
 ATTR_STARTUP = "startup"

supervisor/hardware/helper.py

@@ -19,6 +19,10 @@ _RE_BOOT_TIME: re.Pattern = re.compile(r"btime (\d+)")
 
 _RE_HIDE_SYSFS: re.Pattern = re.compile(r"/sys/devices/virtual/(?:tty|block|vc)/.*")
 
+_MOUNTINFO: Path = Path("/proc/self/mountinfo")
+_BLOCK_DEVICE_CLASS = "/sys/class/block/{}"
+_BLOCK_DEVICE_EMMC_LIFE_TIME = "/sys/block/{}/device/life_time"
+
 
 class HwHelper(CoreSysAttributes):
     """Representation of an interface to procfs, sysfs and udev."""
@@ -78,3 +82,76 @@ class HwHelper(CoreSysAttributes):
         """Return free space (GiB) on disk for path."""
         _, _, free = shutil.disk_usage(path)
         return round(free / (1024.0 ** 3), 1)
+
+    def _get_mountinfo(self, path: str) -> str:
+        mountinfo = _MOUNTINFO.read_text()
+        for line in mountinfo.splitlines():
+            mountinfoarr = line.split()
+            if mountinfoarr[4] == path:
+                return mountinfoarr
+        return None
+
+    def _get_mount_source(self, path: str) -> str:
+        mountinfoarr = self._get_mountinfo(path)
+
+        if mountinfoarr is None:
+            return None
+
+        # Find optional field separator
+        optionsep = 6
+        while mountinfoarr[optionsep] != "-":
+            optionsep += 1
+        return mountinfoarr[optionsep + 2]
+
+    def _try_get_emmc_life_time(self, device_name: str) -> float:
+        # Get eMMC life_time
+        life_time_path = Path(_BLOCK_DEVICE_EMMC_LIFE_TIME.format(device_name))
+
+        if not life_time_path.exists():
+            return None
+
+        # JEDEC health status DEVICE_LIFE_TIME_EST_TYP_A/B
+        emmc_life_time = life_time_path.read_text().split()
+
+        if len(emmc_life_time) < 2:
+            return None
+
+        # Type B life time estimate represents the user partition.
+        life_time_value = int(emmc_life_time[1], 16)
+
+        # 0=Not defined, 1-10=0-100% device life time used, 11=Exceeded
+        if life_time_value == 0:
+            return None
+
+        if life_time_value == 11:
+            logging.warning(
+                "eMMC reports that its estimated life-time has been exceeded!"
+            )
+            return 100.0
+
+        # Return the pessimistic estimate (0x02 -> 10%-20%, return 20%)
+        return life_time_value * 10.0
+
+    def get_disk_life_time(self, path: Union[str, Path]) -> float:
+        """Return life time estimate of the underlying SSD drive."""
+        mount_source = self._get_mount_source(str(path))
+        if mount_source == "overlay":
+            return None
+
+        mount_source_path = Path(mount_source)
+        if not mount_source_path.is_block_device():
+            return None
+
+        # This looks a bit funky but it is more or less what lsblk is doing to get
+        # the parent dev reliably
+
+        # Get class device...
+        mount_source_device_part = Path(
+            _BLOCK_DEVICE_CLASS.format(mount_source_path.name)
+        )
+
+        # ... resolve symlink and get parent device from that path.
+        mount_source_device_name = mount_source_device_part.resolve().parts[-2]
+
+        # Currently only eMMC block devices supported
+        return self._try_get_emmc_life_time(mount_source_device_name)

supervisor/host/info.py

@@ -72,6 +72,13 @@ class InfoCenter(CoreSysAttributes):
             self.coresys.config.path_supervisor
         )
 
+    @property
+    def disk_life_time(self) -> float:
+        """Return the estimated life-time usage (in %) of the SSD storing the data directory."""
+        return self.sys_hardware.helper.get_disk_life_time(
+            self.coresys.config.path_supervisor
+        )
+
     async def get_dmesg(self) -> bytes:
         """Return host dmesg output."""
         proc = await asyncio.create_subprocess_shell(

tests/hardware/test_helper.py

@@ -1,4 +1,5 @@
 """Test hardware utils."""
+# pylint: disable=protected-access
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
@@ -98,3 +99,29 @@ def test_used_space(coresys):
         used = coresys.hardware.helper.get_disk_used_space("/data")
 
     assert used == 8.0
+
+
+def test_get_mountinfo(coresys):
+    """Test mountinfo helper."""
+    mountinfo = coresys.hardware.helper._get_mountinfo("/proc")
+    assert mountinfo[4] == "/proc"
+
+
+def test_get_mount_source(coresys):
+    """Test mount source helper."""
+    # For /proc the mount source is known to be "proc"...
+    mount_source = coresys.hardware.helper._get_mount_source("/proc")
+    assert mount_source == "proc"
+
+
+def test_try_get_emmc_life_time(coresys, tmp_path):
+    """Test eMMC life time helper."""
+    fake_life_time = tmp_path / "fake-mmcblk0-lifetime"
+    fake_life_time.write_text("0x01 0x02\n")
+
+    with patch(
+        "supervisor.hardware.helper._BLOCK_DEVICE_EMMC_LIFE_TIME",
+        str(tmp_path / "fake-{}-lifetime"),
+    ):
+        value = coresys.hardware.helper._try_get_emmc_life_time("mmcblk0")
+    assert value == 20.0