You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by sh...@apache.org on 2016/10/27 15:34:15 UTC
airavata git commit: Refactored RemoteCluster copyTo,
copyFrom and makeDirectory
Repository: airavata
Updated Branches:
refs/heads/develop b4fd4ff8c -> c50470c13
Refactored RemoteCluster copyTo, copyFrom and makeDirectory
Project: http://git-wip-us.apache.org/repos/asf/airavata/repo
Commit: http://git-wip-us.apache.org/repos/asf/airavata/commit/c50470c1
Tree: http://git-wip-us.apache.org/repos/asf/airavata/tree/c50470c1
Diff: http://git-wip-us.apache.org/repos/asf/airavata/diff/c50470c1
Branch: refs/heads/develop
Commit: c50470c137e1883a3c222514bf1be3c84fffca85
Parents: b4fd4ff
Author: Shameera Rathnayaka <sh...@gmail.com>
Authored: Thu Oct 27 11:33:34 2016 -0400
Committer: Shameera Rathnayaka <sh...@gmail.com>
Committed: Thu Oct 27 11:33:34 2016 -0400
----------------------------------------------------------------------
.../gfac/core/cluster/RemoteCluster.java | 6 +-
.../airavata/gfac/impl/BESRemoteCluster.java | 6 +-
.../airavata/gfac/impl/HPCRemoteCluster.java | 19 +-
.../airavata/gfac/impl/LocalRemoteCluster.java | 9 +-
.../org/apache/airavata/gfac/impl/SSHUtils.java | 198 ++++++++++---------
.../airavata/gfac/impl/task/ArchiveTask.java | 9 +-
.../gfac/impl/task/BESJobSubmissionTask.java | 5 +-
.../airavata/gfac/impl/task/DataStageTask.java | 13 +-
.../gfac/impl/task/EnvironmentSetupTask.java | 4 +-
.../gfac/impl/task/SCPDataStageTask.java | 44 ++---
.../gfac/impl/task/utils/StreamData.java | 43 ++--
11 files changed, 178 insertions(+), 178 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/airavata/blob/c50470c1/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/core/cluster/RemoteCluster.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/core/cluster/RemoteCluster.java b/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/core/cluster/RemoteCluster.java
index 5f8d0ec..f59a9e3 100644
--- a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/core/cluster/RemoteCluster.java
+++ b/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/core/cluster/RemoteCluster.java
@@ -52,7 +52,7 @@ public interface RemoteCluster { // FIXME: replace SSHApiException with suitable
* @param remoteFile remote file location, this can be a directory too
* @throws SSHApiException throws exception during error
*/
- void copyTo(String localFile, String remoteFile) throws SSHApiException;
+ void copyTo(String localFile, String remoteFile, SessionConsumer<Session> sessionSessionConsumer) throws SSHApiException;
/**
* This will copy a remote file in path rFile to local file lFile
@@ -60,7 +60,7 @@ public interface RemoteCluster { // FIXME: replace SSHApiException with suitable
* @param remoteFile remote file path, this has to be a full qualified path
* @param localFile This is the local file to copy, this can be a directory too
*/
- void copyFrom(String remoteFile, String localFile) throws SSHApiException;
+ void copyFrom(String remoteFile, String localFile, SessionConsumer<Session> sessionSessionConsumer) throws SSHApiException;
/**
* This wil copy source remote file to target remote file.
@@ -77,7 +77,7 @@ public interface RemoteCluster { // FIXME: replace SSHApiException with suitable
* @param directoryPath the full qualified path for the directory user wants to create
* @throws SSHApiException throws during error
*/
- void makeDirectory(String directoryPath) throws SSHApiException;
+ void makeDirectory(String directoryPath, SessionConsumer<Session> sessionConsumer) throws SSHApiException;
/**
* This will delete the given job from the queue
http://git-wip-us.apache.org/repos/asf/airavata/blob/c50470c1/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/BESRemoteCluster.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/BESRemoteCluster.java b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/BESRemoteCluster.java
index 0f517b5..828a34e 100644
--- a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/BESRemoteCluster.java
+++ b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/BESRemoteCluster.java
@@ -43,12 +43,12 @@ public class BESRemoteCluster extends AbstractRemoteCluster{
}
@Override
- public void copyTo(String localFile, String remoteFile) throws SSHApiException {
+ public void copyTo(String localFile, String remoteFile, SessionConsumer<Session> sessionConsumer) throws SSHApiException {
}
@Override
- public void copyFrom(String remoteFile, String localFile) throws SSHApiException {
+ public void copyFrom(String remoteFile, String localFile, SessionConsumer<Session> sessionConsumer) throws SSHApiException {
}
@@ -57,7 +57,7 @@ public class BESRemoteCluster extends AbstractRemoteCluster{
}
@Override
- public void makeDirectory(String directoryPath) throws SSHApiException {
+ public void makeDirectory(String directoryPath, SessionConsumer<Session> sessionConsumer) throws SSHApiException {
}
http://git-wip-us.apache.org/repos/asf/airavata/blob/c50470c1/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/HPCRemoteCluster.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/HPCRemoteCluster.java b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/HPCRemoteCluster.java
index c3566b8..b322cef 100644
--- a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/HPCRemoteCluster.java
+++ b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/HPCRemoteCluster.java
@@ -86,7 +86,8 @@ public class HPCRemoteCluster extends AbstractRemoteCluster{
@Override
public JobSubmissionOutput submitBatchJob(String jobScriptFilePath, String workingDirectory) throws SSHApiException {
JobSubmissionOutput jsoutput = new JobSubmissionOutput();
- copyTo(jobScriptFilePath, workingDirectory); // scp script file to working directory
+ copyTo(jobScriptFilePath, workingDirectory,
+ session -> SSHUtils.scpTo(jobScriptFilePath, workingDirectory, session)); // scp script file to working directory
RawCommandInfo submitCommand = jobManagerConfiguration.getSubmitCommand(workingDirectory, jobScriptFilePath);
submitCommand.setRawCommand("cd " + workingDirectory + "; " + submitCommand.getRawCommand());
StandardOutReader reader = new StandardOutReader();
@@ -112,13 +113,13 @@ public class HPCRemoteCluster extends AbstractRemoteCluster{
}
@Override
- public void copyTo(String localFile, String remoteFile) throws SSHApiException {
+ public void copyTo(String localFile, String remoteFile, SessionConsumer<Session> sessionConsumer) throws SSHApiException {
int retry = 3;
while (retry > 0) {
try {
session = Factory.getSSHSession(authenticationInfo, serverInfo);
log.info("Transferring localhost:" + localFile + " to " + serverInfo.getHost() + ":" + remoteFile);
- SSHUtils.scpTo(localFile, remoteFile, session);
+ sessionConsumer.consume(session);
retry = 0;
} catch (Exception e) {
retry--;
@@ -139,13 +140,13 @@ public class HPCRemoteCluster extends AbstractRemoteCluster{
}
@Override
- public void copyFrom(String remoteFile, String localFile) throws SSHApiException {
+ public void copyFrom(String remoteFile, String localFile, SessionConsumer<Session> sessionConsumer) throws SSHApiException {
int retry = 3;
while(retry>0) {
try {
session = Factory.getSSHSession(authenticationInfo, serverInfo);
log.info("Transferring " + serverInfo.getHost() + ":" + remoteFile + " To localhost:" + localFile);
- SSHUtils.scpFrom(remoteFile, localFile, session);
+ sessionConsumer.consume(session);
retry=0;
} catch (Exception e) {
retry--;
@@ -190,7 +191,7 @@ public class HPCRemoteCluster extends AbstractRemoteCluster{
}
@Override
- public void makeDirectory(String directoryPath) throws SSHApiException {
+ public void makeDirectory(String directoryPath, SessionConsumer<Session> sessionConsumer) throws SSHApiException {
int retryCount = 0;
try {
while (retryCount < MAX_RETRY_COUNT) {
@@ -198,9 +199,9 @@ public class HPCRemoteCluster extends AbstractRemoteCluster{
session = Factory.getSSHSession(authenticationInfo, serverInfo);
log.info("Creating directory: " + serverInfo.getHost() + ":" + directoryPath);
try {
- SSHUtils.makeDirectory(directoryPath, session);
+ sessionConsumer.consume(session);
break; // Exit while loop
- } catch (JSchException e) {
+ } catch (DataStagingException e) {
if (retryCount == MAX_RETRY_COUNT) {
log.error("Retry count " + MAX_RETRY_COUNT + " exceeded for creating directory: "
+ serverInfo.getHost() + ":" + directoryPath, e);
@@ -210,7 +211,7 @@ public class HPCRemoteCluster extends AbstractRemoteCluster{
log.error("Issue with jsch, Retry creating directory: " + serverInfo.getHost() + ":" + directoryPath);
}
}
- } catch (JSchException | AiravataException | IOException e) {
+ } catch (AiravataException | DataStagingException e) {
throw new SSHApiException("Failed to create directory " + serverInfo.getHost() + ":" + directoryPath, e);
}
}
http://git-wip-us.apache.org/repos/asf/airavata/blob/c50470c1/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/LocalRemoteCluster.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/LocalRemoteCluster.java b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/LocalRemoteCluster.java
index d5422d2..aa79a0c 100644
--- a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/LocalRemoteCluster.java
+++ b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/LocalRemoteCluster.java
@@ -60,7 +60,8 @@ public class LocalRemoteCluster extends AbstractRemoteCluster {
public JobSubmissionOutput submitBatchJob(String jobScriptFilePath, String workingDirectory) throws SSHApiException {
try {
JobSubmissionOutput jsoutput = new JobSubmissionOutput();
- copyTo(jobScriptFilePath, workingDirectory + File.separator + new File(jobScriptFilePath).getName()); // scp script file to working directory
+ String remoteFile = workingDirectory + File.separator + new File(jobScriptFilePath).getName();
+ copyTo(jobScriptFilePath, remoteFile, session -> SSHUtils.scpTo(jobScriptFilePath, remoteFile, session)); // scp script file to working directory
RawCommandInfo submitCommand = jobManagerConfiguration.getSubmitCommand(workingDirectory, jobScriptFilePath);
submitCommand.setRawCommand(submitCommand.getRawCommand());
LocalCommandOutput localCommandOutput = new LocalCommandOutput();
@@ -76,7 +77,7 @@ public class LocalRemoteCluster extends AbstractRemoteCluster {
}
@Override
- public void copyTo(String localFile, String remoteFile) throws SSHApiException {
+ public void copyTo(String localFile, String remoteFile, SessionConsumer<Session> sessionConsumer) throws SSHApiException {
Path sourcePath = Paths.get(localFile);
Path targetPath = Paths.get(remoteFile);
try {
@@ -88,7 +89,7 @@ public class LocalRemoteCluster extends AbstractRemoteCluster {
}
@Override
- public void copyFrom(String remoteFile, String localFile) throws SSHApiException {
+ public void copyFrom(String remoteFile, String localFile, SessionConsumer<Session> sessionConsumer) throws SSHApiException {
Path sourcePath = Paths.get(remoteFile);
Path targetPath = Paths.get(localFile);
try {
@@ -125,7 +126,7 @@ public class LocalRemoteCluster extends AbstractRemoteCluster {
}
@Override
- public void makeDirectory(String directoryPath) throws SSHApiException {
+ public void makeDirectory(String directoryPath, SessionConsumer<Session> sessionConsumer) throws SSHApiException {
Path dirPath = Paths.get(directoryPath);
Set<PosixFilePermission> perms = new HashSet<>();
// add permission as rwxr--r-- 744
http://git-wip-us.apache.org/repos/asf/airavata/blob/c50470c1/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/SSHUtils.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/SSHUtils.java b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/SSHUtils.java
index cd5651e..07e1799 100644
--- a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/SSHUtils.java
+++ b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/SSHUtils.java
@@ -24,6 +24,7 @@ import com.jcraft.jsch.Channel;
import com.jcraft.jsch.ChannelExec;
import com.jcraft.jsch.JSchException;
import com.jcraft.jsch.Session;
+import org.apache.airavata.gfac.core.DataStagingException;
import org.apache.airavata.gfac.core.SSHApiException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -53,97 +54,100 @@ public class SSHUtils {
* @param localFile Local file to transfer, this can be a directory
* @return returns the final remote file path, so that users can use the new file location
*/
- public static String scpTo(String localFile, String remoteFile, Session session) throws IOException,
- JSchException, SSHApiException {
+ public static String scpTo(String localFile, String remoteFile, Session session) throws DataStagingException {
FileInputStream fis = null;
String prefix = null;
if (new File(localFile).isDirectory()) {
prefix = localFile + File.separator;
}
boolean ptimestamp = true;
-
+ Channel channel = null;
// exec 'scp -t rfile' remotely
String command = "scp " + (ptimestamp ? "-p" : "") + " -t " + remoteFile;
- Channel channel = session.openChannel("exec");
+ try {
+ channel = session.openChannel("exec");
- StandardOutReader stdOutReader = new StandardOutReader();
- ((ChannelExec) channel).setErrStream(stdOutReader.getStandardError());
- ((ChannelExec) channel).setCommand(command);
+ StandardOutReader stdOutReader = new StandardOutReader();
+ ((ChannelExec) channel).setErrStream(stdOutReader.getStandardError());
+ ((ChannelExec) channel).setCommand(command);
- // get I/O streams for remote scp
- OutputStream out = channel.getOutputStream();
- InputStream in = channel.getInputStream();
+ // get I/O streams for remote scp
+ try (OutputStream out = channel.getOutputStream();
+ InputStream in = channel.getInputStream()) {
- channel.connect();
+ channel.connect();
+ if (checkAck(in) != 0) {
+ String error = "Error Reading input Stream";
+ log.error(error);
+ throw new DataStagingException(error);
+ }
- if (checkAck(in) != 0) {
- String error = "Error Reading input Stream";
- log.error(error);
- throw new SSHApiException(error);
- }
+ File _lfile = new File(localFile);
+
+ if (ptimestamp) {
+ command = "T" + (_lfile.lastModified() / 1000) + " 0";
+ // The access time should be sent here,
+ // but it is not accessible with JavaAPI ;-<
+ command += (" " + (_lfile.lastModified() / 1000) + " 0\n");
+ out.write(command.getBytes());
+ out.flush();
+ if (checkAck(in) != 0) {
+ String error = "Error Reading input Stream";
+ log.error(error);
+ throw new DataStagingException(error);
+ }
+ }
- File _lfile = new File(localFile);
+ // send "C0644 filesize filename", where filename should not include '/'
+ long filesize = _lfile.length();
+ command = "C0644 " + filesize + " ";
+ if (localFile.lastIndexOf('/') > 0) {
+ command += localFile.substring(localFile.lastIndexOf('/') + 1);
+ } else {
+ command += localFile;
+ }
+ command += "\n";
+ out.write(command.getBytes());
+ out.flush();
+ if (checkAck(in) != 0) {
+ String error = "Error Reading input Stream";
+ log.error(error);
+ throw new DataStagingException(error);
+ }
- if (ptimestamp) {
- command = "T" + (_lfile.lastModified() / 1000) + " 0";
- // The access time should be sent here,
- // but it is not accessible with JavaAPI ;-<
- command += (" " + (_lfile.lastModified() / 1000) + " 0\n");
- out.write(command.getBytes());
- out.flush();
- if (checkAck(in) != 0) {
- String error = "Error Reading input Stream";
- log.error(error);
- throw new SSHApiException(error);
+ // send a content of localFile
+ fis = new FileInputStream(localFile);
+ byte[] buf = new byte[1024];
+ while (true) {
+ int len = fis.read(buf, 0, buf.length);
+ if (len <= 0) break;
+ out.write(buf, 0, len); //out.flush();
+ }
+ fis.close();
+ fis = null;
+ // send '\0'
+ buf[0] = 0;
+ out.write(buf, 0, 1);
+ out.flush();
+ if (checkAck(in) != 0) {
+ String error = "Error Reading input Stream";
+ log.error(error);
+ throw new DataStagingException(error);
+ }
+ }
+ stdOutReader.onOutput(channel);
+ if (stdOutReader.getStdErrorString().contains("scp:")) {
+ throw new DataStagingException(stdOutReader.getStdErrorString());
+ }
+ //since remote file is always a file we just return the file
+ return remoteFile;
+ } catch (IOException | JSchException e) {
+ throw new DataStagingException(e);
+ } finally {
+ if (channel != null && channel.isConnected()) {
+ channel.disconnect();
}
}
-
- // send "C0644 filesize filename", where filename should not include '/'
- long filesize = _lfile.length();
- command = "C0644 " + filesize + " ";
- if (localFile.lastIndexOf('/') > 0) {
- command += localFile.substring(localFile.lastIndexOf('/') + 1);
- } else {
- command += localFile;
- }
- command += "\n";
- out.write(command.getBytes());
- out.flush();
- if (checkAck(in) != 0) {
- String error = "Error Reading input Stream";
- log.error(error);
- throw new SSHApiException(error);
- }
-
- // send a content of localFile
- fis = new FileInputStream(localFile);
- byte[] buf = new byte[1024];
- while (true) {
- int len = fis.read(buf, 0, buf.length);
- if (len <= 0) break;
- out.write(buf, 0, len); //out.flush();
- }
- fis.close();
- fis = null;
- // send '\0'
- buf[0] = 0;
- out.write(buf, 0, 1);
- out.flush();
- if (checkAck(in) != 0) {
- String error = "Error Reading input Stream";
- log.error(error);
- throw new SSHApiException(error);
- }
- out.close();
- stdOutReader.onOutput(channel);
-
-
- channel.disconnect();
- if (stdOutReader.getStdErrorString().contains("scp:")) {
- throw new SSHApiException(stdOutReader.getStdErrorString());
- }
- //since remote file is always a file we just return the file
- return remoteFile;
}
/**
@@ -153,8 +157,7 @@ public class SSHUtils {
* @param localFile This is the local file to copy, this can be a directory too
* @return returns the final local file path of the new file came from the remote resource
*/
- public static void scpFrom(String remoteFile, String localFile, Session session) throws IOException,
- JSchException, SSHApiException {
+ public static void scpFrom(String remoteFile, String localFile, Session session) throws DataStagingException {
FileOutputStream fos = null;
try {
String prefix = null;
@@ -255,6 +258,7 @@ public class SSHUtils {
} catch (Exception e) {
log.error(e.getMessage(), e);
+ throw new DataStagingException(e);
} finally {
try {
if (fos != null) fos.close();
@@ -272,8 +276,11 @@ public class SSHUtils {
* @param destinationSession JSch Session for target
* @return returns the final local file path of the new file came from the remote resource
*/
- public static void scpThirdParty(String sourceFile, Session sourceSession, String destinationFile, Session destinationSession, boolean ignoreEmptyFile) throws
- IOException, JSchException {
+ public static void scpThirdParty(String sourceFile,
+ Session sourceSession,
+ String destinationFile,
+ Session destinationSession,
+ boolean ignoreEmptyFile) throws DataStagingException {
OutputStream sout = null;
InputStream sin = null;
OutputStream dout = null;
@@ -401,7 +408,7 @@ public class SSHUtils {
} catch (Exception e) {
log.error(e.getMessage(), e);
- throw new JSchException(e.getMessage());
+ throw new DataStagingException(e.getMessage());
} finally {
try {
if (dout != null) dout.close();
@@ -426,35 +433,32 @@ public class SSHUtils {
}
}
- public static void makeDirectory(String path, Session session) throws IOException, JSchException, SSHApiException {
+ public static void makeDirectory(String path, Session session) throws DataStagingException {
- // exec 'scp -t rfile' remotely
+ Channel channel = null;
String command = "mkdir -p " + path;
- Channel channel = session.openChannel("exec");
- StandardOutReader stdOutReader = new StandardOutReader();
-
- ((ChannelExec) channel).setCommand(command);
-
-
- ((ChannelExec) channel).setErrStream(stdOutReader.getStandardError());
try {
+ // exec 'scp -t rfile' remotely
+ channel = session.openChannel("exec");
+ StandardOutReader stdOutReader = new StandardOutReader();
channel.connect();
+ stdOutReader.onOutput(channel);
+ if (stdOutReader.getStdErrorString().contains("mkdir:")) {
+ throw new DataStagingException(stdOutReader.getStdErrorString());
+ }
+
} catch (JSchException e) {
- channel.disconnect();
// session.disconnect();
log.error("Unable to retrieve command output. Command - " + command +
" on server - " + session.getHost() + ":" + session.getPort() +
" connecting user name - "
+ session.getUserName());
- throw e;
- }
- stdOutReader.onOutput(channel);
- if (stdOutReader.getStdErrorString().contains("mkdir:")) {
- throw new SSHApiException(stdOutReader.getStdErrorString());
+ throw new DataStagingException(e);
+ }finally {
+ if(channel != null && channel.isConnected())
+ channel.disconnect();
}
-
- channel.disconnect();
}
public static List<String> listDirectory(String path, Session session) throws IOException, JSchException,
http://git-wip-us.apache.org/repos/asf/airavata/blob/c50470c1/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/ArchiveTask.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/ArchiveTask.java b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/ArchiveTask.java
index 88661f8..85d3fd2 100644
--- a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/ArchiveTask.java
+++ b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/ArchiveTask.java
@@ -138,13 +138,8 @@ public class ArchiveTask implements Task {
// move tar to storage resource
remoteCluster.execute(commandInfo);
URI destinationURI = TaskUtils.getDestinationURI(taskContext, hostName, inputPath, archiveTar);
- remoteCluster.thirdPartyTransfer(resourceAbsTarFilePath ,destinationURI.getPath(), session -> {
- try {
- SSHUtils.scpThirdParty(sourceURI.getPath(),session, destinationURI.getPath(), sshSession, true);
- } catch (Exception e) {
- throw new DataStagingException("Error while transferring " + sourceURI.getPath() + " to " + destinationURI.getPath());
- }
- });
+ remoteCluster.thirdPartyTransfer(resourceAbsTarFilePath, destinationURI.getPath(),
+ session -> SSHUtils.scpThirdParty(sourceURI.getPath(), session, destinationURI.getPath(), sshSession, true));
// delete tar in remote computer resource
commandInfo = new RawCommandInfo("rm " + resourceAbsTarFilePath);
http://git-wip-us.apache.org/repos/asf/airavata/blob/c50470c1/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/BESJobSubmissionTask.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/BESJobSubmissionTask.java b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/BESJobSubmissionTask.java
index 990b9ea..1811286 100644
--- a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/BESJobSubmissionTask.java
+++ b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/BESJobSubmissionTask.java
@@ -31,6 +31,7 @@ import de.fzj.unicore.wsrflite.xmlbeans.WSUtilities;
import eu.unicore.util.httpclient.DefaultClientConfiguration;
import org.apache.airavata.common.exception.AiravataException;
import org.apache.airavata.common.exception.ApplicationSettingsException;
+import org.apache.airavata.gfac.core.DataStagingException;
import org.apache.airavata.gfac.core.GFacException;
import org.apache.airavata.gfac.core.GFacUtils;
import org.apache.airavata.gfac.core.SSHApiException;
@@ -273,7 +274,7 @@ public class BESJobSubmissionTask implements JobSubmissionTask {
break;
}
}
- } catch (IOException | JSchException | AiravataException | SSHApiException | URISyntaxException e) {
+ } catch (DataStagingException | AiravataException | URISyntaxException e) {
log.error("Error while coping local file " + localFilePath + " to remote " + remoteFilePath, e);
throw new GFacException("Error while scp output files to remote storage file location", e);
}
@@ -317,7 +318,7 @@ public class BESJobSubmissionTask implements JobSubmissionTask {
input.setValue("file:/" + localFilePath);
}
}
- } catch (IOException | JSchException | AiravataException | SSHApiException | URISyntaxException e) {
+ } catch ( AiravataException | DataStagingException| URISyntaxException e) {
log.error("Error while coping remote file " + remoteFilePath + " to local " + localFilePath, e);
throw new GFacException("Error while scp input files to local file location", e);
}
http://git-wip-us.apache.org/repos/asf/airavata/blob/c50470c1/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/DataStageTask.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/DataStageTask.java b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/DataStageTask.java
index 8c6a125..a171f7f 100644
--- a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/DataStageTask.java
+++ b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/DataStageTask.java
@@ -24,6 +24,7 @@ import org.apache.airavata.gfac.core.SSHApiException;
import org.apache.airavata.gfac.core.context.TaskContext;
import org.apache.airavata.gfac.core.task.Task;
import org.apache.airavata.gfac.core.task.TaskException;
+import org.apache.airavata.gfac.impl.SSHUtils;
import org.apache.airavata.model.commons.ErrorModel;
import org.apache.airavata.model.status.ProcessState;
import org.apache.airavata.model.status.TaskState;
@@ -65,14 +66,18 @@ public class DataStageTask implements Task {
/**
* copy local file to compute resource.
*/
- taskContext.getParentProcessContext().getDataMovementRemoteCluster().copyTo(sourceURI.getPath(), destinationURI
- .getPath());
+ taskContext.getParentProcessContext().getDataMovementRemoteCluster().copyTo(
+ sourceURI.getPath(),
+ destinationURI.getPath(),
+ session -> SSHUtils.scpTo(sourceURI.getPath(), destinationURI.getPath(), session));
} else if (processState == ProcessState.OUTPUT_DATA_STAGING) {
/**
* copy remote file from compute resource.
*/
- taskContext.getParentProcessContext().getDataMovementRemoteCluster().copyFrom(sourceURI.getPath(), destinationURI
- .getPath());
+ taskContext.getParentProcessContext().getDataMovementRemoteCluster().copyFrom(
+ sourceURI.getPath(),
+ destinationURI.getPath(),
+ session -> SSHUtils.scpFrom(sourceURI.getPath(), destinationURI.getPath(), session));
}
status.setReason("Successfully staged data");
} catch (SSHApiException e) {
http://git-wip-us.apache.org/repos/asf/airavata/blob/c50470c1/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/EnvironmentSetupTask.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/EnvironmentSetupTask.java b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/EnvironmentSetupTask.java
index 7de0282..bbe77d8 100644
--- a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/EnvironmentSetupTask.java
+++ b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/EnvironmentSetupTask.java
@@ -25,6 +25,7 @@ import org.apache.airavata.gfac.core.cluster.RemoteCluster;
import org.apache.airavata.gfac.core.context.TaskContext;
import org.apache.airavata.gfac.core.task.Task;
import org.apache.airavata.gfac.core.task.TaskException;
+import org.apache.airavata.gfac.impl.SSHUtils;
import org.apache.airavata.model.commons.ErrorModel;
import org.apache.airavata.model.status.TaskState;
import org.apache.airavata.model.status.TaskStatus;
@@ -48,7 +49,8 @@ public class EnvironmentSetupTask implements Task {
TaskStatus status = new TaskStatus(TaskState.COMPLETED);
try {
RemoteCluster remoteCluster = taskContext.getParentProcessContext().getJobSubmissionRemoteCluster();
- remoteCluster.makeDirectory(taskContext.getParentProcessContext().getWorkingDir());
+ String workingDir = taskContext.getParentProcessContext().getWorkingDir();
+ remoteCluster.makeDirectory(workingDir, session -> SSHUtils.makeDirectory(workingDir, session));
status.setReason("Successfully created environment");
} catch (SSHApiException e) {
String msg = "Error while environment setup";
http://git-wip-us.apache.org/repos/asf/airavata/blob/c50470c1/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/SCPDataStageTask.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/SCPDataStageTask.java b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/SCPDataStageTask.java
index 2788535..4189f81 100644
--- a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/SCPDataStageTask.java
+++ b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/SCPDataStageTask.java
@@ -182,7 +182,7 @@ public class SCPDataStageTask implements Task {
errorModel.setUserFriendlyMessage(msg);
taskContext.getTaskModel().setTaskErrors(Arrays.asList(errorModel));
return status;
- } catch (ApplicationSettingsException | FileNotFoundException e) {
+ } catch (ApplicationSettingsException e) {
String msg = "Failed while reading credentials";
log.error(msg, e);
status.setState(TaskState.FAILED);
@@ -219,7 +219,7 @@ public class SCPDataStageTask implements Task {
errorModel.setActualErrorMessage(e.getMessage());
errorModel.setUserFriendlyMessage(msg);
taskContext.getTaskModel().setTaskErrors(Arrays.asList(errorModel));
- } catch (JSchException | IOException e) {
+ } catch (DataStagingException e) {
String msg = "Failed to do scp with client";
log.error(msg, e);
status.setState(TaskState.FAILED);
@@ -249,39 +249,35 @@ public class SCPDataStageTask implements Task {
}
private void inputDataStaging(TaskContext taskContext, Session sshSession, URI sourceURI, URI
- destinationURI) throws SSHApiException, IOException, JSchException {
+ destinationURI) throws SSHApiException {
/**
* scp third party file transfer 'to' compute resource.
*/
- taskContext.getParentProcessContext().getDataMovementRemoteCluster()
- .thirdPartyTransfer(sourceURI.getPath(), destinationURI.getPath(), session -> {
- try {
- SSHUtils.scpThirdParty(sourceURI.getPath(), sshSession, destinationURI.getPath(), session, false);
- } catch (Exception e) {
- throw new DataStagingException("Error while file staging, from " + sourceURI.getPath()
- + " to " + destinationURI.getPath());
- }
- });
+ taskContext.getParentProcessContext().getDataMovementRemoteCluster().thirdPartyTransfer(
+ sourceURI.getPath(),
+ destinationURI.getPath(),
+ session -> SSHUtils.scpThirdParty(sourceURI.getPath(), sshSession, destinationURI.getPath(), session, false));
}
private void outputDataStaging(TaskContext taskContext, Session sshSession, URI sourceURI, URI destinationURI)
- throws SSHApiException, AiravataException, IOException, JSchException, GFacException {
+ throws SSHApiException, AiravataException, GFacException {
/**
* scp third party file transfer 'from' comute resource.
*/
- taskContext.getParentProcessContext().getDataMovementRemoteCluster()
- .thirdPartyTransfer(sourceURI.getPath(), destinationURI.getPath(), session -> {
- try {
- SSHUtils.scpThirdParty(sourceURI.getPath(), session, destinationURI.getPath(), sshSession, true);
- } catch (Exception e) {
- throw new DataStagingException("Error while file staging, from " + sourceURI.getPath()
- + " to " + destinationURI.getPath());
- }
- });
+ taskContext.getParentProcessContext().getDataMovementRemoteCluster().thirdPartyTransfer(
+ sourceURI.getPath(),
+ destinationURI.getPath(),
+ session -> SSHUtils.scpThirdParty(sourceURI.getPath(), session, destinationURI.getPath(), sshSession, true));
// update output locations
- GFacUtils.saveExperimentOutput(taskContext.getParentProcessContext(), taskContext.getProcessOutput().getName(), destinationURI.toString());
- GFacUtils.saveProcessOutput(taskContext.getParentProcessContext(), taskContext.getProcessOutput().getName(), destinationURI.toString());
+ GFacUtils.saveExperimentOutput(
+ taskContext.getParentProcessContext(),
+ taskContext.getProcessOutput().getName(),
+ destinationURI.toString());
+ GFacUtils.saveProcessOutput(
+ taskContext.getParentProcessContext(),
+ taskContext.getProcessOutput().getName(),
+ destinationURI.toString());
}
http://git-wip-us.apache.org/repos/asf/airavata/blob/c50470c1/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/utils/StreamData.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/utils/StreamData.java b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/utils/StreamData.java
index fccce0d..f5fd14d 100644
--- a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/utils/StreamData.java
+++ b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/utils/StreamData.java
@@ -21,10 +21,8 @@
package org.apache.airavata.gfac.impl.task.utils;
-import com.jcraft.jsch.JSchException;
import com.jcraft.jsch.Session;
import org.apache.airavata.common.exception.AiravataException;
-import org.apache.airavata.credential.store.store.CredentialStoreException;
import org.apache.airavata.gfac.core.DataStagingException;
import org.apache.airavata.gfac.core.GFacException;
import org.apache.airavata.gfac.core.GFacUtils;
@@ -32,7 +30,6 @@ import org.apache.airavata.gfac.core.SSHApiException;
import org.apache.airavata.gfac.core.authentication.AuthenticationInfo;
import org.apache.airavata.gfac.core.cluster.CommandInfo;
import org.apache.airavata.gfac.core.cluster.RawCommandInfo;
-import org.apache.airavata.gfac.core.cluster.RemoteCluster;
import org.apache.airavata.gfac.core.cluster.ServerInfo;
import org.apache.airavata.gfac.core.context.TaskContext;
import org.apache.airavata.gfac.impl.Factory;
@@ -44,7 +41,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
-import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.TimerTask;
@@ -79,25 +75,23 @@ public class StreamData extends TimerTask {
}
// output staging should end when the job is complete
- if (jobStatus != null && jobStatus.getJobState().equals(JobState.COMPLETE) || jobStatus.getJobState().equals(JobState.CANCELED) || jobStatus.getJobState().equals(JobState.FAILED)){
+ if (jobStatus != null && jobStatus.getJobState().equals(JobState.COMPLETE)
+ || jobStatus.getJobState().equals(JobState.CANCELED)
+ || jobStatus.getJobState().equals(JobState.FAILED)){
this.cancel();
}
} catch (URISyntaxException e) {
log.error("expId: {}, processId:{}, taskId: {}:- Couldn't stage file {} , Erroneous path specified",
taskContext.getExperimentId(), taskContext.getProcessId(), taskContext.getTaskId(),
taskContext.getProcessOutput().getName());
- } catch (IllegalAccessException | InstantiationException | AiravataException | IOException | JSchException | SSHApiException e) {
+ } catch (AiravataException | SSHApiException e) {
log.error("expId: {}, processId:{}, taskId: {}:- Couldn't stage file {} , Error occurred while streaming data",
taskContext.getExperimentId(), taskContext.getProcessId(), taskContext.getTaskId(),
taskContext.getProcessOutput().getName());
- } catch (CredentialStoreException e) {
- log.error("expId: {}, processId:{}, taskId: {}:- Couldn't stage file {} , Error occurred while connecting with credential store",
- taskContext.getExperimentId(), taskContext.getProcessId(), taskContext.getTaskId(),
- taskContext.getProcessOutput().getName());
}
}
- public void runOutputStaging() throws URISyntaxException, IllegalAccessException, InstantiationException, CredentialStoreException, AiravataException, IOException, JSchException, SSHApiException {
+ public void runOutputStaging() throws URISyntaxException, AiravataException, SSHApiException {
try {
AuthenticationInfo authenticationInfo = null;
@@ -123,7 +117,7 @@ public class StreamData extends TimerTask {
String targetPath = destinationURI.getPath().substring(0, destinationURI.getPath().lastIndexOf('/'));
SSHUtils.makeDirectory(targetPath, sshSession);
outputDataStaging(taskContext, sshSession, sourceURI, destinationURI);
- } catch (GFacException e) {
+ } catch (DataStagingException | GFacException e) {
log.error("expId: {}, processId:{}, taskId: {}:- Couldn't stage file {} , Error while output staging",
taskContext.getExperimentId(), taskContext.getProcessId(), taskContext.getTaskId(),
taskContext.getProcessOutput().getName());
@@ -147,23 +141,24 @@ public class StreamData extends TimerTask {
}
private void outputDataStaging(TaskContext taskContext, Session sshSession, URI sourceURI, URI destinationURI)
- throws SSHApiException, AiravataException, IOException, JSchException, GFacException {
+ throws SSHApiException, GFacException {
/**
* scp third party file transfer 'from' comute resource.
*/
- taskContext.getParentProcessContext().getDataMovementRemoteCluster()
- .thirdPartyTransfer(sourceURI.getPath(), destinationURI.getPath(), session -> {
- try {
- SSHUtils.scpThirdParty(sourceURI.getPath(), session, destinationURI.getPath(), sshSession, true);
- } catch (Exception e) {
- throw new DataStagingException("Error while file staging, from " + sourceURI.getPath()
- + " to " + destinationURI.getPath());
- }
- });
+ taskContext.getParentProcessContext().getDataMovementRemoteCluster().thirdPartyTransfer(
+ sourceURI.getPath(),
+ destinationURI.getPath(),
+ session -> SSHUtils.scpThirdParty(sourceURI.getPath(), session, destinationURI.getPath(), sshSession, true));
// update output locations
- GFacUtils.saveExperimentOutput(taskContext.getParentProcessContext(), taskContext.getProcessOutput().getName(), destinationURI.getPath());
- GFacUtils.saveProcessOutput(taskContext.getParentProcessContext(), taskContext.getProcessOutput().getName(), destinationURI.getPath());
+ GFacUtils.saveExperimentOutput(
+ taskContext.getParentProcessContext(),
+ taskContext.getProcessOutput().getName(),
+ destinationURI.getPath());
+ GFacUtils.saveProcessOutput(
+ taskContext.getParentProcessContext(),
+ taskContext.getProcessOutput().getName(),
+ destinationURI.getPath());
}