[reboot-cause] Use UTC to ensure consistent sorting (#293)

* Use UTC to ensure consistent sorting

Previously reboot cause uses local timezone. This could result in the
timestamp being in UTC, IDT, or any other timezone depending on the
system configuration. When reboot cause history files are sorted,
mixing different timezones can lead to incorrect chronological
order and cause reboot cause test failure.
Now change the code to use datetime.datetime.utcnow() for
reboot_cause_gen_time, ensuring that all timestamps are consistently in
UTC.

Signed-off-by: Jianyue Wu <[email protected]>

Author: jianyuewu

scripts/determine-reboot-cause

@@ -257,7 +257,7 @@ def main():
     previous_reboot_cause, additional_reboot_info = determine_reboot_cause()
 
     # Current time
-    reboot_cause_gen_time = str(datetime.datetime.now().strftime('%Y_%m_%d_%H_%M_%S'))
+    reboot_cause_gen_time = str(datetime.datetime.utcnow().strftime('%Y_%m_%d_%H_%M_%S'))
 
     # Save the previous cause info into its history file as json format
     reboot_cause_dict = get_reboot_cause_dict(previous_reboot_cause, additional_reboot_info, reboot_cause_gen_time)

scripts/procdockerstatsd

@@ -245,7 +245,7 @@ class ProcDockerStats(daemon_base.DaemonBase):
 
         while True:
             self.update_dockerstats_command()
-            datetimeobj = datetime.now()
+            datetimeobj = datetime.utcnow()
             # Adding key to store latest update time.
             self.update_state_db('DOCKER_STATS|LastUpdateTime', 'lastupdate', str(datetimeobj))
             self.update_processstats_command()

tests/determine-reboot-cause_test.py

@@ -3,6 +3,8 @@
 import shutil
 import pytest
 import json
+import datetime
+import re
 
 from swsscommon import swsscommon
 from sonic_py_common.general import load_module_from_source
@@ -271,3 +273,21 @@ def test_check_and_create_dpu_dirs(
         # Assert that reboot-cause.txt was created for each DPU
         mock_open.assert_any_call(os.path.join(REBOOT_CAUSE_MODULE_DIR, "dpu0", "reboot-cause.txt"), 'w')
         mock_open.assert_any_call(os.path.join(REBOOT_CAUSE_MODULE_DIR, "dpu1", "reboot-cause.txt"), 'w')
+
+    def test_reboot_cause_gen_time_is_utc(self):
+        # Patch datetime.datetime.utcnow to return a known value
+        class FixedDatetime(datetime.datetime):
+            @classmethod
+            def utcnow(cls):
+                return cls(2024, 7, 1, 12, 34, 56)
+        orig_datetime = determine_reboot_cause.datetime
+        determine_reboot_cause.datetime = FixedDatetime
+
+        try:
+            gen_time = determine_reboot_cause.datetime.utcnow().strftime('%Y_%m_%d_%H_%M_%S')
+            # Should match the fixed UTC time
+            assert gen_time == "2024_07_01_12_34_56"
+            # Optionally, check the format
+            assert re.match(r"\d{4}_\d{2}_\d{2}_\d{2}_\d{2}_\d{2}", gen_time)
+        finally:
+            determine_reboot_cause.datetime = orig_datetime        

tests/procdockerstatsd_test.py

@@ -163,3 +163,68 @@ def cpu_percent(self):
 
         assert 1234 in pdstatsd.all_process_obj
         assert 9999 not in pdstatsd.all_process_obj
+
+    def test_datetime_utcnow_usage(self):
+        """Test that datetime.utcnow() is used instead of datetime.now() for consistent UTC timestamps"""
+        pdstatsd = procdockerstatsd.ProcDockerStats(procdockerstatsd.SYSLOG_IDENTIFIER)
+        
+        # Mock datetime.utcnow to return a fixed time for testing
+        fixed_time = datetime(2025, 7, 1, 12, 34, 56)
+        
+        with patch('procdockerstatsd.datetime') as mock_datetime:
+            mock_datetime.utcnow.return_value = fixed_time
+            
+            # Test the update_fipsstats_command method which uses datetime.utcnow()
+            pdstatsd.update_fipsstats_command()
+            
+            # Verify that utcnow() was called
+            mock_datetime.utcnow.assert_called_once()
+            
+            # Verify that now() was NOT called (ensuring we're using UTC)
+            mock_datetime.now.assert_not_called()
+            
+            # Test the main run loop datetime usage
+            with patch.object(pdstatsd, 'update_dockerstats_command'):
+                with patch.object(pdstatsd, 'update_processstats_command'):
+                    with patch.object(pdstatsd, 'update_fipsstats_command'):
+                        with patch('time.sleep'):  # Prevent actual sleep
+                            # Mock the first iteration of the run loop
+                            pdstatsd.update_dockerstats_command()
+                            datetimeobj = mock_datetime.utcnow()
+                            pdstatsd.update_state_db('DOCKER_STATS|LastUpdateTime', 'lastupdate', str(datetimeobj))
+                            pdstatsd.update_processstats_command()
+                            pdstatsd.update_state_db('PROCESS_STATS|LastUpdateTime', 'lastupdate', str(datetimeobj))
+                            pdstatsd.update_fipsstats_command()
+                            pdstatsd.update_state_db('FIPS_STATS|LastUpdateTime', 'lastupdate', str(datetimeobj))
+                            
+                            # Verify utcnow() was called multiple times as expected
+                            assert mock_datetime.utcnow.call_count >= 2        
+
+    def test_run_method_executes_with_utcnow(self):
+        """Test that run method executes and uses datetime.utcnow()"""
+        pdstatsd = procdockerstatsd.ProcDockerStats(procdockerstatsd.SYSLOG_IDENTIFIER)
+        
+        # Mock all dependencies but allow datetime.utcnow() to run normally
+        with patch.object(pdstatsd, 'update_dockerstats_command'):
+            with patch.object(pdstatsd, 'update_processstats_command'):
+                with patch.object(pdstatsd, 'update_fipsstats_command'):
+                    with patch('time.sleep', side_effect=Exception("Stop after first iteration")):
+                        with patch('os.getuid', return_value=0):  # Mock as root
+                            with patch.object(pdstatsd, 'log_info'):
+                                with patch.object(pdstatsd, 'update_state_db') as mock_update_db:
+                                    # This will actually call run() method
+                                    try:
+                                        pdstatsd.run()
+                                    except Exception as e:
+                                        if "Stop after first iteration" in str(e):
+                                            # Verify that update_state_db was called
+                                            assert mock_update_db.call_count >= 3
+                                            # Verify that timestamps were passed
+                                            for call in mock_update_db.call_args_list:
+                                                args = call[0]
+                                                if len(args) >= 3 and 'lastupdate' in args[1]:
+                                                    timestamp_str = args[2]
+                                                    assert isinstance(timestamp_str, str)
+                                                    assert len(timestamp_str) > 0
+                                        else:
+                                            raise