fixed data staging issues, supporting local stageins to remote job working directory, use of delegation tokens while job runs
Project: http://git-wip-us.apache.org/repos/asf/airavata/repo Commit: http://git-wip-us.apache.org/repos/asf/airavata/commit/f4a32fd0 Tree: http://git-wip-us.apache.org/repos/asf/airavata/tree/f4a32fd0 Diff: http://git-wip-us.apache.org/repos/asf/airavata/diff/f4a32fd0 Branch: refs/heads/master Commit: f4a32fd04e1cf2dbc24b59303cba9fcdb6855a9d Parents: 576c87d Author: msmemon <[email protected]> Authored: Fri Dec 5 17:23:45 2014 +0100 Committer: msmemon <[email protected]> Committed: Fri Dec 5 17:23:45 2014 +0100 ---------------------------------------------------------------------- .../client/samples/CreateLaunchExperiment.java | 38 ++- .../tools/RegisterSampleApplications.java | 22 +- .../gfac/bes/provider/impl/BESProvider.java | 9 +- .../bes/security/UNICORESecurityContext.java | 2 +- .../gfac/bes/utils/ApplicationProcessor.java | 21 +- .../gfac/bes/utils/DataStagingProcessor.java | 236 ------------------- .../gfac/bes/utils/DataTransferrer.java | 4 +- .../airavata/gfac/bes/utils/JSDLGenerator.java | 3 +- .../airavata/gfac/bes/utils/SecurityUtils.java | 7 +- .../gfac/bes/utils/UASDataStagingProcessor.java | 10 +- .../org/apache/airavata/gfac/Scheduler.java | 7 + .../airavata/gfac/core/cpi/BetterGfacImpl.java | 16 +- modules/gfac/pom.xml | 2 +- 13 files changed, 85 insertions(+), 292 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/airavata/blob/f4a32fd0/airavata-api/airavata-client-sdks/java-client-samples/src/main/java/org/apache/airavata/client/samples/CreateLaunchExperiment.java ---------------------------------------------------------------------- diff --git a/airavata-api/airavata-client-sdks/java-client-samples/src/main/java/org/apache/airavata/client/samples/CreateLaunchExperiment.java b/airavata-api/airavata-client-sdks/java-client-samples/src/main/java/org/apache/airavata/client/samples/CreateLaunchExperiment.java index 34d419c..5c3ef19 100644 --- a/airavata-api/airavata-client-sdks/java-client-samples/src/main/java/org/apache/airavata/client/samples/CreateLaunchExperiment.java +++ b/airavata-api/airavata-client-sdks/java-client-samples/src/main/java/org/apache/airavata/client/samples/CreateLaunchExperiment.java @@ -57,7 +57,7 @@ public class CreateLaunchExperiment { private static final String DEFAULT_GATEWAY = "default.registry.gateway"; private static Airavata.Client airavataClient; - private static String echoAppId = "Echo_36fbb479-5b41-4f48-a9c5-382ee910ac6b"; + private static String echoAppId = "Echo_753d7cf6-f79a-4f7f-8ada-5d707e90c383"; private static String wrfAppId = "WRF_7ad5da38-c08b-417c-a9ea-da9298839762"; private static String amberAppId = "Amber_49b16f6f-93ab-4885-9971-6ab2ab5eb3d3"; private static String gromacsAppId = "GROMACS_05622038-9edd-4cb1-824e-0b7cb993364b"; @@ -93,9 +93,9 @@ public class CreateLaunchExperiment { try { for (int i = 0; i < 1; i++) { // final String expId = createExperimentForSSHHost(airavata); -// final String expId = createEchoExperimentForFSD(airavataClient); + final String expId = createEchoExperimentForFSD(airavataClient); // final String expId = createEchoExperimentForStampede(airavataClient); - final String expId = createEchoExperimentForTrestles(airavataClient); +// final String expId = createEchoExperimentForTrestles(airavataClient); // final String expId = createExperimentEchoForLocalHost(airavataClient); // final String expId = createExperimentWRFTrestles(airavataClient); // final String expId = createExperimentForBR2(airavataClient); @@ -161,7 +161,6 @@ public class CreateLaunchExperiment { ucrJobSubmission.setSecurityProtocol(securityProtocol); ucrJobSubmission.setUnicoreEndPointURL(unicoreEndPointURL); - return jobSubmission.getJobSubmissionInterfaceId(); } @@ -180,7 +179,13 @@ public class CreateLaunchExperiment { output.setType(DataType.STRING); output.setValue(""); exOut.add(output); - + + OutputDataObjectType output2 = new OutputDataObjectType(); + output2.setName("Echoed_Output2"); + output2.setType(DataType.URI); + output2.setValue("file:///tmp/test.txt"); + exOut.add(output2); + Experiment simpleExperiment = ExperimentModelUtil.createSimpleExperiment("default", "admin", "echoExperiment", "SimpleEcho2", echoAppId, exInputs); simpleExperiment.setExperimentOutputs(exOut); @@ -223,23 +228,32 @@ public class CreateLaunchExperiment { InputDataObjectType input = new InputDataObjectType(); input.setName("Input_to_Echo"); input.setType(DataType.STRING); - input.setValue("Echoed_Output=Hello World"); - exInputs.add(input); + input.setValue("Hello World"); + + InputDataObjectType i2 = new InputDataObjectType(); - i2.setName("Input_to_Echo1"); + i2.setName("Input_to_Echo2"); i2.setType(DataType.URI); - i2.setValue("http://shrib.com/22QmrrX4";); + i2.setValue("http://www.textfiles.com/100/ad.txt";); + + InputDataObjectType i3 = new InputDataObjectType(); + i3.setName("Input_to_Echo3"); + i3.setType(DataType.URI); + i3.setValue("file:///tmp/test.txt"); + + exInputs.add(input); exInputs.add(i2); + exInputs.add(i3); List<OutputDataObjectType> exOut = new ArrayList<OutputDataObjectType>(); + OutputDataObjectType output = new OutputDataObjectType(); output.setName("Echoed_Output"); output.setType(DataType.STRING); - output.setValue("22QmrrX4"); + output.setValue("test.txt"); exOut.add(output); - Experiment simpleExperiment = ExperimentModelUtil.createSimpleExperiment("default", "admin", "echoExperiment", "SimpleEcho2", echoAppId, exInputs); simpleExperiment.setExperimentOutputs(exOut); @@ -251,7 +265,7 @@ public class CreateLaunchExperiment { for (String id : computeResources.keySet()) { String resourceName = computeResources.get(id); if (resourceName.equals(unicoreHostName)) { - ComputationalResourceScheduling scheduling = ExperimentModelUtil.createComputationResourceScheduling(id, 1, 1, 1, "normal", 30, 0, 1, "sds128"); + ComputationalResourceScheduling scheduling = ExperimentModelUtil.createComputationResourceScheduling(id, 1, 1, 1, "normal", 30, 0, 1048576, "sds128"); UserConfigurationData userConfigurationData = new UserConfigurationData(); userConfigurationData.setAiravataAutoSchedule(false); userConfigurationData.setOverrideManualScheduledParams(false); http://git-wip-us.apache.org/repos/asf/airavata/blob/f4a32fd0/airavata-api/airavata-client-sdks/java-client-samples/src/main/java/org/apache/airavata/client/tools/RegisterSampleApplications.java ---------------------------------------------------------------------- diff --git a/airavata-api/airavata-client-sdks/java-client-samples/src/main/java/org/apache/airavata/client/tools/RegisterSampleApplications.java b/airavata-api/airavata-client-sdks/java-client-samples/src/main/java/org/apache/airavata/client/tools/RegisterSampleApplications.java index 41020a0..ce26814 100644 --- a/airavata-api/airavata-client-sdks/java-client-samples/src/main/java/org/apache/airavata/client/tools/RegisterSampleApplications.java +++ b/airavata-api/airavata-client-sdks/java-client-samples/src/main/java/org/apache/airavata/client/tools/RegisterSampleApplications.java @@ -210,7 +210,6 @@ public class RegisterSampleApplications { throw new AiravataClientException(); System.out.println("FSD Compute ResourceID: "+fsdResourceId); - JobSubmissionInterface jobSubmission = RegisterSampleApplicationsUtils.createJobSubmissionInterface(fsdResourceId, protocol, 2); UnicoreJobSubmission ucrJobSubmission = new UnicoreJobSubmission(); ucrJobSubmission.setSecurityProtocol(securityProtocol); @@ -222,11 +221,6 @@ public class RegisterSampleApplications { return jobSubmission.getJobSubmissionInterfaceId(); } - public void registerfsd(){ - System.out.println("\n #### Registering XSEDE Computational Resources #### \n"); - - } - public void registerAppModules() { try { System.out.println("\n #### Registering Application Modules #### \n"); @@ -529,17 +523,25 @@ public class RegisterSampleApplications { List<String> appModules = new ArrayList<String>(); appModules.add(echoModuleId); - InputDataObjectType input1 = RegisterSampleApplicationsUtils.createAppInput("Input_to_Echo", "Hello World", + InputDataObjectType input1 = RegisterSampleApplicationsUtils.createAppInput("Input_to_Echo", "", DataType.STRING, null, false, "A test string to Echo", null); + InputDataObjectType input2 = RegisterSampleApplicationsUtils.createAppInput("Input_to_Echo2", "", + DataType.URI, null, false, "A sample input remote file", null); + + InputDataObjectType input3 = RegisterSampleApplicationsUtils.createAppInput("Input_to_Echo3", "file:///tmp/test.txt", + DataType.URI, null, false, "A sample input local file", null); + + List<InputDataObjectType> applicationInputs = new ArrayList<InputDataObjectType>(); applicationInputs.add(input1); OutputDataObjectType output1 = RegisterSampleApplicationsUtils.createAppOutput("Echoed_Output", - "22QmrrX4", DataType.STRING); - + "", DataType.STRING); + List<OutputDataObjectType> applicationOutputs = new ArrayList<OutputDataObjectType>(); applicationOutputs.add(output1); + echoInterfaceId = airavataClient.registerApplicationInterface( RegisterSampleApplicationsUtils.createApplicationInterfaceDescription(echoName, echoDescription, @@ -1104,7 +1106,7 @@ public class RegisterSampleApplications { "/N/dc2/scratch/cgateway/gta-work-dirs"); ComputeResourcePreference fsdResourcePreferences = RegisterSampleApplicationsUtils. - createComputeResourcePreference(fsdResourceId, null, false, null, null, null,null); + createComputeResourcePreference(fsdResourceId, null, false, null, JobSubmissionProtocol.UNICORE, DataMovementProtocol.UNICORE_STORAGE_SERVICE,null); GatewayResourceProfile gatewayResourceProfile = new GatewayResourceProfile(); gatewayResourceProfile.setGatewayID(DEFAULT_GATEWAY); http://git-wip-us.apache.org/repos/asf/airavata/blob/f4a32fd0/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/provider/impl/BESProvider.java ---------------------------------------------------------------------- diff --git a/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/provider/impl/BESProvider.java b/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/provider/impl/BESProvider.java index 6fdadfb..d490269 100644 --- a/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/provider/impl/BESProvider.java +++ b/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/provider/impl/BESProvider.java @@ -167,9 +167,9 @@ public class BESProvider extends AbstractProvider implements GFacProvider, .toString())); jobExecutionContext.getNotifier().publish(new UnicoreJobIDEvent(jobId)); - GFacUtils.saveJobStatus(jobExecutionContext, details, JobState.SUBMITTED); +// GFacUtils.saveJobStatus(jobExecutionContext, details, JobState.SUBMITTED); - factory.getActivityStatus(activityEpr); + log.info(formatStatusMessage(activityEpr.getAddress() .getStringValue(), factory.getActivityStatus(activityEpr) .toString())); @@ -202,8 +202,7 @@ public class BESProvider extends AbstractProvider implements GFacProvider, + applicationJobStatus; jobExecutionContext.getNotifier().publish( new StatusChangeEvent(jobStatusMessage)); - GFacUtils.updateJobStatus(jobExecutionContext, jobDetails, - applicationJobStatus); + GFacUtils.updateJobStatus(jobExecutionContext, jobDetails, applicationJobStatus); throw new GFacProviderException( jobExecutionContext.getExperimentID() + "Job Canceled"); } else if (activityStatus.getState() == ActivityStateEnumeration.FINISHED) { @@ -412,7 +411,7 @@ public class BESProvider extends AbstractProvider implements GFacProvider, ActivityStatusType activityStatus = getStatus(factory, activityEpr); JobState applicationJobStatus = getApplicationJobStatus(activityStatus); String jobStatusMessage = "Status of job " + jobId + "is " + applicationJobStatus; - GFacUtils.updateJobStatus(jobExecutionContext, jobDetails, applicationJobStatus); +// GFacUtils.updateJobStatus(jobExecutionContext, jobDetails, applicationJobStatus); jobExecutionContext.getNotifier().publish(new StatusChangeEvent(jobStatusMessage)); http://git-wip-us.apache.org/repos/asf/airavata/blob/f4a32fd0/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/security/UNICORESecurityContext.java ---------------------------------------------------------------------- diff --git a/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/security/UNICORESecurityContext.java b/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/security/UNICORESecurityContext.java index 57fc2d5..e2896e8 100644 --- a/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/security/UNICORESecurityContext.java +++ b/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/security/UNICORESecurityContext.java @@ -52,7 +52,7 @@ public class UNICORESecurityContext extends X509SecurityContext { secProperties.getETDSettings().setExtendTrustDelegation(true); // secProperties.setMessageLogging(true); // secProperties.setDoSignMessage(true); -// secProperties.getETDSettings().setIssuerCertificateChain(secProperties.getCredential().getCertificateChain()); + secProperties.getETDSettings().setIssuerCertificateChain(secProperties.getCredential().getCertificateChain()); http://git-wip-us.apache.org/repos/asf/airavata/blob/f4a32fd0/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/ApplicationProcessor.java ---------------------------------------------------------------------- diff --git a/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/ApplicationProcessor.java b/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/ApplicationProcessor.java index 91c27f9..7c7a59e 100644 --- a/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/ApplicationProcessor.java +++ b/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/ApplicationProcessor.java @@ -38,9 +38,9 @@ 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"; - } +// if (userName.equalsIgnoreCase("admin")){ +// userName = "CN=zdv575, O=Ultrascan Gateway, C=DE"; +// } ApplicationDeploymentDescription appDep= context.getApplicationContext().getApplicationDeploymentDescription(); String appname = context.getApplicationContext().getApplicationInterfaceDescription().getApplicationName(); @@ -53,9 +53,20 @@ public class ApplicationProcessor { // createApplicationEnvironment(value, appDep.getSetEnvironment(), parallelism); // } // - String stdout = context.getStandardOutput(); - String stderr = context.getStandardError(); + String stdout = context.getStandardOutput(); + String stderr = context.getStandardError(); + if(stdout != null) { + stdout = stdout.substring(stdout.lastIndexOf('/')+1); + } + + if(stderr != null) { + stderr = stderr.substring(stderr.lastIndexOf('/')+1); + } + + stdout = (stdout == null || stdout.equals("")) ? "stdout":stdout; + stderr = (stdout == null || stderr.equals("")) ? "stderr":stderr; + if (appDep.getExecutablePath() != null) { FileNameType fNameType = FileNameType.Factory.newInstance(); fNameType.setStringValue(appDep.getExecutablePath()); http://git-wip-us.apache.org/repos/asf/airavata/blob/f4a32fd0/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/DataStagingProcessor.java ---------------------------------------------------------------------- diff --git a/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/DataStagingProcessor.java b/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/DataStagingProcessor.java deleted file mode 100644 index 0541d47..0000000 --- a/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/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.bes.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.core.context.JobExecutionContext; -import org.apache.airavata.gfac.core.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 = URIUtils.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 = URIUtils.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 = URIUtils.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/f4a32fd0/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/DataTransferrer.java ---------------------------------------------------------------------- diff --git a/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/DataTransferrer.java b/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/DataTransferrer.java index 4aa6cc1..cb036fa 100644 --- a/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/DataTransferrer.java +++ b/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/DataTransferrer.java @@ -63,7 +63,7 @@ public class DataTransferrer { if (uri.startsWith("file")) { try { String uriWithoutProtocol = uri.substring(uri.lastIndexOf("://") + 1, uri.length()); - FileUploader fileUploader = new FileUploader(uriWithoutProtocol,"input/" + fileName,Mode.overwrite); + FileUploader fileUploader = new FileUploader(uriWithoutProtocol,fileName,Mode.overwrite); fileUploader.perform(storageClient); } catch (FileNotFoundException e3) { throw new GFacProviderException( @@ -139,7 +139,7 @@ public class DataTransferrer { String scriptExitCodeFName = "UNICORE_SCRIPT_EXIT_CODE"; String scriptCodeLocation = downloadLocation+File.separator+scriptExitCodeFName; f1.setFrom(scriptExitCodeFName); - f1.setTo(downloadLocation+File.separator+scriptCodeLocation); + f1.setTo(scriptCodeLocation); f1.perform(storageClient); log.info("UNICORE_SCRIPT_EXIT_CODE downloaded to "+scriptCodeLocation); } http://git-wip-us.apache.org/repos/asf/airavata/blob/f4a32fd0/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/JSDLGenerator.java ---------------------------------------------------------------------- diff --git a/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/JSDLGenerator.java b/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/JSDLGenerator.java index 9755bc7..4743522 100644 --- a/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/JSDLGenerator.java +++ b/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/JSDLGenerator.java @@ -55,8 +55,7 @@ public class JSDLGenerator implements BESConstants { ApplicationProcessor.generateJobSpecificAppElements(value, context); - DataStagingProcessor.generateDataStagingElements(value, context); - + return jobDefDoc; } http://git-wip-us.apache.org/repos/asf/airavata/blob/f4a32fd0/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/SecurityUtils.java ---------------------------------------------------------------------- diff --git a/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/SecurityUtils.java b/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/SecurityUtils.java index ac79f33..1da0bac 100644 --- a/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/SecurityUtils.java +++ b/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/SecurityUtils.java @@ -16,7 +16,6 @@ import javax.security.auth.x500.X500Principal; import org.apache.airavata.common.exception.ApplicationSettingsException; import org.apache.airavata.common.utils.ServerSettings; -import org.apache.airavata.commons.gfac.type.HostDescription; import org.apache.airavata.credential.store.store.CredentialReader; import org.apache.airavata.gfac.GFacException; import org.apache.airavata.gfac.RequestData; @@ -24,14 +23,13 @@ import org.apache.airavata.gfac.bes.security.UNICORESecurityContext; import org.apache.airavata.gfac.bes.security.X509SecurityContext; import org.apache.airavata.gfac.core.context.JobExecutionContext; import org.apache.airavata.gfac.core.utils.GFacUtils; -import org.apache.airavata.schemas.gfac.UnicoreHostType; +import org.apache.airavata.model.appcatalog.computeresource.JobSubmissionProtocol; import org.bouncycastle.asn1.ASN1InputStream; import org.bouncycastle.asn1.x500.X500Name; import org.bouncycastle.asn1.x509.AlgorithmIdentifier; import org.bouncycastle.asn1.x509.SubjectPublicKeyInfo; import org.slf4j.Logger; import org.slf4j.LoggerFactory; - 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; @@ -45,8 +43,7 @@ public class SecurityUtils { public static void addSecurityContext(JobExecutionContext jobExecutionContext) throws GFacException { - HostDescription registeredHost = jobExecutionContext.getApplicationContext().getHostDescription(); - if (! (registeredHost.getType() instanceof UnicoreHostType)) { + if (!jobExecutionContext.getPreferredJobSubmissionProtocol().equals(JobSubmissionProtocol.UNICORE)) { logger.error("This is a wrong method to invoke for UNICORE host types,please check your gfac-config.xml"); } else http://git-wip-us.apache.org/repos/asf/airavata/blob/f4a32fd0/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/UASDataStagingProcessor.java ---------------------------------------------------------------------- diff --git a/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/UASDataStagingProcessor.java b/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/UASDataStagingProcessor.java index 9c92789..e82edef 100644 --- a/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/UASDataStagingProcessor.java +++ b/modules/gfac/gfac-bes/src/main/java/org/apache/airavata/gfac/bes/utils/UASDataStagingProcessor.java @@ -46,7 +46,7 @@ public class UASDataStagingProcessor { private static void createInURISMSElement(JobDefinitionType value, String smsUrl, String uri) throws Exception { - String fileName = "input/" + new File(uri).getName(); + String fileName = new File(uri).getName(); if (uri.startsWith("file")) { uri = smsUrl+"#/"+fileName; @@ -113,13 +113,9 @@ public class UASDataStagingProcessor { List<OutputDataObjectType> applicationOutputs = context.getTaskData().getApplicationOutputs(); if (applicationOutputs != null && !applicationOutputs.isEmpty()){ for (OutputDataObjectType output : applicationOutputs){ - if(output.getType().equals(DataType.URI)) { + if(output.getType().equals(DataType.URI) && !output.getValue().startsWith("file:")) { createOutURIElement(value, output.getValue()); } - else if(output.getType().equals(DataType.STRING)) { - // TODO: remove this check, as out string - createOutStringElements(value, smsUrl, output.getValue()); - } } } return value; @@ -128,7 +124,7 @@ public class UASDataStagingProcessor { private static void buildDataStagingFromInputContext(JobExecutionContext context, JobDefinitionType value, String smsUrl) throws Exception { - List<InputDataObjectType> applicationInputs = context.getApplicationContext().getApplicationInterfaceDescription().getApplicationInputs(); + List<InputDataObjectType> applicationInputs = context.getTaskData().getApplicationInputs(); if (applicationInputs != null && !applicationInputs.isEmpty()){ for (InputDataObjectType input : applicationInputs){ http://git-wip-us.apache.org/repos/asf/airavata/blob/f4a32fd0/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/Scheduler.java ---------------------------------------------------------------------- diff --git a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/Scheduler.java b/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/Scheduler.java index a9940b8..853ffc8 100644 --- a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/Scheduler.java +++ b/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/Scheduler.java @@ -32,6 +32,7 @@ import org.apache.airavata.model.appcatalog.computeresource.JobSubmissionInterfa import org.apache.airavata.model.appcatalog.computeresource.JobSubmissionProtocol; 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.slf4j.Logger; import org.slf4j.LoggerFactory; import org.w3c.dom.Document; @@ -42,6 +43,7 @@ import javax.xml.parsers.DocumentBuilder; import javax.xml.parsers.DocumentBuilderFactory; import javax.xml.parsers.ParserConfigurationException; import javax.xml.xpath.XPathExpressionException; + import java.io.File; import java.io.IOException; import java.net.URL; @@ -118,6 +120,7 @@ public class Scheduler { JobSubmissionProtocol jobSubmissionProtocol = jobExecutionContext.getPreferredJobSubmissionProtocol(); SSHJobSubmission sshJobSubmission; LOCALSubmission localSubmission; + UnicoreJobSubmission unicoreSubmission; String securityProtocol = null; try { AppCatalog appCatalog = jobExecutionContext.getAppCatalog(); @@ -130,6 +133,10 @@ public class Scheduler { }else if (jobSubmissionProtocol == JobSubmissionProtocol.LOCAL) { localSubmission = appCatalog.getComputeResource().getLocalJobSubmission(jobExecutionContext.getPreferredJobSubmissionInterface().getJobSubmissionInterfaceId()); } + else if (jobSubmissionProtocol == JobSubmissionProtocol.UNICORE) { + unicoreSubmission = appCatalog.getComputeResource().getUNICOREJobSubmission(jobExecutionContext.getPreferredJobSubmissionInterface().getJobSubmissionInterfaceId()); + securityProtocol = unicoreSubmission.getSecurityProtocol().toString(); + } List<Element> elements = GFacUtils.getElementList(GFacConfiguration.getHandlerDoc(), Constants.XPATH_EXPR_PROVIDER_ON_SUBMISSION + jobSubmissionProtocol + "']"); for (Element element : elements) { String security = element.getAttribute(Constants.GFAC_CONFIG_SECURITY_ATTRIBUTE); http://git-wip-us.apache.org/repos/asf/airavata/blob/f4a32fd0/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/core/cpi/BetterGfacImpl.java ---------------------------------------------------------------------- diff --git a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/core/cpi/BetterGfacImpl.java b/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/core/cpi/BetterGfacImpl.java index aaa5461..8403f8c 100644 --- a/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/core/cpi/BetterGfacImpl.java +++ b/modules/gfac/gfac-core/src/main/java/org/apache/airavata/gfac/core/cpi/BetterGfacImpl.java @@ -370,12 +370,16 @@ public class BetterGfacImpl implements GFac,Watcher { jobExecutionContext.setPreferredDataMovementInterface(jobExecutionContext.getHostPrioritizedDataMovementInterfaces().get(0)); jobExecutionContext.setPreferredDataMovementProtocol(jobExecutionContext.getPreferredDataMovementInterface().getDataMovementProtocol()); } else { - for (DataMovementInterface dataMovementInterface : dataMovementInterfaces) { - if (gatewayResourcePreferences.getPreferredDataMovementProtocol() == dataMovementInterface.getDataMovementProtocol()) { - jobExecutionContext.setPreferredDataMovementInterface(dataMovementInterface); - break; - } - } + // this check is to avoid NPE when job submission endpoints do + // not contain any data movement interfaces. + if((dataMovementInterfaces != null) && (!dataMovementInterfaces.isEmpty())) { + for (DataMovementInterface dataMovementInterface : dataMovementInterfaces) { + if (gatewayResourcePreferences.getPreferredDataMovementProtocol() == dataMovementInterface.getDataMovementProtocol()) { + jobExecutionContext.setPreferredDataMovementInterface(dataMovementInterface); + break; + } + } + } } } else { setUpWorkingLocation(jobExecutionContext, applicationInterface, "/tmp"); http://git-wip-us.apache.org/repos/asf/airavata/blob/f4a32fd0/modules/gfac/pom.xml ---------------------------------------------------------------------- diff --git a/modules/gfac/pom.xml b/modules/gfac/pom.xml index 34253b9..2e7df07 100644 --- a/modules/gfac/pom.xml +++ b/modules/gfac/pom.xml @@ -38,7 +38,7 @@ <!--<module>gfac-hadoop</module>--> <!--<module>gfac-gram</module>--> <module>gfac-gsissh</module> - <!--<module>gfac-bes</module>--> + <module>gfac-bes</module> <module>gfac-monitor</module> <module>airavata-gfac-service</module> <module>airavata-gfac-stubs</module>
