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 <[email protected]> Authored: Fri Jun 26 00:20:08 2015 -0400 Committer: Shameera Rathanyaka <[email protected]> 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 {
