(airavata) branch refactor-single-module updated: Create ssh data dir before moving data

[di...@apache.org]

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 37a2d299a0 Create ssh data dir before moving data
37a2d299a0 is described below

commit 37a2d299a0e5c6a3f5e11ae65630ce40627def60
Author: DImuthuUpe <di...@gmail.com>
AuthorDate: Sat Sep 12 00:28:53 2026 -0400

    Create ssh data dir before moving data
---
 internal/orchestration/data.go | 16 ++++++++++++
 internal/orchestration/ssh.go  | 57 ++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 73 insertions(+)

diff --git a/internal/orchestration/data.go b/internal/orchestration/data.go
index 4e6e12ab5e..fd619cfa55 100644
--- a/internal/orchestration/data.go
+++ b/internal/orchestration/data.go
@@ -6,6 +6,7 @@ import (
 	"os"
 
 	"github.com/google/uuid"
+	"path/filepath"
 
 	datamodel "github.com/apache/airavata/api/data/model"
 )
@@ -67,6 +68,13 @@ func (a *ExecutionEngine) copyData(ctx context.Context, executionContext *Execut
 			return nil, err
 		}
 
+		// Ensure the destination directory exists before attempting the SCP upload.
+		if err := CreateDirectorySSH(ctx, *s.HostName, *s.Port, *s.LoginUser, *s.SSHKey, filepath.Dir(*dataStagingTask.DestinationPath)); err != nil {
+			slog.Error("Failed to create destination directory through SSH", "taskId", dataStagingTask.ID, "processId", *dataStagingTask.ProcessID,
+				"destinationDirectory", filepath.Dir(*dataStagingTask.DestinationPath), "error", err)
+			return nil, err
+		}
+
 		err = UploadFileToSCP(ctx, *s.HostName, *s.Port, *s.LoginUser, *s.SSHKey, localTempDataPath, *dataStagingTask.DestinationPath)
 		if err != nil {
 			slog.Error("Failed to upload file through SCP", "taskId", dataStagingTask.ID, "processId", *dataStagingTask.ProcessID,
@@ -84,6 +92,14 @@ func (a *ExecutionEngine) copyData(ctx context.Context, executionContext *Execut
 			return nil, err
 		}
 
+		// Ensure the destination directory exists before attempting the SCP upload.
+		if err := CreateDirectorySSH(ctx, s.SlurmCluster.HeadnodeHost, s.SlurmCluster.HeadnodePort, s.LoginUser, *s.SSHKey,
+			filepath.Dir(*dataStagingTask.DestinationPath)); err != nil {
+			slog.Error("Failed to create destination directory on HPC through SSH", "taskId", dataStagingTask.ID, "processId", *dataStagingTask.ProcessID,
+				"destinationDirectory", filepath.Dir(*dataStagingTask.DestinationPath), "error", err)
+			return nil, err
+		}
+
 		err = UploadFileToSCP(ctx, s.SlurmCluster.HeadnodeHost, s.SlurmCluster.HeadnodePort, s.LoginUser,
 			*s.SSHKey, localTempDataPath, *dataStagingTask.DestinationPath)
 		if err != nil {
diff --git a/internal/orchestration/ssh.go b/internal/orchestration/ssh.go
index 75d94c8891..35009f6f7a 100644
--- a/internal/orchestration/ssh.go
+++ b/internal/orchestration/ssh.go
@@ -82,6 +82,23 @@ func UploadFileToSCP(ctx context.Context, host string, port int, username string
 	return nil
 }
 
+// Creates remotePath on the host, including any missing parent directories, and does
+// nothing if it is already there.
+func CreateDirectorySSH(ctx context.Context, host string, port int, username string, key credmodel.SSHKey, remotePath string) error {
+	target := fmt.Sprintf("ssh mkdir %s@%s:%s", username, host, remotePath)
+
+	if strings.TrimSpace(remotePath) == "" {
+		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 {
+		return err
+	}
+
+	slog.Info("Created directory over SSH", "host", host, "remotePath", remotePath)
+	return nil
+}
+
 // Reads the mode, size and name a remote scp announces for a single remote file,
 // without transferring its contents.
 func GetSCPFileMetadata(ctx context.Context, host string, port int, username string, key credmodel.SSHKey, remotePath string) (*FileMetadata, error) {
@@ -130,6 +147,46 @@ const scpOK = 0x00
 // attempt quickly rather than hold a worker slot until the OS gives up on the socket.
 const scpDialTimeout = 30 * time.Second
 
+// 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 {
+	client, err := dialSSH(ctx, host, port, username, key)
+	if err != nil {
+		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)
+	}
+	defer session.Close()
+
+	var stderr bytes.Buffer
+	session.Stderr = &stderr
+
+	done := make(chan struct{})
+	defer close(done)
+	go func() {
+		select {
+		case <-ctx.Done():
+			session.Close()
+		case <-done:
+		}
+	}()
+
+	if err := session.Run(command); err != nil {
+		if ctxErr := ctx.Err(); ctxErr != nil {
+			return fmt.Errorf("%s: %w", target, ctxErr)
+		}
+		if remote := strings.TrimSpace(stderr.String()); remote != "" {
+			return fmt.Errorf("%s: %w (remote: %s)", target, err, remote)
+		}
+		return fmt.Errorf("%s: %w", target, err)
+	}
+	return nil
+}
+
 type scpTransfer struct {
 	client  *ssh.Client
 	session *ssh.Session