Simplify disk detection to single lsblk call with full paths

Replace dual-method detection (lsblk + glob scanning) with single
lsblk -p call that returns full device paths directly. Adds timeout,
returns sorted results for consistency.

Resolves #14

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-10 12:59:49 -05:00
parent ab67d786ce
commit 9f9cc1b763

View File

@@ -1934,42 +1934,26 @@ class SystemHealthMonitor:
# DISK AND STORAGE UTILITY METHODS
# =============================================================================
def _get_all_disks(self) -> List[str]:
"""Get all physical disks using multiple detection methods."""
"""Get all physical disks using lsblk with full device paths."""
disks = set()
# Method 1: Use lsblk to get physical disks, excluding virtual devices
try:
result = subprocess.run(
['lsblk', '-d', '-n', '-o', 'NAME,TYPE'],
['lsblk', '-d', '-n', '-o', 'NAME,TYPE', '-p'],
stdout=subprocess.PIPE,
text=True
text=True,
timeout=10
)
for line in result.stdout.strip().split('\n'):
if line:
parts = line.split()
if len(parts) >= 2:
name, device_type = parts[0], parts[1]
# Only include actual disks, exclude virtual devices
if device_type == 'disk' and not name.startswith('rbd'):
disks.add(f"/dev/{name}")
logger.debug(f"Physical disks found via lsblk: {disks}")
if len(parts) >= 2 and parts[1] == 'disk' and not parts[0].startswith('/dev/rbd'):
disks.add(parts[0])
logger.debug(f"Physical disks found: {disks}")
except subprocess.TimeoutExpired:
logger.error("lsblk timed out during disk detection")
except Exception as e:
logger.debug(f"lsblk detection failed: {e}")
# Method 2: Direct device scanning for physical devices only
for pattern in ['/dev/sd[a-z]', '/dev/nvme[0-9]n[0-9]']:
try:
import glob
matches = glob.glob(pattern)
# Filter out partitions (devices ending in numbers for sd*, already filtered for nvme)
if 'sd' in pattern:
matches = [d for d in matches if not d[-1].isdigit()]
disks.update(matches)
logger.debug(f"Disks found via glob {pattern}: {matches}")
except Exception as e:
logger.debug(f"Glob detection failed for {pattern}: {e}")
return list(disks)
logger.error(f"Failed to detect disks: {e}")
return sorted(disks)
def _is_excluded_mount(self, mountpoint: str) -> bool:
"""Check if a mountpoint should be excluded from monitoring."""