Simplified the logic

Made all logs go to the container
Used logger statements instead of stderr

Author: rishikeshdevsot

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

@@ -1,18 +1,9 @@
-import logging
-
-from celery import Celery, signals
+from celery import Celery
 
 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: object) -> None:
-    add_container_log_handler(logger)
-
-
 if "__main__" == __name__:
     app.start()

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

@@ -4,7 +4,6 @@
 import pathlib
 import shutil
 import subprocess
-import sys
 from contextlib import closing
 from typing import Any
 
@@ -560,12 +559,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
+        # Dump the stderr log so it's visible in container logs
         if stderr_log_path.stat().st_size > 0:
-            sys.stderr.write(
-                f"--- Contents of {stderr_log_path.name} ---\n"
+            logger.error(
+                f"Contents of {stderr_log_path.name}:\n"
                 f"{stderr_log_path.read_text()}"
-                f"--- End of {stderr_log_path.name} ---\n"
             )
 
     worker_output = {

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

@@ -1,18 +1,9 @@
-import logging
-
-from celery import Celery, signals
+from celery import Celery
 
 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: object) -> None:
-    add_container_log_handler(logger)
-
-
 if "__main__" == __name__:
     app.start()

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

@@ -89,12 +89,11 @@ def sigterm_handler(_signo, _stack_frame):
     stdout_data, _ = task_proc.communicate()
     return_code = task_proc.returncode
 
-    # Dump the stderr log to sys.stderr so it's visible in container logs
+    # Dump the stderr log so it's visible in container logs
     if clo_log_path.stat().st_size > 0:
-        sys.stderr.write(
-            f"--- Contents of {clo_log_path.name} ---\n"
+        logger.error(
+            f"Contents of {clo_log_path.name}:\n"
             f"{clo_log_path.read_text()}"
-            f"--- End of {clo_log_path.name} ---\n"
         )
 
     if 0 != return_code:

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

@@ -1,32 +1,13 @@
 """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:
-    """
-    Add 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

@@ -95,13 +95,12 @@ def main(argv: list[str]) -> int:
         logger.info(f"reducer-{i} exited with returncode={reducer.returncode}")
     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
+        # Dump the reducer log so it's visible in container logs
         log_file_path = logs_dir / f"reducer-{i}.log"
         if log_file_path.stat().st_size > 0:
-            sys.stderr.write(
-                f"--- Contents of {log_file_path.name} ---\n"
+            logger.info(
+                f"Contents of {log_file_path.name}:\n"
                 f"{log_file_path.read_text()}"
-                f"--- End of {log_file_path.name} ---\n"
             )
 
     logger.error("All reducers terminated")

tools/deployment/package/docker-compose-all.yaml

@@ -311,7 +311,7 @@ services:
       CLP_HOME: "/opt/clp"
       CLP_LOGGING_LEVEL: "${CLP_COMPRESSION_WORKER_LOGGING_LEVEL:-INFO}"
       CLP_LOGS_DIR: "/var/log/compression_worker"
-      CLP_WORKER_LOG_PATH: "/var/log/compression_worker/worker.log"
+      CLP_WORKER_LOG_PATH: "/dev/stdout"
       PYTHONPATH: "/opt/clp/lib/python3/site-packages"
       RESULT_BACKEND: "redis://default:${CLP_REDIS_PASS:?Please set a value.}\
         @redis:${CLP_REDIS_CONNECT_PORT:-6379}\
@@ -332,7 +332,6 @@ services:
       "worker",
       "--concurrency", "${CLP_COMPRESSION_WORKER_CONCURRENCY:-1}",
       "--loglevel", "WARNING",
-      "-f", "/var/log/compression_worker/worker.log",
       "-Q", "compression",
       "-n", "compression-worker@%h"
     ]
@@ -345,7 +344,7 @@ services:
       CLP_HOME: "/opt/clp"
       CLP_LOGGING_LEVEL: "${CLP_COMPRESSION_WORKER_LOGGING_LEVEL:-INFO}"
       CLP_LOGS_DIR: "/var/log/compression_worker"
-      CLP_WORKER_LOG_PATH: "/var/log/compression_worker/worker.log"
+      CLP_WORKER_LOG_PATH: "/dev/stdout"
       PYTHONPATH: "/opt/clp/lib/python3/site-packages"
       SPIDER_LOG_DIR: "/var/log/compression_worker"
     volumes:
@@ -499,7 +498,7 @@ services:
       CLP_HOME: "/opt/clp"
       CLP_LOGGING_LEVEL: "${CLP_QUERY_WORKER_LOGGING_LEVEL:-INFO}"
       CLP_LOGS_DIR: "/var/log/query_worker"
-      CLP_WORKER_LOG_PATH: "/var/log/query_worker/worker.log"
+      CLP_WORKER_LOG_PATH: "/dev/stdout"
       PYTHONPATH: "/opt/clp/lib/python3/site-packages"
       RESULT_BACKEND: "redis://default:${CLP_REDIS_PASS:?Please set a value.}\
         @redis:${CLP_REDIS_CONNECT_PORT:-6379}\
@@ -519,7 +518,6 @@ services:
       "worker",
       "--concurrency", "${CLP_QUERY_WORKER_CONCURRENCY:-1}",
       "--loglevel", "WARNING",
-      "-f", "/var/log/query_worker/worker.log",
       "-Q", "query",
       "-n", "query-worker@%h"
     ]