1#!/usr/bin/env python3

2"""

3 Script to automate disk SMART testing.

4

5 Copyright (C) 2010-2016 Canonical Ltd.

6

7 Authors

8 Jeff Lane <jeffrey.lane@canonical.com>

9 Brendan Donegan <brendan.donegan@canonical.com>

10 Rod Smith <rod.smith@canonical.com>

11

12 This program is free software: you can redistribute it and/or modify

13 it under the terms of the GNU General Public License version 2,

14 as published by the Free Software Foundation.

15

16 This program is distributed in the hope that it will be useful,

17 but WITHOUT ANY WARRANTY; without even the implied warranty of

18 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

19 GNU General Public License for more details.

20

21 You should have received a copy of the GNU General Public License

22 along with this program. If not, see <http://www.gnu.org/licenses/>.

23

24 The purpose of this script is to simply interact with an onboard hard disk and

25 check for SMART capability and then do a little bit of interaction to make sure

26 we can at least do some limited interaction with the hard disk's SMART

27 functions.

28

29 We assume that SMART is available. The test will fail if this is not the case.

30 The block_device_resource script includes a test of SMART availability.

31 Checkbox tests for SMART availability as part of the disk/smart provider

32 definition, which uses block_device_resource as part of its requires: test.

33

34 This script runs the SMART short self test. It returns 0 if it's all good,

35 and 1 if it fails.

36

37 NOTE: This may not work correctly on systems where the onboard storage is

38 controlled by a hardware RAID controller, on external RAID systems, SAN, and

39 USB/eSATA/eSAS attached storage devices. Such systems should be filtered

40 out by the SMART availability test in block_device_resource.

41

42 Changelog:

43

44 v1.4: Fix script failure on disks with no pre-existing SMART tests

45 v1.3: Fix detection of SMART availability & activate SMART if available

46 but deactivated. Also use smartctl return value rather than string-

47 matching to determine if a test has failed; this should be more

48 robust, as output strings vary between disks.

49 v1.2: Handle multiple output formats for "smartctl -l"

50 v1.1: Put delay before first attempt to acces log, rather than after

51 v1.0: added debugger class and code to allow for verbose debug output if needed

52

53 v0.4: corrected some minor things

54 added option parsing to allow for many disks, or disks other than

55 "/dev/sda"

56

57 V0.3: Removed the arbitrary wait time and implemented a polling method

58 to shorten the test time.

59 Added in Pass/Fail criteria for the final outcome.

60 Added in documentation.

61

62 V0.2: added minor debug routine

63

64 V0.1: Fixed some minor bugs and added the SmartEnabled() function

65

66 V0: First draft

67

68"""

69

70 import os

71 import sys

72 import time

73 import logging

74 import shlex

75

76 from subprocess import Popen, PIPE, check_call, check_output

77 from subprocess import CalledProcessError

78 from argparse import ArgumentParser

79

80# NOTE: If raid_types changes, also change it in block_device_resource script!

81 raid_types = ["megaraid", "cciss", "3ware", "areca"]

82

83

84 class ListHandler(logging.StreamHandler):

85

86 def emit(self, record):

87 if isinstance(record.msg, (list, tuple)):

88 for msg in record.msg:

89 if type(msg) is bytes:

90 msg = msg.decode()

91 logger = logging.getLogger(record.name)

92 new_record = logger.makeRecord(

93 record.name,

94 record.levelno,

95 record.pathname,

96 record.lineno,

97 msg,

98 record.args,

99 record.exc_info,

100 record.funcName,

101 )

102 logging.StreamHandler.emit(self, new_record)

103

104 else:

105 logging.StreamHandler.emit(self, record)

106

107

108 def enable_smart(disk, raid_element, raid_type):

109 """Log data and, if necessary, enable SMART on the specified disk.

110

111 See also smart_support() in block_device_resource script.

112 :param disk:

113 disk device filename (e.g., /dev/sda)

114 :param raid_element:

115 element number to enable in RAID array; undefined if not a RAID device

116 :param raid_type:

117 type of raid device (none, megaraid, etc.)

118 :returns:

119 True if enabling smart was successful, False otherwise

120 """

121 # Check with smartctl to record basic SMART data on the disk

122 if raid_type == "none":

123 command = "smartctl -i {}".format(disk)

124 logging.debug("SMART Info for disk {}".format(disk))

125 else:

