You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by di...@apache.org on 2026/09/19 14:27:41 UTC
(airavata) branch refactor-single-module updated: Parsing batch job id from ssh submission
This is an automated email from the ASF dual-hosted git repository.
DImuthuUpe pushed a commit to branch refactor-single-module
in repository https://gitbox.apache.org/repos/asf/airavata.git
The following commit(s) were added to refs/heads/refactor-single-module by this push:
new 460957f2a4 Parsing batch job id from ssh submission
460957f2a4 is described below
commit 460957f2a445ac7d66cc7129549d7e5cfee2e081
Author: DImuthuUpe <di...@gmail.com>
AuthorDate: Sat Sep 19 10:27:34 2026 -0400
Parsing batch job id from ssh submission
---
internal/orchestration/batch.go | 76 +++++++++++++++++++++++++++++++++---
internal/orchestration/batch_test.go | 45 +++++++++++++++++++++
internal/orchestration/ssh.go | 30 ++++++++------
3 files changed, 134 insertions(+), 17 deletions(-)
diff --git a/internal/orchestration/batch.go b/internal/orchestration/batch.go
index 7f93ff12d7..a9f0119735 100644
--- a/internal/orchestration/batch.go
+++ b/internal/orchestration/batch.go
@@ -5,6 +5,9 @@ import (
"fmt"
"log/slog"
"os"
+ "path"
+ "regexp"
+ "strings"
applicationmodel "github.com/apache/airavata/api/application/model"
model "github.com/apache/airavata/api/process/model"
@@ -56,24 +59,85 @@ func (a *ExecutionEngine) submitBatchJob(ctx context.Context, executionContext *
slog.Info("Retrieved slurm cluster config for batch job", "taskId", taskID, "processId", processID, "clusterConfigId", clusterConfig.ID)
- scriptUploadPath := *jst.WorkingDir + "/" + "script.slurm"
+ if jst.WorkingDir == nil || strings.TrimSpace(*jst.WorkingDir) == "" {
+ err := fmt.Errorf("job submission task %s of process %s names no working directory", taskID, processID)
+ slog.Error("Failed to submit batch job", "taskId", taskID, "processId", processID, "error", err)
+ return nil, err
+ }
+
+ scriptUploadPath := path.Join(*jst.WorkingDir, "script.slurm")
slog.Info("Uploading slurm script to cluster", "taskId", taskID, "processId", processID, "scriptPath", scriptPath, "remotePath", scriptUploadPath)
- UploadFileToSCP(ctx, clusterConfig.SlurmCluster.HeadnodeHost,
+ if err := UploadFileToSCP(ctx, clusterConfig.SlurmCluster.HeadnodeHost,
clusterConfig.SlurmCluster.HeadnodePort, clusterConfig.LoginUser,
- *clusterConfig.SSHKey, scriptPath, scriptUploadPath)
+ *clusterConfig.SSHKey, scriptPath, scriptUploadPath); err != nil {
+ slog.Error("Failed to upload slurm script to cluster", "taskId", taskID, "processId", processID, "remotePath", scriptUploadPath, "error", err)
+ return nil, err
+ }
slog.Info("Uploaded slurm script to cluster", "taskId", taskID, "processId", processID, "scriptPath", scriptPath, "remotePath", scriptUploadPath)
slog.Info("Submitting slurm script to cluster", "taskId", taskID, "processId", processID, "remotePath", scriptUploadPath)
- runSSHCommand(ctx, clusterConfig.SlurmCluster.HeadnodeHost,
+ // sbatch is run from the working directory so that a script writing relative paths —
+ // and the job's own stdout and stderr files — land beside the staged inputs rather
+ // than in the login user's home.
+ submitCommand := "cd " + shellQuote(*jst.WorkingDir) + " && sbatch " + shellQuote(scriptUploadPath)
+ stdout, stderr, err := runSSHCommand(ctx, clusterConfig.SlurmCluster.HeadnodeHost,
clusterConfig.SlurmCluster.HeadnodePort, clusterConfig.LoginUser,
- *clusterConfig.SSHKey, "sbatch"+" "+shellQuote(scriptUploadPath),
+ *clusterConfig.SSHKey, submitCommand,
fmt.Sprintf("ssh sbatch %s@%s:%s", clusterConfig.LoginUser, clusterConfig.SlurmCluster.HeadnodeHost, scriptUploadPath))
+ if err != nil {
+ slog.Error("Failed to submit slurm script to cluster", "taskId", taskID, "processId", processID, "remotePath", scriptUploadPath, "stdout", stdout, "stderr", stderr, "error", err)
+ return nil, err
+ }
+
+ jobID, err := parseSbatchJobID(stdout)
+ if err != nil {
+ // The job may well be queued: sbatch exited zero, so something was accepted and
+ // only its id was not understood. Failing loudly is still right — without an id
+ // nothing downstream can monitor or cancel it — but the output is logged so the
+ // run can be reconciled by hand.
+ slog.Error("Failed to read job id from sbatch output", "taskId", taskID, "processId", processID, "stdout", stdout, "stderr", stderr, "error", err)
+ return nil, err
+ }
- slog.Info("Completed submitting batch job", "taskId", taskID, "processId", processID, "deploymentId", process.BatchProcess.DeploymentID, "JST Id", jst.ID)
+ slog.Info("Submitted batch job", "jobId", jobID, "taskId", taskID, "processId", processID, "jobId", jobID)
+
+ // Recorded in both places the rest of the run reads it from: on the task that did the
+ // submitting, and on the batch process, which is what monitoring and cancellation
+ // look at.
+ jst.JobId = &jobID
+ if err := a.jobSubmissionTasks.Save(ctx, jst); err != nil {
+ slog.Error("Failed to record job id on job submission task", "taskId", taskID, "processId", processID, "jobId", jobID, "error", err)
+ return nil, err
+ }
+
+ process.BatchProcess.JobID = &jobID
+ if err := a.processes.SaveBatchProcess(ctx, process.BatchProcess); err != nil {
+ slog.Error("Failed to record job id on batch process", "taskId", taskID, "processId", processID, "jobId", jobID, "error", err)
+ return nil, err
+ }
+
+ slog.Info("Completed submitting batch job", "taskId", taskID, "processId", processID, "deploymentId", process.BatchProcess.DeploymentID, "JST Id", jst.ID, "jobId", jobID)
return executionContext, nil
}
+// sbatchJobID matches the line sbatch prints when it accepts a script: "Submitted batch
+// job 4242", which on a federated install carries a trailing " on cluster <name>".
+var sbatchJobID = regexp.MustCompile(`Submitted batch job (\d+)`)
+
+// parseSbatchJobID reads the scheduler's job id out of what sbatch printed.
+//
+// The line is searched for rather than the whole output matched: a login shell is free
+// to print a banner, a module load notice or a warning before it, and none of that makes
+// the submission any less successful.
+func parseSbatchJobID(out string) (string, error) {
+ match := sbatchJobID.FindStringSubmatch(out)
+ if match == nil {
+ return "", fmt.Errorf("sbatch announced no job id, and its output was %q", strings.TrimSpace(out))
+ }
+ return match[1], nil
+}
+
// slurmScript gathers what a submission script is built from and renders it.
//
// Each reference is optional on the record it is read from, so it is checked before it
diff --git a/internal/orchestration/batch_test.go b/internal/orchestration/batch_test.go
new file mode 100644
index 0000000000..eb4f153917
--- /dev/null
+++ b/internal/orchestration/batch_test.go
@@ -0,0 +1,45 @@
+package orchestration
+
+import "testing"
+
+// The id has to survive whatever a login shell prints around it, and a submission that
+// announced no id at all has to be a failure rather than an empty job id recorded
+// against the run.
+func TestParseSbatchJobID(t *testing.T) {
+ cases := []struct {
+ name string
+ out string
+ want string
+ wantErr bool
+ }{
+ {name: "plain", out: "Submitted batch job 4242\n", want: "4242"},
+ {name: "no trailing newline", out: "Submitted batch job 4242", want: "4242"},
+ {name: "federated cluster", out: "Submitted batch job 4242 on cluster delta\n", want: "4242"},
+ {
+ name: "after shell noise",
+ out: "Lmod is automatically replacing intel with gcc.\nSubmitted batch job 987654\n",
+ want: "987654",
+ },
+ {name: "empty", out: "", wantErr: true},
+ {name: "unrecognised", out: "sbatch: error: Batch job submission failed: Invalid account\n", wantErr: true},
+ {name: "no digits", out: "Submitted batch job\n", wantErr: true},
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ got, err := parseSbatchJobID(tc.out)
+ if tc.wantErr {
+ if err == nil {
+ t.Fatalf("parseSbatchJobID(%q) = %q, want an error", tc.out, got)
+ }
+ return
+ }
+ if err != nil {
+ t.Fatalf("parseSbatchJobID(%q) returned %v", tc.out, err)
+ }
+ if got != tc.want {
+ t.Errorf("parseSbatchJobID(%q) = %q, want %q", tc.out, got, tc.want)
+ }
+ })
+ }
+}
diff --git a/internal/orchestration/ssh.go b/internal/orchestration/ssh.go
index 396247cda6..0d84a0bd50 100644
--- a/internal/orchestration/ssh.go
+++ b/internal/orchestration/ssh.go
@@ -91,7 +91,7 @@ func CreateDirectorySSH(ctx context.Context, host string, port int, username str
return fmt.Errorf("%s: no remote directory was named", target)
}
- if err := runSSHCommand(ctx, host, port, username, key, "mkdir -p "+shellQuote(remotePath), target); err != nil {
+ if _, _, err := runSSHCommand(ctx, host, port, username, key, "mkdir -p "+shellQuote(remotePath), target); err != nil {
return err
}
@@ -135,22 +135,30 @@ func GetSCPFileMetadata(ctx context.Context, host string, port int, username str
}, nil
}
-// runSSHCommand runs one command on the host and waits for it to finish, folding
-// whatever the command wrote to stderr into the error when it fails.
-func runSSHCommand(ctx context.Context, host string, port int, username string, key credmodel.SSHKey, command, target string) error {
+// runSSHCommand runs one command on the host and waits for it to finish, returning
+// what it wrote to stdout and to stderr.
+//
+// Both streams are returned whether the command succeeded or not: what a failing
+// command printed is usually what explains it, and a caller that reads the output —
+// the job id sbatch announces, say — still wants to see what came back. Whatever went
+// to stderr is folded into the error as well, so a caller that only checks err is no
+// worse off than before. The streams are returned as the command wrote them, trailing
+// newline included; a caller parsing them should trim.
+func runSSHCommand(ctx context.Context, host string, port int, username string, key credmodel.SSHKey, command, target string) (string, string, error) {
client, err := dialSSH(ctx, host, port, username, key)
if err != nil {
- return fmt.Errorf("%s: %w", target, err)
+ return "", "", fmt.Errorf("%s: %w", target, err)
}
defer client.Close()
session, err := client.NewSession()
if err != nil {
- return fmt.Errorf("%s: opening session: %w", target, err)
+ return "", "", fmt.Errorf("%s: opening session: %w", target, err)
}
defer session.Close()
- var stderr bytes.Buffer
+ var stdout, stderr bytes.Buffer
+ session.Stdout = &stdout
session.Stderr = &stderr
done := make(chan struct{})
@@ -165,14 +173,14 @@ func runSSHCommand(ctx context.Context, host string, port int, username string,
if err := session.Run(command); err != nil {
if ctxErr := ctx.Err(); ctxErr != nil {
- return fmt.Errorf("%s: %w", target, ctxErr)
+ return stdout.String(), stderr.String(), fmt.Errorf("%s: %w", target, ctxErr)
}
if remote := strings.TrimSpace(stderr.String()); remote != "" {
- return fmt.Errorf("%s: %w (remote: %s)", target, err, remote)
+ return stdout.String(), stderr.String(), fmt.Errorf("%s: %w (remote: %s)", target, err, remote)
}
- return fmt.Errorf("%s: %w", target, err)
+ return stdout.String(), stderr.String(), fmt.Errorf("%s: %w", target, err)
}
- return nil
+ return stdout.String(), stderr.String(), nil
}
// ========================== Utility section =====================================