Modprobe Tool: Improve failure debugging

If Modprobe fails to remove a module, run some
additional commands to debug why it failed

Author: adityagesh

lisa/tools/dmesg.py

@@ -2,7 +2,7 @@
 # Licensed under the MIT license.
 
 import re
-from typing import List
+from typing import List, Optional
 
 from semver import VersionInfo
 
@@ -49,11 +49,26 @@ def command(self) -> str:
     def _check_exists(self) -> bool:
         return True
 
-    def get_output(self, force_run: bool = False) -> str:
-        command_output = self._run(force_run=force_run)
+    def get_output(
+        self,
+        force_run: bool = False,
+        no_error_log: bool = True,
+        no_info_log: bool = True,
+        no_debug_log: bool = False,
+        tail_lines: Optional[int] = None,
+    ) -> str:
+        command_output = self._run(
+            force_run=force_run,
+            no_error_log=no_error_log,
+            no_info_log=no_info_log,
+            no_debug_log=no_debug_log,
+        )
 
         # Remove the color code from stdout stream
         stdout = filter_ansi_escape(command_output.stdout)
+        if tail_lines is not None:
+            stdout_lines = stdout.splitlines()
+            stdout = "\n".join(stdout_lines[-tail_lines:])
         return stdout
 
     def check_kernel_errors(
@@ -104,12 +119,29 @@ def get_vmbus_version(self) -> VersionInfo:
                 return VersionInfo(int(major), int(minor))
         raise LisaException("No find matched vmbus version in dmesg")
 
-    def _run(self, force_run: bool = False) -> ExecutableResult:
-        # sometime it need sudo, we can retry
-        # so no_error_log for first time
-        result = self.run(force_run=force_run, no_error_log=True)
+    def _run(
+        self,
+        force_run: bool = False,
+        no_error_log: bool = True,
+        no_info_log: bool = True,
+        no_debug_log: bool = False,
+    ) -> ExecutableResult:
+        # no_error_log is set to True for the first run because
+        # it will fail for distros that require sudo to run dmesg.
+        result = self.run(
+            force_run=force_run,
+            no_error_log=True,
+            no_info_log=no_info_log,
+            no_debug_log=no_debug_log,
+        )
         if result.exit_code != 0:
             # may need sudo
-            result = self.run(sudo=True, force_run=force_run)
+            result = self.run(
+                sudo=True,
+                force_run=force_run,
+                no_error_log=no_error_log,
+                no_info_log=no_info_log,
+                no_debug_log=no_debug_log,
+            )
         self._cached_result = result
         return result

lisa/tools/journalctl.py

@@ -53,3 +53,40 @@ def first_n_logs_from_boot(
             expected_exit_code=0,
         )
         return result.stdout
+
+    def filter_logs_by_pattern(
+        self,
+        pattern: str,
+        tail_line_count: int = 1000,
+        no_debug_log: bool = False,
+        no_error_log: bool = False,
+        no_info_log: bool = True,
+        sudo: bool = True,
+    ) -> str:
+        """
+        Filters journalctl logs by a given pattern using grep.
+
+        Args:
+            pattern (str): Passed directly to grep. Use a valid grep pattern
+                (string or regex).
+            tail_line_count (int): Optionally limits the output to the last N lines.
+        """
+        cmd = f"--no-pager | grep '{pattern}'"
+
+        if tail_line_count > 0:
+            cmd = cmd + f" | tail -n {tail_line_count}"
+
+        result = self.run(
+            cmd,
+            force_run=True,
+            shell=True,
+            sudo=sudo,
+            no_debug_log=no_debug_log,
+            no_error_log=no_error_log,
+            no_info_log=no_info_log,
+            expected_exit_code=0,
+            expected_exit_code_failure_message=(
+                f"Failed to filter journalctl logs by pattern '{pattern}'"
+            ),
+        )
+        return result.stdout

lisa/tools/lsmod.py

@@ -2,10 +2,10 @@
 # Licensed under the MIT license.
 
 import re
-from typing import Any
+from typing import Any, Tuple
 
 from lisa.executable import Tool
-from lisa.util import find_patterns_in_lines
+from lisa.util import find_patterns_groups_in_lines, find_patterns_in_lines
 
 
 class Lsmod(Tool):
@@ -14,6 +14,7 @@ class Lsmod(Tool):
     #    fuse                   52176  3
     #   cryptd                 14125  0
     #   aes_generic            32970  1 aes_i586
+    #   ip_tables              32768  3 iptable_filter,iptable_security,iptable_nat
     __output_pattern = re.compile(
         r"^(?P<name>[^\s]+)\s+(?P<size>[^\s]+)\s+(?P<usedby>.*)?$", re.MULTILINE
     )
@@ -34,12 +35,14 @@ def module_exists(
         force_run: bool = False,
         no_info_log: bool = True,
         no_error_log: bool = True,
+        no_debug_log: bool = False,
     ) -> bool:
         result = self.run(
             sudo=True,
             force_run=force_run,
             no_info_log=no_info_log,
             no_error_log=no_error_log,
+            no_debug_log=no_debug_log,
             expected_exit_code=0,
         )
 
@@ -48,3 +51,43 @@ def module_exists(
             return True
 
         return False
+
+    def get_used_by_modules(
+        self,
+        mod_name: str,
+        sudo: bool = True,
+        force_run: bool = False,
+        no_info_log: bool = True,
+        no_error_log: bool = True,
+        no_debug_log: bool = True,
+    ) -> Tuple[int, str]:
+        """
+        Returns a list of modules that are using the given module name.
+        """
+        result = self.run(
+            sudo=sudo,
+            force_run=force_run,
+            no_info_log=no_info_log,
+            no_error_log=no_error_log,
+            no_debug_log=no_debug_log,
+            expected_exit_code=0,
+        )
+
+        module_info = find_patterns_groups_in_lines(
+            result.stdout, [self.__output_pattern]
+        )
+
+        target_module = next(
+            (module for module in module_info[0] if module["name"] == mod_name), None
+        )
+
+        # If the module is not found or has no 'usedby' information,
+        # return 0 and empty string
+        if target_module is None or not target_module.get("usedby"):
+            return 0, ""
+
+        usedby_split = target_module["usedby"].split(maxsplit=1)
+        usedby_count = int(usedby_split[0]) if usedby_split[0].isdigit() else 0
+        usedby_modules = usedby_split[1] if len(usedby_split) > 1 else ""
+
+        return usedby_count, usedby_modules

lisa/tools/modprobe.py

@@ -3,7 +3,10 @@
 from typing import Any, List, Optional, Type, Union
 
 from lisa.executable import ExecutableResult, Tool
+from lisa.tools.dmesg import Dmesg
+from lisa.tools.journalctl import Journalctl
 from lisa.tools.kernel_config import KLDStat
+from lisa.tools.lsmod import Lsmod
 from lisa.util import UnsupportedOperationException
 
 
@@ -74,15 +77,20 @@ def remove(self, mod_names: List[str], ignore_error: bool = False) -> None:
                 self.node.execute(f"rmmod {mod_name}", sudo=True, shell=True)
             else:
                 if self.is_module_loaded(mod_name, force_run=True):
-                    self.run(
-                        f"-r {mod_name}",
-                        force_run=True,
-                        sudo=True,
-                        shell=True,
-                        expected_exit_code=0,
-                        expected_exit_code_failure_message="Fail to remove module "
-                        f"{mod_name}",
-                    )
+                    try:
+                        self.run(
+                            f"-r {mod_name}",
+                            force_run=True,
+                            sudo=True,
+                            shell=True,
+                            expected_exit_code=0,
+                            expected_exit_code_failure_message=(
+                                f"Fail to remove module {mod_name}"
+                            ),
+                        )
+                    except AssertionError as e:
+                        self._collect_logs(mod_name)
+                        raise e
 
     def load(
         self,
@@ -154,6 +162,62 @@ def load_by_file(
             result.assert_exit_code(0, f"failed to load module {file_name}")
         return result
 
+    def _collect_logs(self, mod_name: str) -> None:
+        self._log.info(
+            f"Failed to remove module {mod_name}."
+            " Collecting additional debug information."
+        )
+        self._log_lsmod_status(mod_name)
+        self._tail_dmesg_log()
+        self._tail_journalctl_for_module(mod_name)
+
+    def _log_lsmod_status(self, mod_name: str) -> None:
+        lsmod_tool = self.node.tools[Lsmod]
+        module_exists = lsmod_tool.module_exists(
+            mod_name=mod_name,
+            force_run=True,
+            no_debug_log=True,
+        )
+        if not module_exists:
+            self._log.info(f"Module {mod_name} does not exist in lsmod output.")
+        else:
+            self._log.info(f"Module {mod_name} exists in lsmod output.")
+            usedby_count, usedby_modules = lsmod_tool.get_used_by_modules(
+                mod_name, sudo=True, force_run=True
+            )
+            if usedby_count == 0:
+                self._log.info(f"Module '{mod_name}' is not used by any other modules.")
+            else:
+                self._log.info(
+                    f"Module {mod_name} is used by {usedby_count} modules. "
+                    f"Module names: '{usedby_modules}'"
+                )
+
+    def _tail_dmesg_log(self) -> None:
+        dmesg_tool = self.node.tools[Dmesg]
+        tail_lines = 20
+
+        dmesg_output = dmesg_tool.get_output(
+            force_run=True, no_debug_log=True, tail_lines=tail_lines
+        )
+
+        self._log.info(f"Last {tail_lines} dmesg lines: {dmesg_output}")
+
+    def _tail_journalctl_for_module(self, mod_name: str) -> None:
+        tail_lines = 20
+        journalctl_tool = self.node.tools[Journalctl]
+        journalctl_out = journalctl_tool.filter_logs_by_pattern(
+            mod_name,
+            tail_line_count=tail_lines,
+            no_debug_log=True,
+            no_error_log=True,
+            no_info_log=True,
+        )
+        self._log.info(
+            f"Last {tail_lines} journalctl lines for module {mod_name}:\n"
+            f"{journalctl_out}"
+        )
+
 
 class ModprobeFreeBSD(Modprobe):
     def module_exists(self, modules: Union[str, List[str]]) -> bool: