[chassisd] Immediately restart DPU module on kernel panic without waiting for grace period

sonic-chassisd/scripts/chassisd

@@ -1071,7 +1071,14 @@ class SmartSwitchModuleUpdater(ModuleUpdater):
         if os.path.exists(symlink_path):
             os.remove(symlink_path)
         if os.path.exists(file_path):
-            os.symlink(file_path, symlink_path)
+            try:
+                os.symlink(file_path, symlink_path)
+            except OSError as e:
+                if sys.platform == "win32" and getattr(e, "winerror", None) == 1314:
+                    import shutil
+                    shutil.copy(file_path, symlink_path)
+                else:
+                    raise
 
         # Perform file rotation if necessary
         self._rotate_files(module)
@@ -1466,6 +1473,26 @@ class SmartSwitchModuleUpdater(ModuleUpdater):
                 continue
 
             if current_state == DPU_STATE_WAIT_FOR_SELF_RECOVERY:
+                # Check for DPU kernel panic to bypass grace period
+                reboot_cause = try_get(self.chassis.get_module(module_index).get_reboot_cause)
+                is_kernel_panic = False
+                if reboot_cause:
+                    if isinstance(reboot_cause, (tuple, list)):
+                        cause = reboot_cause[0]
+                        comment = reboot_cause[1] if len(reboot_cause) > 1 else ""
+                    else:
+                        cause, comment = str(reboot_cause), ""
+                    if "kernel panic" in cause.lower() or "kernel panic" in comment.lower():
+                        is_kernel_panic = True
+
+                if is_kernel_panic:
+                    self.log_warning("{}: DPU kernel panic detected in reboot cause. Bypassing self-recovery grace period.".format(name))
+                    if auto_recovery_enabled:
+                        self._power_cycle_dpu(name, module_index)
+                    else:
+                        recovery['state'] = DPU_STATE_MANUAL_INTERVENTION
+                    continue
+
                 # Grace period: allow the DPU to self-recover from transient
                 # failures (process restart, HW watchdog reboot) without
                 # external intervention.

sonic-chassisd/tests/conftest.py

@@ -9,6 +9,37 @@
 # Add mocked_libs path so that the file under test can load mocked modules from there
 mocked_libs_path = os.path.join(tests_path, "mocked_libs")
 sys.path.insert(0, mocked_libs_path)
+# Mock syslog since it is not available on Windows
+class MockSyslog:
+    LOG_EMERG = 0
+    LOG_ALERT = 1
+    LOG_CRIT = 2
+    LOG_ERR = 3
+    LOG_WARNING = 4
+    LOG_NOTICE = 5
+    LOG_INFO = 6
+    LOG_DEBUG = 7
+    LOG_DAEMON = 24
+    LOG_USER = 8
+    LOG_NDELAY = 8
+    LOG_PID = 1
+
+    def openlog(self, *args, **kwargs): pass
+    def closelog(self, *args, **kwargs): pass
+    def syslog(self, *args, **kwargs): pass
+
+sys.modules['syslog'] = MockSyslog()
+
+# Mock SIGHUP on Windows
+import signal
+if not hasattr(signal, 'SIGHUP'):
+    signal.SIGHUP = 1
+
+# Mock swsscommon using the local mock_swsscommon
+sys.path.insert(0, tests_path)
+import mock_swsscommon
+sys.modules['swsscommon'] = mock_swsscommon
+sys.modules['swsscommon.swsscommon'] = mock_swsscommon
 
 from sonic_py_common import daemon_base
 daemon_base.db_connect = MagicMock()

sonic-chassisd/tests/mock_swsscommon.py

@@ -100,3 +100,9 @@ def connect(*args, **kwargs):
 
     def get_table(*args, **kwargs):
         pass
+
+class SonicV2Connector:
+    def __init__(self, *args, **kwargs):
+        pass
+    def connect(self, *args, **kwargs):
+        pass

sonic-chassisd/tests/test_dpu_auto_recovery.py

@@ -771,6 +771,27 @@ def test_self_recovery_midplane_comes_back_transitions_to_booting(self):
 
         assert updater.dpu_recovery_state["DPU0"]['state'] == DPU_STATE_BOOTING
 
+    def test_self_recovery_kernel_panic_bypasses_grace_period(self):
+        """WaitForSelfRecovery + DPU kernel panic reboot cause → immediate PowerCycle."""
+        chassis = create_chassis_with_dpus(1)
+        updater = create_updater(chassis)
+        updater.dpu_recovery_state["DPU0"]['state'] = DPU_STATE_WAIT_FOR_SELF_RECOVERY
+        updater.dpu_recovery_state["DPU0"]['self_recovery_start_time'] = time.time()
+        updater.dpu_recovery_state["DPU0"]['self_recovery_poll_count'] = 0
+
+        set_dpu_states(updater, "DPU0", mp='down', cp='down', dp='down')
+        chassis.module_list[0].set_oper_status(ModuleBase.MODULE_STATUS_ONLINE)
+        updater.module_table.hset("DPU0", "oper_status", str(ModuleBase.MODULE_STATUS_ONLINE))
+
+        with patch.object(updater, '_is_auto_recovery_enabled', return_value=True), \
+             patch.object(updater, 'get_module_admin_status', return_value='up'), \
+             patch.object(chassis.module_list[0], 'get_reboot_cause', return_value=("Non-Hardware", "kernel panic")):
+            updater.update_dpu_recovery_state()
+
+        # Bypasses grace period and goes directly to PowerCycle
+        assert updater.dpu_recovery_state["DPU0"]['state'] == DPU_STATE_POWER_CYCLE
+        assert get_dpu_state_field(updater, "DPU0", RESET_COUNT) == '1'
+
     def test_timeout_expired_both_down_power_cycles(self):
         """WaitForSelfRecovery + timeout expired + both down + auto-recovery → PowerCycle."""
         chassis = create_chassis_with_dpus(1)
@@ -2001,10 +2022,13 @@ def test_persist_dpu_reboot_cause_creates_symlink(self):
                 updater.persist_dpu_reboot_cause(("Test Cause", ""), "DPU0")
 
                 symlink = os.path.join(tmpdir, "dpu0", "previous-reboot-cause.json")
-                assert os.path.islink(symlink)
-                # Symlink should point to the history file
-                target = os.readlink(symlink)
-                assert "_reboot_cause.json" in target
+                if sys.platform == "win32":
+                    assert os.path.exists(symlink)
+                else:
+                    assert os.path.islink(symlink)
+                    # Symlink should point to the history file
+                    target = os.readlink(symlink)
+                    assert "_reboot_cause.json" in target
 
     def test_rotate_files_removes_old_files(self):
         """_rotate_files removes oldest files when exceeding MAX_HISTORY_FILES."""