Add test-case for kernel crash dump threads

tests/linux_kernel/vmcore/__init__.py

@@ -0,0 +1,25 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+from pathlib import Path
+import unittest
+
+import drgn
+from tests import TestCase
+
+VMCORE_PATH = Path("/proc/vmcore")
+
+
+@unittest.skipUnless(VMCORE_PATH.exists(), "not running in kdump")
+class LinuxVMCoreTestCase(TestCase):
+    prog = None
+
+    @classmethod
+    def setUpClass(cls):
+        # We only want to create the Program once for all tests, so it's cached
+        # as a class variable (in the base class).
+        if LinuxVMCoreTestCase.prog is None:
+            prog = drgn.Program()
+            prog.set_core_dump(VMCORE_PATH)
+            prog.load_default_debug_info()
+            LinuxVMCoreTestCase.prog = prog

tests/linux_kernel/vmcore/test_vmcore.py

@@ -1,15 +1,18 @@
-from pathlib import Path
-import unittest
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# SPDX-License-Identifier: GPL-3.0-or-later
 
-from drgn import Program, ProgramFlags
+from drgn import ProgramFlags
+from tests.linux_kernel.vmcore import LinuxVMCoreTestCase
 
-VMCORE_PATH = Path("/proc/vmcore")
 
+class TestVMCore(LinuxVMCoreTestCase):
+    def test_program_flags(self):
+        self.assertFalse(self.prog.flags & ProgramFlags.IS_LIVE)
+        self.assertTrue(self.prog.flags & ProgramFlags.IS_LINUX_KERNEL)
 
-@unittest.skipUnless(VMCORE_PATH.exists(), "not running in kdump")
-class TestAttachToVMCore(unittest.TestCase):
-    def test_attach_to_vmcore(self):
-        prog = Program()
-        prog.set_core_dump("/proc/vmcore")
-        self.assertFalse(prog.flags & ProgramFlags.IS_LIVE)
-        self.assertTrue(prog.flags & ProgramFlags.IS_LINUX_KERNEL)
+    def test_threads(self):
+        crashed_thread_tid = self.prog.crashed_thread().tid
+        self.assertGreater(crashed_thread_tid, 0)
+        self.assertIn(1, (thread.tid for thread in self.prog.threads()))
+        self.assertEqual(self.prog.thread(crashed_thread_tid).tid, crashed_thread_tid)
+        self.assertEqual(self.prog.thread(1).tid, 1)

vmtest/enter_kdump.py

@@ -8,6 +8,7 @@
 import ctypes
 import os
 import re
+import subprocess
 
 SYS_kexec_file_load = 320  # On x86-64.
 KEXEC_FILE_ON_CRASH = 2
@@ -19,10 +20,15 @@
 with open("/proc/cmdline", "rb") as f:
     cmdline = f.read().rstrip(b"\n")
     cmdline = re.sub(rb"(^|\s)crashkernel=\S+", b"", cmdline)
-    # `nosmp` is required to avoid QEMU sporadically failing an internal assertion
     # `nokaslr` is required to avoid sporadically failing to reserve space for the
     # capture kernel
-    cmdline += b" nosmp nokaslr"
+    cmdline += b" nokaslr"
+    if (
+        subprocess.run(("systemd-detect-virt",), stdout=subprocess.PIPE).stdout
+        != b"kvm\n"
+    ):
+        # `nosmp` is required to avoid QEMU sporadically failing an internal assertion
+        cmdline += b" nosmp"
 
 with open(f"/lib/modules/{os.uname().release}/vmlinuz", "rb") as kernel:
     if syscall(