KVM: add dmesg check for both host and guest

KVM/README.md

@@ -44,6 +44,24 @@ lkvs KVM is a seperate test provider for avocado/avocado-vt.
     avocado run boot_td
 ```
 
+## LKVS dmesg verification switches
+
+LKVS qemu tests support two cfg parameters to route dmesg verification to
+LKVS implementation (`KVM/qemu/provider/utils_misc.py`):
+
+```
+lkvs_verify_host_dmesg = yes
+lkvs_verify_guest_dmesg = yes
+```
+
+These switches can be set in any LKVS qemu case cfg file.
+
+LKVS qemu uses explicit import mode:
+
+- Shared module: `KVM/qemu/provider/dmesg_router.py`
+- Requirement: each test module under `KVM/qemu/tests/*.py` needs
+    `from provider import dmesg_router  # pylint: disable=unused-import`
+
 ## Contributions quick start guide
 
 1) Fork this repo on github

KVM/qemu/provider/dmesg_router.py

@@ -0,0 +1,103 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# Copyright (c) 2026 Intel Corporation
+
+from virttest import env_process
+from virttest import virt_vm
+from virttest import utils_misc as vt_utils_misc
+from virttest.test_setup import verify as vt_verify
+
+from . import utils_misc as lkvs_utils_misc
+
+
+_ORIG_PREPROCESS = env_process.preprocess
+_ORIG_VERIFY_HOST_SETUP = vt_verify.VerifyHostDMesg.setup
+_ORIG_VERIFY_HOST_CLEANUP = vt_verify.VerifyHostDMesg.cleanup
+_ORIG_VM_VERIFY_DMESG = virt_vm.BaseVM.verify_dmesg
+
+
+def _patched_preprocess(test, params, env, _orig_preprocess=_ORIG_PREPROCESS):
+    if params.get("lkvs_verify_host_dmesg", "no") == "yes":
+        params["verify_host_dmesg"] = "yes"
+    if params.get("lkvs_verify_guest_dmesg", "no") == "yes":
+        params["verify_guest_dmesg"] = "yes"
+    return _orig_preprocess(test, params, env)
+
+
+def _patched_verify_host_setup(
+    self,
+    _orig_setup=_ORIG_VERIFY_HOST_SETUP,
+    _lkvs_utils_misc=lkvs_utils_misc,
+):
+    if self.params.get("lkvs_verify_host_dmesg", "no") != "yes":
+        return _orig_setup(self)
+    self.params["_lkvs_host_dmesg_start_time"] = _lkvs_utils_misc.dmesg_time()
+    return _lkvs_utils_misc.verify_dmesg(
+        self.params["_lkvs_host_dmesg_start_time"], ignore_result=True
+    )
+
+
+def _patched_verify_host_cleanup(
+    self,
+    _orig_cleanup=_ORIG_VERIFY_HOST_CLEANUP,
+    _lkvs_utils_misc=lkvs_utils_misc,
+    _vt_utils_misc=vt_utils_misc,
+):
+    if self.params.get("lkvs_verify_host_dmesg", "no") != "yes":
+        return _orig_cleanup(self)
+    dmesg_log_file = self.params.get("host_dmesg_logfile", "host_dmesg.log")
+    level = self.params.get("host_dmesg_level", 3)
+    expected_host_dmesg = self.params.get("expected_host_dmesg", "")
+    ignore_result = self.params.get("host_dmesg_ignore", "no") == "yes"
+    dmesg_log_file = _vt_utils_misc.get_path(self.test.debugdir, dmesg_log_file)
+    start_time = self.params.get("_lkvs_host_dmesg_start_time", "full")
+    return _lkvs_utils_misc.verify_dmesg(
+        time=start_time,
+        dmesg_log_file=dmesg_log_file,
+        ignore_result=ignore_result,
+        level_check=level,
+        expected_dmesg=expected_host_dmesg,
+    )
+
+
+@virt_vm.session_handler
+def _patched_vm_verify_dmesg(
+    self,
+    dmesg_log_file=None,
+    connect_uri=None,
+    _orig_vm_verify=_ORIG_VM_VERIFY_DMESG,
+    _lkvs_utils_misc=lkvs_utils_misc,
+):
+    if self.params.get("lkvs_verify_guest_dmesg", "no") != "yes":
+        return _orig_vm_verify(self, dmesg_log_file, connect_uri)
+
+    level = self.params.get("guest_dmesg_level", 3)
+    ignore_result = self.params.get("guest_dmesg_ignore", "no") == "yes"
+    serial_login = self.params.get("serial_login", "no") == "yes"
+    if serial_login:
+        self.session = self.wait_for_serial_login()
+    elif (
+        len(self.virtnet) > 0
+        and self.virtnet[0].nettype != "macvtap"
+        and not connect_uri
+    ):
+        self.session = self.wait_for_login()
+    expected_guest_dmesg = self.params.get("expected_guest_dmesg", "")
+    verify_time = _lkvs_utils_misc.dmesg_time(self.session)
+    return _lkvs_utils_misc.verify_dmesg(
+        time=verify_time,
+        dmesg_log_file=dmesg_log_file,
+        ignore_result=ignore_result,
+        level_check=level,
+        session=self.session,
+        expected_dmesg=expected_guest_dmesg,
+    )
+
+
+if env_process.preprocess is not _patched_preprocess:
+    env_process.preprocess = _patched_preprocess
+if vt_verify.VerifyHostDMesg.setup is not _patched_verify_host_setup:
+    vt_verify.VerifyHostDMesg.setup = _patched_verify_host_setup
+if vt_verify.VerifyHostDMesg.cleanup is not _patched_verify_host_cleanup:
+    vt_verify.VerifyHostDMesg.cleanup = _patched_verify_host_cleanup
+if virt_vm.BaseVM.verify_dmesg is not _patched_vm_verify_dmesg:
+    virt_vm.BaseVM.verify_dmesg = _patched_vm_verify_dmesg

KVM/qemu/provider/utils_misc.py

@@ -0,0 +1,191 @@
+# This project includes code from the Avocado-VT project, which is licensed under the GNU General Public License, version 2 or later (GPLv2+). The original code can be found at https://github.com/avocado-framework/avocado-vt.
+#
+# Modifications made by: Farrah Chen <farrah.chen@intel.com> - Mar. 2026
+# Original License: GNU General Public License, version 2 or later (GPLv2+)
+# Modified Code License: GNU General Public License, version 2 or later (GPLv2+)
+
+import re
+import logging
+from datetime import datetime
+from six.moves import xrange
+from avocado.core import exceptions
+from avocado.utils import process
+
+LOG = logging.getLogger("avocado." + __name__)
+
+
+def _get_kernel_messages(time, level_check=3, session=None):
+    """
+    Reads kernel messages for all levels up to certain level.
+    See details in function `verify_dmesg`
+
+    :param time: timestamp for dmesg --since
+    :param level_check: level of severity of issues to be checked
+    :param session: guest
+    :return: 3-tuple (environ, output, status)
+            environ: (guest|host) indicating where the messages
+                     have been read from
+            output: multi-line string containing all read messages
+            status: exit code of read command
+    """
+    full_dmesg_cmd = "dmesg -T -l %s |grep . --color=never" % ",".join(
+        map(str, xrange(0, int(level_check))))
+    if time == 0:
+        cmd = full_dmesg_cmd
+        LOG.warning("Timestamp from /dev/kmsg is overwritten, check full dmesg from boot")
+    elif time == "full":
+        cmd = full_dmesg_cmd
+    else:
+        cmd = "dmesg -T -l %s --since '%s' |grep . --color=never" % (",".join(
+            map(str, xrange(0, int(level_check)))), time
+        )
+
+    if session:
+        environ = "guest"
+        status, output = session.cmd_status_output(cmd)
+    else:
+        environ = "host"
+        out = process.run(
+            cmd, timeout=30, ignore_status=True, verbose=False, shell=True
+        )
+        status = out.exit_status
+        output = out.stdout_text
+    return environ, output, status
+
+
+def _intersec(list1, list2):
+    """
+    Returns a new list containing the elements that are present
+    in both without any specific order.
+
+    :param list1: some list
+    :param list2: some list
+    :return: unordered list, the intersection of the input lists
+    """
+
+    return [x for x in list1 if x in list2]
+
+
+def _remove_dmesg_matches(messages="", expected_dmesg=""):
+    """
+    Removes all messages that match certain regular expressions
+
+    :param messages: single (possibly multi-line) string
+    :param expected_dmesg: single string comma separated list of
+                           regular expressions whose matches are to be
+                           ignored during verification, e.g. "'x.*', 'y.*'"
+    :return: The subset of `messages` that doesn't match
+    """
+    if not expected_dmesg:
+        return messages
+
+    __messages = messages.split("\n")
+    if "" in __messages:
+        __messages.remove("")
+
+    __expected = expected_dmesg.strip('" ').split(",")
+    expected_messages = [x.strip(" '") for x in __expected]
+    filtered_messages = __messages
+
+    for expected_regex in expected_messages:
+        not_matching = [x for x in __messages if not re.findall(expected_regex, x)]
+        filtered_messages = _intersec(filtered_messages, not_matching)
+    return filtered_messages
+
+
+def _log_full_dmesg(dmesg_log_file, environ, output):
+    """
+    Logs all dmesg messages
+
+    :param dmesg_log_file: if given, messages will be written into this file
+    :param environ: (guest|host)
+    :param output: messages
+    :return: string for test log
+    """
+    err = "Found unexpected failures in %s dmesg log." % environ
+    d_log = "dmesg log:\n%s" % output
+    if dmesg_log_file:
+        with open(dmesg_log_file, "w+") as log_f:
+            log_f.write(d_log)
+        err += " Please check %s dmesg log %s." % (environ, dmesg_log_file)
+    else:
+        err += " Please check %s dmesg log in debug log." % environ
+        LOG.debug(d_log)
+    return err
+
+
+def dmesg_time(session=None):
+    """
+    The timestamp in dmesg is not aligned with the timestamp from date in OS.
+    To check the dmesg during the specified time period by "dmesg --since",
+    input a marker into /dev/kmsg to show and record the initial timestamp.
+    :param session: session object to guest
+    :return: string for formatted timestamp from dmesg marker
+    """
+    kmsg_cmd = "echo 'Input from /dev/kmsg: to show the timestamp' > /dev/kmsg"
+    show_time_cmd = "dmesg -T | tail -n 1|grep timestamp"
+    if session:
+        formatted_time = "full"
+        return formatted_time
+    else:
+        marker_status = process.system(kmsg_cmd, shell=True)
+        time_status, time_output = process.getstatusoutput(show_time_cmd, shell=True)
+    if marker_status:
+        LOG.error("/dev/kmsg is not available, failed to capture timestamp, please check your Kconfig and dmesg, continue test!")
+        formatted_time = "unknown"
+        return formatted_time
+    if time_status:
+        LOG.error("Timestamp from /dev/kmsg is overwritten, check full dmesg from boot")
+        formatted_time = 0
+        return formatted_time
+    raw_timestamp = re.search(r"\[(\w+ \w+  ?\d+ \d{2}:\d{2}:\d{2} \d{4})\]", time_output).group(1)
+    raw_format = "%a %b %d %H:%M:%S %Y"
+    out_format = "%Y-%m-%d %H:%M:%S"
+    timestamp = datetime.strptime(raw_timestamp, raw_format)
+    formatted_time = timestamp.strftime(out_format)
+    return formatted_time
+
+
+def verify_dmesg(
+    time,
+    dmesg_log_file=None,
+    ignore_result=False,
+    level_check=3,
+    session=None,
+    expected_dmesg="",
+):
+    """
+    Find host/guest call trace in dmesg log.
+
+    :param time: timestamp for dmesg --since
+    :param dmesg_log_file: The file used to save host dmesg. If None, will save
+                           guest/host dmesg to logging.debug.
+    :param ignore_result: True or False, whether to fail test case on issues
+    :param level_check: level of severity of issues to be checked
+                        1 - emerg
+                        2 - emerg,alert
+                        3 - emerg,alert,crit
+                        4 - emerg,alert,crit,err
+                        5 - emerg,alert,crit,err,warn
+    :param session: session object to guest
+    :param expected_dmesg: single string comma separated list of
+                           regular expressions whose matches are to be
+                           ignored during verification, e.g. "'x.*', 'y.*'"
+    :param return: if ignore_result=True, return True if no errors/crash
+                   observed, False otherwise.
+    :param raise: if ignore_result=False, raise TestFail exception on
+                  observing errors/crash
+    """
+    if time == "unknown":
+        LOG.error("/dev/kmsg is not available, failed to capture timestamp, please check your Kconfig and dmesg, continue test!")
+        return True
+    environ, output, status = _get_kernel_messages(time, level_check, session)
+    if status == 0:
+        unexpected_messages = _remove_dmesg_matches(output, expected_dmesg)
+        err = _log_full_dmesg(dmesg_log_file, environ, output)
+        if not ignore_result and unexpected_messages:
+            raise exceptions.TestFail(err)
+        if unexpected_messages:
+            LOG.debug(err)
+        return False
+    return True

KVM/qemu/tests/apic_timer_v.py

@@ -7,6 +7,7 @@
 #
 # History:  Jan. 2026 - Xudong Hao - creation
 
+from provider import dmesg_router  # pylint: disable=unused-import
 import os
 import time
 from avocado.utils import process, cpu

KVM/qemu/tests/boot.py

@@ -11,6 +11,7 @@
 # Copyright: Red Hat (c) 2024 and Avocado contributors
 # Copy from tp-qemu
 
+from provider import dmesg_router  # pylint: disable=unused-import
 import time
 
 from virttest import error_context

KVM/qemu/tests/boot_check.py

@@ -7,6 +7,7 @@
 #
 # History:  July. 2024 - Xudong Hao - creation
 
+from provider import dmesg_router  # pylint: disable=unused-import
 from avocado.utils import cpu
 from virttest import env_process
 from virttest import error_context

KVM/qemu/tests/cpu_model.py

@@ -7,6 +7,7 @@
 #
 # History:  Nov. 2025 - Xudong Hao - creation
 
+from provider import dmesg_router  # pylint: disable=unused-import
 import os
 import sys
 from avocado.utils import process

KVM/qemu/tests/cpu_pku.py

@@ -7,6 +7,7 @@
 #
 # History:  June. 2025 - Xudong Hao - creation
 
+from provider import dmesg_router  # pylint: disable=unused-import
 from virttest import cpu, env_process, error_context
 
 

KVM/qemu/tests/ept_5lp_basic.py

@@ -7,6 +7,7 @@
 #
 # History:  May. 2025 - Xudong Hao - creation
 
+from provider import dmesg_router  # pylint: disable=unused-import
 from avocado.utils import process, cpu
 from virttest import error_context, env_process
 from provider.cpu_utils import check_vmx_flags

KVM/qemu/tests/feature_test.py

@@ -7,6 +7,7 @@
 #
 # History: Sept. 2024 - Xudong Hao - creation
 
+from provider import dmesg_router  # pylint: disable=unused-import
 import os
 import re
 from avocado.core import exceptions

KVM/qemu/tests/host_cpu_offline_online.py

@@ -7,6 +7,7 @@
 #
 # History: Dec. 2024 - Xudong Hao - creation
 
+from provider import dmesg_router  # pylint: disable=unused-import
 import random
 from avocado.utils import cpu
 

KVM/qemu/tests/ras.py

@@ -7,6 +7,7 @@
 #
 # History: Nov. 2025 - Farrah Chen - creation
 
+from provider import dmesg_router  # pylint: disable=unused-import
 import os
 import re
 from avocado.utils import process