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