PBSProvider to report stdout/err paths when jobs fails (#3094)

* Updated PBSProvider to report job stdout/err via JobStatus when job reaches terminal state
* Adding a simple test to confirm that stdout/err paths are included in internal table

Author: yadudoc

parsl/providers/pbspro/pbspro.py

@@ -119,13 +119,17 @@ def _status(self):
 
                 job_state = job.get('job_state', JobState.UNKNOWN)
                 state = translate_table.get(job_state, JobState.UNKNOWN)
-                self.resources[job_id]['status'] = JobStatus(state)
+                self.resources[job_id]['status'] = JobStatus(state,
+                                                             stdout_path=self.resources[job_id]['job_stdout_path'],
+                                                             stderr_path=self.resources[job_id]['job_stderr_path'])
                 jobs_missing.remove(job_id)
 
         # squeue does not report on jobs that are not running. So we are filling in the
         # blanks for missing jobs, we might lose some information about why the jobs failed.
         for missing_job in jobs_missing:
-            self.resources[missing_job]['status'] = JobStatus(JobState.COMPLETED)
+            self.resources[missing_job]['status'] = JobStatus(JobState.COMPLETED,
+                                                              stdout_path=self.resources[missing_job]['job_stdout_path'],
+                                                              stderr_path=self.resources[missing_job]['job_stderr_path'])
 
     def submit(self, command, tasks_per_node, job_name="parsl"):
         """Submits the command job.
@@ -149,7 +153,11 @@ def submit(self, command, tasks_per_node, job_name="parsl"):
 
         job_name = "{0}.{1}".format(job_name, time.time())
 
-        script_path = os.path.abspath("{0}/{1}.submit".format(self.script_dir, job_name))
+        assert self.script_dir, "Expected script_dir to be set"
+        script_path = os.path.join(self.script_dir, job_name)
+        script_path = os.path.abspath(script_path)
+        job_stdout_path = script_path + ".stdout"
+        job_stderr_path = script_path + ".stderr"
 
         logger.debug("Requesting {} nodes_per_block, {} tasks_per_node".format(
             self.nodes_per_block, tasks_per_node)
@@ -163,6 +171,8 @@ def submit(self, command, tasks_per_node, job_name="parsl"):
         job_config["scheduler_options"] = self.scheduler_options
         job_config["worker_init"] = self.worker_init
         job_config["user_script"] = command
+        job_config["job_stdout_path"] = job_stdout_path
+        job_config["job_stderr_path"] = job_stderr_path
 
         # Add a colon to select_options if one isn't included
         if self.select_options and not self.select_options.startswith(":"):
@@ -194,7 +204,11 @@ def submit(self, command, tasks_per_node, job_name="parsl"):
             for line in stdout.split('\n'):
                 if line.strip():
                     job_id = line.strip()
-                    self.resources[job_id] = {'job_id': job_id, 'status': JobStatus(JobState.PENDING)}
+                    self.resources[job_id] = {'job_id': job_id,
+                                              'status': JobStatus(JobState.PENDING),
+                                              'job_stdout_path': job_stdout_path,
+                                              'job_stderr_path': job_stderr_path,
+                                              }
         else:
             message = "Command '{}' failed with return code {}".format(launch_cmd, retcode)
             if (stdout is not None) and (stderr is not None):

parsl/providers/pbspro/template.py

@@ -5,8 +5,8 @@
 #PBS -m n
 #PBS -l walltime=$walltime
 #PBS -l select=${nodes_per_block}:ncpus=${ncpus}${select_options}
-#PBS -o ${submit_script_dir}/${jobname}.submit.stdout
-#PBS -e ${submit_script_dir}/${jobname}.submit.stderr
+#PBS -o ${job_stdout_path}
+#PBS -e ${job_stderr_path}
 ${scheduler_options}
 
 ${worker_init}

parsl/tests/test_providers/test_pbspro_template.py

@@ -0,0 +1,28 @@
+import random
+
+from unittest import mock
+import pytest
+
+from parsl.channels import LocalChannel
+from parsl.providers import PBSProProvider
+
+
+@pytest.mark.local
+def test_submit_script_basic(tmp_path):
+    """Test slurm resources table"""
+
+    provider = PBSProProvider(
+        queue="debug", channel=LocalChannel(script_dir=tmp_path)
+    )
+    provider.script_dir = tmp_path
+    job_id = str(random.randint(55000, 59000))
+    provider.execute_wait = mock.Mock(spec=PBSProProvider.execute_wait)
+    provider.execute_wait.return_value = (0, job_id, "")
+    result_job_id = provider.submit("test", tasks_per_node=1)
+    assert job_id == result_job_id
+    provider.execute_wait.assert_called()
+    assert job_id in provider.resources
+
+    job_info = provider.resources[job_id]
+    assert "job_stdout_path" in job_info
+    assert "job_stderr_path" in job_info