diff --git a/AppImage/scripts/build_appimage.sh b/AppImage/scripts/build_appimage.sh index edb51cdb..65213e91 100755 --- a/AppImage/scripts/build_appimage.sh +++ b/AppImage/scripts/build_appimage.sh @@ -124,6 +124,7 @@ cp "$SCRIPT_DIR/post_install_versions.py" "$APP_DIR/usr/bin/" 2>/dev/null || ech cp "$SCRIPT_DIR/mount_monitor.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ mount_monitor.py not found" cp "$SCRIPT_DIR/lxc_mount_points.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ lxc_mount_points.py not found" cp "$SCRIPT_DIR/disk_temperature_history.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ disk_temperature_history.py not found" +cp "$SCRIPT_DIR/smartctl_resolver.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ smartctl_resolver.py not found" cp "$SCRIPT_DIR/health_thresholds.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ health_thresholds.py not found" cp "$SCRIPT_DIR/managed_installs.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ managed_installs.py not found" cp "$SCRIPT_DIR/lxc_apps.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ lxc_apps.py not found" diff --git a/AppImage/scripts/disk_temperature_history.py b/AppImage/scripts/disk_temperature_history.py index b5c3c336..5c392837 100644 --- a/AppImage/scripts/disk_temperature_history.py +++ b/AppImage/scripts/disk_temperature_history.py @@ -12,7 +12,7 @@ don't add another background thread. Performance — three caches keep the steady-state cost flat on big JBODs: - * ``_disk_list_cache`` — lsblk + USB filter, refreshed every 5 min. + * ``_disk_list_cache`` — physical disk inventory, refreshed every 5 min. * ``_disk_probe_cache`` — remembers which ``smartctl -d `` variant works for each disk so we skip the 4-attempt fallback chain. @@ -36,6 +36,8 @@ import time from concurrent.futures import ThreadPoolExecutor from typing import Any, Optional +from smartctl_resolver import smartctl_probe_types, smartctl_result_is_standby + # Use the same DB the CPU temperature pipeline writes to so we share # the WAL file and the periodic vacuum that flask_server already runs. _DB_DIR = "/usr/local/share/proxmenux" @@ -55,7 +57,7 @@ _SMARTCTL_TIMEOUT = 5 # On a 24-disk host the naive sampler can spend several seconds per minute # just iterating smartctl. Three caches keep the steady-state cost flat: # -# _disk_list_cache — the (lsblk + USB filter) result. Disks don't +# _disk_list_cache — the physical disk inventory. Disks don't # appear/disappear between samples, so we only # re-enumerate every _DISK_LIST_TTL seconds. # @@ -81,7 +83,7 @@ _MAX_WORKERS = 16 # cap concurrency for huge JBODs _cache_lock = threading.Lock() _disk_list_cache: Optional[tuple[float, list[str]]] = None -# Maps disk_name -> probe key: 'auto' | 'nvme' | 'ata' | 'sat'. +# Maps disk_name -> the working smartctl device type. # Only successful probes get cached. _disk_probe_cache: dict[str, str] = {} # Maps disk_name -> consecutive_failures count (cleared on success). @@ -176,26 +178,8 @@ def init_disk_temperature_db() -> bool: # Disk enumeration + temperature read # --------------------------------------------------------------------------- -# Match the modal's filter: USB drives are excluded. The hardware tab -# already hides them in the per-disk list and the user's cluster -# storage doesn't run on USB-attached disks anyway. Including them -# would clutter the history table for thumbdrives plugged in once -# during a recovery session. -def _is_usb_disk(disk_name: str) -> bool: - """Return True for disks attached over USB. Mirrors the heuristic - in `get_disk_connection_type` in flask_server — checks the realpath - of /sys/block/ for `usb` in the bus chain.""" - try: - link = os.path.realpath(f"/sys/block/{disk_name}") - return "/usb" in link - except OSError: - return False - - def _enumerate_target_disks() -> list[str]: - """Run ``lsblk`` + USB filter. The expensive part is the realpath - walks in ``_is_usb_disk``; both are short-lived but we still amortise - them via the disk-list cache so they only run every few minutes.""" + """Enumerate physical disks for temperature sampling.""" out: list[str] = [] try: proc = subprocess.run( @@ -214,8 +198,6 @@ def _enumerate_target_disks() -> list[str]: # Skip virtual/loop devices that lsblk still reports as type=disk. if name.startswith("loop") or name.startswith("zd"): continue - if _is_usb_disk(name): - continue out.append(name) except (subprocess.TimeoutExpired, OSError): pass @@ -237,22 +219,6 @@ def _list_target_disks() -> list[str]: return fresh -def _is_disk_usb(disk_name: str) -> bool: - """True if the disk sits behind a USB bus, checked via the resolved - sysfs device path. USB-NVMe bridges (ASMedia, JMicron, Realtek) and - plain USB-HDDs both report `/sys/block//removable = 0`, so the - older removable-flag heuristic missed them and the temperature - poller never tried the snt* driver variants that are the only way - to reach the NVMe controller behind those bridges.""" - try: - base = disk_name[5:] if disk_name.startswith('/dev/') else disk_name - real = os.path.realpath(f'/sys/block/{base}') - return any(seg.startswith('usb') and (len(seg) == 3 or seg[3:].isdigit()) - for seg in real.split('/')) - except Exception: - return False - - def _smartctl_cmd_for(disk_name: str, probe: str) -> list[str]: """Build the smartctl invocation for a given probe key. @@ -293,7 +259,7 @@ def _try_probe(disk_name: str, probe: str) -> Optional[float]: # the backoff and stop polling that drive forever) — surface it # as the dedicated _STANDBY sentinel so the caller skips the # update cleanly. - if proc.returncode == 2: + if smartctl_result_is_standby(proc.returncode, proc.stdout, proc.stderr): return _STANDBY # type: ignore[return-value] # smartctl returns non-zero on warnings (bit 0x40 etc.) even when # JSON is fully populated. Don't gate on returncode — parse the @@ -369,15 +335,9 @@ def _read_temperature(disk_name: str) -> Optional[float]: return temp # Cached probe stopped working — fall through and re-detect. - # Slow path: try every probe and remember the first one that works. - # For USB-attached disks we prepend the three snt* driver variants — - # USB-NVMe bridges (ASMedia / JMicron / Realtek) don't answer the - # plain probes with real SMART; only snt* passes through to the NVMe - # controller so temperature actually comes back. Non-USB disks skip - # them, so this adds zero overhead on internal drives. - probes: tuple[str, ...] = ("auto", "nvme", "ata", "sat") - if _is_disk_usb(disk_name): - probes = ("sntasmedia", "sntjmicron", "sntrealtek") + probes + # Slow path: try the transport-specific order and remember the first + # probe that returns an actual temperature. + probes = smartctl_probe_types(disk_name) for probe in probes: if probe == cached_probe: continue # already tried above diff --git a/AppImage/scripts/flask_server.py b/AppImage/scripts/flask_server.py index a32723bc..ae51c2ee 100644 --- a/AppImage/scripts/flask_server.py +++ b/AppImage/scripts/flask_server.py @@ -69,6 +69,14 @@ BASE_DIR = os.path.dirname(os.path.abspath(__file__)) if BASE_DIR not in sys.path: sys.path.insert(0, BASE_DIR) +from smartctl_resolver import ( # noqa: E402 + is_usb_disk as resolver_is_usb_disk, + probe_smartctl_json, + resolve_smartctl_probe, + smartctl_command, + smartctl_probe_types, + smartctl_type_args, +) from flask_script_runner import script_runner import threading from proxmox_storage_monitor import proxmox_storage_monitor @@ -3602,26 +3610,8 @@ def is_disk_removable(disk_name): def is_disk_usb(disk_name): - """Return True if the disk is attached via USB, using the sysfs device - path. Reliable for USB-attached HDDs and USB-NVMe bridges (which - both report `/sys/block//removable = 0` even though they ARE - USB) — the previous heuristic based on the removable flag missed - them entirely, so snt* pass-through was never attempted and the - bridge's own identity + missing temperature/hours were cached. - - Uses `os.path.realpath` (no subprocess) so it's cheap enough to be - called on every SMART probe. - """ - try: - real = os.path.realpath(f'/sys/block/{disk_name}') - # sysfs path segment like `usb1`, `usb2`, ... always precedes a - # USB-attached block device. Match on the segment prefix rather - # than a substring so a directory happening to contain "usb" in - # its literal name can't false-positive. - return any(seg.startswith('usb') and (len(seg) == 3 or seg[3:].isdigit()) - for seg in real.split('/')) - except Exception: - return False + """Return True when sysfs places the disk behind a USB bus.""" + return resolver_is_usb_disk(disk_name) def _is_system_mount(mountpoint): @@ -4548,6 +4538,7 @@ def _smart_default_payload() -> dict: 'family': None, 'sata_version': None, 'form_factor': None, + '_details_found': False, # Internal flag — True if smartctl confirmed the disk type # (HDD with RPM, or SSD via "Solid State Device" / JSON # rotation_rate=0). Stripped before the dict reaches the API. @@ -4571,6 +4562,25 @@ def _smart_data_useful(d: dict) -> bool: ) +def _smart_payload_score(data: dict) -> int: + return ( + (100 if data.get('_details_found') else 0) + + (12 if (data.get('temperature') or 0) > 0 else 0) + + (8 if (data.get('power_on_hours') or 0) > 0 else 0) + + (5 if data.get('model', 'Unknown') != 'Unknown' else 0) + + (5 if data.get('serial', 'Unknown') != 'Unknown' else 0) + + (2 if data.get('smart_status', 'unknown') not in ('unknown', '') else 0) + ) + + +def _smart_probe_complete(disk_name: str, data: dict) -> bool: + if data.get('_details_found'): + return True + if is_disk_usb(disk_name): + return False + return data.get('model', 'Unknown') != 'Unknown' and data.get('serial', 'Unknown') != 'Unknown' + + _standby_cache: dict[str, tuple] = {} _STANDBY_CACHE_TTL = 15 @@ -4601,11 +4611,13 @@ def _hdd_in_standby(disk_name: str) -> bool: return False parked = False try: - r = subprocess.run( - ['smartctl', '-n', 'standby', '-i', f'/dev/{disk_name}'], - capture_output=True, text=True, timeout=_SMART_TIMEOUT, + result = probe_smartctl_json( + disk_name, + ('-n', 'standby', '-i', '-j'), + timeout=_SMART_TIMEOUT, + require_telemetry=False, ) - parked = r.returncode == 2 + parked = bool(result.get('standby')) except (subprocess.SubprocessError, OSError): parked = False _standby_cache[disk_name] = (now, parked) @@ -4713,51 +4725,16 @@ def _get_smart_data_uncached(disk_name): `get_smart_data` for caching. Probe-cache lives inside this fn so a successful run also remembers which command worked.""" smart_data = _smart_default_payload() + best_smart_data = dict(smart_data) + best_score = -1 try: - all_commands = [ - ['smartctl', '-a', '-j', f'/dev/{disk_name}'], # JSON auto-detect (preferred) - ['smartctl', '-a', '-j', '-d', 'scsi', f'/dev/{disk_name}'], # JSON SCSI/SAS (early for SAS disks) - ['smartctl', '-a', '-d', 'ata', f'/dev/{disk_name}'], # JSON with ATA device type - ['smartctl', '-a', '-d', 'sat', f'/dev/{disk_name}'], # JSON with SAT device type - ['smartctl', '-a', f'/dev/{disk_name}'], # Text output (fallback) - ['smartctl', '-a', '-d', 'ata', f'/dev/{disk_name}'], # Text with ATA device type - ['smartctl', '-a', '-d', 'sat', f'/dev/{disk_name}'], # Text with SAT device type - ['smartctl', '-i', '-H', '-A', f'/dev/{disk_name}'], # Info + Health + Attributes - ['smartctl', '-i', '-H', '-A', '-d', 'ata', f'/dev/{disk_name}'], # With ATA - ['smartctl', '-i', '-H', '-A', '-d', 'sat', f'/dev/{disk_name}'], # With SAT - ['smartctl', '-a', '-j', '-d', 'sat,12', f'/dev/{disk_name}'], # SAT with 12-byte commands - ['smartctl', '-a', '-j', '-d', 'sat,16', f'/dev/{disk_name}'], # SAT with 16-byte commands - ['smartctl', '-a', '-d', 'sat,12', f'/dev/{disk_name}'], # Text SAT with 12-byte commands - ['smartctl', '-a', '-d', 'sat,16', f'/dev/{disk_name}'], # Text SAT with 16-byte commands - ] - - # USB-NVMe bridges (ASMedia ASM2362/ASM2464PD, JMicron JMS583/JMS586, - # Realtek RTL9210): the plain `-a` variant answers with the *bridge* - # identity (e.g. "ASMT 2462 NVME") and no temperature, because the - # bridge exposes itself as generic USB storage. Only `-d snt*` - # passes through to the actual NVMe controller and returns real - # model, serial, temperature and health. - # - # Restricted to disks whose kernel node is `nvmeXnY` — the SNT - # drivers are NVMe-Storage-Namespace-Transport, they only make - # sense on NVMe hardware. For `sdX` USB-SATA (typical setups: - # TerraMaster DAS with HDDs, USB-to-SATA HDD/SSD enclosures) - # every snt* probe returns nothing but eats a 5 s smartctl - # timeout each; three of them stack to ~15 s and can push the - # whole cascade past the failure-backoff threshold BEFORE the - # plain `smartctl -a -j` fallback ever runs — so temperature - # and POH would silently disappear from the UI (GH #293). - # For internal SATA/NVMe the cascade is unchanged. - is_nvme_class = disk_name.startswith('nvme') - if is_nvme_class and (is_disk_usb(disk_name) or is_disk_removable(disk_name)): - all_commands = [ - ['smartctl', '-a', '-j', '-d', 'sntasmedia', f'/dev/{disk_name}'], - ['smartctl', '-a', '-j', '-d', 'sntjmicron', f'/dev/{disk_name}'], - ['smartctl', '-a', '-j', '-d', 'sntrealtek', f'/dev/{disk_name}'], - ] + all_commands + probes = smartctl_probe_types(disk_name) + all_commands = [smartctl_command(disk_name, ('-a', '-j'), probe) for probe in probes] + all_commands += [smartctl_command(disk_name, ('-a',), probe) for probe in probes] + all_commands += [smartctl_command(disk_name, ('-i', '-H', '-A'), probe) for probe in probes] # Probe-cache: if we already know which command works for this # disk, try that first. The fallback chain is still kept after @@ -4772,6 +4749,7 @@ def _get_smart_data_uncached(disk_name): process = None # Initialize process to None for cmd_index, cmd in enumerate(commands_to_try): + smart_data = _smart_default_payload() # print(f"[v0] Attempt {cmd_index + 1}/{len(commands_to_try)}: Running command: {' '.join(cmd)}") pass try: @@ -4827,10 +4805,12 @@ def _get_smart_data_uncached(disk_name): # Extract temperature if 'temperature' in data and 'current' in data['temperature']: smart_data['temperature'] = data['temperature']['current'] + smart_data['_details_found'] = True # Parse NVMe SMART data if 'nvme_smart_health_information_log' in data: + smart_data['_details_found'] = True nvme_data = data['nvme_smart_health_information_log'] if 'temperature' in nvme_data: @@ -4860,6 +4840,7 @@ def _get_smart_data_uncached(disk_name): # Parse SCSI/SAS SMART data (no ATA attribute IDs) device_protocol = data.get('device', {}).get('protocol', '') if device_protocol == 'SCSI' or 'scsi_error_counter_log' in data: + smart_data['_details_found'] = True # Temperature if 'temperature' in data and 'current' in data['temperature']: smart_data['temperature'] = data['temperature']['current'] @@ -4889,6 +4870,7 @@ def _get_smart_data_uncached(disk_name): # Parse ATA SMART attributes elif 'ata_smart_attributes' in data and 'table' in data['ata_smart_attributes']: + smart_data['_details_found'] = bool(data['ata_smart_attributes']['table']) for attr in data['ata_smart_attributes']['table']: attr_id = attr.get('id') @@ -4973,13 +4955,6 @@ def _get_smart_data_uncached(disk_name): except (ValueError, TypeError): pass - # If we got good data, break out of the loop and - # memoise this command so subsequent calls skip - # straight to it instead of re-trying the chain. - if smart_data['model'] != 'Unknown' and smart_data['serial'] != 'Unknown': - _smart_probe_cache[disk_name] = list(cmd) - break - except json.JSONDecodeError as e: # print(f"[v0] JSON parse failed: {e}, trying text parsing...") pass @@ -5047,6 +5022,7 @@ def _get_smart_data_uncached(disk_name): try: temp_str = line.split(':')[1].strip().split()[0] smart_data['temperature'] = int(temp_str) + smart_data['_details_found'] = True # print(f"[v0] Found temperature: {smart_data['temperature']}°C") pass except (ValueError, IndexError): @@ -5059,6 +5035,7 @@ def _get_smart_data_uncached(disk_name): if 'ID# ATTRIBUTE_NAME' in line or 'ID#' in line and 'ATTRIBUTE_NAME' in line: in_attributes = True + smart_data['_details_found'] = True # print(f"[v0] Found SMART attributes table") pass continue @@ -5149,14 +5126,15 @@ def _get_smart_data_uncached(disk_name): pass continue - # If we got complete data, break and memoise the - # winning command for the next call. - if smart_data['model'] != 'Unknown' and smart_data['serial'] != 'Unknown': - _smart_probe_cache[disk_name] = list(cmd) - break - elif smart_data['model'] != 'Unknown' or smart_data['serial'] != 'Unknown': - # print(f"[v0] Extracted partial data from text output, continuing to next attempt...") - pass + score = _smart_payload_score(smart_data) + if score > best_score: + best_score = score + best_smart_data = dict(smart_data) + + if _smart_probe_complete(disk_name, smart_data): + _smart_probe_cache[disk_name] = list(cmd) + best_smart_data = dict(smart_data) + break else: # print(f"[v0] No usable output (return code {result_code}), trying next command...") pass @@ -5183,7 +5161,7 @@ def _get_smart_data_uncached(disk_name): except Exception as kill_err: # print(f"[v0] Error killing process: {kill_err}") pass - + smart_data = best_smart_data if smart_data['reallocated_sectors'] > 0 or smart_data['pending_sectors'] > 0: if smart_data['health'] == 'healthy': @@ -5250,9 +5228,6 @@ def _get_smart_data_uncached(disk_name): smart_data['rotation_rate'] = -1 # HDD, RPM unknown except Exception: pass - smart_data.pop('_rotation_known', None) - - except FileNotFoundError: # print(f"[v0] ERROR: smartctl not found - install smartmontools for disk monitoring.") pass @@ -5263,6 +5238,10 @@ def _get_smart_data_uncached(disk_name): traceback.print_exc() + # These fields only guide transport selection internally. Keep them out + # of the public API even when smartctl is missing or a probe raises. + smart_data.pop('_rotation_known', None) + smart_data.pop('_details_found', None) return smart_data # ─── Proxmox storage cache (Sprint 14 perf pass) ───────────────────────────── @@ -8941,16 +8920,18 @@ def _get_hardware_info_uncached(): # two separate calls that parsed the same output. sata_version = None form_factor = None + model_family = None try: - result_smart = subprocess.run( - ['smartctl', '-n', 'standby', '-i', f'/dev/{disk_name}'], - capture_output=True, text=True, timeout=5) - if result_smart.returncode == 0: - for line in result_smart.stdout.split('\n'): - if 'SATA Version is:' in line: - sata_version = line.split(':', 1)[1].strip() - elif 'Form Factor:' in line: - form_factor = line.split(':', 1)[1].strip() + identity_probe = probe_smartctl_json( + disk_name, + ('-n', 'standby', '-i', '-j'), + timeout=5, + require_telemetry=False, + ) + identity_data = identity_probe.get('data', {}) + sata_version = identity_data.get('sata_version', {}).get('string') + form_factor = identity_data.get('form_factor', {}).get('name') + model_family = identity_data.get('model_family') except: pass @@ -8976,17 +8957,8 @@ def _get_hardware_info_uncached(): if pcie_info: storage_device.update(pcie_info) - # Add family if available (from smartctl) - try: - result_smart = subprocess.run(['smartctl', '-i', f'/dev/{disk_name}'], - capture_output=True, text=True, timeout=5) - if result_smart.returncode == 0: - for line in result_smart.stdout.split('\n'): - if 'Model Family:' in line: - storage_device['family'] = line.split(':', 1)[1].strip() - break - except: - pass + if model_family: + storage_device['family'] = model_family storage_devices.append(storage_device) @@ -10147,12 +10119,16 @@ def api_smart_status(disk_name): # Get device identity via smartctl (works for both NVMe and SATA) _sctl_identity = {} + _sctl_data = {} + _sctl_probe = {} try: - _sctl_proc = subprocess.run( - ['smartctl', '-a', '--json=c', device], - capture_output=True, text=True, timeout=15 + _sctl_probe = probe_smartctl_json( + disk_name, + ('-a', '--json=c'), + timeout=15, + require_telemetry=not is_nvme, ) - _sctl_data = json.loads(_sctl_proc.stdout) + _sctl_data = _sctl_probe.get('data', {}) _sctl_identity = { 'model': _sctl_data.get('model_name', ''), 'serial': _sctl_data.get('serial_number', ''), @@ -10366,16 +10342,17 @@ def api_smart_status(disk_name): result['nvme_error'] = str(e) else: # SATA/SAS/SSD: Single JSON call gives all data at once - proc = subprocess.run( - ['smartctl', '-a', '--json=c', device], - capture_output=True, text=True, timeout=30 - ) - # Parse JSON regardless of exit code — smartctl uses bit-flags for non-fatal conditions - data = {} - try: - data = json.loads(proc.stdout) - except (json.JSONDecodeError, ValueError): - pass + data = _sctl_data + resolved_type_args = smartctl_type_args(_sctl_probe.get('probe')) + if not data: + smart_probe = probe_smartctl_json( + disk_name, + ('-a', '--json=c'), + timeout=30, + require_telemetry=True, + ) + data = smart_probe.get('data', {}) + resolved_type_args = smartctl_type_args(smart_probe.get('probe')) # --- Detect device protocol (ATA vs SCSI/SAS) --- device_protocol = data.get('device', {}).get('protocol', '') @@ -10409,7 +10386,12 @@ def api_smart_status(disk_name): # Fallback text detection in case JSON misses it if result['status'] != 'running': try: - cproc = subprocess.run(['smartctl', '-c', device], capture_output=True, text=True, timeout=10) + cproc = subprocess.run( + ['smartctl', '-c', *resolved_type_args, device], + capture_output=True, + text=True, + timeout=10, + ) if 'Self-test routine in progress' in cproc.stdout or '% of test remaining' in cproc.stdout: result['status'] = 'running' match = re.search(r'(\d+)% of test remaining', cproc.stdout) @@ -10777,7 +10759,12 @@ def api_smart_status(disk_name): # Fallback: if JSON gave no attributes, try text parser if not attrs: try: - aproc = subprocess.run(['smartctl', '-A', device], capture_output=True, text=True, timeout=10) + aproc = subprocess.run( + ['smartctl', '-A', *resolved_type_args, device], + capture_output=True, + text=True, + timeout=10, + ) if aproc.returncode == 0: attrs = _parse_smart_attributes(aproc.stdout.split('\n')) except Exception: @@ -11046,8 +11033,10 @@ def api_smart_run_test(disk_name): return jsonify({'error': 'smartmontools not installed. Please run: apt-get install smartmontools'}), 400 test_flag = '-t short' if test_type == 'short' else '-t long' + smart_probe = resolve_smartctl_probe(disk_name, timeout=10) + smart_type_args = smartctl_type_args(smart_probe) proc = subprocess.run( - ['smartctl'] + test_flag.split() + [device], + ['smartctl'] + test_flag.split() + smart_type_args + [device], capture_output=True, text=True, timeout=30 ) @@ -11070,13 +11059,16 @@ def api_smart_run_test(disk_name): # Start background monitor to save JSON when test completes sleep_interval = 10 if test_type == 'short' else 60 + smart_check_cmd = shlex.join(['smartctl', '-c', *smart_type_args, device]) + smart_report_cmd = shlex.join(['smartctl', '-a', '--json=c', *smart_type_args, device]) + json_path_quoted = shlex.quote(json_path) subprocess.Popen( f''' sleep 5 - while smartctl -c {device} 2>/dev/null | grep -qiE 'Self-test routine in progress|[1-9][0-9]?% of test remaining'; do + while {smart_check_cmd} 2>/dev/null | grep -qiE 'Self-test routine in progress|[1-9][0-9]?% of test remaining'; do sleep {sleep_interval} done - smartctl -a --json=c {device} > {json_path} 2>/dev/null + {smart_report_cmd} > {json_path_quoted} 2>/dev/null ''', shell=True, start_new_session=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL diff --git a/AppImage/scripts/health_monitor.py b/AppImage/scripts/health_monitor.py index fd472242..f6a0560b 100644 --- a/AppImage/scripts/health_monitor.py +++ b/AppImage/scripts/health_monitor.py @@ -19,6 +19,11 @@ from collections import defaultdict import re from health_persistence import health_persistence, disk_base_name +from smartctl_resolver import ( + is_usb_disk as resolver_is_usb_disk, + probe_smartctl_json, + smart_json_has_telemetry, +) try: from proxmox_storage_monitor import proxmox_storage_monitor @@ -100,14 +105,6 @@ def _fmt_entity_and_summary(items, singular: str, plural: str, limit: int = _NAM return title_entity, reason -# USB-NVMe bridges (ASMedia, JMicron, Realtek) answer plain smartctl with -# the *bridge* identity — model shows as "ASMT 2462 NVME" and there is no -# temperature. Only `-d snt*` passes through to the actual NVMe controller -# behind the bridge. For removable disks we try the snt* variants first -# so both identity and health reflect the drive, not the enclosure. -_USB_NVME_DRIVERS = ('sntasmedia', 'sntjmicron', 'sntrealtek') - - def _disk_base_for_sysfs(name: str) -> str: """Normalize `/dev/sda` / `sda` to just `sda` for `/sys/block/` lookups.""" if name.startswith('/dev/'): @@ -142,10 +139,13 @@ def _hdd_in_standby(disk_name: str) -> bool: return False parked = False try: - r = subprocess.run( - ['smartctl', '-n', 'standby', '-i', f'/dev/{base}'], - capture_output=True, text=True, timeout=5) - parked = r.returncode == 2 + result = probe_smartctl_json( + base, + ('-n', 'standby', '-i', '-j'), + timeout=5, + require_telemetry=False, + ) + parked = bool(result.get('standby')) except Exception: parked = False _standby_cache[base] = (now, parked) @@ -195,19 +195,8 @@ def _is_disk_removable(disk_name: str) -> bool: def _is_disk_usb(disk_name: str) -> bool: - """True if the disk sits behind a USB bus. Reads the resolved sysfs - device path — reliable for USB-NVMe bridges and USB-attached HDDs - that report `removable=0` even though they ARE USB (so the older - `_is_disk_removable` heuristic skipped snt* driver probes and left - NVMe-behind-a-bridge disks with the bridge's own chatter cached - forever).""" - try: - base = _disk_base_for_sysfs(disk_name) - real = os.path.realpath(f'/sys/block/{base}') - return any(seg.startswith('usb') and (len(seg) == 3 or seg[3:].isdigit()) - for seg in real.split('/')) - except Exception: - return False + """True when sysfs places the disk behind a USB bus.""" + return resolver_is_usb_disk(disk_name) class HealthMonitor: """ @@ -2057,21 +2046,9 @@ class HealthMonitor: is_usb = tran == 'USB' is_nvme = disk_name.startswith('nvme') - # Get serial from smartctl - serial = '' - model = '' - try: - smart_result = subprocess.run( - ['smartctl', '-i', '-j', f'/dev/{disk_name}'], - capture_output=True, text=True, timeout=5 - ) - if smart_result.returncode in (0, 4): # 4 = SMART not available but info OK - import json - smart_data = json.loads(smart_result.stdout) - serial = smart_data.get('serial_number', '') - model = smart_data.get('model_name', '') or smart_data.get('model_family', '') - except Exception: - pass + identity = self._get_disk_identity(disk_name) + serial = identity.get('serial', '') + model = identity.get('model', '') physical_disks[disk_name] = { 'serial': serial, @@ -2582,41 +2559,15 @@ class HealthMonitor: try: dev_path = f'/dev/{disk_name}' if not disk_name.startswith('/') else disk_name - # USB-attached disks may sit behind an NVMe bridge: try the - # snt* driver variants first so identity reflects the drive - # (Samsung 990 PRO) rather than the enclosure (ASMT 2462 NVME). - # If all snt* fail, fall through to the plain call — that's - # still correct for USB-SATA sticks and non-USB devices. - # USB detection is by sysfs path (`_is_disk_usb`) rather than - # the `removable` flag, since USB-NVMe and USB-HDD both report - # `removable=0` even though they ARE USB. - attempts = [] - # SNT drivers are NVMe-Storage-Namespace-Transport — only - # meaningful when the underlying device is NVMe. Restricting - # to `nvme*` kernel nodes stops `sd*` USB-SATA (TerraMaster - # DAS, USB HDD/SSD enclosures) from paying 3×5 s of dead - # smartctl timeouts before the plain probe runs (GH #293). - is_nvme_class = disk_name.startswith('nvme') - if is_nvme_class and (_is_disk_usb(disk_name) or _is_disk_removable(disk_name)): - for drv in _USB_NVME_DRIVERS: - attempts.append(['smartctl', '-i', '-j', '-d', drv, dev_path]) - attempts.append(['smartctl', '-i', '-j', dev_path]) - - import json as _json - for cmd in attempts: - proc = subprocess.run(cmd, capture_output=True, text=True, timeout=5) - if proc.returncode not in (0, 4): - continue - try: - data = _json.loads(proc.stdout) - except Exception: - continue - serial = data.get('serial_number', '') - model = data.get('model_name', '') or data.get('model_family', '') - if serial or model: - result['serial'] = serial - result['model'] = model - break + probe = probe_smartctl_json( + dev_path, + ('-i', '-j'), + timeout=5, + require_telemetry=False, + ) + data = probe.get('data', {}) + result['serial'] = data.get('serial_number', '') + result['model'] = data.get('model_name', '') or data.get('model_family', '') except Exception: pass @@ -2649,53 +2600,24 @@ class HealthMonitor: try: dev_path = f'/dev/{disk_name}' if not disk_name.startswith('/') else disk_name - # `-n standby` skips the command (exit code 2, no disk I/O) - # when the drive is parked, preventing the health poller - # from spinning up HDDs that hdparm / hd-idle just put to - # sleep — issue #232. The "UNKNOWN" branch below correctly - # keeps the previous cached result alive on exit code 2. - # - # USB-attached disks may sit behind an NVMe bridge: try snt* - # drivers first so health reflects the actual NVMe controller. - # A bridge that fakes "PASSED" while the drive behind it is - # failing is exactly the false-negative we want to avoid. - # USB detection uses the sysfs path so USB-NVMe bridges (which - # report removable=0) are caught too. - attempts = [] - # Same NVMe-class guard as `_get_disk_identity` — see the - # comment there. Prevents USB-SATA drives from wasting - # timeouts on drivers that will never respond (GH #293). - is_nvme_class = disk_name.startswith('nvme') - if is_nvme_class and (_is_disk_usb(disk_name) or _is_disk_removable(disk_name)): - for drv in _USB_NVME_DRIVERS: - attempts.append(['smartctl', '-n', 'standby', '--health', '-j', '-d', drv, dev_path]) - attempts.append(['smartctl', '-n', 'standby', '--health', '-j', dev_path]) + probe = probe_smartctl_json( + dev_path, + ('-n', 'standby', '-a', '-j'), + timeout=5, + require_telemetry=True, + ) + if probe.get('standby'): + return cached['result'] if cached else 'UNKNOWN' - import json as _json - smart_result = None - for cmd in attempts: - result = subprocess.run(cmd, capture_output=True, text=True, timeout=5) - if result.returncode == 2: - # Drive in standby — reuse the previous health state - # if we have one, otherwise report UNKNOWN. Either way, - # don't refresh the cache TTL so we retry on the next - # cycle (a drive can come out of standby at any time). - if cached: - return cached['result'] - return 'UNKNOWN' - try: - data = _json.loads(result.stdout) - except Exception: - continue - passed = data.get('smart_status', {}).get('passed', None) - if passed is True: - smart_result = 'PASSED' - break - if passed is False: - smart_result = 'FAILED' - break - # No opinion yet — next attempt (fallthrough to plain). - if smart_result is None: + data = probe.get('data', {}) + passed = data.get('smart_status', {}).get('passed') + if _is_disk_usb(disk_name) and not smart_json_has_telemetry(data): + smart_result = 'UNKNOWN' + elif passed is True: + smart_result = 'PASSED' + elif passed is False: + smart_result = 'FAILED' + else: smart_result = 'UNKNOWN' # Cache the result with the device fingerprint for hot-swap invalidation @@ -4358,14 +4280,8 @@ class HealthMonitor: try: obs_serial = None try: - sm = subprocess.run( - ['smartctl', '-i', f'/dev/{base_device}'], - capture_output=True, text=True, timeout=3) - if sm.returncode in (0, 4): - for sline in sm.stdout.split('\n'): - if 'Serial Number' in sline or 'Serial number' in sline: - obs_serial = sline.split(':')[-1].strip() - break + identity = self._get_disk_identity(base_device) + obs_serial = identity.get('serial') or None except Exception: pass health_persistence.record_disk_observation( diff --git a/AppImage/scripts/smartctl_resolver.py b/AppImage/scripts/smartctl_resolver.py new file mode 100644 index 00000000..684ef75d --- /dev/null +++ b/AppImage/scripts/smartctl_resolver.py @@ -0,0 +1,201 @@ +#!/usr/bin/env python3 + +import json +import os +import subprocess +from typing import Any, Iterable, Optional + + +USB_NVME_DRIVERS = ("sntasmedia", "sntjmicron", "sntrealtek") +USB_SATA_DRIVERS = ("sat", "sat,12", "sat,16") + +_probe_cache: dict[str, tuple[tuple[Any, ...], str]] = {} + + +def disk_name(device: str) -> str: + return os.path.basename(device.rstrip("/")) + + +def disk_fingerprint(device: str) -> tuple[Any, ...]: + name = disk_name(device) + dev_path = f"/dev/{name}" + try: + stat = os.stat(dev_path) + return (os.path.realpath(f"/sys/block/{name}"), stat.st_rdev, stat.st_ctime_ns) + except OSError: + return (os.path.realpath(f"/sys/block/{name}"),) + + +def is_usb_disk(device: str) -> bool: + try: + real_path = os.path.realpath(f"/sys/block/{disk_name(device)}") + return any( + segment.startswith("usb") + and (len(segment) == 3 or segment[3:].isdigit()) + for segment in real_path.split("/") + ) + except OSError: + return False + + +def smartctl_probe_types(device: str) -> tuple[str, ...]: + name = disk_name(device) + if is_usb_disk(name): + if name.startswith("nvme"): + return USB_NVME_DRIVERS + ("auto", "nvme") + return USB_SATA_DRIVERS + ("auto", "scsi", "ata") + if name.startswith("nvme"): + return ("auto", "nvme") + return ("auto", "scsi", "ata", "sat", "sat,12", "sat,16") + + +def smartctl_type_args(probe: Optional[str]) -> list[str]: + if not probe or probe == "auto": + return [] + return ["-d", probe] + + +def smartctl_command(device: str, options: Iterable[str], probe: Optional[str]) -> list[str]: + path = device if device.startswith("/dev/") else f"/dev/{disk_name(device)}" + return ["smartctl", *options, *smartctl_type_args(probe), path] + + +def smartctl_result_is_standby(returncode: int, stdout: str = "", stderr: str = "") -> bool: + """Distinguish a real low-power response from another exit-code-2 error.""" + if returncode != 2: + return False + message = f"{stdout}\n{stderr}".lower() + return any( + marker in message + for marker in ("standby", "sleep mode", "low-power mode", "low power mode") + ) + + +def smart_json_has_telemetry(data: dict[str, Any]) -> bool: + ata_attributes = data.get("ata_smart_attributes", {}).get("table", []) + return bool( + ata_attributes + or data.get("nvme_smart_health_information_log") + or data.get("scsi_error_counter_log") + or "scsi_grown_defect_list" in data + or data.get("ata_smart_data") + or data.get("temperature", {}).get("current") is not None + or data.get("power_on_time") + or data.get("power_cycle_count") is not None + ) + + +def _smart_json_score(data: dict[str, Any]) -> int: + score = 100 if smart_json_has_telemetry(data) else 0 + score += 12 if data.get("temperature", {}).get("current") is not None else 0 + score += 8 if data.get("power_on_time") else 0 + score += 5 if data.get("model_name") or data.get("model_family") else 0 + score += 5 if data.get("serial_number") else 0 + score += 2 if data.get("smart_status", {}).get("passed") is not None else 0 + return score + + +def probe_smartctl_json( + device: str, + options: Iterable[str], + *, + timeout: int = 5, + require_telemetry: bool = True, +) -> dict[str, Any]: + name = disk_name(device) + fingerprint = disk_fingerprint(name) + probes = list(smartctl_probe_types(name)) + cached = _probe_cache.get(name) + if cached and cached[0] == fingerprint and cached[1] in probes: + probes.remove(cached[1]) + probes.insert(0, cached[1]) + + best: dict[str, Any] = { + "data": {}, + "probe": None, + "command": [], + "returncode": None, + "stdout": "", + "stderr": "", + "standby": False, + } + best_score = -1 + + for probe in probes: + command = smartctl_command(name, options, probe) + try: + proc = subprocess.run( + command, + capture_output=True, + text=True, + timeout=timeout, + ) + except (OSError, subprocess.SubprocessError): + continue + + if "-n" in command and smartctl_result_is_standby( + proc.returncode, + proc.stdout, + proc.stderr, + ): + return { + "data": {}, + "probe": probe, + "command": command, + "returncode": proc.returncode, + "stdout": proc.stdout, + "stderr": proc.stderr, + "standby": True, + } + + try: + data = json.loads(proc.stdout) if proc.stdout else {} + except (TypeError, json.JSONDecodeError): + data = {} + if not isinstance(data, dict) or not data: + continue + + score = _smart_json_score(data) + candidate = { + "data": data, + "probe": probe, + "command": command, + "returncode": proc.returncode, + "stdout": proc.stdout, + "stderr": proc.stderr, + "standby": False, + } + if score > best_score: + best = candidate + best_score = score + + identity = bool(data.get("model_name") or data.get("model_family") or data.get("serial_number")) + status = data.get("smart_status", {}).get("passed") is not None + complete = smart_json_has_telemetry(data) if require_telemetry else (identity or status) + if complete: + _probe_cache[name] = (fingerprint, probe) + return candidate + + return best + + +def resolve_smartctl_probe(device: str, timeout: int = 5) -> Optional[str]: + name = disk_name(device) + cached = _probe_cache.get(name) + fingerprint = disk_fingerprint(name) + if cached and cached[0] == fingerprint: + return cached[1] + result = probe_smartctl_json( + name, + ("-i", "-j"), + timeout=timeout, + require_telemetry=False, + ) + return result.get("probe") + + +def clear_smartctl_probe(device: Optional[str] = None) -> None: + if device is None: + _probe_cache.clear() + else: + _probe_cache.pop(disk_name(device), None) diff --git a/scripts/storage/smart-disk-test.sh b/scripts/storage/smart-disk-test.sh index 915a5719..a4e37b36 100644 --- a/scripts/storage/smart-disk-test.sh +++ b/scripts/storage/smart-disk-test.sh @@ -58,6 +58,25 @@ _smart_is_nvme() { [[ "$1" == *nvme* ]] } +SMARTCTL_TYPE_ARGS=() + +_smart_resolve_type() { + local disk="$1" base real dtype output + SMARTCTL_TYPE_ARGS=() + _smart_is_nvme "$disk" && return + base=$(basename "$disk") + real=$(readlink -f "/sys/block/$base" 2>/dev/null) + [[ "$real" != */usb* ]] && return + + for dtype in sat sat,12 sat,16; do + output=$(smartctl -i -j -d "$dtype" "$disk" 2>/dev/null) + if printf '%s' "$output" | grep -qE '"(model_name|model_family|serial_number)"'; then + SMARTCTL_TYPE_ARGS=(-d "$dtype") + return + fi + done +} + _smart_disk_label() { local disk="$1" local model size @@ -159,6 +178,7 @@ SELECTED_DISK=$(dialog --backtitle "$BACKTITLE" \ # ── Steps 3+: Action loop for the selected disk ─────────── DISK_LABEL=$(_smart_disk_label "$SELECTED_DISK") +_smart_resolve_type "$SELECTED_DISK" mkdir -p "$SMART_DIR" while true; do @@ -221,8 +241,8 @@ while true; do fi else msg_info "$(translate 'Reading SMART data...')" - HEALTH=$(smartctl -H "$SELECTED_DISK" 2>/dev/null | grep -i "overall-health") - ATTRS=$(smartctl -A "$SELECTED_DISK" 2>/dev/null) + HEALTH=$(smartctl -H "${SMARTCTL_TYPE_ARGS[@]}" "$SELECTED_DISK" 2>/dev/null | grep -i "overall-health") + ATTRS=$(smartctl -A "${SMARTCTL_TYPE_ARGS[@]}" "$SELECTED_DISK" 2>/dev/null) stop_spinner if [[ -z "$HEALTH" ]]; then msg_error "$(translate 'Could not read SMART data from') $SELECTED_DISK" @@ -246,7 +266,7 @@ while true; do nvme smart-log "$SELECTED_DISK" > "$TMPFILE" 2>/dev/null nvme id-ctrl "$SELECTED_DISK" >> "$TMPFILE" 2>/dev/null else - smartctl -x "$SELECTED_DISK" > "$TMPFILE" 2>/dev/null + smartctl -x "${SMARTCTL_TYPE_ARGS[@]}" "$SELECTED_DISK" > "$TMPFILE" 2>/dev/null fi stop_spinner if [[ -s "$TMPFILE" ]]; then @@ -274,7 +294,7 @@ while true; do fi else msg_info "$(translate 'Starting SMART short self-test...')" - OUTPUT=$(smartctl -t short "$SELECTED_DISK" 2>/dev/null) + OUTPUT=$(smartctl -t short "${SMARTCTL_TYPE_ARGS[@]}" "$SELECTED_DISK" 2>/dev/null) stop_spinner if echo "$OUTPUT" | grep -qi "Test will complete"; then msg_ok "$(translate 'Short self-test started on') $SELECTED_DISK" @@ -294,6 +314,10 @@ while true; do _smart_cleanup_old_jsons "$SELECTED_DISK" DISK_SAFE=$(printf '%q' "$SELECTED_DISK") JSON_SAFE=$(printf '%q' "$JSON_PATH") + SMARTCTL_TYPE_SAFE="" + if [[ ${#SMARTCTL_TYPE_ARGS[@]} -gt 0 ]]; then + printf -v SMARTCTL_TYPE_SAFE '%q ' "${SMARTCTL_TYPE_ARGS[@]}" + fi if _smart_is_nvme "$SELECTED_DISK"; then msg_info "$(translate 'Starting NVMe long self-test...')" @@ -330,7 +354,7 @@ while true; do fi else msg_info "$(translate 'Starting SMART long self-test...')" - OUTPUT=$(smartctl -t long "$SELECTED_DISK" 2>/dev/null) + OUTPUT=$(smartctl -t long "${SMARTCTL_TYPE_ARGS[@]}" "$SELECTED_DISK" 2>/dev/null) stop_spinner if echo "$OUTPUT" | grep -qi "Test will complete"; then msg_ok "$(translate 'Long self-test started on') $SELECTED_DISK" @@ -342,15 +366,15 @@ while true; do DISK_LABEL_SAFE=$(printf '%q' "$DISK_LABEL") NOTIFY_SCRIPT="/usr/bin/notification_manager.py" nohup bash -c " - while smartctl -c ${DISK_SAFE} 2>/dev/null | grep -qiE 'Self-test routine in progress|[1-9][0-9]?% of test remaining'; do + while smartctl ${SMARTCTL_TYPE_SAFE} -c ${DISK_SAFE} 2>/dev/null | grep -qiE 'Self-test routine in progress|[1-9][0-9]?% of test remaining'; do sleep 60 done - smartctl -a --json=c ${DISK_SAFE} > ${JSON_SAFE} 2>/dev/null + smartctl ${SMARTCTL_TYPE_SAFE} -a --json=c ${DISK_SAFE} > ${JSON_SAFE} 2>/dev/null # Send notification when test completes if [[ -f \"${NOTIFY_SCRIPT}\" ]]; then HOSTNAME=\$(hostname -s) - TEST_RESULT=\$(smartctl -l selftest ${DISK_SAFE} 2>/dev/null | grep -E '^# ?1') + TEST_RESULT=\$(smartctl ${SMARTCTL_TYPE_SAFE} -l selftest ${DISK_SAFE} 2>/dev/null | grep -E '^# ?1') if echo \"\$TEST_RESULT\" | grep -qi 'Completed without error'; then python3 \"${NOTIFY_SCRIPT}\" --action send-raw --severity INFO \ --title \"\${HOSTNAME}: SMART Long Test Completed\" \ @@ -388,9 +412,9 @@ while true; do else msg_info "$(translate 'Reading SMART self-test log...')" # Active test: only "X% of test remaining" appears when a test is actually running - ACTIVE=$(smartctl -c "$SELECTED_DISK" 2>/dev/null | grep -iE "[1-9][0-9]?% of test remaining|Self-test routine in progress") + ACTIVE=$(smartctl -c "${SMARTCTL_TYPE_ARGS[@]}" "$SELECTED_DISK" 2>/dev/null | grep -iE "[1-9][0-9]?% of test remaining|Self-test routine in progress") # Log: grab only result rows (^# N ...) and the column header (^Num) - LOG_OUT=$(smartctl -l selftest "$SELECTED_DISK" 2>/dev/null) + LOG_OUT=$(smartctl -l selftest "${SMARTCTL_TYPE_ARGS[@]}" "$SELECTED_DISK" 2>/dev/null) LOG_HEADER=$(echo "$LOG_OUT" | grep -E "^Num") LOG_ENTRIES=$(echo "$LOG_OUT" | grep -E "^# ?[0-9]") stop_spinner @@ -428,7 +452,7 @@ while true; do if _smart_is_nvme "$SELECTED_DISK"; then nvme smart-log -o json "$SELECTED_DISK" > "$JSON_PATH" 2>/dev/null else - smartctl -a --json=c "$SELECTED_DISK" > "$JSON_PATH" 2>/dev/null + smartctl -a --json=c "${SMARTCTL_TYPE_ARGS[@]}" "$SELECTED_DISK" > "$JSON_PATH" 2>/dev/null fi [[ -s "$JSON_PATH" ]] || rm -f "$JSON_PATH" fi diff --git a/scripts/storage/smart-scheduled-test.sh b/scripts/storage/smart-scheduled-test.sh index 1c7f331f..69174957 100644 --- a/scripts/storage/smart-scheduled-test.sh +++ b/scripts/storage/smart-scheduled-test.sh @@ -64,6 +64,25 @@ _is_nvme() { [[ "$1" == *nvme* ]] } +SMARTCTL_TYPE_ARGS=() + +_resolve_smart_type() { + local disk="$1" base real dtype output + SMARTCTL_TYPE_ARGS=() + _is_nvme "$disk" && return + base=$(basename "$disk") + real=$(readlink -f "/sys/block/$base" 2>/dev/null) + [[ "$real" != */usb* ]] && return + + for dtype in sat sat,12 sat,16; do + output=$(smartctl -i -j -d "$dtype" "$disk" 2>/dev/null) + if printf '%s' "$output" | grep -qE '"(model_name|model_family|serial_number)"'; then + SMARTCTL_TYPE_ARGS=(-d "$dtype") + return + fi + done +} + _get_json_path() { local disk="$1" local test_type="$2" @@ -125,9 +144,11 @@ _run_test() { # SATA/SAS test local test_flag="-t short" [[ "$test_type" == "long" ]] && test_flag="-t long" + _resolve_smart_type "$disk" - smartctl $test_flag "$disk" 2>/dev/null - if [[ $? -ne 0 && $? -ne 4 ]]; then + smartctl $test_flag "${SMARTCTL_TYPE_ARGS[@]}" "$disk" 2>/dev/null + local test_rc=$? + if [[ $test_rc -ne 0 && $test_rc -ne 4 ]]; then log "ERROR: Failed to start SMART test on $disk" return 1 fi @@ -137,12 +158,12 @@ _run_test() { [[ "$test_type" == "long" ]] && sleep_interval=60 sleep 5 - while smartctl -c "$disk" 2>/dev/null | grep -qiE 'Self-test routine in progress|[1-9][0-9]?% of test remaining'; do + while smartctl -c "${SMARTCTL_TYPE_ARGS[@]}" "$disk" 2>/dev/null | grep -qiE 'Self-test routine in progress|[1-9][0-9]?% of test remaining'; do sleep $sleep_interval done # Save results - smartctl -a --json=c "$disk" > "$json_path" 2>/dev/null + smartctl -a --json=c "${SMARTCTL_TYPE_ARGS[@]}" "$disk" > "$json_path" 2>/dev/null fi log "Test completed on $disk, results saved to $json_path"