1# 2# SPDX-License-Identifier: MIT 3# 4 5import os 6import sys 7import json 8import errno 9import datetime 10import itertools 11from .commands import runCmd 12 13class BaseDumper(object): 14 """ Base class to dump commands from host/target """ 15 16 def __init__(self, cmds, parent_dir): 17 self.cmds = [] 18 # Some testing doesn't inherit testimage, so it is needed 19 # to set some defaults. 20 self.parent_dir = parent_dir 21 dft_cmds = """ top -bn1 22 iostat -x -z -N -d -p ALL 20 2 23 ps -ef 24 free 25 df 26 memstat 27 dmesg 28 ip -s link 29 netstat -an""" 30 if not cmds: 31 cmds = dft_cmds 32 for cmd in cmds.split('\n'): 33 cmd = cmd.lstrip() 34 if not cmd or cmd[0] == '#': 35 continue 36 self.cmds.append(cmd) 37 38 def create_dir(self, dir_suffix): 39 dump_subdir = ("%s_%s" % ( 40 datetime.datetime.now().strftime('%Y%m%d%H%M'), 41 dir_suffix)) 42 dump_dir = os.path.join(self.parent_dir, dump_subdir) 43 try: 44 os.makedirs(dump_dir) 45 except OSError as err: 46 if err.errno != errno.EEXIST: 47 raise err 48 self.dump_dir = dump_dir 49 50 def _write_dump(self, command, output): 51 if isinstance(self, HostDumper): 52 prefix = "host" 53 elif isinstance(self, TargetDumper): 54 prefix = "target" 55 elif isinstance(self, MonitorDumper): 56 prefix = "qmp" 57 else: 58 prefix = "unknown" 59 for i in itertools.count(): 60 filename = "%s_%02d_%s" % (prefix, i, command) 61 fullname = os.path.join(self.dump_dir, filename) 62 if not os.path.exists(fullname): 63 break 64 if isinstance(self, MonitorDumper): 65 with open(fullname, 'w') as json_file: 66 json.dump(output, json_file, indent=4) 67 else: 68 with open(fullname, 'w') as dump_file: 69 dump_file.write(output) 70 71class HostDumper(BaseDumper): 72 """ Class to get dumps from the host running the tests """ 73 74 def __init__(self, cmds, parent_dir): 75 super(HostDumper, self).__init__(cmds, parent_dir) 76 77 def dump_host(self, dump_dir=""): 78 if dump_dir: 79 self.dump_dir = dump_dir 80 env = os.environ.copy() 81 env['PATH'] = '/usr/sbin:/sbin:/usr/bin:/bin' 82 env['COLUMNS'] = '9999' 83 for cmd in self.cmds: 84 result = runCmd(cmd, ignore_status=True, env=env) 85 self._write_dump(cmd.split()[0], result.output) 86 87class TargetDumper(BaseDumper): 88 """ Class to get dumps from target, it only works with QemuRunner """ 89 90 def __init__(self, cmds, parent_dir, runner): 91 super(TargetDumper, self).__init__(cmds, parent_dir) 92 self.runner = runner 93 94 def dump_target(self, dump_dir=""): 95 if dump_dir: 96 self.dump_dir = dump_dir 97 for cmd in self.cmds: 98 # We can continue with the testing if serial commands fail 99 try: 100 (status, output) = self.runner.run_serial(cmd) 101 self._write_dump(cmd.split()[0], output) 102 except: 103 print("Tried to dump info from target but " 104 "serial console failed") 105 print("Failed CMD: %s" % (cmd)) 106 107class MonitorDumper(BaseDumper): 108 """ Class to get dumps via the Qemu Monitor, it only works with QemuRunner """ 109 110 def __init__(self, cmds, parent_dir, runner): 111 super(MonitorDumper, self).__init__(cmds, parent_dir) 112 self.runner = runner 113 114 def dump_monitor(self, dump_dir=""): 115 if self.runner is None: 116 return 117 if dump_dir: 118 self.dump_dir = dump_dir 119 for cmd in self.cmds: 120 try: 121 output = self.runner.run_monitor(cmd) 122 self._write_dump(cmd, output) 123 except: 124 print("Failed to dump QMP CMD: %s" % (cmd)) 125