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