feat(job-orchestration): Make stderr logs visible in container logs

- Dump subprocess stderr logs to container stderr after process completion
  in compression_task, query/utils, and reducer, so logs appear in
  `kubectl logs` / `docker compose logs` even on ephemeral storage
- Add container log handler (StreamHandler on fd 2) for Celery workers
  via after_setup_logger signal, bypassing Celery's -f redirect

Author: rishikeshdevsot

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

@@ -1,9 +1,18 @@
-from celery import Celery
+import logging
+
+from celery import Celery, signals
 
 from job_orchestration.executor.compress import celeryconfig
+from job_orchestration.executor.utils import add_container_log_handler
 
 app = Celery("compress")
 app.config_from_object(celeryconfig)
 
+
+@signals.after_setup_logger.connect
+def _on_after_setup_logger(logger: logging.Logger, **kwargs) -> None:
+    add_container_log_handler(logger)
+
+
 if "__main__" == __name__:
     app.start()

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

@@ -4,6 +4,7 @@
 import pathlib
 import shutil
 import subprocess
+import sys
 from contextlib import closing
 from typing import Any
 
@@ -559,6 +560,11 @@ def cleanup_temporary_files():
     finally:
         cleanup_temporary_files()
         stderr_log_file.close()
+        # Dump the stderr log to sys.stderr so it's visible in container logs
+        if stderr_log_path.stat().st_size > 0:
+            print(f"--- Contents of {stderr_log_path.name} ---", file=sys.stderr)
+            sys.stderr.write(stderr_log_path.read_text())
+            print(f"--- End of {stderr_log_path.name} ---", file=sys.stderr)
 
     worker_output = {
         "total_uncompressed_size": total_uncompressed_size,

components/job-orchestration/job_orchestration/executor/query/celery.py

@@ -1,9 +1,18 @@
-from celery import Celery
+import logging
+
+from celery import Celery, signals
 
 from job_orchestration.executor.query import celeryconfig
+from job_orchestration.executor.utils import add_container_log_handler
 
 app = Celery("query")
 app.config_from_object(celeryconfig)
 
+
+@signals.after_setup_logger.connect
+def _on_after_setup_logger(logger: logging.Logger, **kwargs) -> None:
+    add_container_log_handler(logger)
+
+
 if "__main__" == __name__:
     app.start()

components/job-orchestration/job_orchestration/executor/query/utils.py

@@ -88,6 +88,13 @@ def sigterm_handler(_signo, _stack_frame):
     # stdout/stderr.
     stdout_data, _ = task_proc.communicate()
     return_code = task_proc.returncode
+
+    # Dump the stderr log to sys.stderr so it's visible in container logs
+    if clo_log_path.stat().st_size > 0:
+        print(f"--- Contents of {clo_log_path.name} ---", file=sys.stderr)
+        sys.stderr.write(clo_log_path.read_text())
+        print(f"--- End of {clo_log_path.name} ---", file=sys.stderr)
+
     if 0 != return_code:
         task_status = QueryTaskStatus.FAILED
         logger.error(

components/job-orchestration/job_orchestration/executor/utils.py

@@ -1,13 +1,31 @@
 """Utilities shared by job-orchestration executors."""
 
+import logging
 import os
 from logging import Logger
 from pathlib import Path
 
+from celery.app.defaults import DEFAULT_PROCESS_LOG_FMT
 from clp_py_utils.clp_config import ClpConfig, WorkerConfig
 from clp_py_utils.core import read_yaml_config_file
 
 
+def add_container_log_handler(logger: logging.Logger) -> None:
+    """Adds a StreamHandler that writes to the container's real stderr (fd 2).
+
+    In Celery workers, ``-f`` redirects ``sys.stdout`` and ``sys.stderr`` to
+    the log file, so regular ``StreamHandler(sys.stderr)`` would write to the
+    file instead of the container's stderr.  This handler writes directly to
+    file descriptor 2, ensuring output reaches ``docker compose logs`` /
+    ``kubectl logs``.
+    """
+    # dup(2) so the handler owns its own fd and doesn't close the original
+    stderr_stream = os.fdopen(os.dup(2), "w")
+    handler = logging.StreamHandler(stderr_stream)
+    handler.setFormatter(logging.Formatter(DEFAULT_PROCESS_LOG_FMT))
+    logger.addHandler(handler)
+
+
 def load_clp_config_from_config_path_env_var() -> ClpConfig:
     """
     Loads the CLP config from `CLP_CONFIG_PATH`.

components/job-orchestration/job_orchestration/reducer/reducer.py

@@ -93,8 +93,14 @@ def main(argv: list[str]) -> int:
     for i, reducer in enumerate(reducers):
         reducer.communicate()
         logger.info(f"reducer-{i} exited with returncode={reducer.returncode}")
-    for reducer_log_file in reducer_log_files:
+    for i, reducer_log_file in enumerate(reducer_log_files):
         reducer_log_file.close()
+        # Dump the reducer log to sys.stderr so it's visible in container logs
+        log_file_path = logs_dir / f"reducer-{i}.log"
+        if log_file_path.stat().st_size > 0:
+            print(f"--- Contents of {log_file_path.name} ---", file=sys.stderr)
+            sys.stderr.write(log_file_path.read_text())
+            print(f"--- End of {log_file_path.name} ---", file=sys.stderr)
 
     logger.error("All reducers terminated")