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
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
|
#
# Copyright OpenEmbedded Contributors
#
# SPDX-License-Identifier: MIT
#
import collections
import os
import sys
from shutil import rmtree
from oeqa.runtime.case import OERuntimeTestCase
from oeqa.core.decorator.depends import OETestDepends
# importlib.resources.open_text in Python <3.10 doesn't search all directories
# when a package is split across multiple directories. Until we can rely on
# 3.10+, reimplement the searching logic.
if sys.version_info < (3, 10):
def _open_text(package, resource):
import importlib, pathlib
module = importlib.import_module(package)
for path in module.__path__:
candidate = pathlib.Path(path) / resource
if candidate.exists():
return candidate.open(encoding='utf-8')
raise FileNotFoundError
else:
from importlib.resources import open_text as _open_text
class ParseLogsTest(OERuntimeTestCase):
# Which log files should be collected
log_locations = ["/var/log/", "/var/log/dmesg", "/tmp/dmesg_output.log"]
# The keywords that identify error messages in the log files
errors = ["error", "cannot", "can't", "failed"]
# A list of error messages that should be ignored
ignore_errors = []
@classmethod
def setUpClass(cls):
# When systemd is enabled we need to notice errors on
# circular dependencies in units.
if 'systemd' in cls.td.get('DISTRO_FEATURES'):
cls.errors.extend([
'Found ordering cycle on',
'Breaking ordering cycle by deleting job',
'deleted to break ordering cycle',
'Ordering cycle found, skipping',
])
cls.errors = [s.casefold() for s in cls.errors]
cls.load_machine_ignores()
@classmethod
def load_machine_ignores(cls):
# Add TARGET_ARCH explicitly as not every machine has that in MACHINEOVERRDES (eg qemux86-64)
for candidate in ["common", cls.td.get("TARGET_ARCH")] + cls.td.get("MACHINEOVERRIDES").split(":"):
try:
name = f"parselogs-ignores-{candidate}.txt"
for line in _open_text("oeqa.runtime.cases", name):
line = line.strip()
if line and not line.startswith("#"):
cls.ignore_errors.append(line.casefold())
except FileNotFoundError:
pass
# Go through the log locations provided and if it's a folder
# create a list with all the .log files in it, if it's a file
# just add it to that list.
def getLogList(self, log_locations):
logs = []
for location in log_locations:
status, _ = self.target.run('test -f %s' % location)
if status == 0:
logs.append(location)
else:
status, _ = self.target.run('test -d %s' % location)
if status == 0:
cmd = 'find %s -name \\*.log -maxdepth 1 -type f' % location
status, output = self.target.run(cmd)
if status == 0:
output = output.splitlines()
for logfile in output:
logs.append(os.path.join(location, logfile))
return logs
# Copy the log files to be parsed locally
def transfer_logs(self, log_list):
workdir = self.td.get('WORKDIR')
self.target_logs = workdir + '/' + 'target_logs'
target_logs = self.target_logs
if os.path.exists(target_logs):
rmtree(self.target_logs)
os.makedirs(target_logs)
for f in log_list:
self.target.copyFrom(str(f), target_logs)
# Get the local list of logs
def get_local_log_list(self, log_locations):
self.transfer_logs(self.getLogList(log_locations))
list_dir = os.listdir(self.target_logs)
dir_files = [os.path.join(self.target_logs, f) for f in list_dir]
logs = [f for f in dir_files if os.path.isfile(f)]
return logs
def get_context(self, lines, index, before=6, after=3):
"""
Given a set of lines and the index of the line that is important, return
a number of lines surrounding that line.
"""
last = len(lines)
start = index - before
end = index + after + 1
if start < 0:
end -= start
start = 0
if end > last:
start -= end - last
end = last
return lines[start:end]
def test_get_context(self):
"""
A test case for the test case.
"""
lines = list(range(0,10))
self.assertEqual(self.get_context(lines, 0, 2, 1), [0, 1, 2, 3])
self.assertEqual(self.get_context(lines, 5, 2, 1), [3, 4, 5, 6])
self.assertEqual(self.get_context(lines, 9, 2, 1), [6, 7, 8, 9])
def parse_logs(self, logs, lines_before=10, lines_after=10):
"""
Search the log files @logs looking for error lines (marked by
@self.errors), ignoring anything listed in @self.ignore_errors.
Returns a dictionary of log filenames to a dictionary of error lines to
the error context (controlled by @lines_before and @lines_after).
"""
results = collections.defaultdict(dict)
for log in logs:
with open(log) as f:
lines = f.readlines()
for i, line in enumerate(lines):
line = line.strip()
line_lower = line.casefold()
if any(keyword in line_lower for keyword in self.errors):
if not any(ignore in line_lower for ignore in self.ignore_errors):
results[log][line] = "".join(self.get_context(lines, i, lines_before, lines_after))
return results
# Get the output of dmesg and write it in a file.
# This file is added to log_locations.
def write_dmesg(self):
(status, dmesg) = self.target.run('dmesg > /tmp/dmesg_output.log')
@OETestDepends(['ssh.SSHTest.test_ssh'])
def test_parselogs(self):
self.write_dmesg()
log_list = self.get_local_log_list(self.log_locations)
result = self.parse_logs(log_list)
errcount = 0
self.msg = ""
for log in result:
self.msg += 'Log: ' + log + '\n'
self.msg += '-----------------------\n'
for error in result[log]:
errcount += 1
self.msg += 'Central error: ' + error + '\n'
self.msg += '***********************\n'
self.msg += result[log][error] + '\n'
self.msg += '***********************\n'
self.msg += '%s errors found in logs.' % errcount
self.assertEqual(errcount, 0, msg=self.msg)
|