126 command = "smartctl -i {} -d {},{}".format(

127 disk, raid_type, raid_element

128 )

129 logging.debug(

130 "SMART Info for disk {}, element {}".format(disk, raid_element)

131 )

132 diskinfo_bytes = Popen(command, stdout=PIPE, shell=True).communicate()[0]

133 diskinfo = diskinfo_bytes.decode(

134 encoding="utf-8", errors="ignore"

135 ).splitlines()

136 logging.debug(diskinfo)

137 if len(diskinfo) > 2 and not any(

138 "SMART support is" in s and "Enabled" in s for s in diskinfo

139 ):

140 logging.debug("SMART disabled; attempting to enable it.")

141 if raid_type == "none":

142 command = "smartctl -s on {}".format(disk)

143 else:

144 command = "smartctl -s on {} -d {},{}".format(

145 disk, raid_type, raid_element

146 )

147 try:

148 check_call(shlex.split(command))

149 return True

150 except CalledProcessError:

151 if raid_type == "none":

152 logging.warning(

153 "SMART could not be enabled on {}".format(disk)

154 )

155 else:

156 logging.warning(

157 "SMART could not be enabled on {}, element "

158 "{}".format(disk, raid_element)

159 )

160 return False

161 return True

162

163

164 def count_raid_disks(disk):

165 """Count the disks in a RAID array.

166

167 :param disk:

168 Disk device filename (e.g., /dev/sda)

169 :returns:

170 Number of disks in array (0 for non-RAID disk)

171 Type of RAID (none, megaraid, 3ware, areca, or cciss; note that only

172 none and megaraid are tested, as of Jan. 2016)

173 """

174 raid_element = 0

175 raid_type = "none"

176 command = "smartctl -i {}".format(disk)

177 diskinfo_bytes = Popen(command, stdout=PIPE, shell=True).communicate()[0]

178 diskinfo = diskinfo_bytes.decode(

179 encoding="utf-8", errors="ignore"

180 ).splitlines()

181 for type in raid_types:

182 if any("-d {},N".format(type) in s for s in diskinfo):

183 logging.info("Found RAID controller of type {}".format(type))

184 raid_type = type

185 break

186 if raid_type != "none":

187 # This is a hardware RAID controller, so count individual disks....

188 disk_exists = True

189 while disk_exists:

190 command = "smartctl -i {} -d {},{}".format(

191 disk, raid_type, raid_element

192 )

193 try:

194 check_output(shlex.split(command))

195 raid_element += 1

196 except CalledProcessError:

197 disk_exists = False

198 logging.info(

199 "Counted {} RAID disks on {}\n".format(raid_element, disk)

200 )

201 return raid_element, raid_type

202

203

204 def initiate_smart_test(disk, raid_element, raid_type, type="short"):

205 # Note, '-t force' ensures we abort any existing smart test in progress

206 # and start a clean run.

207 if raid_type == "none":

208 ctl_command = "smartctl -t {} -t force {}".format(type, disk)

209 else:

210 ctl_command = "smartctl -t {} -t force {} -d {},{}".format(

211 type, disk, raid_type, raid_element

212 )

213 logging.debug("Beginning test with {}".format(ctl_command))

214

215 smart_proc = Popen(

216 ctl_command,

217 stderr=PIPE,

218 stdout=PIPE,

219 universal_newlines=True,

220 shell=True,

221 )

222 ctl_output, ctl_error = smart_proc.communicate()

223

224 logging.debug(ctl_error + ctl_output)

225

226 return smart_proc.returncode

227

228

229 def get_smart_entries(disk, raid_element, raid_type, verbose=False):

230 entries = []

231 returncode = 0

232 try:

233 if raid_type == "none":

234 stdout = check_output(

235 ["smartctl", "-l", "selftest", disk], universal_newlines=True

236 )

237 else:

238 stdout = check_output(

239 [

240 "smartctl",

241 "-l",

242 "selftest",

243 disk,

244 "-d",

245 "{},{}".format(raid_type, raid_element),

246 ],

247 universal_newlines=True,

248 )

249 except CalledProcessError as err:

250 if verbose:

251 logging.error("Error encountered checking SMART Log")

252 logging.error("\tsmartctl returned: {}".format(err.returncode))

253 logging.error("\tSee 'man smartctl' for info on return codes")

254 stdout = err.output

255 returncode = err.returncode

256

257 # Skip intro lines

258 stdout_lines = iter(stdout.splitlines())

259 for line in stdout_lines:

260 if line.startswith("SMART") or line.startswith(

261 "No self-tests have been logged"

262 ):

263 break

264

265 # Get lengths from header

266 try:

267 line = next(stdout_lines)

268 except StopIteration:

269 logging.info("No entries found in log")

270 if not line.startswith("Num"):

271 entries.append("No entries found in log yet")

272 return entries, stdout, returncode

273 columns = [

274 "number",

275 "description",

276 "status",

277 "remaining",

278 "lifetime",

279 "lba",

280 ]

281 lengths = [line.index(i) for i in line.split()]

282 lengths[columns.index("remaining")] += len("Remaining") - len("100%")

283 lengths.append(len(line))

284

285 # Get remaining lines

286 entries = []

287 for line in stdout_lines:

288 if line.startswith("#"):

289 entry = {}

290 for i, column in enumerate(columns):

291 entry[column] = line[lengths[i] : lengths[i + 1]].strip()

292

293 # Convert some columns to integers

294 entry["number"] = int(entry["number"][1:])

295 entries.append(entry)

296

297 return entries, stdout, returncode

298

299

300 def in_progress(current_entries):

301 """Check to see if the test is in progress.

302

303 :param current_entries:

304 Output of smartctl command to be checked for status indicator.

305 :returns:

306 True if an "in-progress" message is found, False otherwise

307 """

308 # LP:1612220 Only check first log entry for status to avoid false triggers

309 # on older interrupted tests that may still show an "in progress" status.

310 statuses = [

311 entry

312 for entry in current_entries

313 if isinstance(entry, dict)

314 and "status" in entry

315 and entry["number"] == 1

316 and (

317 entry["status"] == "Self-test routine in progress"

318 or "Self test in progress" in entry["status"]

319 )

320 ]

321 if statuses:

322 for entry in statuses:

323 logging.debug(

324 "%s %s %s %s"

325 % (

326 entry["number"],

327 entry["description"],

328 entry["status"],

329 entry["remaining"],

330 )

331 )

332 return True

333 else:

334 return False

335

336

337 def poll_for_status(args, disk, raid_element, raid_type, previous_entries):

338 """Poll a disk for its SMART status.

339

340 Wait for SMART test to complete; return status and return code.

341 Note that different disks return different types of values.

342 Some return no status reports while a test is ongoing; others

343 show a status line at the START of the list of tests, and

344 others show a status line at the END of the list of tests

345 (and then move it to the top once the tests are done).

346 :param args:

347 Script's command-line arguments

348 :param disk:

349 Disk device (e.g., /dev/sda)

350 :param raid_element:

351 RAID disk number (undefined for non-RAID disk)

352 :param raid_type:

353 Type of RAID device (megaraid, etc.)

354 :param previous_entries:

355 Previous SMART output; used to spot a change

356 :returns:

357 Current output and return code

358 """

359 # Priming read... this is here in case our test is finished or fails

360 # immediate after it beginsAccording to.

361 logging.debug("Polling SMART selftest log for status")

362 keep_going = True

363

364 while keep_going:

365 # Poll every sleep seconds until test is complete

366 time.sleep(args.sleep)

367

368 current_entries, output, returncode = get_smart_entries(

369 disk, raid_element, raid_type

370 )

371 if current_entries != previous_entries:

372 if not in_progress(current_entries):

373 logging.debug(

374 "Current log entries differ from starting log"

375 " entries. Stopping polling."

376 )

377 keep_going = False

378

379 if args.timeout is not None:

380 if args.timeout <= 0:

381 logging.debug("Polling timed out")

382 return "Polling timed out", 1

383 else:

384 args.timeout -= args.sleep

385

386 if isinstance(current_entries[0], str):

387 return current_entries[0], returncode

388 else:

389 return current_entries[0]["status"], returncode

390

391

392 def run_smart_test(args, disk, raid_element, raid_type):

393 """Run a test on a single disk device (possibly multiple RAID elements).

394

395 :param args:

396 Command-line arguments passed to script

397 :param disk:

398 Disk device filename (e.g., /dev/sda)

399 :param raid_element:

400 Number of RAID array element or undefined for non-RAID disk

401 :param raid_type:

402 Type of RAID device (e.g., megaraid)

403 :returns:

404 True for success, False for failure

405 """

406 previous_entries, output, returncode = get_smart_entries(

407 disk, raid_element, raid_type

408 )

409 if raid_type == "none":

410 logging.info("Starting SMART self-test on {}".format(disk))

411 else:

412 logging.info(

413 "Starting SMART self-test on {}, element {}".format(

414 disk, raid_element

415 )

416 )

417 if initiate_smart_test(disk, raid_element, raid_type) != 0:

418 logging.error("Error reported during smartctl test")

419 return False

420

421 if len(previous_entries) > 20:

422 # Abort the previous instance

423 # so that polling can identify the difference

424

425 # The proper way to kill the test is using -X

426 # kill_smart_test(disk, raid_element, raid_type)

427 # Then re-initiate the test

428 logging.debug(

429 "Log is 20+ entries long. Restarting test to add an"

430 " abort message to make the log diff easier"

431 )

432 initiate_smart_test(disk, raid_element, raid_type)

433 previous_entries, output, returncode = get_smart_entries(

434 disk, raid_element, raid_type

435 )

436

437 status, returncode = poll_for_status(

438 args, disk, raid_element, raid_type, previous_entries

439 )

440

441 if returncode != 0:

442 log, output, returncode = get_smart_entries(

443 disk, raid_element, raid_type, True

444 )

445 logging.error(

446 "FAIL: SMART Self-Test appears to have failed " "for some reason."

447 )

448 logging.error("\tLast smartctl return code: %d", returncode)

449 logging.error("\tLast smartctl run status: %s", status)

450 if raid_type == "none":

451 logging.error("\t'smartctl -l selftest {}' output:".format(disk))

452 else:

453 logging.error(

454 "\t'smartctl -l selftest {} -d {},{}' output:".format(

455 disk, raid_type, raid_element

456 )

457 )

458 logging.error("\n%s", output)

459 return False

460 else:

461 if raid_type == "none":

462 logging.info(

463 "PASS: SMART Self-Test on {} completed without error".format(

464 disk

465 )

466 )

467 else:

468 logging.info(

469 "PASS: SMART Self-Test on {}, element {} completed "

470 "without error\n".format(disk, raid_element)

471 )

472 return True

473

474

475 def main():

476 """Test SMART capabilities on disks that support SMART functions."""

477 description = (

478 "Tests SMART capabilities on disks that support " "SMART functions."

479 )

480 parser = ArgumentParser(description=description)

481 parser.add_argument(

482 "-b",

483 "--block-dev",

484 metavar="DISK",

485 default="/dev/sda",

486 help=("the DISK to run this test against " "[default: %(default)s]"),

487 )

488 parser.add_argument(

489 "-d",

490 "--debug",

491 action="store_true",

492 default=False,

493 help="prints some debug info",

494 )

495 parser.add_argument(

496 "-s",

497 "--sleep",

498 type=int,

499 default=5,

500 help=(

501 "number of seconds to sleep between checks "

502 "[default: %(default)s]."

503 ),

504 )

505 parser.add_argument(

506 "-t",

507 "--timeout",

508 type=int,

509 help="number of seconds to timeout from sleeping.",

510 )

511 args = parser.parse_args()

512

513 # Set logging

514 format = "%(levelname)-8s %(message)s"

515 handler = ListHandler()

516 handler.setFormatter(logging.Formatter(format))

517 logger = logging.getLogger()

518 logger.addHandler(handler)

519

520 if args.debug:

521 logger.setLevel(logging.DEBUG)

522 else:

523 logger.setLevel(logging.INFO)

524

525 # Make sure we're root, because smartctl doesn't work otherwise.

526 if not os.geteuid() == 0:

527 parser.error("You must be root to run this program")

528

529 disk = args.block_dev

530 num_disks, raid_type = count_raid_disks(disk)

531 if num_disks == 0:

532 success = enable_smart(disk, -1, raid_type)

533 success = success and run_smart_test(args, disk, -1, raid_type)

534 else:

535 success = True

536 for raid_element in range(0, num_disks):

537 if enable_smart(disk, raid_element, raid_type):

538 success = (

539 run_smart_test(args, disk, raid_element, raid_type)

540 and success

541 )

542 else:

543 success = False

544 if success is False:

545 return 1

546 else:

547 return 0

548

549

550 if __name__ == "__main__":

551 sys.exit(main())