different logs for converter and compressor

Author: rishikeshdevsot

components/job-orchestration/job_orchestration/executor/compress/compression_task.py

@@ -429,18 +429,24 @@ def cleanup_temporary_files():
                     converted_inputs_dir,
                 )
 
-    # Open stderr log file
-    stderr_log_path = logs_dir / f"{instance_id_str}-stderr.log"
-    stderr_log_file = open(stderr_log_path, "w")
+    # Open log files
+    compression_log_path = logs_dir / f"{instance_id_str}-stderr.log"
+    compression_log_file = open(compression_log_path, "w")
+    conversion_log_path = logs_dir / f"{instance_id_str}-conversion-stderr.log"
+    conversion_log_file = open(conversion_log_path, "w")
 
     conversion_return_code = 0
     if conversion_cmd is not None:
         logger.debug("Execute log-converter with command: %s", conversion_cmd)
         conversion_proc = subprocess.Popen(
-            conversion_cmd, stdout=subprocess.DEVNULL, stderr=stderr_log_file, env=conversion_env
+            conversion_cmd,
+            stdout=subprocess.DEVNULL,
+            stderr=conversion_log_file,
+            env=conversion_env,
         )
         conversion_return_code = conversion_proc.wait()
-
+    conversion_log_file.close()
+        
     try:
         if conversion_return_code != 0:
             logger.error(
@@ -449,15 +455,17 @@ def cleanup_temporary_files():
             worker_output = {
                 "total_uncompressed_size": 0,
                 "total_compressed_size": 0,
-                "error_message": f"Check logs in {stderr_log_path}",
+                "error_message": f"Check logs in {conversion_log_path}",
             }
+            log_file_contents(logger, conversion_log_path)
+            
             return CompressionTaskStatus.FAILED, worker_output
 
         # Start compression
         logger.debug("Compressing...")
         compression_successful = False
         proc = subprocess.Popen(
-            compression_cmd, stdout=subprocess.PIPE, stderr=stderr_log_file, env=compression_env
+            compression_cmd, stdout=subprocess.PIPE, stderr=compression_log_file, env=compression_env
         )
 
         # Compute the total amount of data compressed
@@ -536,7 +544,7 @@ def cleanup_temporary_files():
                             subprocess.run(
                                 indexer_cmd,
                                 stdout=subprocess.DEVNULL,
-                                stderr=stderr_log_file,
+                                stderr=compression_log_file,
                                 check=True,
                                 env=indexer_env,
                             )
@@ -559,8 +567,8 @@ def cleanup_temporary_files():
         logger.debug("Compressed.")
     finally:
         cleanup_temporary_files()
-        stderr_log_file.close()
-        log_file_contents(logger, stderr_log_path)
+        compression_log_file.close()
+        log_file_contents(logger, compression_log_path)
 
     worker_output = {
         "total_uncompressed_size": total_uncompressed_size,
@@ -571,7 +579,7 @@ def cleanup_temporary_files():
         return CompressionTaskStatus.SUCCEEDED, worker_output
     error_msgs = []
     if compression_successful is False:
-        error_msgs.append(f"See logs {stderr_log_path}")
+        error_msgs.append(f"See logs {compression_log_path}")
     if s3_error is not None:
         error_msgs.append(s3_error)
     worker_output["error_message"] = "\n".join(error_msgs)