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 2015/06/26 06:27:48 UTC
[3/4] airavata git commit: Changed registry code to work with
subTaskModel thrift binary parameter,
implement scp data staging task to work with DataStagingTaskModel
Changed registry code to work with subTaskModel thrift binary parameter, implement scp data staging task to work with DataStagingTaskModel
Project: http://git-wip-us.apache.org/repos/asf/airavata/repo
Commit: http://git-wip-us.apache.org/repos/asf/airavata/commit/7f4f1826
Tree: http://git-wip-us.apache.org/repos/asf/airavata/tree/7f4f1826
Diff: http://git-wip-us.apache.org/repos/asf/airavata/diff/7f4f1826
Branch: refs/heads/master
Commit: 7f4f1826c46b79267e27b3b57f941656863ac1f6
Parents: 455b435
Author: Shameera Rathanyaka <sh...@gmail.com>
Authored: Fri Jun 26 00:20:08 2015 -0400
Committer: Shameera Rathanyaka <sh...@gmail.com>
Committed: Fri Jun 26 00:20:08 2015 -0400
----------------------------------------------------------------------
modules/commons/pom.xml | 5 +++
.../airavata/common/utils/ThriftUtils.java | 33 ++++++++++++++++----
.../gfac/impl/task/AbstractSCPTask.java | 1 +
.../gfac/impl/task/SCPInputDataStageTask.java | 18 ++++++++---
.../gfac/impl/task/SCPOutputDataStatgeTask.java | 16 ++++++++--
.../catalog/impl/ExperimentRegistry.java | 4 +--
.../core/experiment/catalog/model/Task.java | 12 +++----
.../catalog/resources/TaskResource.java | 12 +++----
.../experiment/catalog/resources/Utils.java | 2 +-
.../utils/ThriftDataModelConversion.java | 2 +-
.../experiment/catalog/JobResourceTest.java | 3 +-
.../experiment/catalog/TaskResourceTest.java | 3 +-
.../airavata-api/task_model.thrift | 2 +-
13 files changed, 81 insertions(+), 32 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/airavata/blob/7f4f1826/modules/commons/pom.xml
----------------------------------------------------------------------
diff --git a/modules/commons/pom.xml b/modules/commons/pom.xml
index f6c34d4..7f9db2f 100644
--- a/modules/commons/pom.xml
+++ b/modules/commons/pom.xml
@@ -25,6 +25,11 @@
<dependencies>
<dependency>
+ <groupId>org.apache.airavata</groupId>
+ <artifactId>airavata-model-utils</artifactId>
+ <version>${parent.version}</version>
+ </dependency>
+ <dependency>
<groupId>xerces</groupId>
<artifactId>xercesImpl</artifactId>
<version>2.9.1</version>
http://git-wip-us.apache.org/repos/asf/airavata/blob/7f4f1826/modules/commons/src/main/java/org/apache/airavata/common/utils/ThriftUtils.java
----------------------------------------------------------------------
diff --git a/modules/commons/src/main/java/org/apache/airavata/common/utils/ThriftUtils.java b/modules/commons/src/main/java/org/apache/airavata/common/utils/ThriftUtils.java
index ee86f74..b13ef24 100644
--- a/modules/commons/src/main/java/org/apache/airavata/common/utils/ThriftUtils.java
+++ b/modules/commons/src/main/java/org/apache/airavata/common/utils/ThriftUtils.java
@@ -21,17 +21,38 @@
package org.apache.airavata.common.utils;
+import org.apache.airavata.model.task.DataStagingTaskModel;
+import org.apache.airavata.model.task.TaskModel;
import org.apache.thrift.TBase;
import org.apache.thrift.TDeserializer;
import org.apache.thrift.TException;
import org.apache.thrift.TSerializer;
public class ThriftUtils {
- public static byte[] serializeThriftObject(TBase object) throws TException {
- return new TSerializer().serialize(object);
- }
+ public static byte[] serializeThriftObject(TBase object) throws TException {
+ return new TSerializer().serialize(object);
+ }
- public static void createThriftFromBytes(byte []bytes, TBase object) throws TException {
- new TDeserializer().deserialize(object, bytes);
- }
+ public static void createThriftFromBytes(byte[] bytes, TBase object) throws TException {
+ new TDeserializer().deserialize(object, bytes);
+ }
+
+ public static Object getSubTaskModel(TaskModel taskModel) throws TException {
+ switch (taskModel.getTaskType()) {
+ case DATA_STAGING:
+ DataStagingTaskModel dataStagingTaskModel = new DataStagingTaskModel();
+ ThriftUtils.createThriftFromBytes(taskModel.getSubTaskModel(), dataStagingTaskModel);
+ return dataStagingTaskModel;
+ case ENV_SETUP:
+ // TODO return Environment Setup task model
+ case JOB_SUBMISSION:
+ // TODO return Job Submission task model
+ case MONITORING:
+ // TODO return Job Monitoring task model
+ case ENV_CLEANUP:
+ // TODO return Environment Clean up task model
+ default:
+ return null;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/airavata/blob/7f4f1826/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/AbstractSCPTask.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/AbstractSCPTask.java b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/AbstractSCPTask.java
index 9abc380..ed9ddca 100644
--- a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/AbstractSCPTask.java
+++ b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/AbstractSCPTask.java
@@ -59,4 +59,5 @@ public class AbstractSCPTask implements Task {
}
+
}
http://git-wip-us.apache.org/repos/asf/airavata/blob/7f4f1826/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/SCPInputDataStageTask.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/SCPInputDataStageTask.java b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/SCPInputDataStageTask.java
index 8a74b3d..8c649a2 100644
--- a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/SCPInputDataStageTask.java
+++ b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/SCPInputDataStageTask.java
@@ -23,12 +23,15 @@ package org.apache.airavata.gfac.impl.task;
import com.jcraft.jsch.JSch;
import com.jcraft.jsch.JSchException;
import com.jcraft.jsch.Session;
+import org.apache.airavata.common.utils.ThriftUtils;
import org.apache.airavata.gfac.core.SSHApiException;
import org.apache.airavata.gfac.core.context.TaskContext;
import org.apache.airavata.gfac.core.task.TaskException;
import org.apache.airavata.gfac.impl.SSHUtils;
import org.apache.airavata.model.status.TaskState;
import org.apache.airavata.model.task.DataStagingTaskModel;
+import org.apache.airavata.model.task.TaskTypes;
+import org.apache.thrift.TException;
import java.io.IOException;
import java.net.MalformedURLException;
@@ -41,14 +44,19 @@ public class SCPInputDataStageTask extends AbstractSCPTask {
@Override
public TaskState execute(TaskContext taskContext) throws TaskException {
- DataStagingTaskModel dataStagingTaskModel = new DataStagingTaskModel();
+
+ if (taskContext.getTaskModel().getTaskType() != TaskTypes.DATA_STAGING) {
+ throw new TaskException("Invalid task call, expected " + TaskTypes.DATA_STAGING.toString() + " but found "
+ + taskContext.getTaskModel().getTaskType().toString());
+ }
try {
- URL sourceURL = new URL(dataStagingTaskModel.getSource());
- URL destinationURL = new URL(dataStagingTaskModel.getDestination());
+ DataStagingTaskModel subTaskModel = (DataStagingTaskModel) ThriftUtils.getSubTaskModel(taskContext.getTaskModel());
+ URL sourceURL = new URL(subTaskModel.getSource());
+ URL destinationURL = new URL(subTaskModel.getDestination());
if (sourceURL.getProtocol().equalsIgnoreCase("file")) { // local --> Airavata --> RemoteCluster
taskContext.getParentProcessContext().getRemoteCluster().scpTo(sourceURL.getPath(),
- dataStagingTaskModel.getDestination());
+ subTaskModel.getDestination());
} else { // PGA(client) --> Airavata --> RemoteCluster
// PGA(client) --> Airavata
JSch jsch = new JSch();
@@ -65,6 +73,8 @@ public class SCPInputDataStageTask extends AbstractSCPTask {
throw new TaskException("Scp attempt failed", e);
} catch (JSchException | IOException e) {
throw new TaskException("Scp failed", e);
+ } catch (TException e) {
+ throw new TaskException("Invalid task invocation");
}
return null;
}
http://git-wip-us.apache.org/repos/asf/airavata/blob/7f4f1826/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/SCPOutputDataStatgeTask.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/SCPOutputDataStatgeTask.java b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/SCPOutputDataStatgeTask.java
index d9cf1ba..30619c0 100644
--- a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/SCPOutputDataStatgeTask.java
+++ b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/impl/task/SCPOutputDataStatgeTask.java
@@ -23,12 +23,15 @@ package org.apache.airavata.gfac.impl.task;
import com.jcraft.jsch.JSch;
import com.jcraft.jsch.JSchException;
import com.jcraft.jsch.Session;
+import org.apache.airavata.common.utils.ThriftUtils;
import org.apache.airavata.gfac.core.SSHApiException;
import org.apache.airavata.gfac.core.context.TaskContext;
import org.apache.airavata.gfac.core.task.TaskException;
import org.apache.airavata.gfac.impl.SSHUtils;
import org.apache.airavata.model.status.TaskState;
import org.apache.airavata.model.task.DataStagingTaskModel;
+import org.apache.airavata.model.task.TaskTypes;
+import org.apache.thrift.TException;
import java.io.IOException;
import java.net.MalformedURLException;
@@ -39,10 +42,15 @@ public class SCPOutputDataStatgeTask extends AbstractSCPTask {
@Override
public TaskState execute(TaskContext taskContext) throws TaskException {
- DataStagingTaskModel dataStagingTaskModel = new DataStagingTaskModel();
+ if (taskContext.getTaskModel().getTaskType() != TaskTypes.DATA_STAGING) {
+ throw new TaskException("Invalid task call, expected " + TaskTypes.DATA_STAGING.toString() + " but found "
+ + taskContext.getTaskModel().getTaskType().toString());
+ }
try {
- URL sourceURL = new URL(dataStagingTaskModel.getSource());
- URL destinationURL = new URL(dataStagingTaskModel.getDestination());
+ DataStagingTaskModel subTaskModel = (DataStagingTaskModel) ThriftUtils.getSubTaskModel(taskContext
+ .getTaskModel());
+ URL sourceURL = new URL(subTaskModel.getSource());
+ URL destinationURL = new URL(subTaskModel.getDestination());
JSch jsch = new JSch();
jsch.addIdentity(privateKeyPath, passPhrase);
Session session = jsch.getSession(userName, hostName, DEFAULT_SSH_PORT);
@@ -59,6 +67,8 @@ public class SCPOutputDataStatgeTask extends AbstractSCPTask {
throw new TaskException("Scp attempt failed", e);
} catch (JSchException | IOException e) {
throw new TaskException("Scp failed", e);
+ } catch (TException e) {
+ throw new TaskException("Invalid task invocation");
}
return null;
}
http://git-wip-us.apache.org/repos/asf/airavata/blob/7f4f1826/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/impl/ExperimentRegistry.java
----------------------------------------------------------------------
diff --git a/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/impl/ExperimentRegistry.java b/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/impl/ExperimentRegistry.java
index 75218b7..90f53bb 100644
--- a/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/impl/ExperimentRegistry.java
+++ b/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/impl/ExperimentRegistry.java
@@ -399,7 +399,7 @@ public class ExperimentRegistry {
taskResource.setCreationTime(AiravataUtils.getTime(task.getCreationTime()));
taskResource.setLastUpdateTime(AiravataUtils.getTime(task.getLastUpdateTime()));
taskResource.setTaskDetail(task.getTaskDetail());
- taskResource.setTaskInternalStore(task.getTaskInternalStore());
+ taskResource.setSubTaskModel(task.getSubTaskModel());
taskResource.save();
TaskStatus taskStatus = new TaskStatus();
@@ -783,7 +783,7 @@ public class ExperimentRegistry {
taskResource.setCreationTime(AiravataUtils.getTime(task.getCreationTime()));
taskResource.setLastUpdateTime(AiravataUtils.getTime(task.getLastUpdateTime()));
taskResource.setTaskDetail(task.getTaskDetail());
- taskResource.setTaskInternalStore(task.getTaskInternalStore());
+ taskResource.setSubTaskModel(task.getSubTaskModel());
taskResource.save();
if(task.getTaskError() != null) {
http://git-wip-us.apache.org/repos/asf/airavata/blob/7f4f1826/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/model/Task.java
----------------------------------------------------------------------
diff --git a/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/model/Task.java b/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/model/Task.java
index 5982044..b6386c3 100644
--- a/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/model/Task.java
+++ b/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/model/Task.java
@@ -38,7 +38,7 @@ public class Task {
private Timestamp creationTime;
private Timestamp lastUpdateTime;
private String taskDetail;
- private Byte taskInternalStore;
+ private byte[] setSubTaskModel;
private Process process;
private Collection<TaskError> taskErrors;
private Collection<TaskStatus> taskStatuses;
@@ -104,13 +104,13 @@ public class Task {
}
@Basic
- @Column(name = "TASK_INTERNAL_STORE")
- public Byte getTaskInternalStore() {
- return taskInternalStore;
+ @Column(name = "SUB_TASK_MODEL")
+ public byte[] getSetSubTaskModel() {
+ return setSubTaskModel;
}
- public void setTaskInternalStore(Byte taskInternalStore) {
- this.taskInternalStore = taskInternalStore;
+ public void setSetSubTaskModel(byte[] taskInternalStore) {
+ this.setSubTaskModel = taskInternalStore;
}
// @Override
http://git-wip-us.apache.org/repos/asf/airavata/blob/7f4f1826/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/resources/TaskResource.java
----------------------------------------------------------------------
diff --git a/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/resources/TaskResource.java b/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/resources/TaskResource.java
index 5d90e4f..1b8a0b4 100644
--- a/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/resources/TaskResource.java
+++ b/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/resources/TaskResource.java
@@ -47,7 +47,7 @@ public class TaskResource extends AbstractExpCatResource {
private Timestamp creationTime;
private Timestamp lastUpdateTime;
private String taskDetail;
- private Byte taskInternalStore;
+ private byte[] subTaskModel;
public String getTaskId() {
return taskId;
@@ -97,12 +97,12 @@ public class TaskResource extends AbstractExpCatResource {
this.taskDetail = taskDetail;
}
- public Byte getTaskInternalStore() {
- return taskInternalStore;
+ public byte[] getSubTaskModel() {
+ return subTaskModel;
}
- public void setTaskInternalStore(Byte taskInternalStore) {
- this.taskInternalStore = taskInternalStore;
+ public void setSubTaskModel(byte[] subTaskModel) {
+ this.subTaskModel = subTaskModel;
}
public ExperimentCatResource create(ResourceType type) throws RegistryException {
@@ -305,7 +305,7 @@ public class TaskResource extends AbstractExpCatResource {
task.setCreationTime(creationTime);
task.setLastUpdateTime(lastUpdateTime);
task.setTaskDetail(taskDetail);
- task.setTaskInternalStore(taskInternalStore);
+ task.setSetSubTaskModel(subTaskModel);
em.persist(task);
em.getTransaction().commit();
em.close();
http://git-wip-us.apache.org/repos/asf/airavata/blob/7f4f1826/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/resources/Utils.java
----------------------------------------------------------------------
diff --git a/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/resources/Utils.java b/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/resources/Utils.java
index 90fb348..baae694 100644
--- a/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/resources/Utils.java
+++ b/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/resources/Utils.java
@@ -583,7 +583,7 @@ public class Utils {
taskResource.setCreationTime(o.getCreationTime());
taskResource.setLastUpdateTime(o.getLastUpdateTime());
taskResource.setTaskDetail(o.getTaskDetail());
- taskResource.setTaskInternalStore(o.getTaskInternalStore());
+ taskResource.setSubTaskModel(o.getSetSubTaskModel());
}
return taskResource;
}
http://git-wip-us.apache.org/repos/asf/airavata/blob/7f4f1826/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/utils/ThriftDataModelConversion.java
----------------------------------------------------------------------
diff --git a/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/utils/ThriftDataModelConversion.java b/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/utils/ThriftDataModelConversion.java
index a149dd6..bdaae52 100644
--- a/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/utils/ThriftDataModelConversion.java
+++ b/modules/registry/registry-core/src/main/java/org/apache/airavata/registry/core/experiment/catalog/utils/ThriftDataModelConversion.java
@@ -370,7 +370,7 @@ public class ThriftDataModelConversion {
model.setCreationTime(taskResource.getCreationTime().getTime());
model.setLastUpdateTime(taskResource.getLastUpdateTime().getTime());
model.setTaskDetail(taskResource.getTaskDetail());
- model.setTaskInternalStore(taskResource.getTaskInternalStore());
+ model.setSubTaskModel(taskResource.getSubTaskModel());
model.setTaskStatus(getTaskStatus(taskResource.getTaskStatus()));
model.setTaskError(getErrorModel(taskResource.getTaskError()));
http://git-wip-us.apache.org/repos/asf/airavata/blob/7f4f1826/modules/registry/registry-core/src/test/java/org/apache/airavata/experiment/catalog/JobResourceTest.java
----------------------------------------------------------------------
diff --git a/modules/registry/registry-core/src/test/java/org/apache/airavata/experiment/catalog/JobResourceTest.java b/modules/registry/registry-core/src/test/java/org/apache/airavata/experiment/catalog/JobResourceTest.java
index 7ee72cb..228b721 100644
--- a/modules/registry/registry-core/src/test/java/org/apache/airavata/experiment/catalog/JobResourceTest.java
+++ b/modules/registry/registry-core/src/test/java/org/apache/airavata/experiment/catalog/JobResourceTest.java
@@ -19,6 +19,7 @@
package org.apache.airavata.experiment.catalog;
+import org.apache.airavata.model.task.DataStagingTaskModel;
import org.apache.airavata.model.task.TaskTypes;
import org.apache.airavata.registry.core.experiment.catalog.ResourceType;
import org.apache.airavata.registry.core.experiment.catalog.resources.ExperimentResource;
@@ -69,7 +70,7 @@ public class JobResourceTest extends AbstractResourceTest{
taskResource.setParentProcessId(processId);
taskResource.setTaskType(TaskTypes.DATA_STAGING.toString());
taskResource.setTaskDetail("task detail");
- taskResource.setTaskInternalStore((byte)2);
+ taskResource.setSubTaskModel(new DataStagingTaskModel("source","destination").toString().getBytes());
taskResource.save();
jobResource = (JobResource)taskResource.create(ResourceType.JOB);
http://git-wip-us.apache.org/repos/asf/airavata/blob/7f4f1826/modules/registry/registry-core/src/test/java/org/apache/airavata/experiment/catalog/TaskResourceTest.java
----------------------------------------------------------------------
diff --git a/modules/registry/registry-core/src/test/java/org/apache/airavata/experiment/catalog/TaskResourceTest.java b/modules/registry/registry-core/src/test/java/org/apache/airavata/experiment/catalog/TaskResourceTest.java
index 8ff32eb..85d3501 100644
--- a/modules/registry/registry-core/src/test/java/org/apache/airavata/experiment/catalog/TaskResourceTest.java
+++ b/modules/registry/registry-core/src/test/java/org/apache/airavata/experiment/catalog/TaskResourceTest.java
@@ -19,6 +19,7 @@
package org.apache.airavata.experiment.catalog;
+import org.apache.airavata.model.task.DataStagingTaskModel;
import org.apache.airavata.model.task.TaskTypes;
import org.apache.airavata.registry.core.experiment.catalog.ResourceType;
import org.apache.airavata.registry.core.experiment.catalog.resources.ExperimentResource;
@@ -66,7 +67,7 @@ public class TaskResourceTest extends AbstractResourceTest{
taskResource.setParentProcessId(processId);
taskResource.setTaskType(TaskTypes.DATA_STAGING.toString());
taskResource.setTaskDetail("task detail");
- taskResource.setTaskInternalStore((byte)2);
+ taskResource.setSubTaskModel(new DataStagingTaskModel("source","destination").toString().getBytes());
taskResource.save();
}
http://git-wip-us.apache.org/repos/asf/airavata/blob/7f4f1826/thrift-interface-descriptions/airavata-api/task_model.thrift
----------------------------------------------------------------------
diff --git a/thrift-interface-descriptions/airavata-api/task_model.thrift b/thrift-interface-descriptions/airavata-api/task_model.thrift
index be454a5..c503614 100644
--- a/thrift-interface-descriptions/airavata-api/task_model.thrift
+++ b/thrift-interface-descriptions/airavata-api/task_model.thrift
@@ -44,7 +44,7 @@ enum TaskTypes {
* taskDetail:
* A friendly description of the task, usally used to communicate information to users.
*
- * taskInternalStore:
+ * subTaskModel:
* A generic byte object for the Task developer to store internal serialized data into registry catalogs.
*/
struct TaskModel {