errors/linux_kernel: handle incomplete kernel panics

logspec/errors/linux_kernel.py

@@ -357,7 +357,15 @@ def _parse(self, text):
             report_end = match.start()
             self._report = text[:report_end]
         else:
-            return None
+            # If we couldn't find the end marker, we probably rebooted. So
+            # match sequential lines starting with timestamp.
+            lines = text.split('\n')
+            i = 0
+            while i < len(lines) and re.match(LINUX_TIMESTAMP, lines[i]):
+                i += 1
+            report_end = len('\n'.join(lines[:i]))
+            self._report = text[:report_end]
+
         text = text[:report_end]
 
         match_end = 0

logspec/errors/test.py

@@ -18,3 +18,22 @@ def _parse(self, text):
         caller code working if it calls parse() to generate the error
         signature"""
         pass
+
+
+class KselftestError(Error):
+    """ Parser for kserlftest errors."""
+    def __init__(self):
+        super().__init__()
+        self.error_type = "linux.kselftest"
+
+    def _parse(self, text):
+        """Dummy parse function. The purpose of this is to keep the
+        caller code working if it calls parse() to generate the error
+        signature"""
+        match = re.search(r'(?P<message>not ok \d+ selftests:.*+)', text)
+        if not match:
+            return None
+        self.error_summary = match.group('message')
+        report_end = match.end()
+        self._report = text[:report_end]
+        return report_end

logspec/parser_defs.yaml

@@ -3,6 +3,19 @@ version: 0.0.1
 # Parser definitions
 
 parsers:
+  test_kselftest:
+    states:
+      - name: generic_boot.generic_boot
+        transitions:
+          - function: linux.linux_start_detected
+            state: linux_kernel.kernel_load
+      - name: linux_kernel.kernel_load
+        transitions:
+          - function: linux.linux_prompt_detected
+            state: test_kselftest.test_kselftest
+      - name: test_kselftest.test_kselftest
+    start_state: generic_boot.generic_boot
+
   generic_linux_boot:
     # Parse a Linux boot from kernel startup until it reaches a
     # command-line prompt.

logspec/states/test_kselftest.py

@@ -0,0 +1,80 @@
+# SPDX-License-Identifier: LGPL-2.1-or-later
+#
+# Copyright (C) 2024 Collabora Limited
+# Author: Helen Koike <[email protected]>
+
+import re
+from logspec.parser_classes import State
+from logspec.utils.test_kselftest_errors import find_test_kselftest_error
+from logspec.parser_loader import register_state
+
+MODULE_NAME = 'test_kselftest'
+
+
+# State functions
+
+def detect_test_kselftest(text, start=None, end=None):
+    start_tags = [
+        'kselftest.sh',
+    ]
+    if start or end:
+        text = text[start:end]
+    data = {
+        '_signature_fields': [
+            'test.kselftest.script_call',
+            'test.kselftest.start',
+        ],
+    }
+    regex = '|'.join(start_tags)
+
+    # Check for test start
+    match = re.search(regex, text)
+    if not match:
+        data['test.kselftest.script_call'] = False
+        data['test.kselftest.start'] = False
+        data['_match_end'] = end if end else len(text)
+        data['_summary'] = "Kselftest not detected"
+        return data
+
+    test_start = match.end()
+    test_end = None
+    data['test.kselftest.script_call'] = True
+    data['_summary'] = "Kselftest started"
+
+    # Check for test end, consider the last line starting with "ok \d+ selftests:"
+    # or "not ok \d+ selftests:"
+    regex = r'(?:not )?ok \d+ selftests:'
+    matches = list(re.finditer(regex, text[test_start:]))
+    match = matches[-1] if matches else None
+    if match:
+        data['test.kselftest.start'] = True
+        test_end = test_start + match.end()
+        data['_match_end'] = test_end + start if start else test_end
+    else:
+        data['test.kselftest.start'] = False
+        # TODO: check if this is correct
+        data['_match_end'] = end if end else len(text)
+
+
+    # Check for linux-specific errors in the log. If the `done'
+    # condition was found, search only before it. Otherwise search in
+    # the full log.
+    data['errors'] = []
+    while True:
+        error = find_test_kselftest_error(text[test_start:test_end])
+        if not error:
+            break
+        data['errors'].append(error['error'])
+        test_start += error['_end']
+    return data
+
+
+# Create and register states
+
+register_state(
+    MODULE_NAME,
+    State(
+        name="Kselftest test",
+        description="Search and process a kseftest test",
+        function=detect_test_kselftest),
+    'test_kselftest')

logspec/utils/test_kselftest_errors.py

@@ -0,0 +1,20 @@
+# SPDX-License-Identifier: LGPL-2.1-or-later
+#
+# Copyright (C) 2024 Collabora Limited
+# Author: Ricardo Cañuelo <[email protected]>
+
+from logspec.utils.defs import *
+from logspec.errors.test import *
+
+
+def find_test_kselftest_error(text):
+    error = KselftestError()
+    # Parsing on a generic TestError object simply generates a
+    # signature, we already did the parsing above
+    report_end = error.parse(text)
+    if not report_end:
+        return None
+    return {
+        'error': error,
+        '_end': report_end,
+    }