airavata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From shame...@apache.org
Subject [80/81] [abbrv] airavata git commit: Merge moduleRefactor branch
Date Thu, 04 Jun 2015 20:16:33 GMT
http://git-wip-us.apache.org/repos/asf/airavata/blob/b4ede9cb/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/core/GFacUtils.java
----------------------------------------------------------------------
diff --cc modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/core/GFacUtils.java
index 0000000,b716099..3756140
mode 000000,100644..100644
--- a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/core/GFacUtils.java
+++ b/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/core/GFacUtils.java
@@@ -1,0 -1,747 +1,747 @@@
+ /*
+  *
+  * 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.gfac.core;
+ 
+ import org.airavata.appcatalog.cpi.AppCatalog;
+ import org.airavata.appcatalog.cpi.AppCatalogException;
+ import org.apache.aiaravata.application.catalog.data.impl.AppCatalogFactory;
+ import org.apache.airavata.common.exception.ApplicationSettingsException;
+ import org.apache.airavata.common.utils.AiravataZKUtils;
+ import org.apache.airavata.common.utils.DBUtil;
+ import org.apache.airavata.common.utils.MonitorPublisher;
+ import org.apache.airavata.common.utils.ServerSettings;
+ import org.apache.airavata.credential.store.store.CredentialReader;
+ import org.apache.airavata.credential.store.store.impl.CredentialReaderImpl;
+ import org.apache.airavata.gfac.Constants;
+ import org.apache.airavata.gfac.ExecutionMode;
+ import org.apache.airavata.gfac.GFacConfiguration;
+ import org.apache.airavata.gfac.GFacException;
+ import org.apache.airavata.gfac.core.context.JobExecutionContext;
+ import org.apache.airavata.gfac.core.handler.GFacHandlerException;
+ import org.apache.airavata.gfac.core.states.GfacExperimentState;
+ import org.apache.airavata.gfac.core.states.GfacHandlerState;
+ import org.apache.airavata.model.appcatalog.appinterface.InputDataObjectType;
+ import org.apache.airavata.model.appcatalog.appinterface.OutputDataObjectType;
+ import org.apache.airavata.model.appcatalog.computeresource.LOCALSubmission;
+ import org.apache.airavata.model.appcatalog.computeresource.SSHJobSubmission;
+ import org.apache.airavata.model.appcatalog.computeresource.UnicoreJobSubmission;
+ import org.apache.airavata.model.messaging.event.JobIdentifier;
+ import org.apache.airavata.model.messaging.event.JobStatusChangeRequestEvent;
+ import org.apache.airavata.model.messaging.event.TaskIdentifier;
+ import org.apache.airavata.model.messaging.event.TaskStatusChangeRequestEvent;
+ import org.apache.airavata.model.workspace.experiment.ActionableGroup;
+ import org.apache.airavata.model.workspace.experiment.CorrectiveAction;
+ import org.apache.airavata.model.workspace.experiment.ErrorCategory;
+ import org.apache.airavata.model.workspace.experiment.ErrorDetails;
+ import org.apache.airavata.model.workspace.experiment.Experiment;
+ import org.apache.airavata.model.workspace.experiment.ExperimentState;
+ import org.apache.airavata.model.workspace.experiment.JobDetails;
+ import org.apache.airavata.model.workspace.experiment.JobState;
+ import org.apache.airavata.model.workspace.experiment.JobStatus;
+ import org.apache.airavata.model.workspace.experiment.TaskState;
 -import org.apache.airavata.persistance.registry.jpa.impl.RegistryFactory;
++import org.apache.airavata.experiment.catalog.impl.RegistryFactory;
+ import org.apache.airavata.registry.cpi.ChildDataType;
+ import org.apache.airavata.registry.cpi.CompositeIdentifier;
+ import org.apache.airavata.registry.cpi.Registry;
+ import org.apache.airavata.registry.cpi.RegistryException;
+ import org.apache.airavata.registry.cpi.RegistryModelType;
+ import org.apache.curator.framework.CuratorFramework;
+ import org.apache.curator.utils.ZKPaths;
+ import org.apache.zookeeper.CreateMode;
+ import org.apache.zookeeper.KeeperException;
+ import org.apache.zookeeper.ZooDefs;
+ import org.apache.zookeeper.data.ACL;
+ import org.apache.zookeeper.data.Stat;
+ import org.slf4j.Logger;
+ import org.slf4j.LoggerFactory;
+ import org.w3c.dom.Document;
+ import org.w3c.dom.Element;
+ import org.w3c.dom.Node;
+ import org.w3c.dom.NodeList;
+ 
+ import javax.xml.xpath.XPath;
+ import javax.xml.xpath.XPathConstants;
+ import javax.xml.xpath.XPathExpression;
+ import javax.xml.xpath.XPathExpressionException;
+ import javax.xml.xpath.XPathFactory;
+ import java.io.BufferedReader;
+ import java.io.File;
+ import java.io.FileNotFoundException;
+ import java.io.FileReader;
+ import java.io.IOException;
+ import java.io.InputStream;
+ import java.net.InetAddress;
+ import java.net.URISyntaxException;
+ import java.net.UnknownHostException;
+ import java.nio.ByteBuffer;
+ import java.util.ArrayList;
+ import java.util.Calendar;
+ import java.util.Date;
+ import java.util.HashMap;
+ import java.util.List;
+ import java.util.Map;
+ 
+ //import org.apache.airavata.commons.gfac.type.ActualParameter;
+ 
+ public class GFacUtils {
+ 	private final static Logger log = LoggerFactory.getLogger(GFacUtils.class);
+ 	public static final ArrayList<ACL> OPEN_ACL_UNSAFE = ZooDefs.Ids.OPEN_ACL_UNSAFE;
+ 
+ 	private GFacUtils() {
+ 	}
+ 
+ 	/**
+ 	 * Read data from inputStream and convert it to String.
+ 	 * 
+ 	 * @param in
+ 	 * @return String read from inputStream
+ 	 * @throws java.io.IOException
+ 	 */
+ 	public static String readFromStream(InputStream in) throws IOException {
+ 		try {
+ 			StringBuffer wsdlStr = new StringBuffer();
+ 
+ 			int read;
+ 
+ 			byte[] buf = new byte[1024];
+ 			while ((read = in.read(buf)) > 0) {
+ 				wsdlStr.append(new String(buf, 0, read));
+ 			}
+ 			return wsdlStr.toString();
+ 		} finally {
+ 			if (in != null) {
+ 				try {
+ 					in.close();
+ 				} catch (IOException e) {
+ 					log.warn("Cannot close InputStream: "
+ 							+ in.getClass().getName(), e);
+ 				}
+ 			}
+ 		}
+ 	}
+ 
+ 	/**
+ 	 * This returns true if the give job is finished
+ 	 * otherwise false
+ 	 *
+ 	 * @param job
+ 	 * @return
+ 	 */
+ 	public static boolean isJobFinished(JobDescriptor job) {
+ 		if (org.apache.airavata.gfac.core.cluster.JobStatus.C.toString().equals(job.getStatus())) {
+ 			return true;
+ 		} else {
+ 			return false;
+ 		}
+ 	}
+ 
+ 	/**
+ 	 * This will read
+ 	 *
+ 	 * @param maxWalltime
+ 	 * @return
+ 	 */
+ 	public static String maxWallTimeCalculator(int maxWalltime) {
+ 		if (maxWalltime < 60) {
+ 			return "00:" + maxWalltime + ":00";
+ 		} else {
+ 			int minutes = maxWalltime % 60;
+ 			int hours = maxWalltime / 60;
+ 			return hours + ":" + minutes + ":00";
+ 		}
+ 	}
+ 	public static String maxWallTimeCalculatorForLSF(int maxWalltime) {
+ 		if (maxWalltime < 60) {
+ 			return "00:" + maxWalltime;
+ 		} else {
+ 			int minutes = maxWalltime % 60;
+ 			int hours = maxWalltime / 60;
+ 			return hours + ":" + minutes;
+ 		}
+ 	}
+ 	/**
+ 	 * this can be used to do framework opertaions specific to different modes
+ 	 * 
+ 	 * @param jobExecutionContext
+ 	 * @return
+ 	 */
+ 	public static boolean isSynchronousMode(
+ 			JobExecutionContext jobExecutionContext) {
+ 		GFacConfiguration gFacConfiguration = jobExecutionContext
+ 				.getGFacConfiguration();
+ 		if (ExecutionMode.ASYNCHRONOUS.equals(gFacConfiguration
+ 				.getExecutionMode())) {
+ 			return false;
+ 		}
+ 		return true;
+ 	}
+ 
+ 	public static String readFileToString(String file)
+ 			throws FileNotFoundException, IOException {
+ 		BufferedReader instream = null;
+ 		try {
+ 
+ 			instream = new BufferedReader(new FileReader(file));
+ 			StringBuffer buff = new StringBuffer();
+ 			String temp = null;
+ 			while ((temp = instream.readLine()) != null) {
+ 				buff.append(temp);
+ 				buff.append(Constants.NEWLINE);
+ 			}
+ 			return buff.toString();
+ 		} finally {
+ 			if (instream != null) {
+ 				try {
+ 					instream.close();
+ 				} catch (IOException e) {
+ 					log.warn("Cannot close FileinputStream", e);
+ 				}
+ 			}
+ 		}
+ 	}
+ 
+ 	public static boolean isLocalHost(String appHost)
+ 			throws UnknownHostException {
+ 		String localHost = InetAddress.getLocalHost().getCanonicalHostName();
+ 		return (localHost.equals(appHost)
+ 				|| Constants.LOCALHOST.equals(appHost) || Constants._127_0_0_1
+ 					.equals(appHost));
+ 	}
+ 
+ 	public static String createUniqueNameWithDate(String name) {
+ 		String date = new Date().toString();
+ 		date = date.replaceAll(" ", "_");
+ 		date = date.replaceAll(":", "_");
+ 		return name + "_" + date;
+ 	}
+ 
+     public static List<Element> getElementList(Document doc, String expression) throws XPathExpressionException {
+         XPathFactory xPathFactory = XPathFactory.newInstance();
+         XPath xPath = xPathFactory.newXPath();
+         XPathExpression expr = xPath.compile(expression);
+         NodeList nodeList = (NodeList) expr.evaluate(doc, XPathConstants.NODESET);
+         List<Element> elementList = new ArrayList<Element>();
+         for (int i = 0; i < nodeList.getLength(); i++) {
+             Node item = nodeList.item(i);
+             if (item instanceof Element) {
+                 elementList.add((Element) item);
+             }
+         }
+         return elementList;
+     }
+ 
+ 	public static String createGsiftpURIAsString(String host, String localPath)
+ 			throws URISyntaxException {
+ 		StringBuffer buf = new StringBuffer();
+ 		if (!host.startsWith("gsiftp://"))
+ 			buf.append("gsiftp://");
+ 		buf.append(host);
+ 		if (!host.endsWith("/"))
+ 			buf.append("/");
+ 		buf.append(localPath);
+ 		return buf.toString();
+ 	}
+ 
+ 	public static void saveJobStatus(JobExecutionContext jobExecutionContext,
+                                      JobDetails details, JobState state) throws GFacException {
+ 		try {
+             // first we save job details to the registry for sa and then save the job status.
+             Registry registry = jobExecutionContext.getRegistry();
+             JobStatus status = new JobStatus();
+             status.setJobState(state);
+             details.setJobStatus(status);
+             registry.add(ChildDataType.JOB_DETAIL, details,
+                     new CompositeIdentifier(jobExecutionContext.getTaskData()
+                             .getTaskID(), details.getJobID()));
+             JobIdentifier identifier = new JobIdentifier(details.getJobID(), jobExecutionContext.getTaskData().getTaskID(),
+                     jobExecutionContext.getWorkflowNodeDetails().getNodeInstanceId(), jobExecutionContext.getExperimentID(),
+                     jobExecutionContext.getGatewayID());
+             JobStatusChangeRequestEvent jobStatusChangeRequestEvent = new JobStatusChangeRequestEvent(state, identifier);
+             jobExecutionContext.getMonitorPublisher().publish(jobStatusChangeRequestEvent);
+         } catch (Exception e) {
+ 			throw new GFacException("Error persisting job status"
+ 					+ e.getLocalizedMessage(), e);
+ 		}
+ 	}
+ 
+ 	public static void updateJobStatus(JobExecutionContext jobExecutionContext,
+ 			JobDetails details, JobState state) throws GFacException {
+ 		try {
+ 			Registry registry = jobExecutionContext.getRegistry();
+ 			JobStatus status = new JobStatus();
+ 			status.setJobState(state);
+ 			status.setTimeOfStateChange(Calendar.getInstance()
+ 					.getTimeInMillis());
+ 			details.setJobStatus(status);
+ 			registry.update(
+ 					org.apache.airavata.registry.cpi.RegistryModelType.JOB_DETAIL,
+ 					details, details.getJobID());
+ 		} catch (Exception e) {
+ 			throw new GFacException("Error persisting job status"
+ 					+ e.getLocalizedMessage(), e);
+ 		}
+ 	}
+ 
+ 	public static void saveErrorDetails(
+ 			JobExecutionContext jobExecutionContext, String errorMessage,
+ 			CorrectiveAction action, ErrorCategory errorCatogory)
+ 			throws GFacException {
+ 		try {
+ 			Registry registry = jobExecutionContext.getRegistry();
+ 			ErrorDetails details = new ErrorDetails();
+ 			details.setActualErrorMessage(errorMessage);
+ 			details.setCorrectiveAction(action);
+ 			details.setActionableGroup(ActionableGroup.GATEWAYS_ADMINS);
+ 			details.setCreationTime(Calendar.getInstance().getTimeInMillis());
+ 			details.setErrorCategory(errorCatogory);
+ 			registry.add(ChildDataType.ERROR_DETAIL, details,
+ 					jobExecutionContext.getTaskData().getTaskID());
+ 		} catch (Exception e) {
+ 			throw new GFacException("Error persisting job status"
+ 					+ e.getLocalizedMessage(), e);
+ 		}
+ 	}
+ 
+     public static Map<String, Object> getInputParamMap(List<InputDataObjectType> experimentData) throws GFacException {
+         Map<String, Object> map = new HashMap<String, Object>();
+         for (InputDataObjectType objectType : experimentData) {
+             map.put(objectType.getName(), objectType);
+         }
+         return map;
+     }
+ 
+     public static Map<String, Object> getOuputParamMap(List<OutputDataObjectType> experimentData) throws GFacException {
+         Map<String, Object> map = new HashMap<String, Object>();
+         for (OutputDataObjectType objectType : experimentData) {
+             map.put(objectType.getName(), objectType);
+         }
+         return map;
+     }
+ 
+ 	public static GfacExperimentState getZKExperimentState(CuratorFramework curatorClient,
+ 			JobExecutionContext jobExecutionContext)
+ 			throws Exception {
+ 		String expState = AiravataZKUtils.getExpState(curatorClient, jobExecutionContext
+ 				.getExperimentID());
+         if (expState == null || expState.isEmpty()) {
+             return GfacExperimentState.UNKNOWN;
+         }
+         return GfacExperimentState.findByValue(Integer.valueOf(expState));
+     }
+ 
+ 	public static boolean createHandlerZnode(CuratorFramework curatorClient,
+                                              JobExecutionContext jobExecutionContext, String className)
+ 			throws Exception {
+ 		String expState = AiravataZKUtils.getExpZnodeHandlerPath(
+ 				jobExecutionContext.getExperimentID(), className);
+ 		Stat exists = curatorClient.checkExists().forPath(expState);
+ 		if (exists == null) {
+ 			curatorClient.create().withMode(CreateMode.PERSISTENT).withACL(OPEN_ACL_UNSAFE).forPath(expState, new byte[0]);
+ 			curatorClient.create().withMode(CreateMode.PERSISTENT).withACL(OPEN_ACL_UNSAFE)
+ 					.forPath(expState + File.separator + AiravataZKUtils.ZK_EXPERIMENT_STATE_NODE, new byte[0]);
+ 		} else {
+ 			exists = curatorClient.checkExists().forPath(expState + File.separator + AiravataZKUtils.ZK_EXPERIMENT_STATE_NODE);
+ 			if (exists == null) {
+ 				curatorClient.create().withMode(CreateMode.PERSISTENT).withACL(OPEN_ACL_UNSAFE)
+ 						.forPath(expState + File.separator + AiravataZKUtils.ZK_EXPERIMENT_STATE_NODE, new byte[0]);
+ 			}
+ 		}
+ 
+ 		exists = curatorClient.checkExists().forPath(expState + File.separator + AiravataZKUtils.ZK_EXPERIMENT_STATE_NODE);
+ 		if (exists != null) {
+ 			curatorClient.setData().withVersion(exists.getVersion())
+ 					.forPath(expState + File.separator + AiravataZKUtils.ZK_EXPERIMENT_STATE_NODE,
+ 							String.valueOf(GfacHandlerState.INVOKING.getValue()).getBytes());
+ 		}
+ 		return true;
+ 	}
+ 
+ 	public static boolean createHandlerZnode(CuratorFramework curatorClient,
+                                              JobExecutionContext jobExecutionContext, String className,
+                                              GfacHandlerState state) throws Exception {
+ 		String expState = AiravataZKUtils.getExpZnodeHandlerPath(
+ 				jobExecutionContext.getExperimentID(), className);
+ 		Stat exists = curatorClient.checkExists().forPath(expState);
+ 		if (exists == null) {
+ 			curatorClient.create().withMode(CreateMode.PERSISTENT).withACL(OPEN_ACL_UNSAFE)
+ 					.forPath(expState, new byte[0]);
+ 			curatorClient.create().withMode(CreateMode.PERSISTENT).withACL(OPEN_ACL_UNSAFE)
+ 					.forPath(expState + File.separator + AiravataZKUtils.ZK_EXPERIMENT_STATE_NODE, new byte[0]);
+ 		} else {
+ 			exists = curatorClient.checkExists().forPath(expState + File.separator + AiravataZKUtils.ZK_EXPERIMENT_STATE_NODE);
+ 			if (exists == null) {
+ 				curatorClient.create().withMode(CreateMode.PERSISTENT).withACL(OPEN_ACL_UNSAFE)
+ 						.forPath(expState + File.separator + AiravataZKUtils.ZK_EXPERIMENT_STATE_NODE, new byte[0]);
+ 			}
+ 		}
+ 
+ 		exists = curatorClient.checkExists().forPath(expState + File.separator + AiravataZKUtils.ZK_EXPERIMENT_STATE_NODE);
+ 		if (exists != null) {
+ 			curatorClient.setData().withVersion(exists.getVersion())
+ 					.forPath(expState + File.separator + AiravataZKUtils.ZK_EXPERIMENT_STATE_NODE,
+ 							String.valueOf(state.getValue()).getBytes());
+ 		}
+ 		return true;
+ 	}
+ 
+ 	public static boolean updateHandlerState(CuratorFramework curatorClient,
+                                              JobExecutionContext jobExecutionContext, String className,
+                                              GfacHandlerState state) throws Exception {
+ 		String handlerPath = AiravataZKUtils.getExpZnodeHandlerPath(
+ 				jobExecutionContext.getExperimentID(), className);
+ 		Stat exists = curatorClient.checkExists().forPath(handlerPath + File.separator + AiravataZKUtils.ZK_EXPERIMENT_STATE_NODE);
+ 		if (exists != null) {
+ 			curatorClient.setData().withVersion(exists.getVersion())
+ 					.forPath(handlerPath + File.separator + AiravataZKUtils.ZK_EXPERIMENT_STATE_NODE, String.valueOf(state.getValue()).getBytes());
+ 		} else {
+ 			createHandlerZnode(curatorClient, jobExecutionContext, className, state);
+ 		}
+ 		return false;
+ 	}
+ 
+ 	public static GfacHandlerState getHandlerState(CuratorFramework curatorClient,
+                                                   JobExecutionContext jobExecutionContext, String className) {
+ 		try {
+ 			String handlerPath = AiravataZKUtils.getExpZnodeHandlerPath( jobExecutionContext.getExperimentID(), className);
+ 			Stat exists = curatorClient.checkExists().forPath(handlerPath + File.separator + AiravataZKUtils.ZK_EXPERIMENT_STATE_NODE);
+ 			if (exists != null) {
+ 				String stateVal = new String(curatorClient.getData().storingStatIn(exists)
+ 						.forPath(handlerPath + File.separator + AiravataZKUtils.ZK_EXPERIMENT_STATE_NODE));
+ 				return GfacHandlerState.findByValue(Integer.valueOf(stateVal));
+ 			}
+ 			return GfacHandlerState.UNKNOWN; // if the node doesn't exist or any other error we
+ 							// return false
+ 		} catch (Exception e) {
+ 			log.error("Error occured while getting zk node status", e);
+ 			return null;
+ 		}
+ 	}
+ 
+ 	// This method is dangerous because of moving the experiment data
+ 	public static boolean createExperimentEntryForPassive(String experimentID,
+ 														  String taskID, CuratorFramework curatorClient, String experimentNode,
+ 														  String pickedChild, String tokenId, long deliveryTag) throws Exception {
+ 		String experimentPath = experimentNode + File.separator + pickedChild;
+ 		String newExperimentPath = experimentPath + File.separator + experimentID;
+ 		Stat exists1 = curatorClient.checkExists().forPath(newExperimentPath);
+ 		String oldExperimentPath = GFacUtils.findExperimentEntry(experimentID, curatorClient);
+ 		if (oldExperimentPath == null) {  // this means this is a very new experiment
+ 			// are going to create a new node
+ 			log.info("This is a new Job, so creating all the experiment docs from the scratch");
+ 			curatorClient.create().withMode(CreateMode.PERSISTENT).withACL(OPEN_ACL_UNSAFE).forPath(newExperimentPath, new byte[0]);
+             String stateNodePath = curatorClient.create().withMode(CreateMode.PERSISTENT).withACL(OPEN_ACL_UNSAFE)
+ 					.forPath(newExperimentPath + File.separator + "state",
+ 							String .valueOf(GfacExperimentState.LAUNCHED.getValue()) .getBytes());
+ 
+ 			if(curatorClient.checkExists().forPath(stateNodePath)!=null) {
+ 				log.info("Created the node: " + stateNodePath + " successfully !");
+ 			}else {
+ 				log.error("Error creating node: " + stateNodePath + " successfully !");
+ 			}
+ 			curatorClient.create().withMode(CreateMode.PERSISTENT).withACL(OPEN_ACL_UNSAFE)
+ 					.forPath(newExperimentPath + AiravataZKUtils.DELIVERY_TAG_POSTFIX, longToBytes(deliveryTag));
+ 		} else {
+ 			log.error("ExperimentID: " + experimentID + " taskID: " + taskID + " was running by some Gfac instance,but it failed");
+             removeCancelDeliveryTagNode(oldExperimentPath, curatorClient); // remove previous cancel deliveryTagNode
+             if(newExperimentPath.equals(oldExperimentPath)){
+                 log.info("Re-launch experiment came to the same GFac instance");
+             }else {
+ 				log.info("Re-launch experiment came to a new GFac instance so we are moving data to new gfac node");
+ 				curatorClient.create().withMode(CreateMode.PERSISTENT).withACL(OPEN_ACL_UNSAFE).forPath(newExperimentPath,
+ 						curatorClient.getData().storingStatIn(exists1).forPath(oldExperimentPath)); // recursively copy children
+                 copyChildren(curatorClient, oldExperimentPath, newExperimentPath, 2); // we need to copy children up to depth 2
+ 				String oldDeliveryTag = oldExperimentPath + AiravataZKUtils.DELIVERY_TAG_POSTFIX;
+ 				Stat exists = curatorClient.checkExists().forPath(oldDeliveryTag);
+ 				if(exists!=null) {
+ 					curatorClient.create().withMode(CreateMode.PERSISTENT).withACL(OPEN_ACL_UNSAFE)
+ 							.forPath(newExperimentPath + AiravataZKUtils.DELIVERY_TAG_POSTFIX,
+ 									curatorClient.getData().storingStatIn(exists).forPath(oldDeliveryTag));
+ 					ZKPaths.deleteChildren(curatorClient.getZookeeperClient().getZooKeeper(), oldDeliveryTag, true);
+ 				}
+ 				// After all the files are successfully transfered we delete the // old experiment,otherwise we do
+ 				// not delete a single file
+ 				log.info("After a successful copying of experiment data for an old experiment we delete the old data");
+ 				log.info("Deleting experiment data: " + oldExperimentPath);
+ 				ZKPaths.deleteChildren(curatorClient.getZookeeperClient().getZooKeeper(), oldExperimentPath, true);
+ 			}
+ 		}
+ 		return true;
+ 	}
+ 
+     private static void removeCancelDeliveryTagNode(String experimentPath, CuratorFramework curatorClient) throws Exception {
+         Stat exists = curatorClient.checkExists().forPath(experimentPath + AiravataZKUtils.CANCEL_DELIVERY_TAG_POSTFIX);
+         if (exists != null) {
+ 			ZKPaths.deleteChildren(curatorClient.getZookeeperClient().getZooKeeper(), experimentPath + AiravataZKUtils.CANCEL_DELIVERY_TAG_POSTFIX, true);
+ 		}
+ 	}
+ 
+     private static void copyChildren(CuratorFramework curatorClient, String oldPath, String newPath, int depth) throws Exception {
+         for (String childNode : curatorClient.getChildren().forPath(oldPath)) {
+             String oldChildPath = oldPath + File.separator + childNode;
+             Stat stat = curatorClient.checkExists().forPath(oldChildPath); // no need to check exists
+             String newChildPath = newPath + File.separator + childNode;
+             log.info("Creating new znode: " + newChildPath);
+ 			curatorClient.create().withMode(CreateMode.PERSISTENT).withACL(OPEN_ACL_UNSAFE)
+ 					.forPath(newChildPath, curatorClient.getData().storingStatIn(stat).forPath(oldChildPath));
+ 			if (--depth > 0) {
+                 copyChildren(curatorClient , oldChildPath, newChildPath, depth );
+             }
+         }
+     }
+ 
+ 	/**
+ 	 * This will return a value if the server is down because we iterate through exisiting experiment nodes, not
+ 	 * through gfac-server nodes
+ 	 *
+ 	 * @param experimentID
+ 	 * @param curatorClient
+ 	 * @return
+ 	 * @throws KeeperException
+ 	 * @throws InterruptedException
+ 	 */
+ 	public static String findExperimentEntry(String experimentID, CuratorFramework curatorClient) throws Exception {
+ 		String experimentNode = ServerSettings.getSetting(org.apache.airavata.common.utils.Constants.ZOOKEEPER_GFAC_EXPERIMENT_NODE, "/gfac-experiments");
+ 		List<String> children = curatorClient.getChildren().forPath(experimentNode);
+ 		for (String pickedChild : children) {
+ 			String experimentPath = experimentNode + File.separator + pickedChild;
+ 			String newExpNode = experimentPath + File.separator + experimentID;
+ 			Stat exists = curatorClient.checkExists().forPath(newExpNode);
+ 			if (exists == null) {
+ 				continue;
+ 			} else {
+ 				return newExpNode;
+ 			}
+ 		}
+ 		return null;
+ 	}
+ 
+     public static boolean setExperimentCancel(String experimentId, CuratorFramework curatorClient, long deliveryTag) throws Exception {
+         String experimentEntry = GFacUtils.findExperimentEntry(experimentId, curatorClient);
+         if (experimentEntry == null) {
+             // This should be handle in validation request. Gfac shouldn't get any invalidate experiment.
+             log.error("Cannot find the experiment Entry, so cancel operation cannot be performed. " +
+                     "This happen when experiment completed and already removed from the zookeeper");
+             return false;
+         } else {
+             // check cancel operation is being processed for the same experiment.
+             Stat cancelState = curatorClient.checkExists().forPath(experimentEntry + AiravataZKUtils.CANCEL_DELIVERY_TAG_POSTFIX);
+             if (cancelState != null) {
+                 // another cancel operation is being processed. only one cancel operation can exist for a given experiment.
+                 return false;
+             }
+ 
+ 			curatorClient.create().withMode(CreateMode.PERSISTENT).withACL(OPEN_ACL_UNSAFE)
+ 					.forPath(experimentEntry + AiravataZKUtils.CANCEL_DELIVERY_TAG_POSTFIX, longToBytes(deliveryTag)); // save cancel delivery tag to be acknowledge at the end.
+ 			return true;
+         }
+ 
+     }
+     public static boolean isCancelled(String experimentID, CuratorFramework curatorClient ) throws Exception {
+ 		String experimentEntry = GFacUtils.findExperimentEntry(experimentID, curatorClient);
+         if(experimentEntry == null){
+             return false;
+         }else {
+             Stat exists = curatorClient.checkExists().forPath(experimentEntry);
+             if (exists != null) {
+ 				String operation = new String(curatorClient.getData().storingStatIn(exists).forPath(experimentEntry + File.separator + "operation"));
+ 				if ("cancel".equals(operation)) {
+ 					return true;
+ 				}
+ 			}
+ 		}
+         return false;
+     }
+ 
+     public static void saveHandlerData(JobExecutionContext jobExecutionContext,
+                                        StringBuffer data, String className) throws GFacHandlerException {
+ 		try {
+ 			CuratorFramework curatorClient = jobExecutionContext.getCuratorClient();
+ 			if (curatorClient != null) {
+ 				String expZnodeHandlerPath = AiravataZKUtils
+ 						.getExpZnodeHandlerPath(
+ 								jobExecutionContext.getExperimentID(),
+ 								className);
+ 				Stat exists = curatorClient.checkExists().forPath(expZnodeHandlerPath);
+                 if (exists != null) {
+ 					curatorClient.setData().withVersion(exists.getVersion()).forPath(expZnodeHandlerPath, data.toString().getBytes());
+ 				} else {
+                     log.error("Saving Handler data failed, Stat is null");
+                 }
+             }
+ 		} catch (Exception e) {
+ 			throw new GFacHandlerException(e);
+ 		}
+ 	}
+ 
+ 	public static String getHandlerData(JobExecutionContext jobExecutionContext, String className) throws Exception {
+ 		CuratorFramework curatorClient = jobExecutionContext.getCuratorClient();
+ 		if (curatorClient != null) {
+ 			String expZnodeHandlerPath = AiravataZKUtils
+ 					.getExpZnodeHandlerPath(
+ 							jobExecutionContext.getExperimentID(),
+ 							className);
+ 			Stat exists = curatorClient.checkExists().forPath(expZnodeHandlerPath);
+ 			return new String(jobExecutionContext.getCuratorClient().getData().storingStatIn(exists).forPath(expZnodeHandlerPath));
+ 		}
+ 		return null;
+ 	}
+ 
+ 	public static CredentialReader getCredentialReader()
+ 			throws ApplicationSettingsException, IllegalAccessException,
+ 			InstantiationException {
+ 		try{
+ 		String jdbcUrl = ServerSettings.getCredentialStoreDBURL();
+ 		String jdbcUsr = ServerSettings.getCredentialStoreDBUser();
+ 		String jdbcPass = ServerSettings.getCredentialStoreDBPassword();
+ 		String driver = ServerSettings.getCredentialStoreDBDriver();
+ 		return new CredentialReaderImpl(new DBUtil(jdbcUrl, jdbcUsr, jdbcPass,
+ 				driver));
+ 		}catch(ClassNotFoundException e){
+ 			log.error("Not able to find driver: " + e.getLocalizedMessage());
+ 			return null;	
+ 		}
+ 	}
+ 
+     public static LOCALSubmission getLocalJobSubmission (String submissionId) throws AppCatalogException{
+         try {
+             AppCatalog appCatalog = AppCatalogFactory.getAppCatalog();
+             return appCatalog.getComputeResource().getLocalJobSubmission(submissionId);
+         }catch (Exception e){
+             String errorMsg = "Error while retrieving local job submission with submission id : " + submissionId;
+             log.error(errorMsg, e);
+             throw new AppCatalogException(errorMsg, e);
+         }
+     }
+ 
+     public static UnicoreJobSubmission getUnicoreJobSubmission (String submissionId) throws AppCatalogException{
+         try {
+             AppCatalog appCatalog = AppCatalogFactory.getAppCatalog();
+             return appCatalog.getComputeResource().getUNICOREJobSubmission(submissionId);
+         }catch (Exception e){
+             String errorMsg = "Error while retrieving UNICORE job submission with submission id : " + submissionId;
+             log.error(errorMsg, e);
+             throw new AppCatalogException(errorMsg, e);
+         }
+     }
+ 
+     public static SSHJobSubmission getSSHJobSubmission (String submissionId) throws AppCatalogException{
+         try {
+             AppCatalog appCatalog = AppCatalogFactory.getAppCatalog();
+             return appCatalog.getComputeResource().getSSHJobSubmission(submissionId);
+         }catch (Exception e){
+             String errorMsg = "Error while retrieving SSH job submission with submission id : " + submissionId;
+             log.error(errorMsg, e);
+             throw new AppCatalogException(errorMsg, e);
+         }
+     }
+ 
+     /**
+      * To convert list to separated value
+      * @param listOfStrings
+      * @param separator
+      * @return
+      */
+     public static  String listToCsv(List<String> listOfStrings, char separator) {
+         StringBuilder sb = new StringBuilder();
+ 
+         // all but last
+         for(int i = 0; i < listOfStrings.size() - 1 ; i++) {
+             sb.append(listOfStrings.get(i));
+             sb.append(separator);
+         }
+ 
+         // last string, no separator
+         if(listOfStrings.size() > 0){
+             sb.append(listOfStrings.get(listOfStrings.size()-1));
+         }
+ 
+         return sb.toString();
+     }
+ 
+ 	public static byte[] longToBytes(long x) {
+ 		ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+ 		buffer.putLong(x);
+ 		return buffer.array();
+ 	}
+ 
+ 	public static long bytesToLong(byte[] bytes) {
+ 		ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+ 		buffer.put(bytes);
+ 		buffer.flip();//need flip
+ 		return buffer.getLong();
+ 	}
+ 
+     public static ExperimentState updateExperimentStatus(String experimentId, ExperimentState state) throws RegistryException {
+         Registry airavataRegistry = RegistryFactory.getDefaultRegistry();
+         Experiment details = (Experiment) airavataRegistry.get(RegistryModelType.EXPERIMENT, experimentId);
+         if (details == null) {
+             details = new Experiment();
+             details.setExperimentID(experimentId);
+         }
+         org.apache.airavata.model.workspace.experiment.ExperimentStatus status = new org.apache.airavata.model.workspace.experiment.ExperimentStatus();
+         status.setExperimentState(state);
+         status.setTimeOfStateChange(Calendar.getInstance().getTimeInMillis());
+         if (!ExperimentState.CANCELED.equals(details.getExperimentStatus().getExperimentState()) &&
+                 !ExperimentState.CANCELING.equals(details.getExperimentStatus().getExperimentState())) {
+             status.setExperimentState(state);
+         } else {
+             status.setExperimentState(details.getExperimentStatus().getExperimentState());
+         }
+         details.setExperimentStatus(status);
+         log.info("Updating the experiment status of experiment: " + experimentId + " to " + status.getExperimentState().toString());
+         airavataRegistry.update(RegistryModelType.EXPERIMENT_STATUS, status, experimentId);
+         return details.getExperimentStatus().getExperimentState();
+     }
+ 
+     public static boolean isFailedJob (JobExecutionContext jec) {
+         JobStatus jobStatus = jec.getJobDetails().getJobStatus();
+         if (jobStatus.getJobState() == JobState.FAILED) {
+             return true;
+         }
+         return false;
+     }
+ 
+     public static boolean ackCancelRequest(String experimentId, CuratorFramework curatorClient) throws Exception {
+         String experimentEntry = GFacUtils.findExperimentEntry(experimentId, curatorClient);
+         String cancelNodePath = experimentEntry + AiravataZKUtils.CANCEL_DELIVERY_TAG_POSTFIX;
+         if (experimentEntry == null) {
+             // This should be handle in validation request. Gfac shouldn't get any invalidate experiment.
+             log.error("Cannot find the experiment Entry, so cancel operation cannot be performed. " +
+                     "This happen when experiment completed and already removed from the CuratorFramework");
+         } else {
+             // check cancel operation is being processed for the same experiment.
+             Stat cancelState = curatorClient.checkExists().forPath(cancelNodePath);
+             if (cancelState != null) {
+ 				ZKPaths.deleteChildren(curatorClient.getZookeeperClient().getZooKeeper(), cancelNodePath, true);
+ 				return true;
+ 			}
+ 		}
+         return false;
+     }
+ 
+     public static void publishTaskStatus (JobExecutionContext jobExecutionContext, MonitorPublisher publisher, TaskState state){
+         TaskIdentifier taskIdentity = new TaskIdentifier(jobExecutionContext.getTaskData().getTaskID(),
+                 jobExecutionContext.getWorkflowNodeDetails().getNodeInstanceId(),
+                 jobExecutionContext.getExperimentID(),
+                 jobExecutionContext.getGatewayID());
+         publisher.publish(new TaskStatusChangeRequestEvent(state, taskIdentity));
+     }
+ }

