util: Add a logroll utility.
This utility receives input from stdin or a file, buffers up to "n" lines of it, and prints those lines either when it reaches the end of the file/stdin, or when it receives a SIGUSR1 which can be sent to it using the "kill" utility. When it receives a SIGUSR1, it keeps running and tracking the input. Change-Id: I9eca4514378cf24a31002d37e10e58cc3ee63b5d Reviewed-on: https://gem5-review.googlesource.com/c/public/gem5/+/43266 Maintainer: Gabe Black <gabe.black@gmail.com> Tested-by: kokoro <noreply+kokoro@google.com> Reviewed-by: Giacomo Travaglini <giacomo.travaglini@arm.com>
This commit is contained in:
Gabe Black
245
util/logroll.py
Executable file
245
util/logroll.py
Executable file
@@ -0,0 +1,245 @@
|
||||
#! /usr/bin/env python3
|
||||
#
|
||||
# Copyright 2021 Google Inc.
|
||||
#
|
||||
# Redistribution and use in source and binary forms, with or without
|
||||
# modification, are permitted provided that the following conditions are
|
||||
# met: redistributions of source code must retain the above copyright
|
||||
# notice, this list of conditions and the following disclaimer;
|
||||
# redistributions in binary form must reproduce the above copyright
|
||||
# notice, this list of conditions and the following disclaimer in the
|
||||
# documentation and/or other materials provided with the distribution;
|
||||
# neither the name of the copyright holders nor the names of its
|
||||
# contributors may be used to endorse or promote products derived from
|
||||
# this software without specific prior written permission.
|
||||
#
|
||||
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
|
||||
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
|
||||
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
|
||||
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
|
||||
# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
|
||||
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
|
||||
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
|
||||
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
|
||||
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
|
||||
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
|
||||
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
|
||||
|
||||
import argparse
|
||||
import copy
|
||||
import signal
|
||||
import sys
|
||||
import unittest
|
||||
import unittest.mock
|
||||
|
||||
parser = argparse.ArgumentParser(
|
||||
description='''Circular buffer for text output.
|
||||
|
||||
To capture the rolling last 25 lines of output from command "command":
|
||||
command | logroll.py -n 25
|
||||
|
||||
While that's running, to see the most recent 25 lines of output without
|
||||
interrupting "command", send SIGUSR1 to the logroll.py process:
|
||||
kill -s USR1 ${PID of logroll.py}''')
|
||||
parser.add_argument('-n', '--lines', default=10, type=int,
|
||||
help='Maximum number of lines to buffer at a time.')
|
||||
parser.add_argument('file', nargs='?', default=sys.stdin,
|
||||
type=argparse.FileType('r', encoding='UTF-8'),
|
||||
help='File to read from, default is stdin')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
def dump_lines(lines, idx):
|
||||
for line in lines[idx:]:
|
||||
print(line, end='')
|
||||
for line in lines[:idx]:
|
||||
print(line, end='')
|
||||
|
||||
def dump_and_exit(lines, idx):
|
||||
dump_lines(lines, idx)
|
||||
sys.exit(0)
|
||||
|
||||
def main(target, incoming):
|
||||
idx = 0
|
||||
lines = []
|
||||
last_idx = target - 1
|
||||
|
||||
signal.signal(signal.SIGUSR1, lambda num, frame: dump_lines(lines, idx))
|
||||
signal.signal(signal.SIGINT, lambda num, frame: dump_and_exit(lines, idx))
|
||||
|
||||
for line in incoming:
|
||||
lines.append(line)
|
||||
if idx == last_idx:
|
||||
idx = 0
|
||||
break
|
||||
else:
|
||||
idx += 1
|
||||
|
||||
for lines[idx] in incoming:
|
||||
idx = 0 if idx == last_idx else idx + 1
|
||||
|
||||
dump_lines(lines, idx)
|
||||
|
||||
if __name__ == '__main__':
|
||||
main(target=args.lines, incoming=args.file)
|
||||
|
||||
|
||||
|
||||
# Unit tests #
|
||||
|
||||
class CopyingMock(unittest.mock.MagicMock):
|
||||
def __call__(self, *args, **kwargs):
|
||||
args = copy.deepcopy(args)
|
||||
kwargs = copy.deepcopy(kwargs)
|
||||
return super(CopyingMock, self).__call__(*args, **kwargs)
|
||||
|
||||
class TestLogroll(unittest.TestCase):
|
||||
# Test data.
|
||||
lines2 = ['First line',
|
||||
'Second line']
|
||||
lines3 = ['First line',
|
||||
'Second line',
|
||||
'Third line']
|
||||
lines8 = ['First line',
|
||||
'Second line',
|
||||
'Third line',
|
||||
'Fourth line',
|
||||
'Fifth line',
|
||||
'Sixth line',
|
||||
'Seventh line',
|
||||
'Eigth line']
|
||||
|
||||
# Generator which returns lines like a file object would.
|
||||
def line_gen(self, lines):
|
||||
for line in lines:
|
||||
yield line
|
||||
|
||||
# Generator like above, but which simulates a signal midway through.
|
||||
def signal_line_gen(self, lines, pos, sig_dict, signal):
|
||||
# Return the first few lines.
|
||||
for line in lines[:pos]:
|
||||
yield line
|
||||
|
||||
# Simulate receiving the signal.
|
||||
self.assertIn(signal, sig_dict)
|
||||
if signal in sig_dict:
|
||||
# Pas in junk for the num and frame arguments.
|
||||
sig_dict[signal](None, None)
|
||||
|
||||
# Return the remaining lines.
|
||||
for line in lines[pos:]:
|
||||
yield line
|
||||
|
||||
# Set up a mock of signal.signal to record handlers in a dict.
|
||||
def mock_signal_dict(self, mock):
|
||||
signal_dict = {}
|
||||
def signal_signal(num, action):
|
||||
signal_dict[num] = action
|
||||
mock.side_effect = signal_signal
|
||||
return signal_dict
|
||||
|
||||
# Actual test methods.
|
||||
def test_filling_dump_lines(self):
|
||||
with unittest.mock.patch('builtins.print') as mock_print:
|
||||
dump_lines(self.lines2, len(self.lines2))
|
||||
calls = list([ unittest.mock.call(line, end='') for
|
||||
line in self.lines2 ])
|
||||
mock_print.assert_has_calls(calls)
|
||||
|
||||
def test_full_dump_lines(self):
|
||||
with unittest.mock.patch('builtins.print') as mock_print:
|
||||
dump_lines(self.lines2, 0)
|
||||
calls = list([ unittest.mock.call(line, end='') for
|
||||
line in self.lines2 ])
|
||||
mock_print.assert_has_calls(calls)
|
||||
|
||||
def test_offset_dump_lines(self):
|
||||
with unittest.mock.patch('builtins.print') as mock_print:
|
||||
dump_lines(self.lines3, 1)
|
||||
calls = [ unittest.mock.call(self.lines3[1], end=''),
|
||||
unittest.mock.call(self.lines3[2], end=''),
|
||||
unittest.mock.call(self.lines3[0], end='') ]
|
||||
mock_print.assert_has_calls(calls)
|
||||
|
||||
def test_dump_and_exit(self):
|
||||
with unittest.mock.patch('sys.exit') as mock_sys_exit, \
|
||||
unittest.mock.patch(__name__ + '.dump_lines',
|
||||
new_callable=CopyingMock) as mock_dump_lines:
|
||||
idx = 1
|
||||
dump_and_exit(self.lines3, idx)
|
||||
mock_dump_lines.assert_called_with(self.lines3, idx)
|
||||
mock_sys_exit.assert_called_with(0)
|
||||
|
||||
def test_filling_main(self):
|
||||
with unittest.mock.patch('builtins.print') as mock_print:
|
||||
main(5, self.line_gen(self.lines3))
|
||||
calls = list([ unittest.mock.call(line, end='') for
|
||||
line in self.lines3 ])
|
||||
mock_print.assert_has_calls(calls)
|
||||
|
||||
def test_full_main(self):
|
||||
with unittest.mock.patch('builtins.print') as mock_print:
|
||||
main(5, self.line_gen(self.lines8))
|
||||
calls = list([ unittest.mock.call(line, end='') for
|
||||
line in self.lines8[-5:] ])
|
||||
mock_print.assert_has_calls(calls)
|
||||
|
||||
def test_sigusr1_filling_main(self):
|
||||
with unittest.mock.patch('signal.signal') as mock_signal, \
|
||||
unittest.mock.patch(__name__ + '.dump_lines',
|
||||
new_callable=CopyingMock) as mock_dump_lines:
|
||||
|
||||
signal_dict = self.mock_signal_dict(mock_signal)
|
||||
|
||||
main(4, self.signal_line_gen(
|
||||
self.lines8, 3, signal_dict, signal.SIGUSR1))
|
||||
|
||||
mock_dump_lines.assert_has_calls([
|
||||
unittest.mock.call(self.lines8[0:3], 3 % 4),
|
||||
unittest.mock.call(self.lines8[-4:], len(self.lines8) % 4)
|
||||
])
|
||||
|
||||
def test_sigint_filling_main(self):
|
||||
with unittest.mock.patch('signal.signal') as mock_signal, \
|
||||
unittest.mock.patch(__name__ + '.dump_lines',
|
||||
new_callable=CopyingMock) as mock_dump_lines:
|
||||
|
||||
signal_dict = self.mock_signal_dict(mock_signal)
|
||||
|
||||
with self.assertRaises(SystemExit):
|
||||
main(4, self.signal_line_gen(
|
||||
self.lines8, 3, signal_dict, signal.SIGINT))
|
||||
|
||||
mock_dump_lines.assert_has_calls([
|
||||
unittest.mock.call(self.lines8[0:3], 3 % 4),
|
||||
])
|
||||
|
||||
def test_sigusr1_full_main(self):
|
||||
with unittest.mock.patch('signal.signal') as mock_signal, \
|
||||
unittest.mock.patch(__name__ + '.dump_lines',
|
||||
new_callable=CopyingMock) as mock_dump_lines:
|
||||
|
||||
signal_dict = self.mock_signal_dict(mock_signal)
|
||||
|
||||
main(4, self.signal_line_gen(
|
||||
self.lines8, 5, signal_dict, signal.SIGUSR1))
|
||||
|
||||
mock_dump_lines.assert_has_calls([
|
||||
unittest.mock.call(self.lines8[4:5] + self.lines8[1:4], 5 % 4),
|
||||
unittest.mock.call(self.lines8[-4:], len(self.lines8) % 4)
|
||||
])
|
||||
|
||||
def test_sigint_full_main(self):
|
||||
with unittest.mock.patch('signal.signal') as mock_signal, \
|
||||
unittest.mock.patch(__name__ + '.dump_lines',
|
||||
new_callable=CopyingMock) as mock_dump_lines:
|
||||
|
||||
signal_dict = self.mock_signal_dict(mock_signal)
|
||||
|
||||
with self.assertRaises(SystemExit):
|
||||
main(4, self.signal_line_gen(
|
||||
self.lines8, 5, signal_dict, signal.SIGINT))
|
||||
|
||||
mock_dump_lines.assert_has_calls([
|
||||
unittest.mock.call(self.lines8[4:5] + self.lines8[1:4], 5 % 4),
|
||||
])
|
||||
Reference in New Issue
Block a user