[202411] Backport: [reboot-cause] Use UTC to ensure consistent sorting (#306)

[202411] Backport: [reboot-cause] Use UTC to ensure consistent sorting

Author: jianyuewu

scripts/determine-reboot-cause

@@ -229,7 +229,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

@@ -239,7 +239,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

@@ -2,6 +2,8 @@
 import os
 import shutil
 import pytest
+import datetime
+import re
 
 from swsscommon import swsscommon
 from sonic_py_common.general import load_module_from_source
@@ -206,3 +208,21 @@ def test_determine_reboot_cause_main_with_reboot_cause_dir(self):
             determine_reboot_cause.main()
             assert os.path.exists("host/reboot-cause/reboot-cause.txt") == True
             assert os.path.exists("host/reboot-cause/previous-reboot-cause.json") == True   
+
+    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

@@ -143,3 +143,68 @@ def test_update_fipsstats_command(self, mock_cmd):
         pdstatsd.update_fipsstats_command()
         assert pdstatsd.state_db.get('STATE_DB', 'FIPS_STATS|state', 'enforced') == "False"
         assert pdstatsd.state_db.get('STATE_DB', 'FIPS_STATS|state', 'enabled') == "True"
+
+    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