18798702255

Committed 14 Oct 2025 01:29PM UTC coverage: 100.0%. Remained the same

Build # 18798702255

Build Type

push

github

Committed by

mgedmin

Commit Message

Back to development: 1.5.3

Run Details

133 of 133 branches covered (100.0%)

Branch coverage included in aggregate %.

1 of 1 new or added line in 1 file covered. (100.0%)

1 existing line in 1 file now uncovered.

282 of 282 relevant lines covered (100.0%)

15.94 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

100.0

/strace_process_tree.py

#!/usr/bin/python3
# -*- coding: UTF-8 -*-
"""
Usage:
  strace-process-tree filename

Read strace -f output and produce a process tree.

Recommended strace options for best results:

    strace -f -e trace=process -s 1024 -o filename.out command args

"""

import argparse
import os
import re
import string
import sys
from collections import defaultdict, namedtuple
from contextlib import nullcontext
from functools import partial


__version__ = '1.5.3.dev0'
__author__ = 'Marius Gedminas <marius@gedmin.as>'
__url__ = "https://github.com/mgedmin/strace-process-tree"
__licence__ = 'GPL v2 or v3'  # or ask me for MIT


Tree = namedtuple('Tree', 'trunk, fork, end, space')


class Theme(object):

    default_styles = dict(
        tree_style='normal',
        pid='red',
        process='green',
        time_range='blue',
    )

    ascii_tree = Tree(
        '  | ',
        '  |-',
        '  `-',
        '    ',
    )

    unicode_tree = Tree(
        '  │ ',
        '  ├─',
        '  └─',
        '    ',
    )

    def __new__(cls, color=None, unicode=None):
        if cls is Theme:
            if color is None:
                color = cls.should_use_color()
            if color:
                cls = AnsiTheme
            else:
                cls = PlainTheme
        return object.__new__(cls)

    def __init__(self, color=None, unicode=None):
        if unicode is None:
            unicode = self.can_unicode()
        self.tree = self.unicode_tree if unicode else self.ascii_tree
        self.styles = dict(self.default_styles)

    @classmethod
    def should_use_color(cls):
        return (
            cls.is_terminal()
            and cls.terminal_supports_color()
            and not cls.user_dislikes_color()
        )

    @classmethod
    def is_terminal(cls):
        return hasattr(sys.stdout, 'isatty') and sys.stdout.isatty()

    @classmethod
    def terminal_supports_color(cls):
        return (os.environ.get('TERM') or 'dumb') != 'dumb'

    @classmethod
    def user_dislikes_color(cls):
        # https://no-color.org/
        return bool(os.environ.get('NO_COLOR'))

    @classmethod
    def can_unicode(cls):
        return getattr(sys.stdout, 'encoding', None) == 'UTF-8'

    def _format(self, prefix, suffix, text):
        if not text:
            return ''
        return '{}{}{}'.format(prefix, text, suffix)

    def _no_format(self, text):
        return text or ''

    def __getattr__(self, attr):
        if attr not in self.styles:
            raise AttributeError(attr)
        style = self.styles[attr]
        if style == 'normal':
            _format = self._no_format
        else:
            prefix = self.ctlseq[style]
            suffix = self.ctlseq['normal']
            _format = partial(self._format, prefix, suffix)
        setattr(self, attr, _format)
        return _format


class PlainTheme(Theme):

    def __getattr__(self, attr):
        if attr not in self.styles:
            raise AttributeError(attr)
        _format = self._no_format
        setattr(self, attr, _format)
        return _format


class AnsiTheme(Theme):

    ctlseq = dict(
        normal='\033[m',
        red='\033[31m',
        green='\033[32m',
        blue='\033[34m',
    )


Event = namedtuple('Event', 'pid, timestamp, event')


def parse_timestamp(timestamp):
    if ':' in timestamp:
        h, m, s = timestamp.split(':')
        return (float(h) * 60 + float(m)) * 60 + float(s)
    else:
        return float(timestamp)


RESUMED_PREFIX = re.compile(r'<... \w+ resumed> ?')
UNFINISHED_SUFFIX = ' <unfinished ...>'
DURATION_SUFFIX = re.compile(r' <\d+(?:\.\d+)?>$')
PID = re.compile(r'^\[pid +(\d+)\]')
TIMESTAMP = re.compile(r'^\d+(?::\d+:\d+)?(?:\.\d+)?\s+')
IGNORE = re.compile(r'^$|^strace: Process \d+ attached$')


def events(stream):
    pending = {}
    for n, line in enumerate(stream, 1):
        line = line.strip()
        if line.startswith('[pid'):
            line = PID.sub(r'\1', line)
        pid, space, event = line.partition(' ')
        try:
            pid = int(pid)
        except ValueError:
            if IGNORE.match(line):
                continue
            raise SystemExit(
                "This does not look like a log file produced by strace -f:\n\n"
                "  %s\n\n"
                "There should've been a PID at the beginning of line %d."
                % (line, n))
        event = event.lstrip()
        timestamp = None
        if event[:1].isdigit():
            m = TIMESTAMP.match(event)
            if m is not None:
                timestamp = parse_timestamp(m.group())
                event = event[m.end():]
        if event.endswith('>'):
            e, sp, d = event.rpartition(' <')
            if DURATION_SUFFIX.match(sp + d):
                event = e
        if event.startswith('<...'):
            m = RESUMED_PREFIX.match(event)
            if m is not None:
                pending_event, timestamp = pending.pop(pid)
                event = pending_event + event[m.end():]
        if event.endswith(UNFINISHED_SUFFIX):
            pending[pid] = (event[:-len(UNFINISHED_SUFFIX)], timestamp)
        else:
            yield Event(pid, timestamp, event)


Process = namedtuple('Process', 'pid, seq, name, parent')


class ProcessTree(object):
    def __init__(self):
        self.processes = {}   # map pid to Process
        self.start_time = {}  # map Process to seconds
        self.exit_time = {}   # map Process to seconds
        self.children = defaultdict(set)
        # Invariant: every Process appears exactly once in
        # self.children[some_parent].

    def add_child(self, ppid, pid, name, timestamp):
        parent = self.processes.get(ppid)
        if parent is None:
            # This can happen when we attach to a running process and so miss
            # the initial execve() call that would have given it a name.
            parent = Process(pid=ppid, seq=0, name=None, parent=None)
            self.children[None].add(parent)
        # NB: it's possible that strace saw code executing in the child process
        # before the parent's clone() returned a value, so we might already
        # have a self.processes[pid].
        old_process = self.processes.get(pid)
        if old_process is not None:
            self.children[old_process.parent].remove(old_process)
            child = old_process._replace(parent=parent)
        else:
            # We pass seq=0 here and seq=1 in handle_exec() because
            # conceptually clone() happens before execve(), but we must be
            # ready to handle these two events in either order.
            child = Process(pid=pid, seq=0, name=name, parent=parent)
        self.processes[pid] = child
        self.children[parent].add(child)
        # The timestamp of clone() is always going to be earlier than the
        # timestamp of execve() so we use unconditional assignment here but a
        # setdefault() in handle_exec().
        self.start_time[child] = timestamp

    def handle_exec(self, pid, name, timestamp):
        old_process = self.processes.get(pid)
        if old_process:
            new_process = old_process._replace(seq=old_process.seq + 1,
                                               name=name)
            if old_process.seq == 0 and not self.children[old_process]:
                # Drop the child process if it did nothing interesting between
                # fork() and exec().
                self.children[old_process.parent].remove(old_process)
        else:
            new_process = Process(pid=pid, seq=1, name=name, parent=None)
        self.processes[pid] = new_process
        self.children[new_process.parent].add(new_process)
        self.start_time.setdefault(new_process, timestamp)

    def handle_exit(self, pid, timestamp):
        process = self.processes.get(pid)
        if process:
            # process may be None when we attach to a running process and
            # see it exit before it does any clone()/execve() calls
            self.exit_time[process] = timestamp

    def _format_time_range(self, start_time, exit_time):
        if start_time is not None and exit_time is not None:
            return '[{duration:.1f}s @{start_time:.1f}s]'.format(
                start_time=start_time,
                duration=exit_time - start_time
            )
        elif start_time:  # skip both None and 0 please
            return '[@{start_time:.1f}s]'.format(
                start_time=start_time,
            )
        else:
            return ''

    def _format_process_name(self, theme, name, indent, cs, ccs, padding):
        lines = (name or '').split('\n')
        return '\n{indent}{tree}{padding}'.format(
            indent=indent,
            tree=theme.tree_style(cs + ccs),
            padding=padding,
        ).join(
            theme.process(line)
            for line in lines
        )

    def _format(self, theme, processes, indent='', level=0):
        r = []
        for n, process in enumerate(processes):
            if level == 0:
                s, cs = '', ''
            elif n < len(processes) - 1:
                s, cs = theme.tree.fork, theme.tree.trunk
            else:
                s, cs = theme.tree.end, theme.tree.space
            children = sorted(self.children[process])
            if children:
                ccs = theme.tree.trunk
            else:
                ccs = theme.tree.space
            time_range = self._format_time_range(
                self.start_time.get(process),
                self.exit_time.get(process),
            )
            title = '{pid} {name} {time_range}'.format(
                pid=theme.pid(process.pid or '<unknown>'),
                name=self._format_process_name(
                    theme, process.name, indent, cs, ccs, theme.tree.space),
                time_range=theme.time_range(time_range),
            ).rstrip()
            r.append(indent + (theme.tree_style(s) + title).rstrip() + '\n')
            r.append(self._format(theme, children, indent+cs, level+1))

        return ''.join(r)

    def format(self, theme):
        return self._format(theme, sorted(self.children[None]))

    def __str__(self):
        return self.format(PlainTheme(unicode=True))


