Enumerate physical drives behind RAID via smartctl megaraid passthrough

services/smart.py

@@ -145,6 +145,60 @@ def _parse_smart_json(device: str, data: dict) -> dict:
     return result
 
 
+async def scan_megaraid_drives() -> list[dict]:
+    """Discover physical drives behind MegaRAID controllers via smartctl --scan."""
+    try:
+        proc = await asyncio.create_subprocess_exec(
+            "smartctl", "--scan", "-j",
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        stdout, _ = await proc.communicate()
+        scan_data = json.loads(stdout)
+    except (FileNotFoundError, json.JSONDecodeError) as e:
+        logger.warning("smartctl --scan failed: %s", e)
+        return []
+
+    devices = scan_data.get("devices", [])
+    megaraid_entries = [
+        d for d in devices
+        if "megaraid" in (d.get("type") or "")
+    ]
+
+    if not megaraid_entries:
+        return []
+
+    # Query SMART for each physical drive concurrently
+    async def _query(entry: dict) -> dict | None:
+        dev_path = entry["name"]
+        dev_type = entry["type"]
+        try:
+            proc = await asyncio.create_subprocess_exec(
+                "smartctl", "-a", "-j", "-d", dev_type, dev_path,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+            )
+            stdout, _ = await proc.communicate()
+            if not stdout:
+                return None
+            data = json.loads(stdout)
+        except (FileNotFoundError, json.JSONDecodeError):
+            return None
+
+        # Extract the disk number from type like "sat+megaraid,0"
+        megaraid_id = dev_type.split("megaraid,")[-1] if "megaraid," in dev_type else dev_type
+
+        result = _parse_smart_json(f"megaraid:{megaraid_id}", data)
+        result["megaraid_id"] = megaraid_id
+        result["megaraid_type"] = dev_type
+        result["megaraid_device"] = dev_path
+        return result
+
+    tasks = [_query(e) for e in megaraid_entries]
+    results = await asyncio.gather(*tasks, return_exceptions=True)
+    return [r for r in results if isinstance(r, dict)]
+
+
 def _get_attr_raw(attrs: list[dict], attr_id: int) -> int | None:
     """Get the raw_value for a SMART attribute by ID."""
     for attr in attrs: