Fixed disk_smart test to handle more smartctl -l output formats

1 files changed, 62 insertions, 33 deletions

diff --git a/bin/disk_smart b/bin/disk_smart
index c7f58df..936a926 100755
--- a/bin/disk_smart
+++ b/bin/disk_smart
@@ -7,6 +7,7 @@ Copyright (C) 2010 Canonical Ltd.
 Authors
 Jeff Lane <jeffrey.lane@canonical.com>
 Brendan Donegan <brendan.donegan@canonical.com>
+ Rod Smith <rod.smith@canonical.com>
 
 This program is free software: you can redistribute it and/or modify
 it under the terms of the GNU General Public License version 2,
@@ -34,6 +35,7 @@ USB/eSATA/eSAS attached storage devices.
 
 Changelog:
 
+v1.2: Handle multiple output formats for "smartctl -l"
 v1.1: Put delay before first attempt to acces log, rather than after
 v1.0: added debugger class and code to allow for verbose debug output if needed
 
@@ -72,8 +74,10 @@ class ListHandler(logging.StreamHandler):
 msg = msg.decode()
 logger = logging.getLogger(record.name)
 new_record = logger.makeRecord(record.name, record.levelno,
- record.pathname, record.lineno, msg, record.args,
- record.exc_info, record.funcName)
+ record.pathname, record.lineno,
+ msg, record.args,
+ record.exc_info,
+ record.funcName)
 logging.StreamHandler.emit(self, new_record)
 
 else:
@@ -84,7 +88,7 @@ def is_smart_enabled(disk):
 # Check with smartctl to see if SMART is available and enabled on the disk
 command = 'smartctl -i %s' % disk
 diskinfo_bytes = (Popen(command, stdout=PIPE, shell=True)
- .communicate()[0])
+ .communicate()[0])
 diskinfo = diskinfo_bytes.decode().splitlines()
 
 logging.debug('SMART Info for disk %s', disk)
@@ -149,6 +153,58 @@ def get_smart_entries(disk, type='selftest'):
 
 return entries
 
+
+# Returns True if an "in-progress" message is found in the smartctl
+# output, False if such a message is not found. In the former case,
+# the in-progress message entries are logged.
+def in_progress(current_entries):
+ ret_val = False
+ for entry in current_entries:
+ if not isinstance(entry, str):
+ if entry['status'] == 'Self-test routine in progress':
+ logging.debug('%s %s %s %s' % (entry['number'],
+ entry['description'],
+ entry['status'],
+ entry['remaining']))
+ ret_val = True
+ break
+ return ret_val
+
+
+# Wait for SMART test to complete; no return value.
+# Note that different disks return different types of values.
+# Some return no status reports while a test is ongoing; others
+# show a status line at the START of the list of tests, and
+# others show a status line at the END of the list of tests
+# (and then move it to the top once the tests are done).
+def poll_for_status(args, disk, previous_entries):
+ # Priming read... this is here in case our test is finished or fails
+ # immediate after it begins.
+ logging.debug('Polling selftest.log for status')
+ keep_going = True
+
+ while keep_going:
+ # Poll every sleep seconds until test is complete$
+ time.sleep(args.sleep)
+
+ current_entries = get_smart_entries(disk)
+ if current_entries != previous_entries:
+ if not in_progress(current_entries):
+ keep_going = False
+
+ if args.timeout is not None:
+ if args.timeout <= 0:
+ logging.debug('Polling timed out')
+ return 1
+ else:
+ args.timeout -= args.sleep
+
+ if isinstance(current_entries[0], str):
+ return current_entries[0]
+ else:
+ return current_entries[0]['status']
+
+
 def main():
 description = 'Tests that SMART capabilities on disks that support SMART function.'
 parser = ArgumentParser(description=description)
@@ -184,7 +240,7 @@ def main():
 logger.setLevel(logging.INFO)
 
 # Make sure we're root, because smartctl doesn't work otherwise.
- if not os.geteuid()==0:
+ if not os.geteuid() == 0:
 parser.error("You must be root to run this program")
 
 # If SMART is available and enabled, we proceed. Otherwise, we exit as the
@@ -207,39 +263,12 @@ def main():
 run_smart_test(disk)
 previous_entries = get_smart_entries(disk)
 
- # Priming read... this is here in case our test is finished or fails
- # immediate after it begins.
- logging.debug('Polling selftest.log for status')
-
- while True:
- # Poll every sleep seconds until test is complete$
- time.sleep(args.sleep)
-
- current_entries = get_smart_entries(disk)
- if isinstance(current_entries[0], str):
- logging.debug(current_entries[0])
- else:
- logging.debug('%s %s %s %s' % (current_entries[0]['number'],
- current_entries[0]['description'],
- current_entries[0]['status'],
- current_entries[0]['remaining']))
- if current_entries != previous_entries \
- and current_entries[0]["status"] != 'Self-test routine in progress':
- break
-
- if args.timeout is not None:
- if args.timeout <= 0:
- logging.debug('Polling timed out')
- return 1
- else:
- args.timeout -= args.sleep
-
- status = current_entries[0]['status']
+ status = poll_for_status(args, disk, previous_entries)
 
 if status != 'Completed without error':
 log = get_smart_entries(disk)
 logging.error("FAIL: SMART Self-Test appears to have failed for some reason. "
- "Run 'sudo smartctl -l selftest %s' to see the SMART log" % disk)
+ "Run 'sudo smartctl -l selftest %s' to see the SMART log" % disk)
 logging.debug("Last self-test run status: %s" % status)
 return 1
 else: