You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by ra...@apache.org on 2014/03/26 16:36:23 UTC
[1/3] git commit: Added tests for SSHprovider and fixed bugs.
AIRAVATA-1082
Repository: airavata
Updated Branches:
refs/heads/master 98519a785 -> 007a111e8
Added tests for SSHprovider and fixed bugs. AIRAVATA-1082
Project: http://git-wip-us.apache.org/repos/asf/airavata/repo
Commit: http://git-wip-us.apache.org/repos/asf/airavata/commit/04c502b8
Tree: http://git-wip-us.apache.org/repos/asf/airavata/tree/04c502b8
Diff: http://git-wip-us.apache.org/repos/asf/airavata/diff/04c502b8
Branch: refs/heads/master
Commit: 04c502b8c2dbab66c6f64a05a1433e8a800db11b
Parents: cb826d3
Author: raminder <ra...@apache.org>
Authored: Wed Mar 26 10:42:25 2014 -0400
Committer: raminder <ra...@apache.org>
Committed: Wed Mar 26 10:42:25 2014 -0400
----------------------------------------------------------------------
.../client/samples/CreateLaunchExperiment.java | 45 ++++++
.../airavata/client/tools/DocumentCreator.java | 60 ++++++++
.../org/apache/airavata/gfac/cpi/GFacImpl.java | 16 +--
.../gfac/handler/SCPDirectorySetupHandler.java | 2 +-
.../gfac/provider/impl/SSHProvider.java | 11 +-
.../airavata/job/monitor/AMQPMonitorTest.java | 141 +++++++++++++++++++
.../airavata/job/monitor/QstatMonitorTest.java | 141 +++++++++++++++++++
.../airavata/job/monitor/AMQPMonitorTest.java | 138 ------------------
.../airavata/job/monitor/QstatMonitorTest.java | 138 ------------------
9 files changed, 403 insertions(+), 289 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/airavata/blob/04c502b8/airavata-api/airavata-client-sdks/java-client-samples/src/main/java/org/apache/airavata/client/samples/CreateLaunchExperiment.java
----------------------------------------------------------------------
diff --git a/airavata-api/airavata-client-sdks/java-client-samples/src/main/java/org/apache/airavata/client/samples/CreateLaunchExperiment.java b/airavata-api/airavata-client-sdks/java-client-samples/src/main/java/org/apache/airavata/client/samples/CreateLaunchExperiment.java
index 7c20dff..a178712 100644
--- a/airavata-api/airavata-client-sdks/java-client-samples/src/main/java/org/apache/airavata/client/samples/CreateLaunchExperiment.java
+++ b/airavata-api/airavata-client-sdks/java-client-samples/src/main/java/org/apache/airavata/client/samples/CreateLaunchExperiment.java
@@ -62,6 +62,7 @@ public class CreateLaunchExperiment {
System.out.println("API version is " + airavata.GetAPIVersion());
addDescriptors();
final String expId = createExperimentForLocalHost(airavata);
+// final String expId = createExperimentForSSHHost(airavata);
// final String expId = createUS3ExperimentForTrestles(airavata);
// final String expId = createExperimentForStampede(airavata);
System.out.println("Experiment ID : " + expId);
@@ -124,6 +125,7 @@ public class CreateLaunchExperiment {
try {
DocumentCreator documentCreator = new DocumentCreator(getAiravataAPI());
documentCreator.createLocalHostDocs();
+ documentCreator.createSSHHostDocs();
documentCreator.createGramDocs();
documentCreator.createPBSDocsForOGCE();
documentCreator.createSlurmDocs();
@@ -236,6 +238,49 @@ public class CreateLaunchExperiment {
throw new TException(e);
}
}
+
+ public static String createExperimentForSSHHost(Airavata.Client client) throws TException {
+ try{
+ List<DataObjectType> exInputs = new ArrayList<DataObjectType>();
+ DataObjectType input = new DataObjectType();
+ input.setKey("echo_input");
+ input.setType(DataType.STRING.toString());
+ input.setValue("echo_output=Hello World");
+ exInputs.add(input);
+
+ List<DataObjectType> exOut = new ArrayList<DataObjectType>();
+ DataObjectType output = new DataObjectType();
+ output.setKey("echo_output");
+ output.setType(DataType.STRING.toString());
+ output.setValue("");
+ exOut.add(output);
+
+ Experiment simpleExperiment =
+ ExperimentModelUtil.createSimpleExperiment("project1", "admin", "sshEchoExperiment", "SSHEcho1", "SSHEcho1", exInputs);
+ simpleExperiment.setExperimentOutputs(exOut);
+
+ ComputationalResourceScheduling scheduling = ExperimentModelUtil.createComputationResourceScheduling("gw111.iu.xsede.org", 1, 1, 1, "normal", 0, 0, 1, "sds128");
+ scheduling.setResourceHostId("gw111.iu.xsede.org");
+ UserConfigurationData userConfigurationData = new UserConfigurationData();
+ userConfigurationData.setAiravataAutoSchedule(false);
+ userConfigurationData.setOverrideManualScheduledParams(false);
+ userConfigurationData.setComputationalResourceScheduling(scheduling);
+ simpleExperiment.setUserConfigurationData(userConfigurationData);
+ return client.createExperiment(simpleExperiment);
+ } catch (AiravataSystemException e) {
+ logger.error("Error occured while creating the experiment...", e.getMessage());
+ throw new AiravataSystemException(e);
+ } catch (InvalidRequestException e) {
+ logger.error("Error occured while creating the experiment...", e.getMessage());
+ throw new InvalidRequestException(e);
+ } catch (AiravataClientException e) {
+ logger.error("Error occured while creating the experiment...", e.getMessage());
+ throw new AiravataClientException(e);
+ }catch (TException e) {
+ logger.error("Error occured while creating the experiment...", e.getMessage());
+ throw new TException(e);
+ }
+ }
public static String createExperimentForStampede(Airavata.Client client) throws TException {
try{
List<DataObjectType> exInputs = new ArrayList<DataObjectType>();
http://git-wip-us.apache.org/repos/asf/airavata/blob/04c502b8/modules/airavata-client/src/main/java/org/apache/airavata/client/tools/DocumentCreator.java
----------------------------------------------------------------------
diff --git a/modules/airavata-client/src/main/java/org/apache/airavata/client/tools/DocumentCreator.java b/modules/airavata-client/src/main/java/org/apache/airavata/client/tools/DocumentCreator.java
index 9fbcbd2..dad3e70 100644
--- a/modules/airavata-client/src/main/java/org/apache/airavata/client/tools/DocumentCreator.java
+++ b/modules/airavata-client/src/main/java/org/apache/airavata/client/tools/DocumentCreator.java
@@ -39,6 +39,7 @@ import org.apache.airavata.schemas.gfac.OutputParameterType;
import org.apache.airavata.schemas.gfac.ParameterType;
import org.apache.airavata.schemas.gfac.ProjectAccountType;
import org.apache.airavata.schemas.gfac.QueueType;
+import org.apache.airavata.schemas.gfac.SSHHostType;
public class DocumentCreator {
@@ -112,6 +113,65 @@ public class DocumentCreator {
}
}
+
+ public void createSSHHostDocs() {
+ HostDescription descriptor = new HostDescription(SSHHostType.type);
+ descriptor.getType().setHostName("gw111.iu.xsede.org");
+ descriptor.getType().setHostAddress("gw111.iu.xsede.org");
+ try {
+ airavataAPI.getApplicationManager().saveHostDescription(descriptor);
+ } catch (AiravataAPIInvocationException e) {
+ e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
+ }
+
+ String serviceName = "SSHEcho1";
+ ServiceDescription serviceDescription = new ServiceDescription();
+ List<InputParameterType> inputParameters = new ArrayList<InputParameterType>();
+ List<OutputParameterType> outputParameters = new ArrayList<OutputParameterType>();
+ serviceDescription.getType().setName(serviceName);
+ serviceDescription.getType().setDescription("Echo service");
+ // Creating input parameters
+ InputParameterType parameter = InputParameterType.Factory.newInstance();
+ parameter.setParameterName("echo_input");
+ parameter.setParameterDescription("echo input");
+ ParameterType parameterType = parameter.addNewParameterType();
+ parameterType.setType(DataType.STRING);
+ parameterType.setName("String");
+ inputParameters.add(parameter);
+
+ // Creating output parameters
+ OutputParameterType outputParameter = OutputParameterType.Factory.newInstance();
+ outputParameter.setParameterName("echo_output");
+ outputParameter.setParameterDescription("Echo output");
+ ParameterType outputParaType = outputParameter.addNewParameterType();
+ outputParaType.setType(DataType.STRING);
+ outputParaType.setName("String");
+ outputParameters.add(outputParameter);
+
+ // Setting input and output parameters to serviceDescriptor
+ serviceDescription.getType().setInputParametersArray(inputParameters.toArray(new InputParameterType[]{}));
+ serviceDescription.getType().setOutputParametersArray(outputParameters.toArray(new OutputParameterType[]{}));
+
+ try {
+ airavataAPI.getApplicationManager().saveServiceDescription(serviceDescription);
+ } catch (AiravataAPIInvocationException e) {
+ e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
+ }
+
+ ApplicationDescription applicationDeploymentDescription = new ApplicationDescription();
+ ApplicationDeploymentDescriptionType applicationDeploymentDescriptionType = applicationDeploymentDescription
+ .getType();
+ applicationDeploymentDescriptionType.addNewApplicationName().setStringValue("SSHEchoApplication");
+ applicationDeploymentDescriptionType.setExecutableLocation("/bin/echo");
+ applicationDeploymentDescriptionType.setScratchWorkingDirectory("/tmp");
+
+ try {
+ airavataAPI.getApplicationManager().saveApplicationDescription(serviceName, "gw111.iu.xsede.org", applicationDeploymentDescription);
+ } catch (AiravataAPIInvocationException e) {
+ e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
+ }
+
+ }
public void createGramDocs() {
/*
http://git-wip-us.apache.org/repos/asf/airavata/blob/04c502b8/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/cpi/GFacImpl.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/cpi/GFacImpl.java b/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/cpi/GFacImpl.java
index f79561c..775695a 100644
--- a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/cpi/GFacImpl.java
+++ b/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/cpi/GFacImpl.java
@@ -431,7 +431,6 @@ public class GFacImpl implements GFac {
String sshPassword = configurationProperties.getProperty(Constants.SSH_PASSWORD);
String sshPublicKey = configurationProperties.getProperty(Constants.SSH_PUBLIC_KEY);
SSHSecurityContext sshSecurityContext = new SSHSecurityContext();
- if (((SSHHostType) registeredHost.getType()).getHpcResource()) {
AuthenticationInfo authenticationInfo = null;
// we give higher preference to the password over keypair ssh authentication
if (sshPassword != null) {
@@ -443,8 +442,11 @@ public class GFacImpl implements GFac {
Cluster pbsCluster = null;
try {
- String installedParentPath = ((HpcApplicationDeploymentType)
- jobExecutionContext.getApplicationContext().getApplicationDeploymentDescription().getType()).getInstalledParentPath();
+ String installedParentPath = "/";
+ if(((SSHHostType) registeredHost.getType()).getHpcResource()){
+ installedParentPath = ((HpcApplicationDeploymentType)
+ jobExecutionContext.getApplicationContext().getApplicationDeploymentDescription().getType()).getInstalledParentPath();
+ }
pbsCluster = new PBSCluster(serverInfo, authenticationInfo,
CommonUtils.getPBSJobManager(installedParentPath));
} catch (SSHApiException e) {
@@ -452,13 +454,7 @@ public class GFacImpl implements GFac {
}
sshSecurityContext.setPbsCluster(pbsCluster);
sshSecurityContext.setUsername(sshUserName);
- } else {
- sshSecurityContext = new SSHSecurityContext();
- sshSecurityContext.setUsername(sshUserName);
- sshSecurityContext.setPrivateKeyLoc(sshPrivateKey);
- sshSecurityContext.setKeyPass(sshPrivateKeyPass);
- }
- jobExecutionContext.addSecurityContext(SSHSecurityContext.SSH_SECURITY_CONTEXT, sshSecurityContext);
+ jobExecutionContext.addSecurityContext(SSHSecurityContext.SSH_SECURITY_CONTEXT, sshSecurityContext);
}
}
http://git-wip-us.apache.org/repos/asf/airavata/blob/04c502b8/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/handler/SCPDirectorySetupHandler.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/handler/SCPDirectorySetupHandler.java b/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/handler/SCPDirectorySetupHandler.java
index 7640e8b..ead2ad4 100644
--- a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/handler/SCPDirectorySetupHandler.java
+++ b/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/handler/SCPDirectorySetupHandler.java
@@ -60,7 +60,7 @@ public class SCPDirectorySetupHandler extends AbstractHandler{
}
if (cluster == null) {
throw new GFacHandlerException("Security context is not set properly");
- } else {
+ } else {
log.info("Successfully retrieved the Security Context");
}
ApplicationDeploymentDescriptionType app = jobExecutionContext.getApplicationContext().getApplicationDeploymentDescription().getType();
http://git-wip-us.apache.org/repos/asf/airavata/blob/04c502b8/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/impl/SSHProvider.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/impl/SSHProvider.java b/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/impl/SSHProvider.java
index f857d79..5914dd3 100644
--- a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/impl/SSHProvider.java
+++ b/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/impl/SSHProvider.java
@@ -45,6 +45,7 @@ import org.apache.airavata.gfac.utils.GFacUtils;
import org.apache.airavata.gsi.ssh.api.Cluster;
import org.apache.airavata.gsi.ssh.api.CommandExecutor;
import org.apache.airavata.gsi.ssh.api.SSHApiException;
+import org.apache.airavata.gsi.ssh.api.job.JobDescriptor;
import org.apache.airavata.gsi.ssh.impl.RawCommandInfo;
import org.apache.airavata.gsi.ssh.impl.StandardOutReader;
import org.apache.airavata.model.workspace.experiment.JobState;
@@ -77,7 +78,12 @@ public class SSHProvider extends AbstractProvider implements GFacProvider{
ApplicationDeploymentDescriptionType app = jobExecutionContext.getApplicationContext().getApplicationDeploymentDescription().getType();
String remoteFile = app.getStaticWorkingDirectory() + File.separatorChar + Constants.EXECUTABLE_NAME;
+ details.setJobID(taskID);
details.setJobDescription(remoteFile);
+ jobExecutionContext.setJobDetails(details);
+ JobDescriptor jobDescriptor = GFacUtils.createJobDescriptor(jobExecutionContext, app, null);
+ details.setJobDescription(jobDescriptor.toXML());
+
GFacUtils.saveJobStatus(details, JobState.SETUP, taskID);
log.info(remoteFile);
try {
@@ -101,7 +107,8 @@ public class SSHProvider extends AbstractProvider implements GFacProvider{
*/
String execuable = app.getStaticWorkingDirectory() + File.separatorChar + Constants.EXECUTABLE_NAME;
details.setJobDescription(execuable);
- GFacUtils.updateJobStatus(details, JobState.SUBMITTED);
+
+// GFacUtils.updateJobStatus(details, JobState.SUBMITTED);
RawCommandInfo rawCommandInfo = new RawCommandInfo("/bin/chmod 755 " + execuable + "; " + execuable);
StandardOutReader jobIDReaderCommandOutput = new StandardOutReader();
@@ -111,7 +118,7 @@ public class SSHProvider extends AbstractProvider implements GFacProvider{
log.info("stdout=" + stdOutputString);
- GFacUtils.updateJobStatus(details, JobState.COMPLETE);
+// GFacUtils.updateJobStatus(details, JobState.COMPLETE);
} catch (Exception e) {
throw new GFacProviderException(e.getMessage(), e);
} finally {
http://git-wip-us.apache.org/repos/asf/airavata/blob/04c502b8/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/monitor/AMQPMonitorTest.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/monitor/AMQPMonitorTest.java b/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/monitor/AMQPMonitorTest.java
new file mode 100644
index 0000000..4a1126a
--- /dev/null
+++ b/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/monitor/AMQPMonitorTest.java
@@ -0,0 +1,141 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+*/
+package org.apache.airavata.job.monitor;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.airavata.commons.gfac.type.HostDescription;
+import org.apache.airavata.gsi.ssh.api.Cluster;
+import org.apache.airavata.gsi.ssh.api.SSHApiException;
+import org.apache.airavata.gsi.ssh.api.ServerInfo;
+import org.apache.airavata.gsi.ssh.api.authentication.GSIAuthenticationInfo;
+import org.apache.airavata.gsi.ssh.api.job.JobDescriptor;
+import org.apache.airavata.gsi.ssh.impl.PBSCluster;
+import org.apache.airavata.gsi.ssh.impl.authentication.MyProxyAuthenticationInfo;
+import org.apache.airavata.job.monitor.exception.AiravataMonitorException;
+import org.apache.airavata.job.monitor.impl.push.amqp.AMQPMonitor;
+import org.apache.airavata.schemas.gfac.GsisshHostType;
+import org.junit.Before;
+import org.junit.Test;
+
+public class AMQPMonitorTest {
+ private MonitorManager monitorManager;
+
+ private String myProxyUserName;
+ private String myProxyPassword;
+ private String certificateLocation;
+ private String pbsFilePath;
+ private String workingDirectory;
+ private HostDescription hostDescription;
+
+ @Before
+ public void setUp() throws Exception {
+ System.setProperty("myproxy.user", "ogce");
+ System.setProperty("myproxy.password", "");
+ System.setProperty("basedir", "/Users/lahirugunathilake/work/airavata/sandbox/gsissh");
+ System.setProperty("gsi.working.directory", "/home/ogce");
+ myProxyUserName = System.getProperty("myproxy.user");
+ myProxyPassword = System.getProperty("myproxy.password");
+ workingDirectory = System.getProperty("gsi.working.directory");
+ String pomDirectory = System.getProperty("basedir");
+ certificateLocation = "/Users/lahirugunathilake/Downloads/certificates";
+ if (myProxyUserName == null || myProxyPassword == null || workingDirectory == null) {
+ System.out.println(">>>>>> Please run tests with my proxy user name and password. " +
+ "E.g :- mvn clean install -Dmyproxy.user=xxx -Dmyproxy.password=xxx -Dgsi.working.directory=/path<<<<<<<");
+ throw new Exception("Need my proxy user name password to run tests.");
+ }
+
+ monitorManager = new MonitorManager();
+ AMQPMonitor amqpMonitor = new
+ AMQPMonitor(monitorManager.getMonitorPublisher(),
+ monitorManager.getPullQueue(), monitorManager.getFinishQueue(),"/Users/lahirugunathilake/Downloads/x509up_u503876","xsede_private",
+ Arrays.asList("info1.dyn.teragrid.org,info2.dyn.teragrid.org".split(",")));
+ try {
+ monitorManager.addPushMonitor(amqpMonitor);
+ monitorManager.launchMonitor();
+ } catch (AiravataMonitorException e) {
+ e.printStackTrace();
+ }
+
+ hostDescription = new HostDescription(GsisshHostType.type);
+ hostDescription.getType().setHostAddress("gordon.sdsc.xsede.org");
+ hostDescription.getType().setHostName("gsissh-gordon");
+ }
+
+ @Test
+ public void testAMQPMonitor() throws SSHApiException {
+ /* now have to submit a job to some machine and add that job to the queue */
+ //Create authentication
+ GSIAuthenticationInfo authenticationInfo
+ = new MyProxyAuthenticationInfo(myProxyUserName, myProxyPassword, "myproxy.teragrid.org",
+ 7512, 17280000, certificateLocation);
+
+ // Server info
+ ServerInfo serverInfo = new ServerInfo("ogce", "trestles.sdsc.edu");
+
+
+ Cluster pbsCluster = new
+ PBSCluster(serverInfo, authenticationInfo, org.apache.airavata.gsi.ssh.util.CommonUtils.getPBSJobManager("/opt/torque/bin/"));
+
+
+ // Execute command
+ System.out.println("Target PBS file path: " + workingDirectory);
+ // constructing the job object
+ JobDescriptor jobDescriptor = new JobDescriptor();
+ jobDescriptor.setWorkingDirectory(workingDirectory);
+ jobDescriptor.setShellName("/bin/bash");
+ jobDescriptor.setJobName("GSI_SSH_SLEEP_JOB");
+ jobDescriptor.setExecutablePath("/bin/echo");
+ jobDescriptor.setAllEnvExport(true);
+ jobDescriptor.setMailOptions("n");
+ jobDescriptor.setStandardOutFile(workingDirectory + File.separator + "application.out");
+ jobDescriptor.setStandardErrorFile(workingDirectory + File.separator + "application.err");
+ jobDescriptor.setNodes(1);
+ jobDescriptor.setProcessesPerNode(1);
+ jobDescriptor.setQueueName("normal");
+ jobDescriptor.setMaxWallTime("60");
+ jobDescriptor.setAcountString("sds128");
+ List<String> inputs = new ArrayList<String>();
+ jobDescriptor.setOwner("ogce");
+ inputs.add("Hello World");
+ jobDescriptor.setInputValues(inputs);
+ //finished construction of job object
+ System.out.println(jobDescriptor.toXML());
+ String jobID = pbsCluster.submitBatchJob(jobDescriptor);
+ System.out.println(jobID);
+ try {
+ monitorManager.addAJobToMonitor(new MonitorID(hostDescription, jobID,null,null, "ogce"));
+ }catch (InterruptedException e) {
+ e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
+ }
+ catch (AiravataMonitorException e) {
+ e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
+ }
+ try {
+ Thread.sleep(10000);
+ } catch (InterruptedException e) {
+ e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/airavata/blob/04c502b8/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/monitor/QstatMonitorTest.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/monitor/QstatMonitorTest.java b/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/monitor/QstatMonitorTest.java
new file mode 100644
index 0000000..fdfe841
--- /dev/null
+++ b/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/monitor/QstatMonitorTest.java
@@ -0,0 +1,141 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+*/
+package org.apache.airavata.job.monitor;
+
+import org.apache.airavata.commons.gfac.type.HostDescription;
+import org.apache.airavata.gsi.ssh.api.Cluster;
+import org.apache.airavata.gsi.ssh.api.SSHApiException;
+import org.apache.airavata.gsi.ssh.api.ServerInfo;
+import org.apache.airavata.gsi.ssh.api.authentication.GSIAuthenticationInfo;
+import org.apache.airavata.gsi.ssh.api.job.JobDescriptor;
+import org.apache.airavata.gsi.ssh.impl.PBSCluster;
+import org.apache.airavata.gsi.ssh.impl.authentication.MyProxyAuthenticationInfo;
+import org.apache.airavata.gsi.ssh.util.CommonUtils;
+import org.apache.airavata.job.monitor.exception.AiravataMonitorException;
+import org.apache.airavata.job.monitor.impl.pull.qstat.QstatMonitor;
+import org.apache.airavata.schemas.gfac.GsisshHostType;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.List;
+
+public class QstatMonitorTest {
+ private MonitorManager monitorManager;
+
+ private String myProxyUserName;
+ private String myProxyPassword;
+ private String certificateLocation;
+ private String pbsFilePath;
+ private String workingDirectory;
+ private HostDescription hostDescription;
+
+ @Before
+ public void setUp() throws Exception {
+ System.setProperty("myproxy.user", "ogce");
+ System.setProperty("myproxy.password", "");
+ System.setProperty("basedir", "/Users/lahirugunathilake/work/airavata/sandbox/gsissh");
+ System.setProperty("gsi.working.directory", "/home/ogce");
+ myProxyUserName = System.getProperty("myproxy.user");
+ myProxyPassword = System.getProperty("myproxy.password");
+ workingDirectory = System.getProperty("gsi.working.directory");
+ String pomDirectory = System.getProperty("basedir");
+ certificateLocation = "/Users/lahirugunathilake/Downloads/certificates";
+ if (myProxyUserName == null || myProxyPassword == null || workingDirectory == null) {
+ System.out.println(">>>>>> Please run tests with my proxy user name and password. " +
+ "E.g :- mvn clean install -Dmyproxy.user=xxx -Dmyproxy.password=xxx -Dgsi.working.directory=/path<<<<<<<");
+ throw new Exception("Need my proxy user name password to run tests.");
+ }
+
+ monitorManager = new MonitorManager();
+ QstatMonitor qstatMonitor = new
+ QstatMonitor(monitorManager.getPullQueue(), monitorManager.getMonitorPublisher());
+ try {
+ monitorManager.addPullMonitor(qstatMonitor);
+ monitorManager.launchMonitor();
+ } catch (AiravataMonitorException e) {
+ e.printStackTrace();
+ }
+
+ hostDescription = new HostDescription(GsisshHostType.type);
+ hostDescription.getType().setHostAddress("trestles.sdsc.edu");
+ hostDescription.getType().setHostName("gsissh-gordon");
+ }
+
+ @Test
+ public void testQstatMonitor() throws SSHApiException {
+ /* now have to submit a job to some machine and add that job to the queue */
+ //Create authentication
+ GSIAuthenticationInfo authenticationInfo
+ = new MyProxyAuthenticationInfo(myProxyUserName, myProxyPassword, "myproxy.teragrid.org",
+ 7512, 17280000, certificateLocation);
+
+ // Server info
+ ServerInfo serverInfo = new ServerInfo("ogce", hostDescription.getType().getHostAddress());
+
+
+ Cluster pbsCluster = new PBSCluster(serverInfo, authenticationInfo, CommonUtils.getPBSJobManager("/opt/torque/bin/"));
+
+
+ // Execute command
+ System.out.println("Target PBS file path: " + workingDirectory);
+ // constructing the job object
+ JobDescriptor jobDescriptor = new JobDescriptor();
+ jobDescriptor.setWorkingDirectory(workingDirectory);
+ jobDescriptor.setShellName("/bin/bash");
+ jobDescriptor.setJobName("GSI_SSH_SLEEP_JOB");
+ jobDescriptor.setExecutablePath("/bin/echo");
+ jobDescriptor.setAllEnvExport(true);
+ jobDescriptor.setMailOptions("n");
+ jobDescriptor.setStandardOutFile(workingDirectory + File.separator + "application.out");
+ jobDescriptor.setStandardErrorFile(workingDirectory + File.separator + "application.err");
+ jobDescriptor.setNodes(1);
+ jobDescriptor.setProcessesPerNode(1);
+ jobDescriptor.setQueueName("normal");
+ jobDescriptor.setMaxWallTime("60");
+ jobDescriptor.setAcountString("sds128");
+ List<String> inputs = new ArrayList<String>();
+ jobDescriptor.setOwner("ogce");
+ inputs.add("Hello World");
+ jobDescriptor.setInputValues(inputs);
+ //finished construction of job object
+ System.out.println(jobDescriptor.toXML());
+ for (int i = 0; i < 1; i++) {
+ String jobID = pbsCluster.submitBatchJob(jobDescriptor);
+ MonitorID monitorID = new MonitorID(hostDescription, jobID,null,null, "ogce");
+ monitorID.setAuthenticationInfo(authenticationInfo);
+ try {
+ monitorManager.addAJobToMonitor(monitorID);
+ }catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ catch (AiravataMonitorException e) {
+ e.printStackTrace();
+ }
+ }
+ try {
+ Thread.sleep(10000000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/airavata/blob/04c502b8/tools/job-monitor/src/test/java/org/apache/airavata/job/monitor/AMQPMonitorTest.java
----------------------------------------------------------------------
diff --git a/tools/job-monitor/src/test/java/org/apache/airavata/job/monitor/AMQPMonitorTest.java b/tools/job-monitor/src/test/java/org/apache/airavata/job/monitor/AMQPMonitorTest.java
deleted file mode 100644
index 4e35434..0000000
--- a/tools/job-monitor/src/test/java/org/apache/airavata/job/monitor/AMQPMonitorTest.java
+++ /dev/null
@@ -1,138 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
-*/
-package org.apache.airavata.job.monitor;
-
-import org.apache.airavata.commons.gfac.type.HostDescription;
-import org.apache.airavata.gsi.ssh.api.Cluster;
-import org.apache.airavata.gsi.ssh.api.SSHApiException;
-import org.apache.airavata.gsi.ssh.api.ServerInfo;
-import org.apache.airavata.gsi.ssh.api.authentication.GSIAuthenticationInfo;
-import org.apache.airavata.gsi.ssh.api.job.JobDescriptor;
-import org.apache.airavata.gsi.ssh.impl.PBSCluster;
-import org.apache.airavata.gsi.ssh.impl.authentication.MyProxyAuthenticationInfo;
-import org.apache.airavata.job.monitor.exception.AiravataMonitorException;
-import org.apache.airavata.job.monitor.impl.push.amqp.AMQPMonitor;
-import org.apache.airavata.schemas.gfac.GsisshHostType;
-import org.junit.Before;
-import org.junit.Test;
-
-import java.io.File;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-
-public class AMQPMonitorTest {
- private MonitorManager monitorManager;
-
- private String myProxyUserName;
- private String myProxyPassword;
- private String certificateLocation;
- private String pbsFilePath;
- private String workingDirectory;
- private HostDescription hostDescription;
-
- @Before
- public void setUp() throws Exception {
- System.setProperty("myproxy.user", "ogce");
- System.setProperty("myproxy.password", "");
- System.setProperty("basedir", "/Users/lahirugunathilake/work/airavata/sandbox/gsissh");
- System.setProperty("gsi.working.directory", "/home/ogce");
- myProxyUserName = System.getProperty("myproxy.user");
- myProxyPassword = System.getProperty("myproxy.password");
- workingDirectory = System.getProperty("gsi.working.directory");
- String pomDirectory = System.getProperty("basedir");
- certificateLocation = "/Users/lahirugunathilake/Downloads/certificates";
- if (myProxyUserName == null || myProxyPassword == null || workingDirectory == null) {
- System.out.println(">>>>>> Please run tests with my proxy user name and password. " +
- "E.g :- mvn clean install -Dmyproxy.user=xxx -Dmyproxy.password=xxx -Dgsi.working.directory=/path<<<<<<<");
- throw new Exception("Need my proxy user name password to run tests.");
- }
-
- monitorManager = new MonitorManager();
- AMQPMonitor amqpMonitor = new
- AMQPMonitor(monitorManager.getMonitorPublisher(),
- monitorManager.getPullQueue(), monitorManager.getFinishQueue(),"/Users/lahirugunathilake/Downloads/x509up_u503876","xsede_private",
- Arrays.asList("info1.dyn.teragrid.org,info2.dyn.teragrid.org".split(",")));
- try {
- monitorManager.addPushMonitor(amqpMonitor);
- monitorManager.launchMonitor();
- } catch (AiravataMonitorException e) {
- e.printStackTrace();
- }
-
- hostDescription = new HostDescription(GsisshHostType.type);
- hostDescription.getType().setHostAddress("gordon.sdsc.xsede.org");
- hostDescription.getType().setHostName("gsissh-gordon");
- }
-
- @Test
- public void testAMQPMonitor() throws SSHApiException {
- /* now have to submit a job to some machine and add that job to the queue */
- //Create authentication
- GSIAuthenticationInfo authenticationInfo
- = new MyProxyAuthenticationInfo(myProxyUserName, myProxyPassword, "myproxy.teragrid.org",
- 7512, 17280000, certificateLocation);
-
- // Server info
- ServerInfo serverInfo = new ServerInfo("ogce", "trestles.sdsc.edu");
-
-
- Cluster pbsCluster = new
- PBSCluster(serverInfo, authenticationInfo, org.apache.airavata.gsi.ssh.util.CommonUtils.getPBSJobManager("/opt/torque/bin/"));
-
-
- // Execute command
- System.out.println("Target PBS file path: " + workingDirectory);
- // constructing the job object
- JobDescriptor jobDescriptor = new JobDescriptor();
- jobDescriptor.setWorkingDirectory(workingDirectory);
- jobDescriptor.setShellName("/bin/bash");
- jobDescriptor.setJobName("GSI_SSH_SLEEP_JOB");
- jobDescriptor.setExecutablePath("/bin/echo");
- jobDescriptor.setAllEnvExport(true);
- jobDescriptor.setMailOptions("n");
- jobDescriptor.setStandardOutFile(workingDirectory + File.separator + "application.out");
- jobDescriptor.setStandardErrorFile(workingDirectory + File.separator + "application.err");
- jobDescriptor.setNodes(1);
- jobDescriptor.setProcessesPerNode(1);
- jobDescriptor.setQueueName("normal");
- jobDescriptor.setMaxWallTime("60");
- jobDescriptor.setAcountString("sds128");
- List<String> inputs = new ArrayList<String>();
- jobDescriptor.setOwner("ogce");
- inputs.add("Hello World");
- jobDescriptor.setInputValues(inputs);
- //finished construction of job object
- System.out.println(jobDescriptor.toXML());
- String jobID = pbsCluster.submitBatchJob(jobDescriptor);
- System.out.println(jobID);
- try {
- monitorManager.addAJobToMonitor(new MonitorID(hostDescription, jobID,null,null, "ogce"));
- } catch (AiravataMonitorException e) {
- e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
- }
- try {
- Thread.sleep(10000);
- } catch (InterruptedException e) {
- e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/airavata/blob/04c502b8/tools/job-monitor/src/test/java/org/apache/airavata/job/monitor/QstatMonitorTest.java
----------------------------------------------------------------------
diff --git a/tools/job-monitor/src/test/java/org/apache/airavata/job/monitor/QstatMonitorTest.java b/tools/job-monitor/src/test/java/org/apache/airavata/job/monitor/QstatMonitorTest.java
deleted file mode 100644
index 126b8ae..0000000
--- a/tools/job-monitor/src/test/java/org/apache/airavata/job/monitor/QstatMonitorTest.java
+++ /dev/null
@@ -1,138 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
-*/
-package org.apache.airavata.job.monitor;
-
-import org.apache.airavata.commons.gfac.type.HostDescription;
-import org.apache.airavata.gsi.ssh.api.Cluster;
-import org.apache.airavata.gsi.ssh.api.SSHApiException;
-import org.apache.airavata.gsi.ssh.api.ServerInfo;
-import org.apache.airavata.gsi.ssh.api.authentication.GSIAuthenticationInfo;
-import org.apache.airavata.gsi.ssh.api.job.JobDescriptor;
-import org.apache.airavata.gsi.ssh.impl.PBSCluster;
-import org.apache.airavata.gsi.ssh.impl.authentication.MyProxyAuthenticationInfo;
-import org.apache.airavata.gsi.ssh.util.CommonUtils;
-import org.apache.airavata.job.monitor.exception.AiravataMonitorException;
-import org.apache.airavata.job.monitor.impl.pull.qstat.QstatMonitor;
-import org.apache.airavata.schemas.gfac.GsisshHostType;
-import org.junit.Before;
-import org.junit.Test;
-
-import java.io.File;
-import java.util.ArrayList;
-import java.util.List;
-
-public class QstatMonitorTest {
- private MonitorManager monitorManager;
-
- private String myProxyUserName;
- private String myProxyPassword;
- private String certificateLocation;
- private String pbsFilePath;
- private String workingDirectory;
- private HostDescription hostDescription;
-
- @Before
- public void setUp() throws Exception {
- System.setProperty("myproxy.user", "ogce");
- System.setProperty("myproxy.password", "");
- System.setProperty("basedir", "/Users/lahirugunathilake/work/airavata/sandbox/gsissh");
- System.setProperty("gsi.working.directory", "/home/ogce");
- myProxyUserName = System.getProperty("myproxy.user");
- myProxyPassword = System.getProperty("myproxy.password");
- workingDirectory = System.getProperty("gsi.working.directory");
- String pomDirectory = System.getProperty("basedir");
- certificateLocation = "/Users/lahirugunathilake/Downloads/certificates";
- if (myProxyUserName == null || myProxyPassword == null || workingDirectory == null) {
- System.out.println(">>>>>> Please run tests with my proxy user name and password. " +
- "E.g :- mvn clean install -Dmyproxy.user=xxx -Dmyproxy.password=xxx -Dgsi.working.directory=/path<<<<<<<");
- throw new Exception("Need my proxy user name password to run tests.");
- }
-
- monitorManager = new MonitorManager();
- QstatMonitor qstatMonitor = new
- QstatMonitor(monitorManager.getPullQueue(), monitorManager.getMonitorPublisher());
- try {
- monitorManager.addPullMonitor(qstatMonitor);
- monitorManager.launchMonitor();
- } catch (AiravataMonitorException e) {
- e.printStackTrace();
- }
-
- hostDescription = new HostDescription(GsisshHostType.type);
- hostDescription.getType().setHostAddress("trestles.sdsc.edu");
- hostDescription.getType().setHostName("gsissh-gordon");
- }
-
- @Test
- public void testQstatMonitor() throws SSHApiException {
- /* now have to submit a job to some machine and add that job to the queue */
- //Create authentication
- GSIAuthenticationInfo authenticationInfo
- = new MyProxyAuthenticationInfo(myProxyUserName, myProxyPassword, "myproxy.teragrid.org",
- 7512, 17280000, certificateLocation);
-
- // Server info
- ServerInfo serverInfo = new ServerInfo("ogce", hostDescription.getType().getHostAddress());
-
-
- Cluster pbsCluster = new PBSCluster(serverInfo, authenticationInfo, CommonUtils.getPBSJobManager("/opt/torque/bin/"));
-
-
- // Execute command
- System.out.println("Target PBS file path: " + workingDirectory);
- // constructing the job object
- JobDescriptor jobDescriptor = new JobDescriptor();
- jobDescriptor.setWorkingDirectory(workingDirectory);
- jobDescriptor.setShellName("/bin/bash");
- jobDescriptor.setJobName("GSI_SSH_SLEEP_JOB");
- jobDescriptor.setExecutablePath("/bin/echo");
- jobDescriptor.setAllEnvExport(true);
- jobDescriptor.setMailOptions("n");
- jobDescriptor.setStandardOutFile(workingDirectory + File.separator + "application.out");
- jobDescriptor.setStandardErrorFile(workingDirectory + File.separator + "application.err");
- jobDescriptor.setNodes(1);
- jobDescriptor.setProcessesPerNode(1);
- jobDescriptor.setQueueName("normal");
- jobDescriptor.setMaxWallTime("60");
- jobDescriptor.setAcountString("sds128");
- List<String> inputs = new ArrayList<String>();
- jobDescriptor.setOwner("ogce");
- inputs.add("Hello World");
- jobDescriptor.setInputValues(inputs);
- //finished construction of job object
- System.out.println(jobDescriptor.toXML());
- for (int i = 0; i < 1; i++) {
- String jobID = pbsCluster.submitBatchJob(jobDescriptor);
- MonitorID monitorID = new MonitorID(hostDescription, jobID,null,null, "ogce");
- monitorID.setAuthenticationInfo(authenticationInfo);
- try {
- monitorManager.addAJobToMonitor(monitorID);
- } catch (AiravataMonitorException e) {
- e.printStackTrace();
- }
- }
- try {
- Thread.sleep(10000000);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- }
-}
[3/3] git commit: Merge branch 'master' of
https://git-wip-us.apache.org/repos/asf/airavata
Posted by ra...@apache.org.
Merge branch 'master' of https://git-wip-us.apache.org/repos/asf/airavata
Project: http://git-wip-us.apache.org/repos/asf/airavata/repo
Commit: http://git-wip-us.apache.org/repos/asf/airavata/commit/007a111e
Tree: http://git-wip-us.apache.org/repos/asf/airavata/tree/007a111e
Diff: http://git-wip-us.apache.org/repos/asf/airavata/diff/007a111e
Branch: refs/heads/master
Commit: 007a111e80d832a59c02b3672ba850699ef6c6e5
Parents: 7e845c1 98519a7
Author: raminder <ra...@apache.org>
Authored: Wed Mar 26 11:36:01 2014 -0400
Committer: raminder <ra...@apache.org>
Committed: Wed Mar 26 11:36:01 2014 -0400
----------------------------------------------------------------------
.../job/monitor/impl/push/amqp/AMQPMonitor.java | 39 ++++++++++++--------
.../monitor/impl/push/amqp/BasicConsumer.java | 2 +-
.../airavata/job/monitor/util/CommonUtils.java | 32 +++++++++++++++-
3 files changed, 55 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
[2/3] git commit: merged the changes
Posted by ra...@apache.org.
merged the changes
Project: http://git-wip-us.apache.org/repos/asf/airavata/repo
Commit: http://git-wip-us.apache.org/repos/asf/airavata/commit/7e845c10
Tree: http://git-wip-us.apache.org/repos/asf/airavata/tree/7e845c10
Diff: http://git-wip-us.apache.org/repos/asf/airavata/diff/7e845c10
Branch: refs/heads/master
Commit: 7e845c10441ea7067167f12bfd2e088adb39c060
Parents: 04c502b c1f3429
Author: raminder <ra...@apache.org>
Authored: Wed Mar 26 11:35:53 2014 -0400
Committer: raminder <ra...@apache.org>
Committed: Wed Mar 26 11:35:53 2014 -0400
----------------------------------------------------------------------
modules/distribution/airavata-client/pom.xml | 2 +-
.../src/main/resources/conf/log4j.properties | 1 +
modules/gfac/gfac-monitor/.pom.xml.swp | Bin 16384 -> 0 bytes
modules/gfac/gfac-monitor/pom.xml | 2 +-
.../airavata/job/monitor/MonitorManager.java | 20 ++-
.../airavata/job/monitor/AMQPMonitorTest.java | 36 +++---
.../airavata/job/monitor/QstatMonitorTest.java | 31 ++---
.../orchestrator-client-sdks/pom.xml | 2 +-
modules/orchestrator/orchestrator-core/pom.xml | 2 +-
.../job/monitor/core/MessageParser.java | 4 +-
.../airavata/job/monitor/core/PushMonitor.java | 2 +-
.../job/monitor/event/MonitorPublisher.java | 4 +
.../job/monitor/impl/push/amqp/AMQPMonitor.java | 125 +++++++++++--------
.../monitor/impl/push/amqp/BasicConsumer.java | 34 ++---
.../impl/push/amqp/JSONMessageParser.java | 4 +-
.../impl/push/amqp/UnRegisterThread.java | 76 -----------
.../impl/push/amqp/UnRegisterWorker.java | 68 ++++++++++
.../airavata/job/monitor/util/CommonUtils.java | 1 +
18 files changed, 203 insertions(+), 211 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/airavata/blob/7e845c10/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/monitor/AMQPMonitorTest.java
----------------------------------------------------------------------
diff --cc modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/monitor/AMQPMonitorTest.java
index 4a1126a,107f0dc..1085dd0
--- a/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/monitor/AMQPMonitorTest.java
+++ b/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/monitor/AMQPMonitorTest.java
@@@ -20,11 -20,6 +20,13 @@@
*/
package org.apache.airavata.job.monitor;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
++import java.util.concurrent.BlockingQueue;
++import java.util.concurrent.LinkedBlockingQueue;
+
import org.apache.airavata.commons.gfac.type.HostDescription;
import org.apache.airavata.gsi.ssh.api.Cluster;
import org.apache.airavata.gsi.ssh.api.SSHApiException;
@@@ -33,22 -28,27 +35,24 @@@ import org.apache.airavata.gsi.ssh.api.
import org.apache.airavata.gsi.ssh.api.job.JobDescriptor;
import org.apache.airavata.gsi.ssh.impl.PBSCluster;
import org.apache.airavata.gsi.ssh.impl.authentication.MyProxyAuthenticationInfo;
++import org.apache.airavata.job.monitor.event.MonitorPublisher;
import org.apache.airavata.job.monitor.exception.AiravataMonitorException;
import org.apache.airavata.job.monitor.impl.push.amqp.AMQPMonitor;
import org.apache.airavata.schemas.gfac.GsisshHostType;
import org.junit.Before;
import org.junit.Test;
- public class AMQPMonitorTest {
- private MonitorManager monitorManager;
-import java.io.File;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
++import com.google.common.eventbus.EventBus;
+ public class AMQPMonitorTest {
private String myProxyUserName;
private String myProxyPassword;
private String certificateLocation;
-- private String pbsFilePath;
private String workingDirectory;
private HostDescription hostDescription;
--
++ private BlockingQueue<MonitorID> amqpQueue;
++ private BlockingQueue<MonitorID> finishQueue;
++ private AMQPMonitor amqpMonitor;
@Before
public void setUp() throws Exception {
System.setProperty("myproxy.user", "ogce");
@@@ -65,19 -65,19 +69,12 @@@
"E.g :- mvn clean install -Dmyproxy.user=xxx -Dmyproxy.password=xxx -Dgsi.working.directory=/path<<<<<<<");
throw new Exception("Need my proxy user name password to run tests.");
}
--
-- monitorManager = new MonitorManager();
-- AMQPMonitor amqpMonitor = new
-- AMQPMonitor(monitorManager.getMonitorPublisher(),
- monitorManager.getPullQueue(), monitorManager.getFinishQueue(),"/Users/lahirugunathilake/Downloads/x509up_u503876","xsede_private",
- monitorManager.getPushQueue(), monitorManager.getFinishQueue(),"/Users/lahirugunathilake/Downloads/x509up_u503876","xsede_private",
++ amqpQueue = new LinkedBlockingQueue<MonitorID>();
++ finishQueue = new LinkedBlockingQueue<MonitorID>();
++ amqpMonitor = new
++ AMQPMonitor(new MonitorPublisher(new EventBus()),
++ amqpQueue,finishQueue,"/Users/lahirugunathilake/Downloads/x509up_u503876","xsede_private",
Arrays.asList("info1.dyn.teragrid.org,info2.dyn.teragrid.org".split(",")));
-- try {
-- monitorManager.addPushMonitor(amqpMonitor);
-- monitorManager.launchMonitor();
-- } catch (AiravataMonitorException e) {
-- e.printStackTrace();
-- }
--
hostDescription = new HostDescription(GsisshHostType.type);
hostDescription.getType().setHostAddress("gordon.sdsc.xsede.org");
hostDescription.getType().setHostName("gsissh-gordon");
@@@ -125,12 -125,11 +122,9 @@@
String jobID = pbsCluster.submitBatchJob(jobDescriptor);
System.out.println(jobID);
try {
-- monitorManager.addAJobToMonitor(new MonitorID(hostDescription, jobID,null,null, "ogce"));
- }catch (InterruptedException e) {
- e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
- }
- catch (AiravataMonitorException e) {
++ amqpMonitor.registerListener(new MonitorID(hostDescription,jobID,null,null,myProxyUserName));
+ } catch (AiravataMonitorException e) {
e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
- } catch (InterruptedException e) {
- e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
}
try {
Thread.sleep(10000);
http://git-wip-us.apache.org/repos/asf/airavata/blob/7e845c10/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/monitor/QstatMonitorTest.java
----------------------------------------------------------------------
diff --cc modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/monitor/QstatMonitorTest.java
index fdfe841,68460d4..bc6c4c5
--- a/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/monitor/QstatMonitorTest.java
+++ b/modules/gfac/gfac-monitor/src/test/java/org/apache/airavata/job/monitor/QstatMonitorTest.java
@@@ -20,6 -20,6 +20,7 @@@
*/
package org.apache.airavata.job.monitor;
++import com.google.common.eventbus.EventBus;
import org.apache.airavata.commons.gfac.type.HostDescription;
import org.apache.airavata.gsi.ssh.api.Cluster;
import org.apache.airavata.gsi.ssh.api.SSHApiException;
@@@ -28,9 -28,9 +29,10 @@@ import org.apache.airavata.gsi.ssh.api.
import org.apache.airavata.gsi.ssh.api.job.JobDescriptor;
import org.apache.airavata.gsi.ssh.impl.PBSCluster;
import org.apache.airavata.gsi.ssh.impl.authentication.MyProxyAuthenticationInfo;
--import org.apache.airavata.gsi.ssh.util.CommonUtils;
++import org.apache.airavata.job.monitor.event.MonitorPublisher;
import org.apache.airavata.job.monitor.exception.AiravataMonitorException;
import org.apache.airavata.job.monitor.impl.pull.qstat.QstatMonitor;
++import org.apache.airavata.job.monitor.util.CommonUtils;
import org.apache.airavata.schemas.gfac.GsisshHostType;
import org.junit.Before;
import org.junit.Test;
@@@ -38,17 -38,16 +40,17 @@@
import java.io.File;
import java.util.ArrayList;
import java.util.List;
++import java.util.concurrent.BlockingQueue;
public class QstatMonitorTest {
-- private MonitorManager monitorManager;
-
private String myProxyUserName;
private String myProxyPassword;
private String certificateLocation;
private String pbsFilePath;
private String workingDirectory;
private HostDescription hostDescription;
--
++ private BlockingQueue<UserMonitorData> q;
++ private QstatMonitor qstatMonitor;
@Before
public void setUp() throws Exception {
System.setProperty("myproxy.user", "ogce");
@@@ -66,19 -65,19 +68,13 @@@
throw new Exception("Need my proxy user name password to run tests.");
}
-- monitorManager = new MonitorManager();
-- QstatMonitor qstatMonitor = new
-- QstatMonitor(monitorManager.getPullQueue(), monitorManager.getMonitorPublisher());
-- try {
-- monitorManager.addPullMonitor(qstatMonitor);
-- monitorManager.launchMonitor();
-- } catch (AiravataMonitorException e) {
-- e.printStackTrace();
-- }
++ qstatMonitor = new
++ QstatMonitor(q, new MonitorPublisher(new EventBus()));
hostDescription = new HostDescription(GsisshHostType.type);
hostDescription.getType().setHostAddress("trestles.sdsc.edu");
hostDescription.getType().setHostName("gsissh-gordon");
++ qstatMonitor.startPulling();
}
@Test
@@@ -93,7 -92,7 +89,7 @@@
ServerInfo serverInfo = new ServerInfo("ogce", hostDescription.getType().getHostAddress());
-- Cluster pbsCluster = new PBSCluster(serverInfo, authenticationInfo, CommonUtils.getPBSJobManager("/opt/torque/bin/"));
++ Cluster pbsCluster = new PBSCluster(serverInfo, authenticationInfo, org.apache.airavata.gsi.ssh.util.CommonUtils.getPBSJobManager("/opt/torque/bin/"));
// Execute command
@@@ -124,12 -123,11 +120,9 @@@
MonitorID monitorID = new MonitorID(hostDescription, jobID,null,null, "ogce");
monitorID.setAuthenticationInfo(authenticationInfo);
try {
-- monitorManager.addAJobToMonitor(monitorID);
- }catch (InterruptedException e) {
- e.printStackTrace();
- }
- catch (AiravataMonitorException e) {
++ CommonUtils.addMonitortoQueue(q,monitorID);
+ } catch (AiravataMonitorException e) {
e.printStackTrace();
- } catch (InterruptedException e) {
- e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
}
}
try {