def simplify_syscall(event):
    # clone(child_stack=0x..., flags=FLAGS, parent_tidptr=..., tls=...,
    #       child_tidptr=...) => clone(FLAGS)
    if event.startswith(('clone(', 'clone3(')):
        event = re.sub('[(].*(?:, |{)flags=([^,]*), .*[)]', r'(\1)', event)
    return event.rstrip()


def extract_command_line(event):
    # execve("/usr/bin/foo", ["foo", "bar"], [/* 45 vars */]) => foo bar
    # execve("/usr/bin/foo", ["foo", "bar"], [/* 1 var */]) => foo bar
    if event.startswith(('clone(', 'clone3(')):
        if 'CLONE_THREAD' in event:
            return '(thread)'
        elif 'flags=CLONE_CHILD_CLEARTID|CLONE_CHILD_SETTID|SIGCHLD' in event:
            return '(fork)'
        else:
            return '...'
    elif event.startswith('execve('):
        command = event.strip()
        command = re.sub(r'^execve\([^[]*\[', '', command)
        command = re.sub(r'\], (0x[0-9a-f]+ )?\[?/\* \d+ vars? \*/\]?\)$', '',
                         command)
        command = parse_argv(command)
        return format_command(command)
    else:
        return event.rstrip()


ESCAPES = {
    'n': '\n',
    'r': '\r',
    't': '\t',
    'b': '\b',
    '0': '\0',
    'a': '\a',
}


def parse_argv(s):
    # '"foo", "bar"..., "baz", "\""' => ['foo', 'bar...', 'baz', '"']
    it = iter(s + ",")
    args = []
    for c in it:
        if c == ' ':
            continue
        if c == '.':
            c = next(it)
            assert c == ".", c
            c = next(it)
            assert c == ".", c
            c = next(it)
            args.append(Ellipsis)
            assert c == ',', (c, s)
            continue
        assert c == '"', c
        arg = []
        for c in it:  # pragma: no branch -- loop will execute at least once
            if c == '"':
                break
            if c == '\\':
                c = next(it)
                arg.append(ESCAPES.get(c, c))
            else:
                arg.append(c)
        c = next(it)
        if c == ".":
            arg.append('...')
            c = next(it)
            assert c == ".", c
            c = next(it)
            assert c == ".", c
            c = next(it)
        args.append(''.join(arg))
        assert c == ',', (c, s)
    return args


SHELL_SAFE_CHARS = set(string.ascii_letters + string.digits + '%+,-./:=@^_~')
SHELL_SAFE_QUOTED = SHELL_SAFE_CHARS | set("!#&'()*;<>?[]{|} \t\n")


def format_command(command):
    return ' '.join(map(pushquote, (
        '...' if arg is Ellipsis else
        arg if all(c in SHELL_SAFE_CHARS for c in arg) else
        '"%s"' % arg if all(c in SHELL_SAFE_QUOTED for c in arg) else
        "'%s'" % arg.replace("'", "'\\''")
        for arg in command
    )))


def pushquote(arg):
    # Change "--foo=bar" to --foo="bar" because that looks better to human eyes
    return re.sub('''^(['"])(--[a-zA-Z0-9_-]+)=''', r'\2=\1', arg)


def parse_stream(event_stream, mogrifier=extract_command_line):
    tree = ProcessTree()
    first_timestamp = None
    for e in event_stream:
        timestamp = e.timestamp
        if timestamp is not None:
            if first_timestamp is None:
                first_timestamp = e.timestamp
            timestamp -= first_timestamp
        if e.event.startswith('execve('):
            args, equal, result = e.event.rpartition(' = ')
            if result == '0':
                name = mogrifier(args)
                tree.handle_exec(e.pid, name, timestamp)
        if e.event.startswith(('clone(', 'clone3(', 'fork(', 'vfork(')):
            args, equal, result = e.event.rpartition(' = ')
            # if clone() fails, the event will look like this:
            #   clone(...) = -1 EPERM (Operation not permitted)
            # and it will fail the result.isdigit() check
            if result.isdigit():
                child_pid = int(result)
                name = mogrifier(args)
                tree.add_child(e.pid, child_pid, name, timestamp)
        if e.event.startswith('+++ exited with '):
            tree.handle_exit(e.pid, timestamp)
    return tree


def open_arg(arg: str):
    if arg == '-':
        return nullcontext(sys.stdin)
    else:
        return open(arg)


def main():
    parser = argparse.ArgumentParser(
        description="""
            Read strace -f output and produce a process tree.

            Recommended strace options for best results:

                strace -f -ttt -e trace=process -s 1024 -o FILENAME COMMAND
            """)
    parser.add_argument('--version', action='version', version=__version__)
    parser.add_argument('-c', '--color', action='store_true', default=None,
                        help='force color output')
    parser.add_argument('-C', '--no-color', action='store_false', dest='color',
                        help='disable color output')
    parser.add_argument('-U', '--unicode', action='store_true', default=None,
                        help='force Unicode output')
    parser.add_argument('-A', '--ascii', action='store_false', dest='unicode',
                        help='force ASCII output')
    parser.add_argument('-v', '--verbose', action='store_true',
                        help='more verbose output')
    parser.add_argument('filename',
                        help='strace log to parse (use - to read stdin)')
    args = parser.parse_args()

    mogrifier = simplify_syscall if args.verbose else extract_command_line

    with open_arg(args.filename) as fp:
        tree = parse_stream(events(fp), mogrifier)

    theme = Theme(color=args.color, unicode=args.unicode)
    print(tree.format(theme).rstrip())


if __name__ == '__main__':
    main()

