Merge pull request #171 from NVIDIA/use_prev_log_file

Use previous logging file when available

Author: namitdhameja

src/nvidia_resiliency_ext/shared_utils/log_node_local_tmp.py

@@ -57,27 +57,33 @@ def emit(self, record: logging.LogRecord):
             sys.stderr.write(f"Log handler error: {record.getMessage()}\n")
             sys.stderr.flush()
 
+    def _get_backup_files(self):
+        """Return sorted list of backup files for this rank/process."""
+        rank_str = str(self.rank_id) if self.rank_id is not None else "unknown"
+        file_prefix = f"rank_{rank_str}_{self.proc_name}.msg."
+        backup_files = [
+            filename
+            for filename in os.listdir(self.file_path)
+            if re.match(rf"{file_prefix}(\d+)", filename)
+        ]
+        backup_files.sort()
+        return backup_files
+
     def _log_file_namer(self):
-        # Use "unknown" for rank_id if it's None
+        backup_files = self._get_backup_files()
+        if self.fname is None and backup_files:
+            return backup_files[0]
         rank_str = str(self.rank_id) if self.rank_id is not None else "unknown"
-        return f"rank_{rank_str}_{self.proc_name}.msg.{int(time.time()*1000)}"
+        file_prefix = f"rank_{rank_str}_{self.proc_name}.msg."
+        return f"{file_prefix}{int(time.time()*1000)}"
 
     def _cleanup_old_backup_files(self):
-        """Clean up old log files, keeping only the most recent one's."""
-        backup_files = []
-        # Use "unknown" for rank_id if it's None
-        rank_str = str(self.rank_id) if self.rank_id is not None else "unknown"
-        for filename in os.listdir(self.file_path):
-            match = re.match(rf"rank_{rank_str}_{self.proc_name}.msg\.(\d+)", filename)
-            if not match:
-                continue
-            backup_files.append(filename)
-        backup_files.sort()
+        """Clean up old log files, keeping only the most recent ones."""
+        backup_files = self._get_backup_files()
         for old_file in backup_files[: -self.max_backup_files]:
             try:
                 os.remove(os.path.join(self.file_path, old_file))
             except (OSError, IOError) as e:
-                # Log the error but don't fail the entire operation
                 sys.stderr.write(f"Failed to remove backup file {old_file}: {e}\n")
                 sys.stderr.flush()
 
@@ -156,7 +162,7 @@ def __init__(self, log_message: str):
                     # Convert asctime to a datetime object, then to a Unix timestamp
                     dt = datetime.strptime(value, '%Y-%m-%d %H:%M:%S,%f')
                     timestamp = int(dt.timestamp())
-                    self.hash_table[key] = value
+                    self.hash_table[key] = timestamp
                 else:
                     self.hash_table[key] = value
 

tests/shared_utils/test_logger.py

@@ -180,8 +180,9 @@ def check_msg(
         log_type="info",
         dbg_on="0",
         max_file_num=None,
+        clean_workspace=True,
     ):
-        log_dir, temp_dir = create_test_workspace(clean=True)
+        log_dir, temp_dir = create_test_workspace(clean=clean_workspace)
         setup_vars(
             global_id=0,
             local_id=0,
@@ -237,6 +238,19 @@ def test_many_msg(self):
     def test_rotation(self):
         self.check_msg(num_msg=900, file_size_kb=10, pm_files=5, is_agg=False)
 
+    def test_rotation_existing_file(self):
+        _, temp_dir = create_test_workspace(clean=True)
+        os.makedirs(temp_dir, exist_ok=True)
+        fname = os.path.join(temp_dir, "rank_0_test.msg.1757013222372")
+        with open(fname, 'x') as f:
+            for i in range(50):
+                f.write(f"My Old Logging Message {i}\n")
+            f.flush()
+        file_size = os.path.getsize(fname)
+        self.check_msg(clean_workspace=False, num_msg=90, file_size_kb=10, pm_files=2, is_agg=False)
+        file_size_after_logging = os.path.getsize(fname)
+        assert file_size_after_logging > file_size, "File size after logging should be > before"
+
     def test_rotation_cleanup(self):
         self.check_msg(num_msg=2000, file_size_kb=10, pm_files=1, is_agg=True, max_file_num=50)