http://git-wip-us.apache.org/repos/asf/airavata/blob/b4ede9cb/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/core/handler/AbstractHandler.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/airavata/blob/b4ede9cb/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/ssh/util/GFACSSHUtils.java
----------------------------------------------------------------------
diff --cc modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/ssh/util/GFACSSHUtils.java
index 0000000,307d8c3..f28b6e4
mode 000000,100644..100644
--- a/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/ssh/util/GFACSSHUtils.java
+++ b/modules/gfac/gfac-impl/src/main/java/org/apache/airavata/gfac/ssh/util/GFACSSHUtils.java
@@@ -1,0 -1,562 +1,562 @@@
+ /*
+  *
+  * 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.gfac.ssh.util;
+ 
+ import org.airavata.appcatalog.cpi.AppCatalog;
+ import org.airavata.appcatalog.cpi.AppCatalogException;
+ import org.apache.airavata.common.exception.ApplicationSettingsException;
+ import org.apache.airavata.common.utils.ServerSettings;
+ import org.apache.airavata.credential.store.credential.impl.ssh.SSHCredential;
+ import org.apache.airavata.gfac.Constants;
+ import org.apache.airavata.gfac.GFacException;
+ import org.apache.airavata.gfac.RequestData;
+ import org.apache.airavata.gfac.core.JobDescriptor;
+ import org.apache.airavata.gfac.core.JobManagerConfiguration;
+ import org.apache.airavata.gfac.core.cluster.Cluster;
+ import org.apache.airavata.gfac.core.cluster.ServerInfo;
+ import org.apache.airavata.gfac.core.context.JobExecutionContext;
+ import org.apache.airavata.gfac.core.context.MessageContext;
+ import org.apache.airavata.gfac.core.handler.GFacHandlerException;
+ import org.apache.airavata.gfac.core.GFacUtils;
+ import org.apache.airavata.gfac.gsi.ssh.impl.GSISSHAbstractCluster;
+ import org.apache.airavata.gfac.gsi.ssh.impl.PBSCluster;
+ import org.apache.airavata.gfac.gsi.ssh.impl.authentication.DefaultPasswordAuthenticationInfo;
+ import org.apache.airavata.gfac.gsi.ssh.util.CommonUtils;
+ import org.apache.airavata.gfac.ssh.context.SSHAuthWrapper;
+ import org.apache.airavata.gfac.ssh.security.SSHSecurityContext;
+ import org.apache.airavata.gfac.ssh.security.TokenizedSSHAuthInfo;
+ import org.apache.airavata.gfac.core.authentication.AuthenticationInfo;
+ import org.apache.airavata.model.appcatalog.appdeployment.ApplicationDeploymentDescription;
+ import org.apache.airavata.model.appcatalog.appdeployment.ApplicationParallelismType;
+ import org.apache.airavata.model.appcatalog.appinterface.DataType;
+ import org.apache.airavata.model.appcatalog.appinterface.InputDataObjectType;
+ import org.apache.airavata.model.appcatalog.appinterface.OutputDataObjectType;
+ import org.apache.airavata.model.appcatalog.computeresource.*;
+ import org.apache.airavata.model.appcatalog.gatewayprofile.ComputeResourcePreference;
+ import org.apache.airavata.model.workspace.experiment.ComputationalResourceScheduling;
+ import org.apache.airavata.model.workspace.experiment.CorrectiveAction;
+ import org.apache.airavata.model.workspace.experiment.ErrorCategory;
+ import org.apache.airavata.model.workspace.experiment.TaskDetails;
+ import org.slf4j.Logger;
+ import org.slf4j.LoggerFactory;
+ 
+ import java.io.File;
+ import java.io.PrintWriter;
+ import java.io.StringWriter;
+ import java.util.*;
+ 
+ public class GFACSSHUtils {
+     private final static Logger logger = LoggerFactory.getLogger(GFACSSHUtils.class);
+ 
+     public static Map<String, List<Cluster>> clusters = new HashMap<String, List<Cluster>>();
+ 
+     public static final String PBS_JOB_MANAGER = "pbs";
+     public static final String SLURM_JOB_MANAGER = "slurm";
+     public static final String SUN_GRID_ENGINE_JOB_MANAGER = "UGE";
+     public static final String LSF_JOB_MANAGER = "LSF";
+ 
+     public static int maxClusterCount = 5;
+ 
+     /**
+      * This method is to add computing resource specific authentication, if its a third party machine, use the other addSecurityContext
+      * @param jobExecutionContext
+      * @throws GFacException
+      * @throws ApplicationSettingsException
+      */
+     public static void addSecurityContext(JobExecutionContext jobExecutionContext) throws GFacException, ApplicationSettingsException {
+         JobSubmissionProtocol preferredJobSubmissionProtocol = jobExecutionContext.getPreferredJobSubmissionProtocol();
+         JobSubmissionInterface preferredJobSubmissionInterface = jobExecutionContext.getPreferredJobSubmissionInterface();
+         if (preferredJobSubmissionProtocol == JobSubmissionProtocol.GLOBUS || preferredJobSubmissionProtocol == JobSubmissionProtocol.UNICORE) {
+             logger.error("This is a wrong method to invoke to non ssh host types,please check your gfac-config.xml");
+         } else if (preferredJobSubmissionProtocol == JobSubmissionProtocol.SSH) {
+             try {
+                 AppCatalog appCatalog = jobExecutionContext.getAppCatalog();
+                 SSHJobSubmission sshJobSubmission = appCatalog.getComputeResource().getSSHJobSubmission(preferredJobSubmissionInterface.getJobSubmissionInterfaceId());
+                 SecurityProtocol securityProtocol = sshJobSubmission.getSecurityProtocol();
 -                if (securityProtocol == SecurityProtocol.GSI || securityProtocol == SecurityProtocol.SSH_KEYS) {
++                if (securityProtocol == SecurityProtocol.GSI || securityProtocol == SecurityProtocol.SSH_KEYS || securityProtocol == SecurityProtocol.USERNAME_PASSWORD) {
+                     SSHSecurityContext sshSecurityContext = new SSHSecurityContext();
+                     String credentialStoreToken = jobExecutionContext.getCredentialStoreToken(); // this is set by the framework
+                     RequestData requestData = new RequestData(jobExecutionContext.getGatewayID());
+                     requestData.setTokenId(credentialStoreToken);
+ 
+                     ServerInfo serverInfo = new ServerInfo(null, jobExecutionContext.getHostName());
+ 
+                     Cluster pbsCluster = null;
+                     try {
+                         AuthenticationInfo tokenizedSSHAuthInfo = new TokenizedSSHAuthInfo(requestData);
+                         String installedParentPath = jobExecutionContext.getResourceJobManager().getJobManagerBinPath();
+                         if (installedParentPath == null) {
+                             installedParentPath = "/";
+                         }
+ 
+                         SSHCredential credentials =((TokenizedSSHAuthInfo)tokenizedSSHAuthInfo).getCredentials();// this is just a call to get and set credentials in to this object,data will be used
 -                        if(credentials.getPrivateKey()==null || credentials.getPublicKey()==null){
++                        if(credentials.getPrivateKey()==null || credentials.getPublicKey()==null || securityProtocol == SecurityProtocol.USERNAME_PASSWORD){
+                             // now we fall back to username password authentication
+                             Properties configurationProperties = ServerSettings.getProperties();
+                             tokenizedSSHAuthInfo = new DefaultPasswordAuthenticationInfo(configurationProperties.getProperty(Constants.SSH_PASSWORD));
+                         }
+                         // This should be the login user name from compute resource preference
+                         String loginUser = jobExecutionContext.getLoginUserName();
+                         if (loginUser == null) {
+                             loginUser = credentials.getPortalUserName();
+                         }
+                         serverInfo.setUserName(loginUser);
+                         jobExecutionContext.getExperiment().setUserName(loginUser);
+ 
+ 
+                         // inside the pbsCluser object
+ 
+                         String key = loginUser + jobExecutionContext.getHostName() + serverInfo.getPort();
+                         boolean recreate = false;
+                         synchronized (clusters) {
+                             if (clusters.containsKey(key) && clusters.get(key).size() < maxClusterCount) {
+                                 recreate = true;
+                             } else if (clusters.containsKey(key)) {
+                                 int i = new Random().nextInt(Integer.MAX_VALUE) % maxClusterCount;
+                                 if (clusters.get(key).get(i).getSession().isConnected()) {
+                                     pbsCluster = clusters.get(key).get(i);
+                                 } else {
+                                     clusters.get(key).remove(i);
+                                     recreate = true;
+                                 }
+                                 if (!recreate) {
+                                     try {
+                                         pbsCluster.listDirectory("~/"); // its hard to trust isConnected method, so we try to connect if it works we are good,else we recreate
+                                     } catch (Exception e) {
+                                         clusters.get(key).remove(i);
+                                         logger.info("Connection found the connection map is expired, so we create from the scratch");
+                                         maxClusterCount++;
+                                         recreate = true; // we make the pbsCluster to create again if there is any exception druing connection
+                                     }
+                                 }
+                                 logger.info("Re-using the same connection used with the connection string:" + key);
+                             } else {
+                                 recreate = true;
+                             }
+                             if (recreate) {
+                             	 JobManagerConfiguration jConfig = null;
+                                  String jobManager = sshJobSubmission.getResourceJobManager().getResourceJobManagerType().toString();
+                                  if (jobManager == null) {
+                                      logger.error("No Job Manager is configured, so we are picking pbs as the default job manager");
+                                      jConfig = CommonUtils.getPBSJobManager(installedParentPath);
+                                  } else {
+                                      if (PBS_JOB_MANAGER.equalsIgnoreCase(jobManager)) {
+                                          jConfig = CommonUtils.getPBSJobManager(installedParentPath);
+                                      } else if (SLURM_JOB_MANAGER.equalsIgnoreCase(jobManager)) {
+                                          jConfig = CommonUtils.getSLURMJobManager(installedParentPath);
+                                      } else if (SUN_GRID_ENGINE_JOB_MANAGER.equalsIgnoreCase(jobManager)) {
+                                          jConfig = CommonUtils.getUGEJobManager(installedParentPath);
+                                      } else if (LSF_JOB_MANAGER.equalsIgnoreCase(jobManager)) {
+                                          jConfig = CommonUtils.getLSFJobManager(installedParentPath);
+                                      }
+                                  }
+ 
+                                 pbsCluster = new PBSCluster(serverInfo, tokenizedSSHAuthInfo,jConfig);
+                                 List<Cluster> pbsClusters = null;
+                                 if (!(clusters.containsKey(key))) {
+                                     pbsClusters = new ArrayList<Cluster>();
+                                 } else {
+                                     pbsClusters = clusters.get(key);
+                                 }
+                                 pbsClusters.add(pbsCluster);
+                                 clusters.put(key, pbsClusters);
+                             }
+                         }
+                     } catch (Exception e) {
+                         throw new GFacException("Error occurred...", e);
+                     }
+                     sshSecurityContext.setPbsCluster(pbsCluster);
+                     jobExecutionContext.addSecurityContext(jobExecutionContext.getHostName(), sshSecurityContext);
+                 }
+             } catch (AppCatalogException e) {
+                 throw new GFacException("Error while getting SSH Submission object from app catalog", e);
+             }
+         }
+     }
+ 
+     /**
+      * This method can be used to add third party resource security contexts
+      * @param jobExecutionContext
+      * @param sshAuth
+      * @throws GFacException
+      * @throws ApplicationSettingsException
+      */
+     public static void addSecurityContext(JobExecutionContext jobExecutionContext,SSHAuthWrapper sshAuth) throws GFacException, ApplicationSettingsException {
+         try {
+             if(sshAuth== null) {
+                 throw new GFacException("Error adding security Context, because sshAuthWrapper is null");
+             }
+             SSHSecurityContext sshSecurityContext = new SSHSecurityContext();
+             AppCatalog appCatalog = jobExecutionContext.getAppCatalog();
+             JobSubmissionInterface preferredJobSubmissionInterface = jobExecutionContext.getPreferredJobSubmissionInterface();
+             SSHJobSubmission sshJobSubmission = null;
+ 			try {
+ 				sshJobSubmission = appCatalog.getComputeResource().getSSHJobSubmission(preferredJobSubmissionInterface.getJobSubmissionInterfaceId());
+ 			} catch (Exception e1) {
+ 				 logger.error("Not able to get SSHJobSubmission from registry");
+ 			}
+ 
+             Cluster pbsCluster = null;
+             String key=sshAuth.getKey();
+             boolean recreate = false;
+             synchronized (clusters) {
+                 if (clusters.containsKey(key) && clusters.get(key).size() < maxClusterCount) {
+                     recreate = true;
+                 } else if (clusters.containsKey(key)) {
+                     int i = new Random().nextInt(Integer.MAX_VALUE) % maxClusterCount;
+                     if (clusters.get(key).get(i).getSession().isConnected()) {
+                         pbsCluster = clusters.get(key).get(i);
+                     } else {
+                         clusters.get(key).remove(i);
+                         recreate = true;
+                     }
+                     if (!recreate) {
+                         try {
+                             pbsCluster.listDirectory("~/"); // its hard to trust isConnected method, so we try to connect if it works we are good,else we recreate
+                         } catch (Exception e) {
+                             clusters.get(key).remove(i);
+                             logger.info("Connection found the connection map is expired, so we create from the scratch");
+                             maxClusterCount++;
+                             recreate = true; // we make the pbsCluster to create again if there is any exception druing connection
+                         }
+                     }
+                     logger.info("Re-using the same connection used with the connection string:" + key);
+                 } else {
+                     recreate = true;
+                 }
+                 if (recreate) {
+                	 JobManagerConfiguration jConfig = null;
+                	 String installedParentPath = null;
+                	 if(jobExecutionContext.getResourceJobManager()!= null){
+                		installedParentPath = jobExecutionContext.getResourceJobManager().getJobManagerBinPath();
+                	 }
+                  if (installedParentPath == null) {
+                      installedParentPath = "/";
+                  }
+ 					if (sshJobSubmission != null) {
+ 						String jobManager = sshJobSubmission.getResourceJobManager().getResourceJobManagerType().toString();
+ 						if (jobManager == null) {
+ 							logger.error("No Job Manager is configured, so we are picking pbs as the default job manager");
+ 							jConfig = CommonUtils.getPBSJobManager(installedParentPath);
+ 						} else {
+ 							if (PBS_JOB_MANAGER.equalsIgnoreCase(jobManager)) {
+ 								jConfig = CommonUtils.getPBSJobManager(installedParentPath);
+ 							} else if (SLURM_JOB_MANAGER.equalsIgnoreCase(jobManager)) {
+ 								jConfig = CommonUtils.getSLURMJobManager(installedParentPath);
+ 							} else if (SUN_GRID_ENGINE_JOB_MANAGER.equalsIgnoreCase(jobManager)) {
+ 								jConfig = CommonUtils.getUGEJobManager(installedParentPath);
+ 							} else if (LSF_JOB_MANAGER.equals(jobManager)) {
+ 								jConfig = CommonUtils.getLSFJobManager(installedParentPath);
+ 							}
+ 						}
+ 					}
+                     pbsCluster = new PBSCluster(sshAuth.getServerInfo(), sshAuth.getAuthenticationInfo(),jConfig);
+                     key = sshAuth.getKey();
+                     List<Cluster> pbsClusters = null;
+                     if (!(clusters.containsKey(key))) {
+                         pbsClusters = new ArrayList<Cluster>();
+                     } else {
+                         pbsClusters = clusters.get(key);
+                     }
+                     pbsClusters.add(pbsCluster);
+                     clusters.put(key, pbsClusters);
+                 }
+             }
+             sshSecurityContext.setPbsCluster(pbsCluster);
+             jobExecutionContext.addSecurityContext(key, sshSecurityContext);
+         } catch (Exception e) {
+             logger.error(e.getMessage(), e);
+             throw new GFacException("Error adding security Context", e);
+         }
+     }
+ 
+ 
+     public static JobDescriptor createJobDescriptor(JobExecutionContext jobExecutionContext, Cluster cluster) throws AppCatalogException, ApplicationSettingsException {
+         JobDescriptor jobDescriptor = new JobDescriptor();
+         TaskDetails taskData = jobExecutionContext.getTaskData();
+ 
+ 
+         // set email based job monitoring email  address if monitor mode is JOB_EMAIL_NOTIFICATION_MONITOR
+         boolean addJobNotifMail = isEmailBasedJobMonitor(jobExecutionContext);
+         String emailIds = null;
+         if (addJobNotifMail) {
+             emailIds = ServerSettings.getEmailBasedMonitorAddress();
+         }
+         // add all configured job notification email addresses.
+         if (ServerSettings.getSetting(ServerSettings.JOB_NOTIFICATION_ENABLE).equalsIgnoreCase("true")) {
+             String flags = ServerSettings.getSetting(ServerSettings.JOB_NOTIFICATION_FLAGS);
+             if (flags != null && jobExecutionContext.getApplicationContext().getComputeResourceDescription().getHostName().equals("stampede.tacc.xsede.org")) {
+                 flags = "ALL";
+             }
+             jobDescriptor.setMailOptions(flags);
+ 
+             String userJobNotifEmailIds = ServerSettings.getSetting(ServerSettings.JOB_NOTIFICATION_EMAILIDS);
+             if (userJobNotifEmailIds != null && !userJobNotifEmailIds.isEmpty()) {
+                 if (emailIds != null && !emailIds.isEmpty()) {
+                     emailIds += ("," + userJobNotifEmailIds);
+                 } else {
+                     emailIds = userJobNotifEmailIds;
+                 }
+             }
+ 
+             if (taskData.isEnableEmailNotification()) {
+                 List<String> emailList = jobExecutionContext.getTaskData().getEmailAddresses();
+                 String elist = GFacUtils.listToCsv(emailList, ',');
+                 if (elist != null && !elist.isEmpty()) {
+                     if (emailIds != null && !emailIds.isEmpty()) {
+                         emailIds = emailIds + "," + elist;
+                     } else {
+                         emailIds = elist;
+                     }
+                 }
+             }
+         }
+         if (emailIds != null && !emailIds.isEmpty()) {
+             logger.info("Email list: " + emailIds);
+             jobDescriptor.setMailAddress(emailIds);
+         }
+         // this is common for any application descriptor
+ 
+         jobDescriptor.setCallBackIp(ServerSettings.getIp());
+         jobDescriptor.setCallBackPort(ServerSettings.getSetting(org.apache.airavata.common.utils.Constants.GFAC_SERVER_PORT, "8950"));
+         jobDescriptor.setInputDirectory(jobExecutionContext.getInputDir());
+         jobDescriptor.setOutputDirectory(jobExecutionContext.getOutputDir());
+         jobDescriptor.setExecutablePath(jobExecutionContext.getApplicationContext()
+                 .getApplicationDeploymentDescription().getExecutablePath());
+         jobDescriptor.setStandardOutFile(jobExecutionContext.getStandardOutput());
+         jobDescriptor.setStandardErrorFile(jobExecutionContext.getStandardError());
+         String computationalProjectAccount = taskData.getTaskScheduling().getComputationalProjectAccount();
+         if (computationalProjectAccount == null){
+             ComputeResourcePreference computeResourcePreference = jobExecutionContext.getApplicationContext().getComputeResourcePreference();
+             if (computeResourcePreference != null) {
+                 computationalProjectAccount = computeResourcePreference.getAllocationProjectNumber();
+             }
+         }
+         if (computationalProjectAccount != null) {
+             jobDescriptor.setAcountString(computationalProjectAccount);
+         }
+         // To make job name alpha numeric
+         jobDescriptor.setJobName("A" + String.valueOf(generateJobName()));
+         jobDescriptor.setWorkingDirectory(jobExecutionContext.getWorkingDir());
+ 
+         List<String> inputValues = new ArrayList<String>();
+         MessageContext input = jobExecutionContext.getInMessageContext();
+ 
+         // sort the inputs first and then build the command ListR
+         Comparator<InputDataObjectType> inputOrderComparator = new Comparator<InputDataObjectType>() {
+             @Override
+             public int compare(InputDataObjectType inputDataObjectType, InputDataObjectType t1) {
+                 return inputDataObjectType.getInputOrder() - t1.getInputOrder();
+             }
+         };
+         Set<InputDataObjectType> sortedInputSet = new TreeSet<InputDataObjectType>(inputOrderComparator);
+         for (Object object : input.getParameters().values()) {
+             if (object instanceof InputDataObjectType) {
+                 InputDataObjectType inputDOT = (InputDataObjectType) object;
+                 sortedInputSet.add(inputDOT);
+             }
+         }
+         for (InputDataObjectType inputDataObjectType : sortedInputSet) {
+             if (!inputDataObjectType.isRequiredToAddedToCommandLine()) {
+                 continue;
+             }
+             if (inputDataObjectType.getApplicationArgument() != null
+                     && !inputDataObjectType.getApplicationArgument().equals("")) {
+                 inputValues.add(inputDataObjectType.getApplicationArgument());
+             }
+ 
+             if (inputDataObjectType.getValue() != null
+                     && !inputDataObjectType.getValue().equals("")) {
+                 if (inputDataObjectType.getType() == DataType.URI) {
+                     // set only the relative path
+                     String filePath = inputDataObjectType.getValue();
+                     filePath = filePath.substring(filePath.lastIndexOf(File.separatorChar) + 1, filePath.length());
+                     inputValues.add(filePath);
+                 }else {
+                     inputValues.add(inputDataObjectType.getValue());
+                 }
+ 
+             }
+         }
+         Map<String, Object> outputParams = jobExecutionContext.getOutMessageContext().getParameters();
+         for (Object outputParam : outputParams.values()) {
+             if (outputParam instanceof OutputDataObjectType) {
+                 OutputDataObjectType output = (OutputDataObjectType) outputParam;
+                 if (output.getApplicationArgument() != null
+                         && !output.getApplicationArgument().equals("")) {
+                     inputValues.add(output.getApplicationArgument());
+                 }
+                 if (output.getValue() != null && !output.getValue().equals("") && output.isRequiredToAddedToCommandLine()) {
+                     if (output.getType() == DataType.URI){
+                         String filePath = output.getValue();
+                         filePath = filePath.substring(filePath.lastIndexOf(File.separatorChar) + 1, filePath.length());
+                         inputValues.add(filePath);
+                     }
+                 }
+             }
+         }
+ 
+         jobDescriptor.setInputValues(inputValues);
+         jobDescriptor.setUserName(((GSISSHAbstractCluster) cluster).getServerInfo().getUserName());
+         jobDescriptor.setShellName("/bin/bash");
+         jobDescriptor.setAllEnvExport(true);
+         jobDescriptor.setOwner(((PBSCluster) cluster).getServerInfo().getUserName());
+ 
+         ResourceJobManager resourceJobManager = jobExecutionContext.getResourceJobManager();
+ 
+ 
+         ComputationalResourceScheduling taskScheduling = taskData.getTaskScheduling();
+         if (taskScheduling != null) {
+             int totalNodeCount = taskScheduling.getNodeCount();
+             int totalCPUCount = taskScheduling.getTotalCPUCount();
+ 
+ 
+             if (taskScheduling.getComputationalProjectAccount() != null) {
+                 jobDescriptor.setAcountString(taskScheduling.getComputationalProjectAccount());
+             }
+             if (taskScheduling.getQueueName() != null) {
+                 jobDescriptor.setQueueName(taskScheduling.getQueueName());
+             }
+ 
+             if (totalNodeCount > 0) {
+                 jobDescriptor.setNodes(totalNodeCount);
+             }
+             if (taskScheduling.getComputationalProjectAccount() != null) {
+                 jobDescriptor.setAcountString(taskScheduling.getComputationalProjectAccount());
+             }
+             if (taskScheduling.getQueueName() != null) {
+                 jobDescriptor.setQueueName(taskScheduling.getQueueName());
+             }
+             if (totalCPUCount > 0) {
+                 int ppn = totalCPUCount / totalNodeCount;
+                 jobDescriptor.setProcessesPerNode(ppn);
+                 jobDescriptor.setCPUCount(totalCPUCount);
+             }
+             if (taskScheduling.getWallTimeLimit() > 0) {
+                 jobDescriptor.setMaxWallTime(String.valueOf(taskScheduling.getWallTimeLimit()));
+                 if(resourceJobManager.getResourceJobManagerType().equals(ResourceJobManagerType.LSF)){
+                     jobDescriptor.setMaxWallTimeForLSF(String.valueOf(taskScheduling.getWallTimeLimit()));
+                 }
+             }
+             if (taskScheduling.getTotalPhysicalMemory() > 0) {
+                 jobDescriptor.setUsedMemory(taskScheduling.getTotalPhysicalMemory() + "");
+             }
+         } else {
+             logger.error("Task scheduling cannot be null at this point..");
+         }
+         ApplicationDeploymentDescription appDepDescription = jobExecutionContext.getApplicationContext().getApplicationDeploymentDescription();
+         List<String> moduleCmds = appDepDescription.getModuleLoadCmds();
+         if (moduleCmds != null) {
+             for (String moduleCmd : moduleCmds) {
+                 jobDescriptor.addModuleLoadCommands(moduleCmd);
+             }
+         }
+         List<String> preJobCommands = appDepDescription.getPreJobCommands();
+         if (preJobCommands != null) {
+             for (String preJobCommand : preJobCommands) {
+                 jobDescriptor.addPreJobCommand(parseCommand(preJobCommand, jobExecutionContext));
+             }
+         }
+ 
+         List<String> postJobCommands = appDepDescription.getPostJobCommands();
+         if (postJobCommands != null) {
+             for (String postJobCommand : postJobCommands) {
+                 jobDescriptor.addPostJobCommand(parseCommand(postJobCommand, jobExecutionContext));
+             }
+         }
+ 
+         ApplicationParallelismType parallelism = appDepDescription.getParallelism();
+         if (parallelism != null){
+             if (parallelism == ApplicationParallelismType.MPI || parallelism == ApplicationParallelismType.OPENMP || parallelism == ApplicationParallelismType.OPENMP_MPI){
+                 Map<JobManagerCommand, String> jobManagerCommands = resourceJobManager.getJobManagerCommands();
+                 if (jobManagerCommands != null && !jobManagerCommands.isEmpty()) {
+                     for (JobManagerCommand command : jobManagerCommands.keySet()) {
+                         if (command == JobManagerCommand.SUBMISSION) {
+                             String commandVal = jobManagerCommands.get(command);
+                             jobDescriptor.setJobSubmitter(commandVal);
+                         }
+                     }
+                 }
+             }
+         }
+         return jobDescriptor;
+     }
+ 
+     public static boolean isEmailBasedJobMonitor(JobExecutionContext jobExecutionContext) throws AppCatalogException {
+         if (jobExecutionContext.getPreferredJobSubmissionProtocol() == JobSubmissionProtocol.SSH) {
+             String jobSubmissionInterfaceId = jobExecutionContext.getPreferredJobSubmissionInterface().getJobSubmissionInterfaceId();
+             SSHJobSubmission sshJobSubmission = jobExecutionContext.getAppCatalog().getComputeResource().getSSHJobSubmission(jobSubmissionInterfaceId);
+             MonitorMode monitorMode = sshJobSubmission.getMonitorMode();
+             return monitorMode != null && monitorMode == MonitorMode.JOB_EMAIL_NOTIFICATION_MONITOR;
+         } else {
+             return false;
+         }
+     }
+ 
+     private static int generateJobName() {
+         Random random = new Random();
+         int i = random.nextInt(Integer.MAX_VALUE);
+         i = i + 99999999;
+         if(i<0) {
+             i = i * (-1);
+         }
+         return i;
+     }
+ 
+     private static String parseCommand(String value, JobExecutionContext jobExecutionContext) {
+         String parsedValue = value.replaceAll("\\$workingDir", jobExecutionContext.getWorkingDir());
+         parsedValue = parsedValue.replaceAll("\\$inputDir", jobExecutionContext.getInputDir());
+         parsedValue = parsedValue.replaceAll("\\$outputDir", jobExecutionContext.getOutputDir());
+         return parsedValue;
+     }
+     /**
+      * This method can be used to set the Security Context if its not set and later use it in other places
+      * @param jobExecutionContext
+      * @param authenticationInfo
+      * @param userName
+      * @param hostName
+      * @param port
+      * @return
+      * @throws GFacException
+      */
+     public static String prepareSecurityContext(JobExecutionContext jobExecutionContext, AuthenticationInfo authenticationInfo
+             , String userName, String hostName, int port) throws GFacException {
+         ServerInfo serverInfo = new ServerInfo(userName, hostName);
+         String key = userName+hostName+port;
+         SSHAuthWrapper sshAuthWrapper = new SSHAuthWrapper(serverInfo, authenticationInfo, key);
+         if (jobExecutionContext.getSecurityContext(key) == null) {
+             try {
+                 GFACSSHUtils.addSecurityContext(jobExecutionContext, sshAuthWrapper);
+             } catch (ApplicationSettingsException e) {
+                 logger.error(e.getMessage());
+                 try {
+                     StringWriter errors = new StringWriter();
+                     e.printStackTrace(new PrintWriter(errors));
+                     GFacUtils.saveErrorDetails(jobExecutionContext, errors.toString(), CorrectiveAction.CONTACT_SUPPORT, ErrorCategory.AIRAVATA_INTERNAL_ERROR);
+                 } catch (GFacException e1) {
+                     logger.error(e1.getLocalizedMessage());
+                 }
+                 throw new GFacHandlerException("Error while creating SSHSecurityContext", e, e.getLocalizedMessage());
+             }
+         }
+         return key;
+     }
+ }

http://git-wip-us.apache.org/repos/asf/airavata/blob/b4ede9cb/modules/gfac/gfac-impl/src/test/java/org/apache/airavata/core/gfac/services/impl/BigRed2TestWithSSHAuth.java
----------------------------------------------------------------------
diff --cc modules/gfac/gfac-impl/src/test/java/org/apache/airavata/core/gfac/services/impl/BigRed2TestWithSSHAuth.java
index 0000000,73a6e4a..38981aa
mode 000000,100644..100644
--- a/modules/gfac/gfac-impl/src/test/java/org/apache/airavata/core/gfac/services/impl/BigRed2TestWithSSHAuth.java
+++ b/modules/gfac/gfac-impl/src/test/java/org/apache/airavata/core/gfac/services/impl/BigRed2TestWithSSHAuth.java
@@@ -1,0 -1,252 +1,252 @@@
+ ///*
+ // *
+ // * 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.core.gfac.services.impl;
+ //
+ //import org.apache.airavata.commons.gfac.type.ActualParameter;
+ //import org.apache.airavata.commons.gfac.type.ApplicationDescription;
+ //import org.apache.airavata.commons.gfac.type.HostDescription;
+ //import org.apache.airavata.commons.gfac.type.ServiceDescription;
+ //import org.apache.airavata.gfac.GFacConfiguration;
+ //import org.apache.airavata.gfac.GFacException;
+ //import org.apache.airavata.gfac.SecurityContext;
+ //import org.apache.airavata.gfac.core.context.ApplicationContext;
+ //import org.apache.airavata.gfac.core.context.JobExecutionContext;
+ //import org.apache.airavata.gfac.core.context.MessageContext;
+ //import org.apache.airavata.gfac.impl.BetterGfacImpl;
+ //import org.apache.airavata.gfac.ssh.security.SSHSecurityContext;
+ //import org.apache.airavata.gfac.ssh.api.Cluster;
+ //import org.apache.airavata.gfac.ssh.api.SSHApiException;
+ //import org.apache.airavata.gfac.ssh.api.ServerInfo;
+ //import AuthenticationInfo;
+ //import org.apache.airavata.gfac.ssh.api.job.JobManagerConfiguration;
+ //import org.apache.airavata.gfac.ssh.impl.PBSCluster;
+ //import org.apache.airavata.gfac.ssh.impl.authentication.DefaultPasswordAuthenticationInfo;
+ //import org.apache.airavata.gfac.ssh.impl.authentication.DefaultPublicKeyFileAuthentication;
+ //import org.apache.airavata.gfac.ssh.util.CommonUtils;
+ //import org.apache.airavata.model.workspace.experiment.TaskDetails;
 -//import org.apache.airavata.persistance.registry.jpa.impl.RegistryFactory;
++//import org.apache.airavata.experiment.registry.jpa.impl.RegistryFactory;
+ //import org.apache.airavata.schemas.gfac.*;
+ //import org.testng.annotations.BeforeClass;
+ //import org.testng.annotations.Test;
+ //
+ //import java.io.File;
+ //import java.net.URL;
+ //import java.util.ArrayList;
+ //import java.util.Date;
+ //import java.util.List;
+ //import java.util.UUID;
+ //
+ //public class BigRed2TestWithSSHAuth {
+ //    private JobExecutionContext jobExecutionContext;
+ //
+ //    private String userName;
+ //    private String password;
+ //    private String passPhrase;
+ //    private String hostName;
+ //    private String workingDirectory;
+ //    private String privateKeyPath;
+ //    private String publicKeyPath;
+ //
+ //    @BeforeClass
+ //    public void setUp() throws Exception {
+ //
+ //        System.out.println("Test case name " + this.getClass().getName());
+ ////        System.setProperty("ssh.host","bigred2.uits.iu.edu");        //default ssh host
+ ////        System.setProperty("ssh.user", "lginnali");
+ ////        System.setProperty("ssh.private.key.path", "/Users/lahirugunathilake/.ssh/id_dsa");
+ ////        System.setProperty("ssh.public.key.path", "/Users/lahirugunathilake/.ssh/id_dsa.pub");
+ ////        System.setProperty("ssh.working.directory", "/tmp");
+ //
+ //        this.hostName = "bigred2.uits.iu.edu";
+ //        this.hostName = System.getProperty("ssh.host");
+ //        this.userName = System.getProperty("ssh.username");
+ //        this.password = System.getProperty("ssh.password");
+ //        this.privateKeyPath = System.getProperty("private.ssh.key");
+ //        this.publicKeyPath = System.getProperty("public.ssh.key");
+ //        this.passPhrase = System.getProperty("ssh.keypass");
+ //        this.workingDirectory = System.getProperty("ssh.working.directory");
+ //
+ //
+ //         if (this.userName == null
+ //                || (this.password==null && (this.publicKeyPath == null || this.privateKeyPath == null)) || this.workingDirectory == null) {
+ //            System.out.println("########### In order to test you have to either username password or private,public keys");
+ //            System.out.println("Use -Dssh.username=xxx -Dssh.password=yyy -Dssh.keypass=zzz " +
+ //                    "-Dprivate.ssh.key -Dpublic.ssh.key -Dssh.working.directory ");
+ //        }
+ //        URL resource = BigRed2TestWithSSHAuth.class.getClassLoader().getResource(org.apache.airavata.common.utils.Constants.GFAC_CONFIG_XML);
+ //        assert resource != null;
+ //        System.out.println(resource.getFile());
+ //        GFacConfiguration gFacConfiguration = GFacConfiguration.create(new File(resource.getPath()), null);
+ //
+ ////        gFacConfiguration.setMyProxyLifeCycle(3600);
+ ////        gFacConfiguration.setMyProxyServer("myproxy.teragrid.org");
+ ////        gFacConfiguration.setMyProxyUser("*****");
+ ////        gFacConfiguration.setMyProxyPassphrase("*****");
+ ////        gFacConfiguration.setTrustedCertLocation("./certificates");
+ ////        //have to set InFlwo Handlers and outFlowHandlers
+ ////        gFacConfiguration.setInHandlers(Arrays.asList(new String[] {"org.apache.airavata.gfac.handler.GramDirectorySetupHandler","org.apache.airavata.gfac.handler.GridFTPInputHandler"}));
+ ////        gFacConfiguration.setOutHandlers(Arrays.asList(new String[] {"org.apache.airavata.gfac.handler.GridFTPOutputHandler"}));
+ //
+ //        /*
+ //        * Host
+ //        */
+ //        HostDescription host = new HostDescription(SSHHostType.type);
+ //        host.getType().setHostAddress(hostName);
+ //        host.getType().setHostName(hostName);
+ //        ((SSHHostType)host.getType()).setHpcResource(true);
+ //        /*
+ //        * App
+ //        */
+ //        ApplicationDescription appDesc = new ApplicationDescription(HpcApplicationDeploymentType.type);
+ //        HpcApplicationDeploymentType app = (HpcApplicationDeploymentType) appDesc.getType();
+ //        ApplicationDeploymentDescriptionType.ApplicationName name = ApplicationDeploymentDescriptionType.ApplicationName.Factory.newInstance();
+ //        name.setStringValue("EchoLocal");
+ //        app.setApplicationName(name);
+ //
+ //        app.setCpuCount(1);
+ //        app.setJobType(JobTypeType.SERIAL);
+ //        app.setNodeCount(1);
+ //        app.setProcessorsPerNode(1);
+ //
+ //        /*
+ //        * Use bat file if it is compiled on Windows
+ //        */
+ //        app.setExecutableLocation("/bin/echo");
+ //
+ //        /*
+ //        * Default tmp location
+ //        */
+ //        String tempDir = "/tmp";
+ //        String date = (new Date()).toString();
+ //        date = date.replaceAll(" ", "_");
+ //        date = date.replaceAll(":", "_");
+ //
+ //        tempDir = tempDir + File.separator
+ //                + "SimpleEcho" + "_" + date + "_" + UUID.randomUUID();
+ //
+ //        System.out.println(tempDir);
+ //        app.setScratchWorkingDirectory(tempDir);
+ //        app.setStaticWorkingDirectory(tempDir);
+ //        app.setInputDataDirectory(tempDir + File.separator + "inputData");
+ //        app.setOutputDataDirectory(tempDir + File.separator + "outputData");
+ //        app.setStandardOutput(tempDir + File.separator + app.getApplicationName().getStringValue() + ".stdout");
+ //        app.setStandardError(tempDir + File.separator + app.getApplicationName().getStringValue() + ".stderr");
+ //        app.setMaxWallTime(5);
+ //        app.setJobSubmitterCommand("aprun -n 1");
+ //        app.setInstalledParentPath("/opt/torque/torque-4.2.3.1/bin/");
+ //
+ //        /*
+ //        * Service
+ //        */
+ //        ServiceDescription serv = new ServiceDescription();
+ //        serv.getType().setName("SimpleEcho");
+ //
+ //        List<InputParameterType> inputList = new ArrayList<InputParameterType>();
+ //
+ //        InputParameterType input = InputParameterType.Factory.newInstance();
+ //        input.setParameterName("echo_input");
+ //        input.setParameterType(StringParameterType.Factory.newInstance());
+ //        inputList.add(input);
+ //
+ //        InputParameterType[] inputParamList = inputList.toArray(new InputParameterType[inputList
+ //
+ //                .size()]);
+ //        List<OutputParameterType> outputList = new ArrayList<OutputParameterType>();
+ //        OutputParameterType output = OutputParameterType.Factory.newInstance();
+ //        output.setParameterName("echo_output");
+ //        output.setParameterType(StringParameterType.Factory.newInstance());
+ //        outputList.add(output);
+ //
+ //        OutputParameterType[] outputParamList = outputList
+ //                .toArray(new OutputParameterType[outputList.size()]);
+ //
+ //        serv.getType().setInputParametersArray(inputParamList);
+ //        serv.getType().setOutputParametersArray(outputParamList);
+ //
+ //        jobExecutionContext = new JobExecutionContext(gFacConfiguration, serv.getType().getName());
+ //        // Adding security context
+ //        jobExecutionContext.addSecurityContext(SSHSecurityContext.SSH_SECURITY_CONTEXT, getSecurityContext(app));
+ //        ApplicationContext applicationContext = new ApplicationContext();
+ //        jobExecutionContext.setApplicationContext(applicationContext);
+ //        applicationContext.setServiceDescription(serv);
+ //        applicationContext.setApplicationDeploymentDescription(appDesc);
+ //        applicationContext.setHostDescription(host);
+ //
+ //        MessageContext inMessage = new MessageContext();
+ //        ActualParameter echo_input = new ActualParameter();
+ //        ((StringParameterType) echo_input.getType()).setValue("echo_output=hello");
+ //        inMessage.addParameter("echo_input", echo_input);
+ //
+ //
+ //        jobExecutionContext.setInMessageContext(inMessage);
+ //
+ //        MessageContext outMessage = new MessageContext();
+ //        ActualParameter echo_out = new ActualParameter();
+ ////		((StringParameterType)echo_input.getType()).setValue("echo_output=hello");
+ //        outMessage.addParameter("echo_output", echo_out);
+ //        jobExecutionContext.setRegistry(RegistryFactory.getLoggingRegistry());
+ //        jobExecutionContext.setTaskData(new TaskDetails("11323"));
+ //        jobExecutionContext.setOutMessageContext(outMessage);
+ //
+ //    }
+ //
+ //
+ //    private SecurityContext getSecurityContext(HpcApplicationDeploymentType app) {
+ //         try {
+ //
+ //        AuthenticationInfo authenticationInfo = null;
+ //        if (password != null) {
+ //            authenticationInfo = new DefaultPasswordAuthenticationInfo(this.password);
+ //        } else {
+ //            authenticationInfo = new DefaultPublicKeyFileAuthentication(this.publicKeyPath, this.privateKeyPath,
+ //                    this.passPhrase);
+ //        }
+ //        // Server info
+ //        ServerInfo serverInfo = new ServerInfo(this.userName, this.hostName);
+ //
+ //        Cluster pbsCluster = null;
+ //        SSHSecurityContext sshSecurityContext = null;
+ //
+ //            JobManagerConfiguration pbsJobManager = CommonUtils.getPBSJobManager(app.getInstalledParentPath());
+ //            pbsCluster = new PBSCluster(serverInfo, authenticationInfo, pbsJobManager);
+ //
+ //
+ //            sshSecurityContext = new SSHSecurityContext();
+ //            sshSecurityContext.setPbsCluster(pbsCluster);
+ //            sshSecurityContext.setUsername(userName);
+ //            sshSecurityContext.setKeyPass(passPhrase);
+ //            sshSecurityContext.setPrivateKeyLoc(privateKeyPath);
+ //             return sshSecurityContext;
+ //        } catch (SSHApiException e) {
+ //            e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
+ //        }
+ //        return null;
+ //    }
+ //
+ //    @Test
+ //    public void testSSHProvider() throws GFacException {
+ //        BetterGfacImpl gFacAPI = new BetterGfacImpl();
+ //        gFacAPI.submitJob(jobExecutionContext.getExperimentID(), jobExecutionContext.getTaskData().getTaskID(), jobExecutionContext.getGatewayID());
+ //        org.junit.Assert.assertNotNull(jobExecutionContext.getJobDetails().getJobDescription());
+ //        org.junit.Assert.assertNotNull(jobExecutionContext.getJobDetails().getJobID());
+ //    }
+ //
+ //}


Mime
View raw message