1	#!/usr/bin/python3
2	# -- coding: UTF-8 --
3	"""	4✔
4	Usage:
5	strace-process-tree filename
6
7	Read strace -f output and produce a process tree.
8
9	Recommended strace options for best results:
10
11	strace -f -e trace=process -s 1024 -o filename.out command args
12
13	"""
14
15	import argparse	16✔
16	import os	16✔
17	import re	16✔
18	import string	16✔
19	import sys	16✔
20	from collections import defaultdict, namedtuple	16✔
21	from contextlib import nullcontext	16✔
22	from functools import partial	16✔
23
24
25	__version__ = '1.5.3.dev0'	16✔
26	__author__ = 'Marius Gedminas <marius@gedmin.as>'	16✔
27	__url__ = "https://github.com/mgedmin/strace-process-tree"	16✔
28	__licence__ = 'GPL v2 or v3' # or ask me for MIT	16✔
29
30
31	Tree = namedtuple('Tree', 'trunk, fork, end, space')	16✔
32
33
34	class Theme(object):	16✔
35
36	default_styles = dict(	16✔
37	tree_style='normal',
38	pid='red',
39	process='green',
40	time_range='blue',
41	)
42
43	ascii_tree = Tree(	16✔
44	' \| ',
45	' \|-',
46	' `-',
47	' ',
48	)
49
50	unicode_tree = Tree(	16✔
51	' │ ',
52	' ├─',
53	' └─',
54	' ',
55	)
56
57	def __new__(cls, color=None, unicode=None):	16✔
58	if cls is Theme:	16✔
59	if color is None:	16✔
60	color = cls.should_use_color()	16✔
61	if color:	16✔
62	cls = AnsiTheme	16✔
63	else:
64	cls = PlainTheme	16✔
65	return object.__new__(cls)	16✔
66
67	def __init__(self, color=None, unicode=None):	16✔
68	if unicode is None:	16✔
69	unicode = self.can_unicode()	16✔
70	self.tree = self.unicode_tree if unicode else self.ascii_tree	16✔
71	self.styles = dict(self.default_styles)	16✔
72
73	@classmethod	16✔
74	def should_use_color(cls):	16✔
75	return (	16✔
76	cls.is_terminal()
77	and cls.terminal_supports_color()
78	and not cls.user_dislikes_color()
79	)
80
81	@classmethod	16✔
82	def is_terminal(cls):	16✔
83	return hasattr(sys.stdout, 'isatty') and sys.stdout.isatty()	16✔
84
85	@classmethod	16✔
86	def terminal_supports_color(cls):	16✔
87	return (os.environ.get('TERM') or 'dumb') != 'dumb'	16✔
88
89	@classmethod	16✔
90	def user_dislikes_color(cls):	16✔
91	# https://no-color.org/
92	return bool(os.environ.get('NO_COLOR'))	16✔
93
94	@classmethod	16✔
95	def can_unicode(cls):	16✔
96	return getattr(sys.stdout, 'encoding', None) == 'UTF-8'	16✔
97
98	def _format(self, prefix, suffix, text):	16✔
99	if not text:	16✔
100	return ''	16✔
101	return '{}{}{}'.format(prefix, text, suffix)	16✔
102
103	def _no_format(self, text):	16✔
104	return text or ''	16✔
105
106	def __getattr__(self, attr):	16✔
107	if attr not in self.styles:	16✔
108	raise AttributeError(attr)	16✔
109	style = self.styles[attr]	16✔
110	if style == 'normal':	16✔
111	_format = self._no_format	16✔
112	else:
113	prefix = self.ctlseq[style]	16✔
114	suffix = self.ctlseq['normal']	16✔
115	_format = partial(self._format, prefix, suffix)	16✔
116	setattr(self, attr, _format)	16✔
117	return _format	16✔
118
119
120	class PlainTheme(Theme):	16✔
121
122	def __getattr__(self, attr):	16✔
123	if attr not in self.styles:	16✔
124	raise AttributeError(attr)	16✔
125	_format = self._no_format	16✔
126	setattr(self, attr, _format)	16✔
127	return _format	16✔
128
129
130	class AnsiTheme(Theme):	16✔
131
132	ctlseq = dict(	16✔
133	normal='\033[m',
134	red='\033[31m',
135	green='\033[32m',
136	blue='\033[34m',
137	)
138
139
140	Event = namedtuple('Event', 'pid, timestamp, event')	16✔
141
142
143	def parse_timestamp(timestamp):	16✔
144	if ':' in timestamp:	16✔
145	h, m, s = timestamp.split(':')	16✔
146	return (float(h) * 60 + float(m)) * 60 + float(s)	16✔
147	else:
148	return float(timestamp)	16✔
149
150
151	RESUMED_PREFIX = re.compile(r'<... \w+ resumed> ?')	16✔
152	UNFINISHED_SUFFIX = ' <unfinished ...>'	16✔
153	DURATION_SUFFIX = re.compile(r' <\d+(?:\.\d+)?>$')	16✔
154	PID = re.compile(r'^\[pid +(\d+)\]')	16✔
155	TIMESTAMP = re.compile(r'^\d+(?::\d+:\d+)?(?:\.\d+)?\s+')	16✔
156	IGNORE = re.compile(r'^$\|^strace: Process \d+ attached$')	16✔
157
158
159	def events(stream):	16✔
160	pending = {}	16✔
161	for n, line in enumerate(stream, 1):	16✔
162	line = line.strip()	16✔
163	if line.startswith('[pid'):	16✔
164	line = PID.sub(r'\1', line)	16✔
165	pid, space, event = line.partition(' ')	16✔
166	try:	16✔
167	pid = int(pid)	16✔
168	except ValueError:	16✔
169	if IGNORE.match(line):	16✔
170	continue	16✔
171	raise SystemExit(	16✔
172	"This does not look like a log file produced by strace -f:\n\n"
173	" %s\n\n"
174	"There should've been a PID at the beginning of line %d."
175	% (line, n))
176	event = event.lstrip()	16✔
177	timestamp = None	16✔
178	if event[:1].isdigit():	16✔
179	m = TIMESTAMP.match(event)	16✔
180	if m is not None:	16✔
181	timestamp = parse_timestamp(m.group())	16✔
182	event = event[m.end():]	16✔
183	if event.endswith('>'):	16✔
184	e, sp, d = event.rpartition(' <')	16✔
185	if DURATION_SUFFIX.match(sp + d):	16✔
186	event = e	16✔
187	if event.startswith('<...'):	16✔
188	m = RESUMED_PREFIX.match(event)	16✔
189	if m is not None:	16✔
190	pending_event, timestamp = pending.pop(pid)	16✔
191	event = pending_event + event[m.end():]	16✔
192	if event.endswith(UNFINISHED_SUFFIX):	16✔
193	pending[pid] = (event[:-len(UNFINISHED_SUFFIX)], timestamp)	16✔
194	else:
195	yield Event(pid, timestamp, event)	16✔
196
197
198	Process = namedtuple('Process', 'pid, seq, name, parent')	16✔
199
200
201	class ProcessTree(object):	16✔
202	def __init__(self):	16✔
203	self.processes = {} # map pid to Process	16✔
204	self.start_time = {} # map Process to seconds	16✔
205	self.exit_time = {} # map Process to seconds	16✔
206	self.children = defaultdict(set)	16✔
207	# Invariant: every Process appears exactly once in
208	# self.children[some_parent].
209
210	def add_child(self, ppid, pid, name, timestamp):	16✔
211	parent = self.processes.get(ppid)	16✔
212	if parent is None:	16✔
213	# This can happen when we attach to a running process and so miss
214	# the initial execve() call that would have given it a name.
215	parent = Process(pid=ppid, seq=0, name=None, parent=None)	16✔
216	self.children[None].add(parent)	16✔
217	# NB: it's possible that strace saw code executing in the child process
218	# before the parent's clone() returned a value, so we might already
219	# have a self.processes[pid].
220	old_process = self.processes.get(pid)	16✔
221	if old_process is not None:	16✔
222	self.children[old_process.parent].remove(old_process)	16✔
223	child = old_process._replace(parent=parent)	16✔
224	else:
225	# We pass seq=0 here and seq=1 in handle_exec() because
226	# conceptually clone() happens before execve(), but we must be
227	# ready to handle these two events in either order.
228	child = Process(pid=pid, seq=0, name=name, parent=parent)	16✔
229	self.processes[pid] = child	16✔
230	self.children[parent].add(child)	16✔
231	# The timestamp of clone() is always going to be earlier than the
232	# timestamp of execve() so we use unconditional assignment here but a
233	# setdefault() in handle_exec().
234	self.start_time[child] = timestamp	16✔
235
236	def handle_exec(self, pid, name, timestamp):	16✔
237	old_process = self.processes.get(pid)	16✔
238	if old_process:	16✔
239	new_process = old_process._replace(seq=old_process.seq + 1,	16✔
240	name=name)
241	if old_process.seq == 0 and not self.children[old_process]:	16✔
242	# Drop the child process if it did nothing interesting between
243	# fork() and exec().
244	self.children[old_process.parent].remove(old_process)	16✔
245	else:
246	new_process = Process(pid=pid, seq=1, name=name, parent=None)	16✔
247	self.processes[pid] = new_process	16✔
248	self.children[new_process.parent].add(new_process)	16✔
249	self.start_time.setdefault(new_process, timestamp)	16✔
250
251	def handle_exit(self, pid, timestamp):	16✔
252	process = self.processes.get(pid)	16✔
253	if process:	16✔
254	# process may be None when we attach to a running process and
255	# see it exit before it does any clone()/execve() calls
256	self.exit_time[process] = timestamp	16✔
257
258	def _format_time_range(self, start_time, exit_time):	16✔
259	if start_time is not None and exit_time is not None:	16✔
260	return '[{duration:.1f}s @{start_time:.1f}s]'.format(	16✔
261	start_time=start_time,
262	duration=exit_time - start_time
263	)
264	elif start_time: # skip both None and 0 please	16✔
265	return '[@{start_time:.1f}s]'.format(	16✔
266	start_time=start_time,
267	)
268	else:
269	return ''	16✔
270
271	def _format_process_name(self, theme, name, indent, cs, ccs, padding):	16✔
272	lines = (name or '').split('\n')	16✔
273	return '\n{indent}{tree}{padding}'.format(	16✔
274	indent=indent,
275	tree=theme.tree_style(cs + ccs),
276	padding=padding,
277	).join(
278	theme.process(line)
279	for line in lines
280	)
281
282	def _format(self, theme, processes, indent='', level=0):	16✔
283	r = []	16✔
284	for n, process in enumerate(processes):	16✔
285	if level == 0:	16✔
286	s, cs = '', ''	16✔
287	elif n < len(processes) - 1:	16✔
288	s, cs = theme.tree.fork, theme.tree.trunk	16✔
289	else:
290	s, cs = theme.tree.end, theme.tree.space	16✔
291	children = sorted(self.children[process])	16✔
292	if children:	16✔
293	ccs = theme.tree.trunk	16✔
294	else:
295	ccs = theme.tree.space	16✔
296	time_range = self._format_time_range(	16✔
297	self.start_time.get(process),
298	self.exit_time.get(process),
299	)
300	title = '{pid} {name} {time_range}'.format(	16✔
301	pid=theme.pid(process.pid or '<unknown>'),
302	name=self._format_process_name(
303	theme, process.name, indent, cs, ccs, theme.tree.space),
304	time_range=theme.time_range(time_range),
305	).rstrip()
306	r.append(indent + (theme.tree_style(s) + title).rstrip() + '\n')	16✔
307	r.append(self._format(theme, children, indent+cs, level+1))	16✔
308
309	return ''.join(r)	16✔
310
311	def format(self, theme):	16✔
312	return self._format(theme, sorted(self.children[None]))	16✔
313
314	def __str__(self):	16✔
315	return self.format(PlainTheme(unicode=True))	16✔
316
317
318	def simplify_syscall(event):	16✔
319	# clone(child_stack=0x..., flags=FLAGS, parent_tidptr=..., tls=...,
320	# child_tidptr=...) => clone(FLAGS)
321	if event.startswith(('clone(', 'clone3(')):	16✔
322	event = re.sub('[(].(?:, \|{)flags=([^,]), .*[)]', r'(\1)', event)	16✔
323	return event.rstrip()	16✔
324
325
326	def extract_command_line(event):	16✔
327	# execve("/usr/bin/foo", ["foo", "bar"], [/* 45 vars */]) => foo bar
328	# execve("/usr/bin/foo", ["foo", "bar"], [/* 1 var */]) => foo bar
329	if event.startswith(('clone(', 'clone3(')):	16✔
330	if 'CLONE_THREAD' in event:	16✔
331	return '(thread)'	16✔
332	elif 'flags=CLONE_CHILD_CLEARTID\|CLONE_CHILD_SETTID\|SIGCHLD' in event:	16✔
333	return '(fork)'	16✔
334	else:
335	return '...'	16✔
336	elif event.startswith('execve('):	16✔
337	command = event.strip()	16✔
338	command = re.sub(r'^execve\([^[]*\[', '', command)	16✔
339	command = re.sub(r'\], (0x[0-9a-f]+ )?\[?/\* \d+ vars? \*/\]?\)$', '',	16✔
340	command)
341	command = parse_argv(command)	16✔
342	return format_command(command)	16✔
343	else:
344	return event.rstrip()	16✔
345
346
347	ESCAPES = {	16✔
348	'n': '\n',
349	'r': '\r',
350	't': '\t',
351	'b': '\b',
352	'0': '\0',
353	'a': '\a',
354	}
355
356
357	def parse_argv(s):	16✔
358	# '"foo", "bar"..., "baz", "\""' => ['foo', 'bar...', 'baz', '"']
359	it = iter(s + ",")	16✔
360	args = []	16✔
361	for c in it:	16✔
362	if c == ' ':	16✔
363	continue	16✔
364	if c == '.':	16✔
365	c = next(it)	16✔
366	assert c == ".", c	16✔
367	c = next(it)	16✔
368	assert c == ".", c	16✔
369	c = next(it)	16✔
370	args.append(Ellipsis)	16✔
371	assert c == ',', (c, s)	16✔
UNCOV 372	continue	12✔
373	assert c == '"', c	16✔
374	arg = []	16✔
375	for c in it: # pragma: no branch -- loop will execute at least once	16✔
376	if c == '"':	16✔
377	break	16✔
378	if c == '\\':	16✔
379	c = next(it)	16✔
380	arg.append(ESCAPES.get(c, c))	16✔
381	else:
382	arg.append(c)	16✔
383	c = next(it)	16✔
384	if c == ".":	16✔
385	arg.append('...')	16✔
386	c = next(it)	16✔
387	assert c == ".", c	16✔
388	c = next(it)	16✔
389	assert c == ".", c	16✔
390	c = next(it)	16✔
391	args.append(''.join(arg))	16✔
392	assert c == ',', (c, s)	16✔
393	return args	16✔
394
395
396	SHELL_SAFE_CHARS = set(string.ascii_letters + string.digits + '%+,-./:=@^_~')	16✔
397	SHELL_SAFE_QUOTED = SHELL_SAFE_CHARS \| set("!#&'()*;<>?[]{\|} \t\n")	16✔
398
399
400	def format_command(command):	16✔
401	return ' '.join(map(pushquote, (	16✔
402	'...' if arg is Ellipsis else
403	arg if all(c in SHELL_SAFE_CHARS for c in arg) else
404	'"%s"' % arg if all(c in SHELL_SAFE_QUOTED for c in arg) else
405	"'%s'" % arg.replace("'", "'\\''")
406	for arg in command
407	)))
408
409
410	def pushquote(arg):	16✔
411	# Change "--foo=bar" to --foo="bar" because that looks better to human eyes
412	return re.sub('''^(['"])(--[a-zA-Z0-9_-]+)=''', r'\2=\1', arg)	16✔
413
414
415	def parse_stream(event_stream, mogrifier=extract_command_line):	16✔
416	tree = ProcessTree()	16✔
417	first_timestamp = None	16✔
418	for e in event_stream:	16✔
419	timestamp = e.timestamp	16✔
420	if timestamp is not None:	16✔
421	if first_timestamp is None:	16✔
422	first_timestamp = e.timestamp	16✔
423	timestamp -= first_timestamp	16✔
424	if e.event.startswith('execve('):	16✔
425	args, equal, result = e.event.rpartition(' = ')	16✔
426	if result == '0':	16✔
427	name = mogrifier(args)	16✔
428	tree.handle_exec(e.pid, name, timestamp)	16✔
429	if e.event.startswith(('clone(', 'clone3(', 'fork(', 'vfork(')):	16✔
430	args, equal, result = e.event.rpartition(' = ')	16✔
431	# if clone() fails, the event will look like this:
432	# clone(...) = -1 EPERM (Operation not permitted)
433	# and it will fail the result.isdigit() check
434	if result.isdigit():	16✔
435	child_pid = int(result)	16✔
436	name = mogrifier(args)	16✔
437	tree.add_child(e.pid, child_pid, name, timestamp)	16✔
438	if e.event.startswith('+++ exited with '):	16✔
439	tree.handle_exit(e.pid, timestamp)	16✔
440	return tree	16✔
441
442
443	def open_arg(arg: str):	16✔
444	if arg == '-':	16✔
445	return nullcontext(sys.stdin)	16✔
446	else:
447	return open(arg)	16✔
448
449
450	def main():	16✔
451	parser = argparse.ArgumentParser(	16✔
452	description="""
453	Read strace -f output and produce a process tree.
454
455	Recommended strace options for best results:
456
457	strace -f -ttt -e trace=process -s 1024 -o FILENAME COMMAND
458	""")
459	parser.add_argument('--version', action='version', version=__version__)	16✔
460	parser.add_argument('-c', '--color', action='store_true', default=None,	16✔
461	help='force color output')
462	parser.add_argument('-C', '--no-color', action='store_false', dest='color',	16✔
463	help='disable color output')
464	parser.add_argument('-U', '--unicode', action='store_true', default=None,	16✔
465	help='force Unicode output')
466	parser.add_argument('-A', '--ascii', action='store_false', dest='unicode',	16✔
467	help='force ASCII output')
468	parser.add_argument('-v', '--verbose', action='store_true',	16✔
469	help='more verbose output')
470	parser.add_argument('filename',	16✔
471	help='strace log to parse (use - to read stdin)')
472	args = parser.parse_args()	16✔
473
474	mogrifier = simplify_syscall if args.verbose else extract_command_line	16✔
475
476	with open_arg(args.filename) as fp:	16✔
477	tree = parse_stream(events(fp), mogrifier)	16✔
478
479	theme = Theme(color=args.color, unicode=args.unicode)	16✔
480	print(tree.format(theme).rstrip())	16✔
481
482
483	if __name__ == '__main__':
484	main()

mgedmin / strace-process-tree / 18798702255

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous