fix: fix incorrect error messages (#483)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit


- **New Features**
- Enhanced error messages for command execution failures, providing
clearer context by including the specific command that failed.
	
- **Bug Fixes**
- Improved error reporting in multiple command handling methods, aiding
in quicker identification of issues during command execution.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Signed-off-by: Jinzhe Zeng <jinzhe.zeng@rutgers.edu>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: njzjz

dpdispatcher/machines/JH_UniScheduler.py

@@ -105,7 +105,7 @@ def check_status(self, job):
         elif ret != 0:
             # just retry when any unknown error raised.
             raise RetrySignal(
-                "Get error code %d in checking status through ssh with job: %s . message: %s"
+                "Get error code %d in checking status with job: %s . message: %s"
                 % (ret, job.job_hash, err_str)
             )
         status_out = stdout.read().decode("utf-8").split("\n")

dpdispatcher/machines/distributed_shell.py

@@ -181,8 +181,8 @@ def do_submit(self, job):
         if ret != 0:
             err_str = stderr.decode("utf-8")
             raise RuntimeError(
-                "Command squeue fails to execute, error message:%s\nreturn code %d\n"
-                % (err_str, ret)
+                "Command %s fails to execute, error message:%s\nreturn code %d\n"
+                % (cmd, err_str, ret)
             )
         job_id = int(stdout.decode("utf-8").strip())
 

dpdispatcher/machines/lsf.py

@@ -129,7 +129,7 @@ def check_status(self, job):
         elif ret != 0:
             # just retry when any unknown error raised.
             raise RetrySignal(
-                "Get error code %d in checking status through ssh with job: %s . message: %s"
+                "Get error code %d in checking status with job: %s . message: %s"
                 % (ret, job.job_hash, err_str)
             )
         status_out = stdout.read().decode("utf-8").split("\n")

dpdispatcher/machines/pbs.py

@@ -76,7 +76,8 @@ def check_status(self, job):
         job_id = job.job_id
         if job_id == "":
             return JobStatus.unsubmitted
-        ret, stdin, stdout, stderr = self.context.block_call("qstat -x " + job_id)
+        command = "qstat -x " + job_id
+        ret, stdin, stdout, stderr = self.context.block_call(command)
         err_str = stderr.read().decode("utf-8")
         if ret != 0:
             if "qstat: Unknown Job Id" in err_str or "Job has finished" in err_str:
@@ -86,8 +87,8 @@ def check_status(self, job):
                     return JobStatus.terminated
             else:
                 raise RuntimeError(
-                    "status command qstat fails to execute. erro info: %s return code %d"
-                    % (err_str, ret)
+                    "status command %s fails to execute. erro info: %s return code %d"
+                    % (command, err_str, ret)
                 )
         status_line = stdout.read().decode("utf-8").split("\n")[-2]
         status_word = status_line.split()[-2]
@@ -126,7 +127,8 @@ def check_status(self, job):
         job_id = job.job_id
         if job_id == "":
             return JobStatus.unsubmitted
-        ret, stdin, stdout, stderr = self.context.block_call("qstat -l " + job_id)
+        command = "qstat -l " + job_id
+        ret, stdin, stdout, stderr = self.context.block_call(command)
         err_str = stderr.read().decode("utf-8")
         if ret != 0:
             if "qstat: Unknown Job Id" in err_str or "Job has finished" in err_str:
@@ -136,8 +138,8 @@ def check_status(self, job):
                     return JobStatus.terminated
             else:
                 raise RuntimeError(
-                    "status command qstat fails to execute. erro info: %s return code %d"
-                    % (err_str, ret)
+                    "status command %s fails to execute. erro info: %s return code %d"
+                    % (command, err_str, ret)
                 )
         status_line = stdout.read().decode("utf-8").split("\n")[-2]
         status_word = status_line.split()[-2]
@@ -263,11 +265,12 @@ def check_status(self, job):
         status_line = None
         if job_id == "":
             return JobStatus.unsubmitted
-        ret, stdin, stdout, stderr = self.context.block_call("qstat")
+        command = "qstat"
+        ret, stdin, stdout, stderr = self.context.block_call(command)
         err_str = stderr.read().decode("utf-8")
         if ret != 0:
             raise RuntimeError(
-                f"status command qstat fails to execute. erro info: {err_str} return code {ret}"
+                f"status command {command} fails to execute. erro info: {err_str} return code {ret}"
             )
         status_text_list = stdout.read().decode("utf-8").split("\n")
         for txt in status_text_list:

dpdispatcher/machines/shell.py

@@ -38,14 +38,13 @@ def do_submit(self, job):
         script_run_str = self.gen_script_command(job)
         script_run_file_name = f"{job.script_file_name}.run"
         self.context.write_file(fname=script_run_file_name, write_str=script_run_str)
-        ret, stdin, stdout, stderr = self.context.block_call(
-            f"cd {shlex.quote(self.context.remote_root)} && {{ nohup bash {script_file_name} 1>>{output_name} 2>>{output_name} & }} && echo $!"
-        )
+        cmd = f"cd {shlex.quote(self.context.remote_root)} && {{ nohup bash {script_file_name} 1>>{output_name} 2>>{output_name} & }} && echo $!"
+        ret, stdin, stdout, stderr = self.context.block_call(cmd)
         if ret != 0:
             err_str = stderr.read().decode("utf-8")
             raise RuntimeError(
-                "status command squeue fails to execute\nerror message:%s\nreturn code %d\n"
-                % (err_str, ret)
+                "status command %s fails to execute\nerror message:%s\nreturn code %d\n"
+                % (cmd, err_str, ret)
             )
         job_id = int(stdout.read().decode("utf-8").strip())
         self.context.write_file(job_id_name, str(job_id))
@@ -73,15 +72,16 @@ def check_status(self, job):
             return JobStatus.unsubmitted
 
         # mark defunct process as terminated
-        ret, stdin, stdout, stderr = self.context.block_call(
+        cmd = (
             r"""command -v ps >/dev/null 2>&1 || { echo >&2 "I require ps but it's not installed. Aborting."; exit 1; };"""
             f"if ps -p {job_id} > /dev/null && ! (ps -o command -p {job_id} | grep defunct >/dev/null) ; then echo 1; fi"
         )
+        ret, stdin, stdout, stderr = self.context.block_call(cmd)
         if ret != 0:
             err_str = stderr.read().decode("utf-8")
             raise RuntimeError(
-                "status command squeue fails to execute\nerror message:%s\nreturn code %d\n"
-                % (err_str, ret)
+                "status command %s fails to execute\nerror message:%s\nreturn code %d\n"
+                % (cmd, err_str, ret)
             )
 
         if_job_exists = bool(stdout.read().decode("utf-8").strip())

dpdispatcher/machines/slurm.py

@@ -83,13 +83,12 @@ def do_submit(self, job):
         script_run_file_name = f"{job.script_file_name}.run"
         self.context.write_file(fname=script_run_file_name, write_str=script_run_str)
         # self.context.write_file(fname=os.path.join(self.context.submission.work_base, script_file_name), write_str=script_str)
-        ret, stdin, stdout, stderr = self.context.block_call(
-            "cd {} && {} {}".format(
-                shlex.quote(self.context.remote_root),
-                "sbatch",
-                shlex.quote(script_file_name),
-            )
+        command = "cd {} && {} {}".format(
+            shlex.quote(self.context.remote_root),
+            "sbatch",
+            shlex.quote(script_file_name),
         )
+        ret, stdin, stdout, stderr = self.context.block_call(command)
         if ret != 0:
             err_str = stderr.read().decode("utf-8")
             if (
@@ -98,7 +97,7 @@ def do_submit(self, job):
             ):
                 # server network error, retry 3 times
                 raise RetrySignal(
-                    "Get error code %d in submitting through ssh with job: %s . message: %s"
+                    "Get error code %d in submitting with job: %s . message: %s"
                     % (ret, job.job_hash, err_str)
                 )
             elif (
@@ -110,8 +109,8 @@ def do_submit(self, job):
                 # job number exceeds, skip the submitting
                 return ""
             raise RuntimeError(
-                "status command squeue fails to execute\nerror message:%s\nreturn code %d\n"
-                % (err_str, ret)
+                "command %s fails to execute\nerror message:%s\nreturn code %d\n"
+                % (command, err_str, ret)
             )
         subret = stdout.readlines()
         # --parsable
@@ -129,9 +128,8 @@ def check_status(self, job):
         job_id = job.job_id
         if job_id == "":
             return JobStatus.unsubmitted
-        ret, stdin, stdout, stderr = self.context.block_call(
-            'squeue -o "%.18i %.2t" -j ' + job_id
-        )
+        command = 'squeue -o "%.18i %.2t" -j ' + job_id
+        ret, stdin, stdout, stderr = self.context.block_call(command)
         if ret != 0:
             err_str = stderr.read().decode("utf-8")
             if "Invalid job id specified" in err_str:
@@ -147,13 +145,13 @@ def check_status(self, job):
             ):
                 # retry 3 times
                 raise RetrySignal(
-                    "Get error code %d in checking status through ssh with job: %s . message: %s"
+                    "Get error code %d in checking status with job: %s . message: %s"
                     % (ret, job.job_hash, err_str)
                 )
             raise RuntimeError(
-                "status command squeue fails to execute."
+                "status command %s fails to execute."
                 "job_id:%s \n error message:%s\n return code %d\n"
-                % (job_id, err_str, ret)
+                % (command, job_id, err_str, ret)
             )
         status_line = stdout.read().decode("utf-8").split("\n")[-2]
         status_word = status_line.split()[-1]
@@ -319,9 +317,8 @@ def check_status(self, job):
         job_id = job.job_id
         if job_id == "":
             return JobStatus.unsubmitted
-        ret, stdin, stdout, stderr = self.context.block_call(
-            'squeue -h -o "%.18i %.2t" -j ' + job_id
-        )
+        command = 'squeue -h -o "%.18i %.2t" -j ' + job_id
+        ret, stdin, stdout, stderr = self.context.block_call(command)
         if ret != 0:
             err_str = stderr.read().decode("utf-8")
             if "Invalid job id specified" in err_str:
@@ -336,13 +333,13 @@ def check_status(self, job):
             ):
                 # retry 3 times
                 raise RetrySignal(
-                    "Get error code %d in checking status through ssh with job: %s . message: %s"
+                    "Get error code %d in checking status with job: %s . message: %s"
                     % (ret, job.job_hash, err_str)
                 )
             raise RuntimeError(
-                "status command squeue fails to execute."
+                "status command %s fails to execute."
                 "job_id:%s \n error message:%s\n return code %d\n"
-                % (job_id, err_str, ret)
+                % (command, job_id, err_str, ret)
             )
         status_lines = stdout.read().decode("utf-8").split("\n")[:-1]
         status = []