You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by la...@apache.org on 2014/04/24 22:52:02 UTC
[06/11] creating gfac-bes and gfac-gram out from gfac-core
http://git-wip-us.apache.org/repos/asf/airavata/blob/13b505ae/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/impl/BESProvider.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/impl/BESProvider.java b/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/impl/BESProvider.java
deleted file mode 100644
index 654c9ec..0000000
--- a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/impl/BESProvider.java
+++ /dev/null
@@ -1,568 +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.gfac.provider.impl;
-
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.math.BigInteger;
-import java.security.InvalidKeyException;
-import java.security.KeyPair;
-import java.security.KeyPairGenerator;
-import java.security.PrivateKey;
-import java.security.cert.X509Certificate;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Calendar;
-import java.util.Date;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.Random;
-import java.util.Set;
-
-import javax.security.auth.x500.X500Principal;
-
-import org.apache.airavata.gfac.Constants;
-import org.apache.airavata.gfac.GFacException;
-import org.apache.airavata.gfac.context.JobExecutionContext;
-import org.apache.airavata.gfac.context.security.GSISecurityContext;
-import org.apache.airavata.gfac.notification.events.StatusChangeEvent;
-import org.apache.airavata.gfac.notification.events.UnicoreJobIDEvent;
-import org.apache.airavata.gfac.provider.GFacProvider;
-import org.apache.airavata.gfac.provider.GFacProviderException;
-import org.apache.airavata.gfac.provider.utils.DataTransferrer;
-import org.apache.airavata.gfac.provider.utils.JSDLGenerator;
-import org.apache.airavata.gfac.provider.utils.StorageCreator;
-import org.apache.airavata.gfac.utils.GFacUtils;
-import org.apache.airavata.model.workspace.experiment.JobState;
-import org.apache.airavata.registry.api.workflow.ApplicationJob;
-import org.apache.airavata.registry.api.workflow.ApplicationJob.ApplicationJobStatus;
-import org.apache.airavata.schemas.gfac.UnicoreHostType;
-import org.apache.xmlbeans.XmlCursor;
-import org.bouncycastle.asn1.ASN1InputStream;
-import org.bouncycastle.asn1.x500.X500Name;
-import org.bouncycastle.asn1.x500.style.BCStyle;
-import org.bouncycastle.asn1.x509.AlgorithmIdentifier;
-import org.bouncycastle.asn1.x509.SubjectPublicKeyInfo;
-import org.ggf.schemas.bes.x2006.x08.besFactory.ActivityStateEnumeration;
-import org.ggf.schemas.bes.x2006.x08.besFactory.ActivityStateEnumeration.Enum;
-import org.ggf.schemas.bes.x2006.x08.besFactory.ActivityStatusType;
-import org.ggf.schemas.bes.x2006.x08.besFactory.CreateActivityDocument;
-import org.ggf.schemas.bes.x2006.x08.besFactory.CreateActivityResponseDocument;
-import org.ggf.schemas.bes.x2006.x08.besFactory.GetActivityStatusesDocument;
-import org.ggf.schemas.bes.x2006.x08.besFactory.GetActivityStatusesResponseDocument;
-import org.ggf.schemas.jsdl.x2005.x11.jsdl.JobDefinitionDocument;
-import org.ggf.schemas.jsdl.x2005.x11.jsdl.JobDefinitionType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.w3.x2005.x08.addressing.EndpointReferenceType;
-
-import de.fzj.unicore.bes.client.FactoryClient;
-import de.fzj.unicore.bes.faults.UnknownActivityIdentifierFault;
-import de.fzj.unicore.uas.client.StorageClient;
-import de.fzj.unicore.wsrflite.xmlbeans.WSUtilities;
-import eu.emi.security.authn.x509.helpers.CertificateHelpers;
-import eu.emi.security.authn.x509.helpers.proxy.X509v3CertificateBuilder;
-import eu.emi.security.authn.x509.impl.CertificateUtils;
-import eu.emi.security.authn.x509.impl.CertificateUtils.Encoding;
-import eu.emi.security.authn.x509.impl.DirectoryCertChainValidator;
-import eu.emi.security.authn.x509.impl.KeyAndCertCredential;
-import eu.emi.security.authn.x509.impl.X500NameUtils;
-import eu.unicore.util.httpclient.DefaultClientConfiguration;
-
-
-
-public class BESProvider extends AbstractProvider{
- protected final Logger log = LoggerFactory.getLogger(this.getClass());
-
- private DefaultClientConfiguration secProperties;
-
- private String jobId;
-
-
-
- public void initialize(JobExecutionContext jobExecutionContext)
- throws GFacProviderException, GFacException {
- log.info("Initializing UNICORE Provider");
- super.initialize(jobExecutionContext);
- initSecurityProperties(jobExecutionContext);
- log.debug("initialized security properties");
- }
-
-
- public void execute(JobExecutionContext jobExecutionContext)
- throws GFacProviderException {
- UnicoreHostType host = (UnicoreHostType) jobExecutionContext.getApplicationContext().getHostDescription()
- .getType();
-
- String factoryUrl = host.getUnicoreBESEndPointArray()[0];
-
- EndpointReferenceType eprt = EndpointReferenceType.Factory.newInstance();
- eprt.addNewAddress().setStringValue(factoryUrl);
-
- String userDN = getUserName(jobExecutionContext);
-
- if (userDN == null || userDN.equalsIgnoreCase("admin")) {
- userDN = "CN=zdv575, O=Ultrascan Gateway, C=DE";
- }
-
- String xlogin = getCNFromUserDN(userDN);
- // create storage
- StorageCreator storageCreator = new StorageCreator(secProperties, factoryUrl, 5, xlogin);
-
- StorageClient sc = null;
- try {
- try {
- sc = storageCreator.createStorage();
- } catch (Exception e2) {
- log.error("Cannot create storage..");
- throw new GFacProviderException("Cannot create storage..", e2);
- }
-
- CreateActivityDocument cad = CreateActivityDocument.Factory.newInstance();
- JobDefinitionDocument jobDefDoc = JobDefinitionDocument.Factory.newInstance();
-
- JobDefinitionType jobDefinition = jobDefDoc.addNewJobDefinition();
- try {
- jobDefinition = JSDLGenerator.buildJSDLInstance(jobExecutionContext, sc.getUrl()).getJobDefinition();
- cad.addNewCreateActivity().addNewActivityDocument().setJobDefinition(jobDefinition);
-
- log.info("JSDL" + jobDefDoc.toString());
- } catch (Exception e1) {
- throw new GFacProviderException("Cannot generate JSDL instance from the JobExecutionContext.", e1);
- }
-
- // upload files if any
- DataTransferrer dt = new DataTransferrer(jobExecutionContext, sc);
- dt.uploadLocalFiles();
-
- FactoryClient factory = null;
- try {
- factory = new FactoryClient(eprt, secProperties);
- } catch (Exception e) {
- throw new GFacProviderException(e.getLocalizedMessage(), e);
- }
-
- CreateActivityResponseDocument response = null;
- try {
- log.info(String.format("Activity Submitting to %s ... \n", factoryUrl));
- response = factory.createActivity(cad);
- log.info(String.format("Activity Submitted to %s \n", factoryUrl));
- } catch (Exception e) {
- throw new GFacProviderException("Cannot create activity.", e);
- }
- EndpointReferenceType activityEpr = response.getCreateActivityResponse().getActivityIdentifier();
-
- log.info("Activity : " + activityEpr.getAddress().getStringValue() + " Submitted.");
-
- // factory.waitWhileActivityIsDone(activityEpr, 1000);
- jobId = WSUtilities.extractResourceID(activityEpr);
- if (jobId == null) {
- jobId = new Long(Calendar.getInstance().getTimeInMillis()).toString();
- }
- log.info("JobID: " + jobId);
- jobExecutionContext.getNotifier().publish(new UnicoreJobIDEvent(jobId));
- saveApplicationJob(jobExecutionContext, jobDefinition, activityEpr.toString());
-
- factory.getActivityStatus(activityEpr);
- log.info(formatStatusMessage(activityEpr.getAddress().getStringValue(),
- factory.getActivityStatus(activityEpr).toString()));
-
- // TODO publish the status messages to the message bus
- while ((factory.getActivityStatus(activityEpr) != ActivityStateEnumeration.FINISHED)
- && (factory.getActivityStatus(activityEpr) != ActivityStateEnumeration.FAILED)
- && (factory.getActivityStatus(activityEpr) != ActivityStateEnumeration.CANCELLED)) {
-
- ActivityStatusType activityStatus = null;
- try {
- activityStatus = getStatus(factory, activityEpr);
- JobState jobStatus = getApplicationJobStatus(activityStatus);
- String jobStatusMessage = "Status of job " + jobId + "is " + jobStatus;
- jobExecutionContext.getNotifier().publish(new StatusChangeEvent(jobStatusMessage));
- details.setJobID(jobId);
- GFacUtils.updateJobStatus(details, jobStatus);
- } catch (UnknownActivityIdentifierFault e) {
- throw new GFacProviderException(e.getMessage(), e.getCause());
- }catch (GFacException e) {
- throw new GFacProviderException(e.getMessage(), e.getCause());
- }
-
- try {
- Thread.sleep(5000);
- } catch (InterruptedException e) {
- }
- continue;
- }
-
- ActivityStatusType activityStatus = null;
- try {
- activityStatus = getStatus(factory, activityEpr);
- } catch (UnknownActivityIdentifierFault e) {
- throw new GFacProviderException(e.getMessage(), e.getCause());
- }
-
- log.info(formatStatusMessage(activityEpr.getAddress().getStringValue(), activityStatus.getState()
- .toString()));
-
- if ((activityStatus.getState() == ActivityStateEnumeration.FAILED)) {
- String error = activityStatus.getFault().getFaultcode().getLocalPart() + "\n"
- + activityStatus.getFault().getFaultstring() + "\n EXITCODE: " + activityStatus.getExitCode();
- log.info(error);
- try {
- Thread.sleep(5000);
- } catch (InterruptedException e) {
- }
- dt.downloadStdOuts();
- } else if (activityStatus.getState() == ActivityStateEnumeration.CANCELLED) {
- String experimentID = (String) jobExecutionContext.getProperty(Constants.PROP_TOPIC);
- JobState jobStatus = JobState.CANCELED;
- String jobStatusMessage = "Status of job " + jobId + "is " + jobStatus;
- jobExecutionContext.getNotifier().publish(new StatusChangeEvent(jobStatusMessage));
- details.setJobID(jobId);
- try {
- GFacUtils.saveJobStatus(details, jobStatus, jobExecutionContext.getTaskData().getTaskID());
- } catch (GFacException e) {
- throw new GFacProviderException(e.getLocalizedMessage(),e);
- }
- throw new GFacProviderException(experimentID + "Job Canceled");
- }
-
- else if (activityStatus.getState() == ActivityStateEnumeration.FINISHED) {
- try {
- Thread.sleep(5000);
- } catch (InterruptedException e) {
- }
- if (activityStatus.getExitCode() == 0) {
- dt.downloadRemoteFiles();
- } else {
- dt.downloadStdOuts();
- }
- }
-
- } catch (UnknownActivityIdentifierFault e1) {
- throw new GFacProviderException(e1.getLocalizedMessage(), e1);
- } finally {
- // destroy sms instance
- try {
- if (sc != null) {
- sc.destroy();
- }
- } catch (Exception e) {
- log.warn("Cannot destroy temporary SMS instance:" + sc.getUrl(), e);
- }
- }
- }
-
- private JobState getApplicationJobStatus(ActivityStatusType activityStatus){
- if (activityStatus == null) {
- return JobState.UNKNOWN;
- }
- Enum state = activityStatus.getState();
- String status = null;
- XmlCursor acursor = activityStatus.newCursor();
- try {
- if (acursor.toFirstChild()) {
- if (acursor.getName().getNamespaceURI().equals("http://schemas.ogf.org/hpcp/2007/01/fs")) {
- status = acursor.getName().getLocalPart();
- }
- }
- if (status != null) {
- if (status.equalsIgnoreCase("Queued") || status.equalsIgnoreCase("Starting")
- || status.equalsIgnoreCase("Ready")) {
- return JobState.QUEUED;
- } else if (status.equalsIgnoreCase("Staging-In")) {
- return JobState.SUBMITTED;
- } else if (status.equalsIgnoreCase("Staging-Out") || status.equalsIgnoreCase("FINISHED")) {
- return JobState.COMPLETE;
- } else if (status.equalsIgnoreCase("Executing")) {
- return JobState.ACTIVE;
- } else if (status.equalsIgnoreCase("FAILED")) {
- return JobState.FAILED;
- } else if (status.equalsIgnoreCase("CANCELLED")) {
- return JobState.CANCELED;
- }
- } else {
- if (ActivityStateEnumeration.CANCELLED.equals(state)) {
- return JobState.CANCELED;
- } else if (ActivityStateEnumeration.FAILED.equals(state)) {
- return JobState.FAILED;
- } else if (ActivityStateEnumeration.FINISHED.equals(state)) {
- return JobState.COMPLETE;
- } else if (ActivityStateEnumeration.RUNNING.equals(state)) {
- return JobState.ACTIVE;
- }
- }
- } finally {
- if (acursor != null)
- acursor.dispose();
- }
- return JobState.UNKNOWN;
- }
-
- private void saveApplicationJob(JobExecutionContext jobExecutionContext, JobDefinitionType jobDefinition,
- String metadata) {
- ApplicationJob appJob = GFacUtils.createApplicationJob(jobExecutionContext);
- appJob.setJobId(jobId);
- appJob.setJobData(jobDefinition.toString());
- appJob.setSubmittedTime(Calendar.getInstance().getTime());
- appJob.setStatus(ApplicationJobStatus.SUBMITTED);
- appJob.setStatusUpdateTime(appJob.getSubmittedTime());
- appJob.setMetadata(metadata);
- GFacUtils.recordApplicationJob(jobExecutionContext, appJob);
- }
-
- public void dispose(JobExecutionContext jobExecutionContext) throws GFacProviderException {
- secProperties = null;
- }
-
- /**
- * EndpointReference need to be saved to make cancel work.
- *
- * @param activityEpr
- * @param jobExecutionContext
- * @throws GFacProviderException
- */
- public void cancelJob(String activityEpr, JobExecutionContext jobExecutionContext) throws GFacProviderException {
- try {
- initSecurityProperties(jobExecutionContext);
- EndpointReferenceType eprt = EndpointReferenceType.Factory.parse(activityEpr);
- UnicoreHostType host = (UnicoreHostType) jobExecutionContext.getApplicationContext().getHostDescription()
- .getType();
-
- String factoryUrl = host.getUnicoreBESEndPointArray()[0];
- EndpointReferenceType epr = EndpointReferenceType.Factory.newInstance();
- epr.addNewAddress().setStringValue(factoryUrl);
-
- FactoryClient factory = new FactoryClient(epr, secProperties);
- factory.terminateActivity(eprt);
- } catch (Exception e) {
- throw new GFacProviderException(e.getLocalizedMessage(),e);
- }
-
- }
-
- protected void downloadOffline(String smsEpr, JobExecutionContext jobExecutionContext) throws GFacProviderException {
- try {
- initSecurityProperties(jobExecutionContext);
- EndpointReferenceType eprt = EndpointReferenceType.Factory.parse(smsEpr);
- StorageClient sms = new StorageClient(eprt, secProperties);
- DataTransferrer dt = new DataTransferrer(jobExecutionContext, sms);
- // there must be output files there
- // this is also possible if client is re-connected, the jobs are
- // still
- // running and no output is produced
- dt.downloadRemoteFiles();
-
- // may be use the below method before downloading for checking
- // the number of entries
- // sms.listDirectory(".");
-
- } catch (Exception e) {
- throw new GFacProviderException(e.getLocalizedMessage(), e);
- }
- }
-
- protected void initSecurityProperties(JobExecutionContext jobExecutionContext) throws GFacProviderException,
- GFacException {
-
- if (secProperties != null)
- return;
-
- GSISecurityContext gssContext = (GSISecurityContext) jobExecutionContext
- .getSecurityContext(GSISecurityContext.GSI_SECURITY_CONTEXT);
-
- try {
- String certLocation = gssContext.getTrustedCertificatePath();
- List<String> trustedCert = new ArrayList<String>();
- trustedCert.add(certLocation + "/*.0");
- trustedCert.add(certLocation + "/*.pem");
-
- DirectoryCertChainValidator dcValidator = new DirectoryCertChainValidator(trustedCert, Encoding.PEM, -1,
- 60000, null);
-
- String userID = getUserName(jobExecutionContext);
-
- if ( userID == null || "".equals(userID) || userID.equalsIgnoreCase("admin") ) {
- userID = "CN=zdv575, O=Ultrascan Gateway, C=DE";
- }
-
- String userDN = userID.replaceAll("^\"|\"$", "");
-
- // TODO: should be changed to default airavata server locations
- KeyAndCertCredential cred = generateShortLivedCertificate(userDN, certLocation
- + "/cacert.pem", certLocation
- + "/cakey.pem", "ultrascan3");
- secProperties = new DefaultClientConfiguration(dcValidator, cred);
-
- // secProperties.doSSLAuthn();
- secProperties.getETDSettings().setExtendTrustDelegation(true);
-
- secProperties.setDoSignMessage(true);
-
- String[] outHandlers = secProperties.getOutHandlerClassNames();
-
- Set<String> outHandlerLst = null;
-
- // timeout in milliseconds
- Properties p = secProperties.getExtraSettings();
- p.setProperty("http.connection.timeout", "300000");
- p.setProperty("http.socket.timeout", "300000");
-
- if (outHandlers == null) {
- outHandlerLst = new HashSet<String>();
- } else {
- outHandlerLst = new HashSet<String>(Arrays.asList(outHandlers));
- }
-
- outHandlerLst.add("de.fzj.unicore.uas.security.ProxyCertOutHandler");
-
- secProperties.setOutHandlerClassNames(outHandlerLst.toArray(new String[outHandlerLst.size()]));
-
- } catch (Exception e) {
- throw new GFacProviderException(e.getMessage(), e);
- }
- }
-
- //FIXME: Get user details
- private String getUserName(JobExecutionContext context) {
-// if (context.getConfigurationData()!= null) {
-// return context.getConfigurationData().getBasicMetadata().getUserName();
-// } else {
- return "";
-// }
- }
-
- protected ActivityStatusType getStatus(FactoryClient fc, EndpointReferenceType activityEpr)
- throws UnknownActivityIdentifierFault {
-
- GetActivityStatusesDocument stats = GetActivityStatusesDocument.Factory.newInstance();
-
- stats.addNewGetActivityStatuses().setActivityIdentifierArray(new EndpointReferenceType[] { activityEpr });
-
- GetActivityStatusesResponseDocument resDoc = fc.getActivityStatuses(stats);
-
- ActivityStatusType activityStatus = resDoc.getGetActivityStatusesResponse().getResponseArray()[0]
- .getActivityStatus();
- return activityStatus;
- }
-
- protected String formatStatusMessage(String activityUrl, String status) {
- return String.format("Activity %s is %s.\n", activityUrl, status);
- }
-
- protected String subStatusAsString(ActivityStatusType statusType) {
-
- StringBuffer sb = new StringBuffer();
-
- sb.append(statusType.getState().toString());
-
- XmlCursor acursor = statusType.newCursor();
- if (acursor.toFirstChild()) {
- do {
- if (acursor.getName().getNamespaceURI().equals("http://schemas.ogf.org/hpcp/2007/01/fs")) {
- sb.append(":");
- sb.append(acursor.getName().getLocalPart());
- }
- } while (acursor.toNextSibling());
- acursor.dispose();
- return sb.toString();
- } else {
- acursor.dispose();
- return sb.toString();
- }
-
- }
-
- public void initProperties(Map<String, String> properties) throws GFacProviderException, GFacException {
-
- }
-
- protected KeyAndCertCredential generateShortLivedCertificate(String userDN, String caCertPath, String caKeyPath,
- String caPwd) throws Exception {
- final long CredentialGoodFromOffset = 1000L * 60L * 15L; // 15 minutes
- // ago
-
- final long startTime = System.currentTimeMillis() - CredentialGoodFromOffset;
- final long endTime = startTime + 30 * 3600 * 1000;
-
- String keyLengthProp = "1024";
- int keyLength = Integer.parseInt(keyLengthProp);
- String signatureAlgorithm = "SHA1withRSA";
-
- KeyAndCertCredential caCred = getCACredential(caCertPath, caKeyPath, caPwd);
-
- KeyPairGenerator kpg = KeyPairGenerator.getInstance(caCred.getKey().getAlgorithm());
- kpg.initialize(keyLength);
- KeyPair pair = kpg.generateKeyPair();
-
- X500Principal subjectDN = new X500Principal(userDN);
- Random rand = new Random();
-
- SubjectPublicKeyInfo publicKeyInfo;
- try {
- publicKeyInfo = SubjectPublicKeyInfo.getInstance(new ASN1InputStream(pair.getPublic().getEncoded())
- .readObject());
- } catch (IOException e) {
- throw new InvalidKeyException("Can not parse the public key"
- + "being included in the short lived certificate", e);
- }
-
- X500Name issuerX500Name = CertificateHelpers.toX500Name(caCred.getCertificate().getSubjectX500Principal());
-
- X500Name subjectX500Name = CertificateHelpers.toX500Name(subjectDN);
-
- X509v3CertificateBuilder certBuilder = new X509v3CertificateBuilder(issuerX500Name, new BigInteger(20, rand),
- new Date(startTime), new Date(endTime), subjectX500Name, publicKeyInfo);
-
- AlgorithmIdentifier sigAlgId = X509v3CertificateBuilder.extractAlgorithmId(caCred.getCertificate());
-
- X509Certificate certificate = certBuilder.build(caCred.getKey(), sigAlgId, signatureAlgorithm, null, null);
-
- certificate.checkValidity(new Date());
- certificate.verify(caCred.getCertificate().getPublicKey());
- KeyAndCertCredential result = new KeyAndCertCredential(pair.getPrivate(), new X509Certificate[] { certificate,
- caCred.getCertificate() });
-
- return result;
- }
-
- private KeyAndCertCredential getCACredential(String caCertPath, String caKeyPath, String password) throws Exception {
- InputStream isKey = new FileInputStream(caKeyPath);
- PrivateKey pk = CertificateUtils.loadPrivateKey(isKey, Encoding.PEM, password.toCharArray());
-
- InputStream isCert = new FileInputStream(caCertPath);
- X509Certificate caCert = CertificateUtils.loadCertificate(isCert, Encoding.PEM);
-
- if (isKey != null)
- isKey.close();
- if (isCert != null)
- isCert.close();
-
- return new KeyAndCertCredential(pk, new X509Certificate[] { caCert });
- }
-
- private String getCNFromUserDN(String userDN) {
- return X500NameUtils.getAttributeValues(userDN, BCStyle.CN)[0];
-
- }
-}
http://git-wip-us.apache.org/repos/asf/airavata/blob/13b505ae/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/impl/GramProvider.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/impl/GramProvider.java b/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/impl/GramProvider.java
deleted file mode 100644
index 4066c00..0000000
--- a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/impl/GramProvider.java
+++ /dev/null
@@ -1,527 +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.gfac.provider.impl;
-
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.io.InputStream;
-import java.net.MalformedURLException;
-import java.util.Map;
-import java.util.MissingResourceException;
-import java.util.Properties;
-import java.util.concurrent.ConcurrentHashMap;
-
-import org.apache.airavata.common.exception.ApplicationSettingsException;
-import org.apache.airavata.common.utils.ServerSettings;
-import org.apache.airavata.gfac.GFacException;
-import org.apache.airavata.gfac.JobSubmissionFault;
-import org.apache.airavata.gfac.context.JobExecutionContext;
-import org.apache.airavata.gfac.context.security.GSISecurityContext;
-import org.apache.airavata.gfac.notification.events.JobIDEvent;
-import org.apache.airavata.gfac.notification.events.StartExecutionEvent;
-import org.apache.airavata.gfac.provider.GFacProviderException;
-import org.apache.airavata.gfac.utils.GFacUtils;
-import org.apache.airavata.gfac.utils.GramJobSubmissionListener;
-import org.apache.airavata.gfac.utils.GramProviderUtils;
-import org.apache.airavata.model.workspace.experiment.CorrectiveAction;
-import org.apache.airavata.model.workspace.experiment.ErrorCategory;
-import org.apache.airavata.model.workspace.experiment.JobState;
-import org.apache.airavata.schemas.gfac.ApplicationDeploymentDescriptionType;
-import org.apache.airavata.schemas.gfac.GlobusHostType;
-import org.globus.gram.GramException;
-import org.globus.gram.GramJob;
-import org.globus.gram.WaitingForCommitException;
-import org.globus.gram.internal.GRAMConstants;
-import org.globus.gram.internal.GRAMProtocolErrorConstants;
-import org.ietf.jgss.GSSCredential;
-import org.ietf.jgss.GSSException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class GramProvider extends AbstractProvider{
- private static final Logger log = LoggerFactory.getLogger(GramJobSubmissionListener.class);
-
- private GramJob job;
- private GramJobSubmissionListener listener;
- private boolean twoPhase = true;
-
- /**
- * If normal job submission fail due to an authorisation failure or script failure we
- * will re-attempt to submit the job. In-order to avoid any recursive loop during a continuous
- * failure we track whether failure paths are tried or not. Following booleans keeps track whether
- * we already tried a failure path or not.
- */
- /**
- * To track job submissions during a authorisation failure while requesting job.
- */
- private boolean renewCredentialsAttempt = false;
- /**
- * To track job submission during a script error situation.
- */
- private boolean reSubmissionInProgress = false;
- /**
- * To track authorisation failures during status monitoring.
- */
- private boolean authorisationFailedAttempt = false;
-
- private static final Map<String, GramJob> currentlyExecutingJobCache
- = new ConcurrentHashMap<String, GramJob>();
-
- private static Properties resources;
-
- static {
- try {
-
- String propFileName = "errors.properties";
- resources = new Properties();
- InputStream inputStream = GramProvider.class.getClassLoader()
- .getResourceAsStream(propFileName);
-
- if (inputStream == null) {
- throw new FileNotFoundException("property file '" + propFileName
- + "' not found in the classpath");
- }
-
- resources.load(inputStream);
-
- } catch (FileNotFoundException mre) {
- log.error("errors.properties not found", mre);
- } catch (IOException e) {
- log.error("Error reading errors.properties file", e);
- }
- }
-
-
- // This method prepare the environment before the application invocation.
- public void initialize(JobExecutionContext jobExecutionContext) throws GFacProviderException, GFacException {
-
- try {
- super.initialize(jobExecutionContext);
- String strTwoPhase = ServerSettings.getSetting("TwoPhase");
- if (strTwoPhase != null) {
- twoPhase = Boolean.parseBoolean(strTwoPhase);
- log.info("Two phase commit is set to " + twoPhase);
- }
- } catch (ApplicationSettingsException e) {
- log.warn("Error reading TwoPhase property from configurations.", e);
- }
-
- job = GramProviderUtils.setupEnvironment(jobExecutionContext, twoPhase);
- listener = new GramJobSubmissionListener(job, jobExecutionContext);
- job.addListener(listener);
- }
-
- public void execute(JobExecutionContext jobExecutionContext) throws GFacProviderException, GFacException{
- jobExecutionContext.getNotifier().publish(new StartExecutionEvent());
- GlobusHostType host = (GlobusHostType) jobExecutionContext.getApplicationContext().
- getHostDescription().getType();
- ApplicationDeploymentDescriptionType app = jobExecutionContext.getApplicationContext().
- getApplicationDeploymentDescription().getType();
-
- StringBuilder stringBuilder = new StringBuilder();
- try {
-
- GSSCredential gssCred = ((GSISecurityContext)jobExecutionContext.
- getSecurityContext(GSISecurityContext.GSI_SECURITY_CONTEXT)).getGssCredentials();
- job.setCredentials(gssCred);
- // We do not support multiple gatekeepers in XBaya GUI, so we simply pick the 0th element in the array
- String gateKeeper = host.getGlobusGateKeeperEndPointArray(0);
- log.info("Request to contact:" + gateKeeper);
-
- stringBuilder.append("Finished launching job, Host = ").append(host.getHostAddress()).append(" RSL = ")
- .append(job.getRSL()).append(" working directory = ").append(app.getStaticWorkingDirectory())
- .append(" temp directory = ").append(app.getScratchWorkingDirectory())
- .append(" Globus GateKeeper Endpoint = ").append(gateKeeper);
-
- log.info(stringBuilder.toString());
-
- submitJobs(gateKeeper, jobExecutionContext, host);
-
- } catch (ApplicationSettingsException e) {
- throw new GFacException(e.getMessage(), e);
- } finally {
- if (job != null) {
- try {
- /*
- * Remove listener
- */
- job.removeListener(listener);
- } catch (Exception e) {
- log.error(e.getMessage());
- }
- }
- }
- }
-
- private void submitJobs(String gateKeeper,
- JobExecutionContext jobExecutionContext,
- GlobusHostType globusHostType) throws GFacException, GFacProviderException {
- boolean applicationSaved=false;
- String taskID = jobExecutionContext.getTaskData().getTaskID();
-
- if (twoPhase) {
- try {
- /*
- * The first boolean is to force communication through SSLv3
- * The second boolean is to specify the job is a batch job - use true for interactive and false for
- * batch.
- * The third boolean is to specify to use the full proxy and not delegate a limited proxy.
- */
- job.request(true, gateKeeper, false, false);
-
- // Single boolean to track all authentication failures, therefore we need to re-initialize
- // this here
- renewCredentialsAttempt = false;
-
- } catch (WaitingForCommitException e) {
- String jobID = job.getIDAsString();
-
- details.setJobID(jobID);
- details.setJobDescription(job.getRSL());
- jobExecutionContext.setJobDetails(details);
- GFacUtils.saveJobStatus(details, JobState.UN_SUBMITTED, taskID);
-
- applicationSaved=true;
- String jobStatusMessage = "Un-submitted JobID= " + jobID;
- log.info(jobStatusMessage);
- jobExecutionContext.getNotifier().publish(new JobIDEvent(jobStatusMessage));
-
- log.info("Two phase commit: sending COMMIT_REQUEST signal; Job id - " + jobID);
-
- try {
- job.signal(GramJob.SIGNAL_COMMIT_REQUEST);
-
- } catch (GramException gramException) {
- throw new GFacException("Error while sending commit request. Job Id - "
- + job.getIDAsString(), gramException);
- } catch (GSSException gssException) {
-
- // User credentials are invalid
- log.error("Error while submitting commit request - Credentials provided are invalid. Job Id - "
- + job.getIDAsString(), e);
- log.info("Attempting to renew credentials and re-submit commit signal...");
- GFacUtils.saveErrorDetails(gssException.getLocalizedMessage(), CorrectiveAction.RETRY_SUBMISSION, ErrorCategory.AIRAVATA_INTERNAL_ERROR, taskID);
- renewCredentials(jobExecutionContext);
-
- try {
- job.signal(GramJob.SIGNAL_COMMIT_REQUEST);
- } catch (GramException e1) {
- GFacUtils.saveErrorDetails(gssException.getLocalizedMessage(), CorrectiveAction.CONTACT_SUPPORT, ErrorCategory.AIRAVATA_INTERNAL_ERROR, taskID);
- throw new GFacException("Error while sending commit request. Job Id - "
- + job.getIDAsString(), e1);
- } catch (GSSException e1) {
- GFacUtils.saveErrorDetails(gssException.getLocalizedMessage(), CorrectiveAction.CONTACT_SUPPORT, ErrorCategory.AIRAVATA_INTERNAL_ERROR, taskID);
- throw new GFacException("Error while sending commit request. Job Id - "
- + job.getIDAsString() + ". Credentials provided invalid", e1);
- }
- }
- GFacUtils.updateJobStatus(details, JobState.SUBMITTED);
- jobStatusMessage = "Submitted JobID= " + job.getIDAsString();
- log.info(jobStatusMessage);
- jobExecutionContext.getNotifier().publish(new JobIDEvent(jobStatusMessage));
-
- } catch (GSSException e) {
- // Renew credentials and re-submit
- GFacUtils.saveErrorDetails(e.getLocalizedMessage(), CorrectiveAction.RETRY_SUBMISSION, ErrorCategory.AIRAVATA_INTERNAL_ERROR, taskID);
-
- reSubmitJob(gateKeeper, jobExecutionContext, globusHostType, e);
-
- } catch (GramException e) {
- GFacUtils.saveErrorDetails(e.getLocalizedMessage(), CorrectiveAction.CONTACT_SUPPORT, ErrorCategory.AIRAVATA_INTERNAL_ERROR, taskID);
-
- throw new GFacException("An error occurred while submitting a job, job id = " + job.getIDAsString(), e);
- }
- } else {
-
- /*
- * The first boolean is to force communication through SSLv3
- * The second boolean is to specify the job is a batch job - use true for interactive and false for
- * batch.
- * The third boolean is to specify to use the full proxy and not delegate a limited proxy.
- */
- try {
-
- job.request(true, gateKeeper, false, false);
- renewCredentialsAttempt = false;
-
- } catch (GramException e) {
- GFacUtils.saveErrorDetails(e.getLocalizedMessage(), CorrectiveAction.CONTACT_SUPPORT, ErrorCategory.AIRAVATA_INTERNAL_ERROR, taskID);
- throw new GFacException("An error occurred while submitting a job, job id = " + job.getIDAsString(), e);
- } catch (GSSException e) {
- GFacUtils.saveErrorDetails(e.getLocalizedMessage(), CorrectiveAction.RETRY_SUBMISSION, ErrorCategory.AIRAVATA_INTERNAL_ERROR, taskID);
- // Renew credentials and re-submit
- reSubmitJob(gateKeeper, jobExecutionContext, globusHostType, e);
- }
-
- String jobStatusMessage = "Un-submitted JobID= " + job.getIDAsString();
- log.info(jobStatusMessage);
- jobExecutionContext.getNotifier().publish(new JobIDEvent(jobStatusMessage));
-
- }
-
- currentlyExecutingJobCache.put(job.getIDAsString(), job);
- /*
- * Wait until job is done
- */
- listener.waitFor();
-
- checkJobStatus(jobExecutionContext, globusHostType, gateKeeper);
-
- }
-
- private void renewCredentials(JobExecutionContext jobExecutionContext) throws GFacException {
-
- renewCredentials(this.job, jobExecutionContext);
- }
-
- private void renewCredentials(GramJob gramJob, JobExecutionContext jobExecutionContext) throws GFacException {
-
- try {
- GSSCredential gssCred = ((GSISecurityContext)jobExecutionContext.
- getSecurityContext(GSISecurityContext.GSI_SECURITY_CONTEXT)).renewCredentials();
- gramJob.renew(gssCred);
- } catch (GramException e1) {
- throw new GFacException("Unable to renew credentials. Job Id - "
- + gramJob.getIDAsString(), e1);
- } catch (GSSException e1) {
- throw new GFacException("Unable to renew credentials. Job Id - "
- + gramJob.getIDAsString(), e1);
- } catch (ApplicationSettingsException e) {
- throw new GFacException(e.getLocalizedMessage(), e);
- }
- }
-
- private void reSubmitJob(String gateKeeper,
- JobExecutionContext jobExecutionContext,
- GlobusHostType globusHostType, Exception e) throws GFacException, GFacProviderException {
-
- if (!renewCredentialsAttempt) {
-
- renewCredentialsAttempt = true;
-
- // User credentials are invalid
- log.error("Error while submitting job - Credentials provided are invalid. Job Id - "
- + job.getIDAsString(), e);
- log.info("Attempting to renew credentials and re-submit jobs...");
-
- // Remove existing listener and register a new listener
- job.removeListener(listener);
- listener = new GramJobSubmissionListener(job, jobExecutionContext);
-
- job.addListener(listener);
-
- renewCredentials(jobExecutionContext);
-
- submitJobs(gateKeeper, jobExecutionContext, globusHostType);
-
- } else {
- throw new GFacException("Error while submitting job - Credentials provided are invalid. Job Id - "
- + job.getIDAsString(), e);
- }
-
- }
-
- private void reSubmitJob(String gateKeeper,
- JobExecutionContext jobExecutionContext,
- GlobusHostType globusHostType) throws GFacException, GFacProviderException {
-
- // User credentials are invalid
- log.info("Attempting to renew credentials and re-submit jobs...");
-
- // Remove existing listener and register a new listener
- job.removeListener(listener);
- listener = new GramJobSubmissionListener(job, jobExecutionContext);
-
- job.addListener(listener);
-
- renewCredentials(jobExecutionContext);
-
- submitJobs(gateKeeper, jobExecutionContext, globusHostType);
-
- }
-
-
-
- public void dispose(JobExecutionContext jobExecutionContext) throws GFacProviderException {
- }
-
- @Override
- public void cancelJob(String jobId, JobExecutionContext jobExecutionContext) throws GFacException {
- cancelSingleJob(jobId, jobExecutionContext);
- }
-
-
- private void cancelSingleJob(String jobId, JobExecutionContext context) throws GFacException {
- // First check whether job id is in the cache
- if (currentlyExecutingJobCache.containsKey(jobId)) {
-
- synchronized (this) {
- GramJob gramJob = currentlyExecutingJobCache.get(jobId);
-
- // Even though we check using containsKey, at this point job could be null
- if (gramJob != null && (gramJob.getStatus() != GRAMConstants.STATUS_DONE ||
- gramJob.getStatus() != GRAMConstants.STATUS_FAILED)) {
- cancelJob(gramJob, context);
- }
- }
-
- } else {
-
- try {
- GSSCredential gssCred = ((GSISecurityContext)context.
- getSecurityContext(GSISecurityContext.GSI_SECURITY_CONTEXT)).getGssCredentials();
-
- GramJob gramJob = new GramJob(null);
- try {
- gramJob.setID(jobId);
- } catch (MalformedURLException e) {
- throw new GFacException("Invalid job id - " + jobId, e);
- }
- gramJob.setCredentials(gssCred);
-
- synchronized (this) {
- if (gramJob.getStatus() != GRAMConstants.STATUS_DONE ||
- gramJob.getStatus() != GRAMConstants.STATUS_FAILED) {
- cancelJob(gramJob, context);
- }
- }
- } catch (ApplicationSettingsException e) {
- throw new GFacException(e);
- }
- }
- }
-
- private void cancelJob(GramJob gramJob, JobExecutionContext context) throws GFacException{
-
- try {
- gramJob.cancel();
- } catch (GramException e) {
- throw new GFacException("Error cancelling job, id - " + gramJob.getIDAsString(), e);
- } catch (GSSException e) {
-
- log.warn("Credentials invalid to cancel job. Attempting to renew credentials and re-try. " +
- "Job id - " + gramJob.getIDAsString());
- renewCredentials(gramJob, context);
-
- try {
- gramJob.cancel();
- gramJob.signal(GramJob.SIGNAL_COMMIT_END);
- } catch (GramException e1) {
- throw new GFacException("Error cancelling job, id - " + gramJob.getIDAsString(), e1);
- } catch (GSSException e1) {
- throw new GFacException("Error cancelling job, invalid credentials. Job id - "
- + gramJob.getIDAsString(), e);
- }
- }
-
- }
-
- public void initProperties(Map<String, String> properties) throws GFacException {
-
- }
-
- private void checkJobStatus(JobExecutionContext jobExecutionContext, GlobusHostType host, String gateKeeper)
- throws GFacProviderException {
- int jobStatus = listener.getCurrentStatus();
-
- if (jobStatus == GramJob.STATUS_FAILED) {
-
- String errorMsg = "Job " + job.getIDAsString() + " on host " + host.getHostAddress() + " Job Exit Code = "
- + listener.getError() + " Error Description = " + getGramErrorString(listener.getError());
-
- if (listener.getError() == GRAMProtocolErrorConstants.INVALID_SCRIPT_REPLY) {
-
- // re-submitting without renewing
- // TODO verify why we re-submit jobs when we get a invalid script reply
- if (!reSubmissionInProgress) {
- reSubmissionInProgress = true;
-
- log.info("Invalid script reply received. Re-submitting job, id - " + job.getIDAsString());
- try {
- reSubmitJob(gateKeeper, jobExecutionContext, host);
- } catch (GFacException e) {
- throw new GFacProviderException
- ("Error during re-submission. Original job submission data - " + errorMsg, e);
- }
- return;
- }
-
- } else if (listener.getError() == GRAMProtocolErrorConstants.ERROR_AUTHORIZATION) {
-
- // re-submit with renewed credentials
- if (!authorisationFailedAttempt) {
- authorisationFailedAttempt = true;
- log.info("Authorisation error contacting provider. Re-submitting job with renewed credentials.");
-
- try {
- renewCredentials(jobExecutionContext);
- reSubmitJob(gateKeeper, jobExecutionContext, host);
- } catch (GFacException e) {
- throw new GFacProviderException
- ("Error during re-submission. Original job submission data - " + errorMsg, e);
- }
-
- return;
- }
-
- } else if (listener.getError() == GRAMProtocolErrorConstants.USER_CANCELLED) {
-
- log.info("User successfully cancelled job id " + job.getIDAsString());
- return;
- }
-
-
-
- log.error(errorMsg);
-
- synchronized (this) {
- currentlyExecutingJobCache.remove(job.getIDAsString());
- }
-
- throw new JobSubmissionFault(new Exception(errorMsg), host.getHostAddress(), gateKeeper,
- job.getRSL(), jobExecutionContext, getGramErrorString(listener.getError()),
- listener.getError());
-
- } else if (jobStatus == GramJob.STATUS_DONE) {
- log.info("Job " + job.getIDAsString() + " on host " + host.getHostAddress() + " is successfully executed.");
-
- synchronized (this) {
- currentlyExecutingJobCache.remove(job.getIDAsString());
- }
- }
- }
-
- public String getGramErrorString(int errorCode) {
-
- if (resources != null) {
- try {
- return resources.getProperty(String.valueOf(errorCode));
- } catch (MissingResourceException mre) {
- log.warn("Error reading globus error descriptions.", mre);
- return "Error code: " + errorCode;
- }
- } else {
- return "Error code: " + errorCode;
- }
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/airavata/blob/13b505ae/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/utils/ApplicationProcessor.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/utils/ApplicationProcessor.java b/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/utils/ApplicationProcessor.java
deleted file mode 100644
index d88cddf..0000000
--- a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/utils/ApplicationProcessor.java
+++ /dev/null
@@ -1,252 +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.gfac.provider.utils;
-
-import org.apache.airavata.gfac.context.JobExecutionContext;
-import org.apache.airavata.schemas.gfac.ExtendedKeyValueType;
-import org.apache.airavata.schemas.gfac.HpcApplicationDeploymentType;
-import org.apache.airavata.schemas.gfac.JobTypeType;
-import org.apache.airavata.schemas.gfac.NameValuePairType;
-import org.ggf.schemas.jsdl.x2005.x11.jsdl.ApplicationType;
-import org.ggf.schemas.jsdl.x2005.x11.jsdl.JobDefinitionType;
-import org.ggf.schemas.jsdl.x2005.x11.jsdlPosix.EnvironmentType;
-import org.ggf.schemas.jsdl.x2005.x11.jsdlPosix.FileNameType;
-import org.ggf.schemas.jsdl.x2005.x11.jsdlPosix.UserNameType;
-import org.ogf.schemas.jsdl.x2007.x02.jsdlSpmd.NumberOfProcessesType;
-import org.ogf.schemas.jsdl.x2007.x02.jsdlSpmd.ProcessesPerHostType;
-import org.ogf.schemas.jsdl.x2007.x02.jsdlSpmd.ThreadsPerProcessType;
-
-import java.io.File;
-
-
-public class ApplicationProcessor {
-
- public static void generateJobSpecificAppElements(JobDefinitionType value, JobExecutionContext context){
-
- String userName = getUserNameFromContext(context);
- if (userName.equalsIgnoreCase("admin")){
- userName = "CN=zdv575, O=Ultrascan Gateway, C=DE";
- }
-
- HpcApplicationDeploymentType appDepType = (HpcApplicationDeploymentType) context
- .getApplicationContext().getApplicationDeploymentDescription()
- .getType();
-
- createGenericApplication(value, appDepType);
-
- if (appDepType.getApplicationEnvironmentArray().length > 0) {
- createApplicationEnvironment(value,
- appDepType.getApplicationEnvironmentArray(), appDepType);
- }
-
-
- if (appDepType.getExecutableLocation() != null) {
- FileNameType fNameType = FileNameType.Factory.newInstance();
- fNameType.setStringValue(appDepType.getExecutableLocation());
- if(isParallelJob(appDepType)) {
- JSDLUtils.getOrCreateSPMDApplication(value).setExecutable(fNameType);
- JSDLUtils.getSPMDApplication(value).setSPMDVariation(getSPMDVariation(appDepType));
-
- if(getValueFromMap(appDepType, JSDLUtils.NUMBEROFPROCESSES)!=null){
- NumberOfProcessesType num = NumberOfProcessesType.Factory.newInstance();
- num.setStringValue(getValueFromMap(appDepType, JSDLUtils.NUMBEROFPROCESSES));
- JSDLUtils.getSPMDApplication(value).setNumberOfProcesses(num);
- }
-
- if(getValueFromMap(appDepType, JSDLUtils.PROCESSESPERHOST)!=null){
- ProcessesPerHostType pph = ProcessesPerHostType.Factory.newInstance();
- pph.setStringValue(getValueFromMap(appDepType, JSDLUtils.PROCESSESPERHOST));
- JSDLUtils.getSPMDApplication(value).setProcessesPerHost(pph);
- }
-
- if(getValueFromMap(appDepType, JSDLUtils.THREADSPERHOST)!=null){
- ThreadsPerProcessType tpp = ThreadsPerProcessType.Factory.newInstance();
- tpp.setStringValue(getValueFromMap(appDepType, JSDLUtils.THREADSPERHOST));
- JSDLUtils.getSPMDApplication(value).setThreadsPerProcess(tpp);
-
- }
-
- if(userName != null) {
- UserNameType userNameType = UserNameType.Factory.newInstance();
- userNameType.setStringValue(userName);
- JSDLUtils.getSPMDApplication(value).setUserName(userNameType);
- }
- }
- else {
- JSDLUtils.getOrCreatePOSIXApplication(value).setExecutable(fNameType);
- if(userName != null) {
- UserNameType userNameType = UserNameType.Factory.newInstance();
- userNameType.setStringValue(userName);
- JSDLUtils.getOrCreatePOSIXApplication(value).setUserName(userNameType);
- }
- }
- }
-
-
- String stdout = (appDepType.getStandardOutput() != null) ? new File(appDepType.getStandardOutput()).getName(): "stdout";
- ApplicationProcessor.setApplicationStdOut(value, appDepType, stdout);
-
-
- String stderr = (appDepType.getStandardError() != null) ? new File(appDepType.getStandardError()).getName() : "stderr";
- ApplicationProcessor.setApplicationStdErr(value, appDepType, stderr);
-
- }
-
- public static String getUserNameFromContext(JobExecutionContext jobContext) {
- if(jobContext.getTaskData() == null)
- return null;
- //FIXME: Discuss to get user and change this
- return "admin";
- }
- public static boolean isParallelJob(HpcApplicationDeploymentType appDepType) {
-
- boolean isParallel = false;
-
- if (appDepType.getJobType() != null) {
- // TODO set data output directory
- int status = appDepType.getJobType().intValue();
-
- switch (status) {
- // TODO: this check should be done outside this class
- case JobTypeType.INT_MPI:
- case JobTypeType.INT_OPEN_MP:
- isParallel = true;
- break;
-
- case JobTypeType.INT_SERIAL:
- case JobTypeType.INT_SINGLE:
- isParallel = false;
- break;
-
- default:
- isParallel = false;
- break;
- }
- }
- return isParallel;
- }
-
-
- public static void createApplicationEnvironment(JobDefinitionType value, NameValuePairType[] nameValuePairs, HpcApplicationDeploymentType appDepType) {
-
- if(isParallelJob(appDepType)) {
- for (NameValuePairType nv : nameValuePairs) {
- EnvironmentType envType = JSDLUtils.getOrCreateSPMDApplication(value).addNewEnvironment();
- envType.setName(nv.getName());
- envType.setStringValue(nv.getValue());
- }
- }
- else {
- for (NameValuePairType nv : nameValuePairs) {
- EnvironmentType envType = JSDLUtils.getOrCreatePOSIXApplication(value).addNewEnvironment();
- envType.setName(nv.getName());
- envType.setStringValue(nv.getValue());
- }
- }
-
- }
-
-
- public static String getSPMDVariation (HpcApplicationDeploymentType appDepType) {
-
- String variation = null;
-
- if (appDepType.getJobType() != null) {
- // TODO set data output directory
- int status = appDepType.getJobType().intValue();
-
- switch (status) {
- // TODO: this check should be done outside this class
- case JobTypeType.INT_MPI:
- variation = SPMDVariations.MPI.value();
- break;
-
- case JobTypeType.INT_OPEN_MP:
- variation = SPMDVariations.OpenMPI.value();
- break;
-
- }
- }
- return variation;
- }
-
-
- public static void addApplicationArgument(JobDefinitionType value, HpcApplicationDeploymentType appDepType, String stringPrm) {
- if(isParallelJob(appDepType))
- JSDLUtils.getOrCreateSPMDApplication(value)
- .addNewArgument().setStringValue(stringPrm);
- else
- JSDLUtils.getOrCreatePOSIXApplication(value)
- .addNewArgument().setStringValue(stringPrm);
-
- }
-
- public static void setApplicationStdErr(JobDefinitionType value, HpcApplicationDeploymentType appDepType, String stderr) {
- FileNameType fName = FileNameType.Factory.newInstance();
- fName.setStringValue(stderr);
- if (isParallelJob(appDepType))
- JSDLUtils.getOrCreateSPMDApplication(value).setError(fName);
- else
- JSDLUtils.getOrCreatePOSIXApplication(value).setError(fName);
- }
-
- public static void setApplicationStdOut(JobDefinitionType value, HpcApplicationDeploymentType appDepType, String stderr) {
- FileNameType fName = FileNameType.Factory.newInstance();
- fName.setStringValue(stderr);
- if (isParallelJob(appDepType))
- JSDLUtils.getOrCreateSPMDApplication(value).setOutput(fName);
- else
- JSDLUtils.getOrCreatePOSIXApplication(value).setOutput(fName);
- }
-
- public static String getApplicationStdOut(JobDefinitionType value, HpcApplicationDeploymentType appDepType) throws RuntimeException {
- if (isParallelJob(appDepType)) return JSDLUtils.getOrCreateSPMDApplication(value).getOutput().getStringValue();
- else return JSDLUtils.getOrCreatePOSIXApplication(value).getOutput().getStringValue();
- }
-
- public static String getApplicationStdErr(JobDefinitionType value, HpcApplicationDeploymentType appDepType) throws RuntimeException {
- if (isParallelJob(appDepType)) return JSDLUtils.getOrCreateSPMDApplication(value).getError().getStringValue();
- else return JSDLUtils.getOrCreatePOSIXApplication(value).getError().getStringValue();
- }
-
- public static void createGenericApplication(JobDefinitionType value, HpcApplicationDeploymentType appDepType) {
- if (appDepType.getApplicationName() != null) {
- ApplicationType appType = JSDLUtils.getOrCreateApplication(value);
- String appName = appDepType.getApplicationName()
- .getStringValue();
- appType.setApplicationName(appName);
- JSDLUtils.getOrCreateJobIdentification(value).setJobName(appName);
- }
- }
-
-
- public static String getValueFromMap(HpcApplicationDeploymentType appDepType, String name) {
- ExtendedKeyValueType[] extended = appDepType.getKeyValuePairsArray();
- for(ExtendedKeyValueType e: extended) {
- if(e.getName().equalsIgnoreCase(name)) {
- return e.getStringValue();
- }
- }
- return null;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/airavata/blob/13b505ae/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/utils/DataStagingProcessor.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/utils/DataStagingProcessor.java b/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/utils/DataStagingProcessor.java
deleted file mode 100644
index 80112df..0000000
--- a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/utils/DataStagingProcessor.java
+++ /dev/null
@@ -1,236 +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.gfac.provider.utils;
-
-import java.io.File;
-import java.net.URI;
-import java.util.Map;
-
-import org.apache.airavata.commons.gfac.type.ActualParameter;
-import org.apache.airavata.gfac.context.JobExecutionContext;
-import org.apache.airavata.gfac.utils.GFacUtils;
-import org.apache.airavata.schemas.gfac.HpcApplicationDeploymentType;
-import org.apache.airavata.schemas.gfac.StringArrayType;
-import org.apache.airavata.schemas.gfac.StringParameterType;
-import org.apache.airavata.schemas.gfac.URIArrayType;
-import org.apache.airavata.schemas.gfac.URIParameterType;
-import org.apache.airavata.schemas.gfac.UnicoreHostType;
-import org.ggf.schemas.jsdl.x2005.x11.jsdl.JobDefinitionType;
-
-public class DataStagingProcessor {
-
- public static void generateDataStagingElements(JobDefinitionType value, JobExecutionContext context) throws Exception{
-
- HpcApplicationDeploymentType appDepType = (HpcApplicationDeploymentType) context
- .getApplicationContext().getApplicationDeploymentDescription()
- .getType();
-
-
- String gridftpEndpoint = ((UnicoreHostType) context.getApplicationContext().getHostDescription().getType())
- .getGridFTPEndPointArray()[0];
-
-
- if (context.getInMessageContext().getParameters().size() > 0) {
- buildDataStagingFromInputContext(context, value, gridftpEndpoint, appDepType);
- }
-
- if (context.getOutMessageContext().getParameters().size() > 0) {
- buildFromOutputContext(context, value, gridftpEndpoint, appDepType);
- }
-
- createStdOutURIs(value, appDepType, gridftpEndpoint, isUnicoreEndpoint(context));
-
- }
-
- private static void createInURIElement(JobDefinitionType value,
- String endpoint, String inputDir, ActualParameter inParam)
- throws Exception {
-
- String uri = ((URIParameterType) inParam.getType()).getValue();
- String fileName = new File(uri).getName();
- if (uri.startsWith("file")) {
- URI gridFTPInputDirURI = GFacUtils.createGsiftpURI(endpoint,
- inputDir);
- String filePath = gridFTPInputDirURI.toString() + File.separator
- + fileName;
- JSDLUtils
- .addDataStagingSourceElement(value, filePath, null, fileName);
- } else if (uri.startsWith("gsiftp") || uri.startsWith("http")
- || uri.startsWith("rns")) {
- // no need to stage-in those files to the input
- // directory
- JSDLUtils.addDataStagingSourceElement(value, uri, null, fileName);
- }
-
- }
-
- private static void createStdOutURIs(JobDefinitionType value,
- HpcApplicationDeploymentType appDepType, String endpoint,
- boolean isUnicore) throws Exception {
-
- URI remoteOutputDir = GFacUtils.createGsiftpURI(endpoint,
- appDepType.getOutputDataDirectory());
-
- String stdout = ApplicationProcessor.getApplicationStdOut(value, appDepType);
-
- String stderr = ApplicationProcessor.getApplicationStdErr(value, appDepType);
-
- String stdoutFileName = (stdout == null || stdout.equals("")) ? "stdout"
- : stdout;
- String stdoutURI = GFacUtils.createGsiftpURIAsString(
- remoteOutputDir.toString(), stdoutFileName);
- JSDLUtils.addDataStagingTargetElement(value, null, stdoutFileName,
- stdoutURI);
-
- String stderrFileName = (stdout == null || stderr.equals("")) ? "stderr"
- : stderr;
- String stderrURI = GFacUtils.createGsiftpURIAsString(
- remoteOutputDir.toString(), stderrFileName);
- JSDLUtils.addDataStagingTargetElement(value, null, stderrFileName,
- stderrURI);
-
- if(isUnicore) {
- String scriptExitCodeFName = "UNICORE_SCRIPT_EXIT_CODE";
- String scriptExitCode = GFacUtils.createGsiftpURIAsString(
- remoteOutputDir.toString(), scriptExitCodeFName);
- JSDLUtils.addDataStagingTargetElement(value, null,
- scriptExitCodeFName, scriptExitCode.toString());
- }
-
- }
-
-
- private static void createOutStringElements(JobDefinitionType value,
- HpcApplicationDeploymentType appDeptype, String endpoint, String prmValue) throws Exception {
-
- if(prmValue == null || "".equals(prmValue)) return;
-
-
- String outputUri = GFacUtils.createGsiftpURIAsString(endpoint, appDeptype.getOutputDataDirectory());
-
- URI finalOutputUri = GFacUtils.createGsiftpURI(outputUri, prmValue);
- JSDLUtils.addDataStagingTargetElement(value, null, prmValue, finalOutputUri.toString());
- }
-
-
- private static void createOutURIElement(JobDefinitionType value,
- String prmValue) throws Exception {
- String fileName = new File(prmValue.toString()).getName();
- JSDLUtils.addDataStagingTargetElement(value, null, fileName, prmValue);
- }
-
-
- private static JobDefinitionType buildFromOutputContext(JobExecutionContext context,
- JobDefinitionType value, String gridftpEndpoint,
- HpcApplicationDeploymentType appDepType) throws Exception {
-
- Map<String, Object> outputParams = context.getOutMessageContext()
- .getParameters();
-
- for (String paramKey : outputParams.keySet()) {
-
- ActualParameter outParam = (ActualParameter) outputParams
- .get(paramKey);
-
- // if single urls then convert each url into jsdl source
- // elements,
- // that are formed by concat of gridftpurl+inputdir+filename
-
- String paramDataType = outParam.getType().getType().toString();
-
- if ("URI".equals(paramDataType)) {
- String uriPrm = ((URIParameterType) outParam.getType())
- .getValue();
- createOutURIElement(value, uriPrm);
- }
-
- // string params are converted into the job arguments
-
- else if (("URIArray").equals(paramDataType)) {
- String[] uriArray = ((URIArrayType) outParam.getType())
- .getValueArray();
- for (String u : uriArray) {
-
- createOutURIElement(value, u);
- }
-
- }
- else if ("String".equals(paramDataType)) {
- String stringPrm = ((StringParameterType) outParam
- .getType()).getValue();
- createOutStringElements(value, appDepType, gridftpEndpoint, stringPrm);
- }
-
- else if ("StringArray".equals(paramDataType)) {
- String[] valueArray = ((StringArrayType) outParam.getType())
- .getValueArray();
- for (String v : valueArray) {
- createOutStringElements(value, appDepType, gridftpEndpoint, v);
- }
- }
- }
-
- return value;
- }
-
-
- private static void buildDataStagingFromInputContext(JobExecutionContext context, JobDefinitionType value, String gridftpEndpoint, HpcApplicationDeploymentType appDepType)
- throws Exception {
-
- // TODO set data directory
- Map<String, Object> inputParams = context.getInMessageContext()
- .getParameters();
-
- for (String paramKey : inputParams.keySet()) {
-
- ActualParameter inParam = (ActualParameter) inputParams
- .get(paramKey);
-
- // if single urls then convert each url into jsdl source
- // elements,
- // that are formed by concat of gridftpurl+inputdir+filename
-
- String paramDataType = inParam.getType().getType().toString();
-
- if ("URI".equals(paramDataType)) {
- createInURIElement(value, gridftpEndpoint,
- appDepType.getInputDataDirectory(), inParam);
- }
-
- // string params are converted into the job arguments
-
- else if ("String".equals(paramDataType)) {
- String stringPrm = ((StringParameterType) inParam.getType())
- .getValue();
- ApplicationProcessor.addApplicationArgument(value, appDepType, stringPrm);
- }
- }
-
- }
-
-
- public static boolean isUnicoreEndpoint(JobExecutionContext context) {
- return ( (context.getApplicationContext().getHostDescription().getType() instanceof UnicoreHostType)?true:false );
- }
-
-
-}
http://git-wip-us.apache.org/repos/asf/airavata/blob/13b505ae/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/utils/DataTransferrer.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/utils/DataTransferrer.java b/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/utils/DataTransferrer.java
deleted file mode 100644
index ff3f9e2..0000000
--- a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/utils/DataTransferrer.java
+++ /dev/null
@@ -1,241 +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.gfac.provider.utils;
-
-import java.io.BufferedReader;
-import java.io.File;
-import java.io.FileNotFoundException;
-import java.io.FileReader;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.airavata.commons.gfac.type.ActualParameter;
-import org.apache.airavata.commons.gfac.type.ApplicationDescription;
-import org.apache.airavata.gfac.Constants;
-import org.apache.airavata.gfac.context.JobExecutionContext;
-import org.apache.airavata.gfac.provider.GFacProviderException;
-import org.apache.airavata.model.workspace.experiment.TaskDetails;
-import org.apache.airavata.schemas.gfac.ApplicationDeploymentDescriptionType;
-import org.apache.airavata.schemas.gfac.HpcApplicationDeploymentType;
-import org.apache.airavata.schemas.gfac.StringArrayType;
-import org.apache.airavata.schemas.gfac.StringParameterType;
-import org.apache.airavata.schemas.gfac.URIParameterType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import de.fzj.unicore.uas.client.StorageClient;
-
-
-public class DataTransferrer {
- protected final Logger log = LoggerFactory.getLogger(this.getClass());
-
- private JobExecutionContext jobContext;
-
- private StorageClient storageClient;
-
- public DataTransferrer(JobExecutionContext jobContext, StorageClient storageClient) {
- this.jobContext = jobContext;
- this.storageClient = storageClient;
- }
-
-
- public void uploadLocalFiles() throws GFacProviderException {
- Map<String, Object> inputParams = jobContext.getInMessageContext()
- .getParameters();
- for (String paramKey : inputParams.keySet()) {
- ActualParameter inParam = (ActualParameter) inputParams
- .get(paramKey);
- String paramDataType = inParam.getType().getType().toString();
- if("URI".equals(paramDataType)) {
- String uri = ((URIParameterType) inParam.getType()).getValue();
- String fileName = new File(uri).getName();
- if (uri.startsWith("file")) {
- try {
- String uriWithoutProtocol = uri.substring(
- uri.lastIndexOf("://") + 1, uri.length());
- FileUploader fileUploader = new FileUploader(
- uriWithoutProtocol, "input/" + fileName,
- Mode.overwrite);
- fileUploader.perform(storageClient);
- } catch (FileNotFoundException e3) {
- throw new GFacProviderException(
- "Error while staging-in, local file "+fileName+" not found", e3);
- } catch (Exception e) {
- throw new GFacProviderException("Cannot upload files", e);
-
- }
-
- }
- }
- }
-
- }
-
- /**
- * This method will download all the remote files specified according to the output
- * context of a job.
- * */
- public void downloadRemoteFiles() throws GFacProviderException {
-
- String downloadLocation = getDownloadLocation();
-
- File file = new File(downloadLocation);
- if(!file.exists()){
- file.mkdirs();
- }
-
- Map<String, ActualParameter> stringMap = new HashMap<String, ActualParameter>();
-
- Map<String, Object> outputParams = jobContext.getOutMessageContext()
- .getParameters();
-
- for (String paramKey : outputParams.keySet()) {
-
- ActualParameter outParam = (ActualParameter) outputParams
- .get(paramKey);
-
- // if single urls then convert each url into jsdl source
- // elements,
- // that are formed by concat of gridftpurl+inputdir+filename
-
- String paramDataType = outParam.getType().getType().toString();
-
- if ("String".equals(paramDataType)) {
- String stringPrm = ((StringParameterType) outParam
- .getType()).getValue();
- String localFileName = null;
- //TODO: why analysis.tar? it wont scale to gateways..
- if(stringPrm == null || stringPrm.isEmpty()){
- localFileName = "analysis-results.tar";
- }else{
- localFileName = stringPrm.substring(stringPrm.lastIndexOf("/")+1);
- }
- String outputLocation = downloadLocation+File.separator+localFileName;
- FileDownloader fileDownloader = new FileDownloader("output/"+stringPrm,outputLocation, Mode.overwrite);
- try {
- fileDownloader.perform(storageClient);
- ((StringParameterType) outParam.getType()).setValue(outputLocation);
- stringMap.put(paramKey, outParam);
- } catch (Exception e) {
- throw new GFacProviderException(e.getLocalizedMessage(),e);
- }
- }
-
- else if ("StringArray".equals(paramDataType)) {
- String[] valueArray = ((StringArrayType) outParam.getType())
- .getValueArray();
- for (String v : valueArray) {
- String localFileName = v.substring(v.lastIndexOf("/")+1);;
- String outputLocation = downloadLocation+File.separator+localFileName;
- FileDownloader fileDownloader = new FileDownloader("output/"+v,outputLocation, Mode.overwrite);
- try {
- fileDownloader.perform(storageClient);
- ((StringParameterType) outParam.getType()).setValue(outputLocation);
- stringMap.put(paramKey, outParam);
- } catch (Exception e) {
- throw new GFacProviderException(e.getLocalizedMessage(),e);
- }
- }
- }
- }
- if (stringMap == null || stringMap.isEmpty()) {
- throw new GFacProviderException("Empty Output returned from the Application, Double check the application" +
- "and ApplicationDescriptor output Parameter Names");
- }
-
- downloadStdOuts();
- }
-
-
- public void downloadStdOuts() throws GFacProviderException{
- String downloadLocation = getDownloadLocation();
- File file = new File(downloadLocation);
- if(!file.exists()){
- file.mkdirs();
- }
-
- HpcApplicationDeploymentType appDepType = (HpcApplicationDeploymentType) jobContext
- .getApplicationContext().getApplicationDeploymentDescription()
- .getType();
-
- String stdout = appDepType.getStandardOutput();
- String stderr = appDepType.getStandardError();
- if(stdout != null) {
- stdout = stdout.substring(stdout.lastIndexOf('/')+1);
- }
-
- if(stderr != null) {
- stderr = stderr.substring(stderr.lastIndexOf('/')+1);
- }
-
- String stdoutFileName = (stdout == null || stdout.equals("")) ? "stdout"
- : stdout;
- String stderrFileName = (stdout == null || stderr.equals("")) ? "stderr"
- : stderr;
-
- ApplicationDescription application = jobContext.getApplicationContext().getApplicationDeploymentDescription();
- ApplicationDeploymentDescriptionType appDesc = application.getType();
-
- String stdoutLocation = downloadLocation+File.separator+stdoutFileName;
- FileDownloader f1 = new FileDownloader("output/"+stdoutFileName,stdoutLocation, Mode.overwrite);
- try {
- f1.perform(storageClient);
- String stdoutput = readFile(stdoutLocation);
- appDesc.setStandardOutput(stdoutput);
- } catch (Exception e) {
- throw new GFacProviderException(e.getLocalizedMessage(),e);
- }
- String stderrLocation = downloadLocation+File.separator+stderrFileName;
- FileDownloader f2 = new FileDownloader("output/"+stderrFileName,stderrLocation, Mode.overwrite);
- try {
- f2.perform(storageClient);
- String stderror = readFile(stderrLocation);
- appDesc.setStandardError(stderror);
- } catch (Exception e) {
- throw new GFacProviderException(e.getLocalizedMessage(),e);
- }
- }
-
- private String readFile(String localFile) throws IOException {
- BufferedReader instream = new BufferedReader(new FileReader(localFile));
- StringBuffer buff = new StringBuffer();
- String temp = null;
- while ((temp = instream.readLine()) != null) {
- buff.append(temp);
- buff.append(Constants.NEWLINE);
- }
-
- log.info("finish read file:" + localFile);
-
- return buff.toString();
- }
-
- private String getDownloadLocation() {
- TaskDetails taskData = jobContext.getTaskData();
- if (taskData != null && taskData.getAdvancedOutputDataHandling() != null) {
- String outputDataDirectory = taskData.getAdvancedOutputDataHandling().getOutputDataDir();
- return outputDataDirectory;
- }
- return null;
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/airavata/blob/13b505ae/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/utils/FileDownloader.java
----------------------------------------------------------------------
diff --git a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/utils/FileDownloader.java b/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/utils/FileDownloader.java
deleted file mode 100644
index 1bbf2b7..0000000
--- a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/provider/utils/FileDownloader.java
+++ /dev/null
@@ -1,256 +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.gfac.provider.utils;
-
-import java.io.File;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.OutputStream;
-import java.util.Map;
-
-import org.unigrids.services.atomic.types.GridFileType;
-import org.unigrids.services.atomic.types.ProtocolType;
-
-import de.fzj.unicore.uas.client.FileTransferClient;
-import de.fzj.unicore.uas.client.StorageClient;
-import de.fzj.unicore.uas.client.UFTPConstants;
-import de.fzj.unicore.uas.client.UFTPFileTransferClient;
-import de.fzj.unicore.uas.fts.FiletransferOptions.IMonitorable;
-import de.fzj.unicore.uas.fts.FiletransferOptions.SupportsPartialRead;
-
-/**
- * helper that exports remote files from a UNICORE Storage
- * to the local client machine.<br/>
- * Simple wildcards ("*" and "?") and download of
- * directories are supported.
- *
- * TODO this should be refactored so the single-file download logic
- * is separated from the wildcard/directory/provided outputStream logic
- *
- * @author schuller
- */
-public class FileDownloader extends FileTransferBase{
-
- private boolean showProgress=true;
-
- private boolean forceFileOnly=false;
-
- private OutputStream targetStream=null;
-
- public FileDownloader(String from, String to, Mode mode){
- this(from,to,mode,true);
- }
-
- public FileDownloader(String from, String to, Mode mode, boolean failOnError){
- this.to=to;
- this.from=from;
- this.mode=mode;
- this.failOnError=failOnError;
- }
-
- public void perform(StorageClient sms)throws Exception{
- boolean isWildcard=hasWildCards(from);
- boolean isDirectory=false;
- GridFileType gridSource=null;
- if(isWildcard){
- performWildCardExport(sms);
- }
- else {
- //check if source is a directory
- gridSource=sms.listProperties(from);
- isDirectory=gridSource.getIsDirectory();
- if(isDirectory){
- if(forceFileOnly){
- throw new IOException("Source is a directory");
- }
- performDirectoryExport(gridSource, new File(to), sms);
- }
- else{
- download(gridSource,new File(to),sms);
- }
- }
- }
-
- protected void performDirectoryExport(GridFileType directory, File targetDirectory, StorageClient sms)throws Exception{
- if(!targetDirectory.exists()|| !targetDirectory.canWrite()){
- throw new IOException("Target directory <"+to+"> does not exist or is not writable!");
- }
- if(!targetDirectory.isDirectory()){
- throw new IOException("Target <"+to+"> is not a directory!");
- }
- GridFileType[]gridFiles=sms.listDirectory(directory.getPath());
- for(GridFileType file: gridFiles){
- if(file.getIsDirectory()){
- if(!recurse) {
- System.out.println("Skipping directory "+file.getPath());
- continue;
- }
- else{
- File newTargetDirectory=new File(targetDirectory,getName(file.getPath()));
- boolean success=newTargetDirectory.mkdirs();
- if(!success)throw new IOException("Can create directory: "+newTargetDirectory.getAbsolutePath());
- performDirectoryExport(file, newTargetDirectory, sms);
- continue;
- }
- }
- download(file, new File(targetDirectory,getName(file.getPath())), sms);
- }
- }
-
- protected void performWildCardExport(StorageClient sms)throws Exception{
- String dir=getDir(from);
- if(dir==null)dir="/";
- GridFileType[] files=sms.find(dir, false, from, false, null, null);
- File targetDir=targetStream==null?new File(to):null;
- if(targetStream==null){
- if(!targetDir.isDirectory())throw new IOException("Target is not a directory.");
- }
- for(GridFileType f: files){
- download(f, targetDir, sms);
- }
- }
-
- private String getDir(String path){
- return new File(path).getParent();
- }
-
- private String getName(String path){
- return new File(path).getName();
- }
-
- /**
- * download a single regular file
- *
- * @param source - grid file descriptor
- * @param localFile - local file or directory to write to
- * @param sms
- * @throws Exception
- */
- private void download(GridFileType source, File localFile, StorageClient sms)throws Exception{
- if(source==null || source.getIsDirectory()){
- throw new IllegalStateException("Source="+source);
- }
-
- OutputStream os=targetStream!=null?targetStream:null;
- FileTransferClient ftc=null;
- try{
- String path=source.getPath();
- if(targetStream==null){
- if(localFile.isDirectory()){
- localFile=new File(localFile,getName(source.getPath()));
- }
- if(mode.equals(Mode.nooverwrite) && localFile.exists()){
- System.out.println("File exists and creation mode was set to 'nooverwrite'.");
- return;
- }
- System.out.println("Downloading remote file '"+sms.getUrl()+"#/"+path+"' -> "+localFile.getAbsolutePath());
- os=new FileOutputStream(localFile.getAbsolutePath(), mode.equals(Mode.append));
- }
-
- chosenProtocol=sms.findSupportedProtocol(preferredProtocols.toArray(new ProtocolType.Enum[preferredProtocols.size()]));
- Map<String,String>extraParameters=makeExtraParameters(chosenProtocol);
- ftc=sms.getExport(path,extraParameters,chosenProtocol);
- configure(ftc, extraParameters);
- System.out.println("DEB:File transfer URL : "+ftc.getUrl());
-// ProgressBar p=null;
- if(ftc instanceof IMonitorable && showProgress){
- long size=ftc.getSourceFileSize();
- if(isRange()){
- size=getRangeSize();
- }
-// p=new ProgressBar(localFile.getName(),size,msg);
-// ((IMonitorable) ftc).setProgressListener(p);
- }
- long startTime=System.currentTimeMillis();
- if(isRange()){
- if(!(ftc instanceof SupportsPartialRead)){
- throw new Exception("Byte range is defined but protocol does not allow " +
- "partial read! Please choose a different protocol!");
- }
- System.out.println("Byte range: "+startByte+" - "+(getRangeSize()>0?endByte:""));
- SupportsPartialRead pReader=(SupportsPartialRead)ftc;
- pReader.readPartial(startByte, endByte-startByte+1, os);
- }
- else{
- ftc.readAllData(os);
- }
-// if(p!=null){
-// p.finish();
-// }
- if(timing){
- long duration=System.currentTimeMillis()-startTime;
- double rate=(double)localFile.length()/(double)duration;
- System.out.println("Rate: " +rate+ " kB/sec.");
- }
- if(targetStream==null)copyProperties(source, localFile);
- }
- finally{
- try{
- if(targetStream==null && os!=null){
- os.close();
- }
- }catch(Exception ignored){}
- if(ftc!=null){
- try{
- ftc.destroy();
- }catch(Exception e1){
-// System.out.println("Could not destroy the filetransfer client",e1);
- }
- }
- }
- }
-
- /**
- * if possible, copy the remote executable flag to the local file
- * @throws Exception
- */
- private void copyProperties(GridFileType source, File localFile)throws Exception{
- try{
- localFile.setExecutable(source.getPermissions().getExecutable());
- }
- catch(Exception ex){
- //TODO: logging
-// ("Can't set 'executable' flag for "+localFile.getName(), ex);
- }
- }
-
- private void configure(FileTransferClient ftc, Map<String,String>params){
- if(ftc instanceof UFTPFileTransferClient){
- UFTPFileTransferClient u=(UFTPFileTransferClient)ftc;
- String secret=params.get(UFTPConstants.PARAM_SECRET);
- u.setSecret(secret);
- }
- }
-
- public void setShowProgress(boolean showProgress) {
- this.showProgress = showProgress;
- }
-
- public void setForceFileOnly(boolean forceFileOnly) {
- this.forceFileOnly = forceFileOnly;
- }
-
- public void setTargetStream(OutputStream targetStream) {
- this.targetStream = targetStream;
- }
-
-}