Fix/job manager logger

python/ray/dashboard/modules/job/job_manager.py

@@ -30,6 +30,7 @@
 from ray.dashboard.modules.job.job_log_storage_client import JobLogStorageClient
 from ray.dashboard.modules.job.job_supervisor import JobSupervisor
 from ray.dashboard.modules.job.utils import get_head_node_id
+from ray.dashboard.utils import close_logger_file_descriptor
 from ray.exceptions import ActorUnschedulableError, RuntimeEnvSetupError
 from ray.job_submission import JobStatus
 from ray.runtime_env import RuntimeEnvConfig
@@ -505,6 +506,8 @@ async def submit_job(
                 "Please use a different submission_id."
             )
 
+        driver_logger = self._get_job_driver_logger(submission_id)
+        driver_logger.info("Runtime env is setting up.")
         # Wait for the actor to start up asynchronously so this call always
         # returns immediately and we can catch errors with the actor starting
         # up.
@@ -525,8 +528,6 @@ async def submit_job(
                     f"Started a ray job {submission_id}.", submission_id=submission_id
                 )
 
-            driver_logger = self._get_job_driver_logger(submission_id)
-            driver_logger.info("Runtime env is setting up.")
             supervisor = self._supervisor_actor_cls.options(
                 lifetime="detached",
                 name=JOB_ACTOR_NAME_TEMPLATE.format(job_id=submission_id),
@@ -559,8 +560,7 @@ async def submit_job(
             )
         except Exception as e:
             tb_str = traceback.format_exc()
-
-            logger.warning(
+            driver_logger.warning(
                 f"Failed to start supervisor actor for job {submission_id}: '{e}'"
                 f". Full traceback:\n{tb_str}"
             )
@@ -572,6 +572,8 @@ async def submit_job(
                     f". Full traceback:\n{tb_str}"
                 ),
             )
+        finally:
+            close_logger_file_descriptor(driver_logger)
 
         return submission_id
 

python/ray/dashboard/modules/job/utils.py

@@ -27,7 +27,6 @@
     Request = None
     Response = None
 
-
 logger = logging.getLogger(__name__)
 
 MAX_CHUNK_LINE_LENGTH = 10

python/ray/dashboard/tests/test_utils.py

@@ -0,0 +1,25 @@
+import logging
+import sys
+
+import pytest
+
+from ray.dashboard.utils import close_logger_file_descriptor
+
+
+def test_close_logger_file_descriptor():
+    logger_format = "%(message)s"
+    logger = logging.getLogger("test_job_id")
+
+    job_driver_log_path = "/tmp/ray.log"
+    job_driver_handler = logging.FileHandler(job_driver_log_path)
+    job_driver_formatter = logging.Formatter(logger_format)
+    job_driver_handler.setFormatter(job_driver_formatter)
+    logger.addHandler(job_driver_handler)
+
+    assert job_driver_handler._closed is False
+    close_logger_file_descriptor(logger)
+    assert job_driver_handler._closed is True
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main(["-v", __file__]))

python/ray/dashboard/utils.py

@@ -692,3 +692,8 @@ def compose_state_message(
         else:
             state_message = death_reason_message
     return state_message
+
+
+def close_logger_file_descriptor(logger_instance):
+    for handler in logger_instance.handlers:
+        handler.close()