Ensure job directory creation for various schedulers (#216)

* Ensure job directory creation for various schedulers

Signed-off-by: Hemil Desai <[email protected]>

* fix

Signed-off-by: Hemil Desai <[email protected]>

---------

Signed-off-by: Hemil Desai <[email protected]>

Author: hemildesai

nemo_run/core/execution/docker.py

@@ -350,6 +350,7 @@ def get_containers(self, client: "DockerClient") -> list["Container"]:
     def save(self) -> None:
         apps = {}
         if not os.path.isfile(DOCKER_JOB_DIRS):
+            os.makedirs(os.path.dirname(DOCKER_JOB_DIRS), exist_ok=True)
             Path(DOCKER_JOB_DIRS).touch()
 
         with open(DOCKER_JOB_DIRS, "r+") as f:

nemo_run/run/torchx_backend/schedulers/dgxcloud.py

@@ -204,6 +204,7 @@ def _save_job_dir(app_id: str, job_status: str, executor: DGXCloudExecutor) -> N
     original_apps = {}
     os.makedirs(os.path.dirname(DGX_JOB_DIRS), exist_ok=True)
     if not os.path.isfile(DGX_JOB_DIRS):
+        os.makedirs(os.path.dirname(DGX_JOB_DIRS), exist_ok=True)
         Path(DGX_JOB_DIRS).touch()
 
     serializer = ZlibJSONSerializer()

nemo_run/run/torchx_backend/schedulers/local.py

@@ -184,6 +184,7 @@ def create_scheduler(
 def _save_job_dir(apps: dict[str, _LocalAppDef]) -> None:
     original_apps = {}
     if not os.path.isfile(LOCAL_JOB_DIRS):
+        os.makedirs(os.path.dirname(LOCAL_JOB_DIRS), exist_ok=True)
         Path(LOCAL_JOB_DIRS).touch()
 
     with open(LOCAL_JOB_DIRS, "r+") as f:

nemo_run/run/torchx_backend/schedulers/skypilot.py

@@ -227,6 +227,7 @@ def create_scheduler(session_name: str, **kwargs: Any) -> SkypilotScheduler:
 def _save_job_dir(app_id: str, job_status: str, log_dir: str) -> None:
     original_apps = {}
     if not os.path.isfile(SKYPILOT_JOB_DIRS):
+        os.makedirs(os.path.dirname(SKYPILOT_JOB_DIRS), exist_ok=True)
         Path(SKYPILOT_JOB_DIRS).touch()
 
     with open(SKYPILOT_JOB_DIRS, "r+") as f:

nemo_run/run/torchx_backend/schedulers/slurm.py

@@ -374,6 +374,7 @@ def create_scheduler(session_name: str, **kwargs: Any) -> SlurmTunnelScheduler:
 def _save_job_dir(
     job_id: str, local_job_dir: str, tunnel: SSHTunnel | LocalTunnel, ls_term: str
 ) -> None:
+    os.makedirs(os.path.dirname(SLURM_JOB_DIRS), exist_ok=True)
     with open(SLURM_JOB_DIRS, "a+") as f:
         f.write(
             f"{job_id} = {ls_term},{local_job_dir},{tunnel.__class__.__name__},{json.dumps(asdict(tunnel))}\n"