Repository: airavata Updated Branches: refs/heads/master a393b735e -> 636a733bf
http://git-wip-us.apache.org/repos/asf/airavata/blob/636a733b/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/validator/impl/BatchQueueValidator.java ---------------------------------------------------------------------- diff --git a/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/validator/impl/BatchQueueValidator.java b/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/validator/impl/BatchQueueValidator.java index 664fcfe..b210a2a 100644 --- a/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/validator/impl/BatchQueueValidator.java +++ b/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/validator/impl/BatchQueueValidator.java @@ -27,7 +27,6 @@ import org.apache.airavata.model.error.ValidatorResult; import org.apache.airavata.model.experiment.*; import org.apache.airavata.model.process.ProcessModel; import org.apache.airavata.model.scheduling.ComputationalResourceSchedulingModel; -import org.apache.airavata.model.task.TaskModel; import org.apache.airavata.orchestrator.core.validator.JobMetadataValidator; import org.apache.airavata.registry.core.experiment.catalog.impl.RegistryFactory; import org.apache.airavata.registry.cpi.*; @@ -54,7 +53,7 @@ public class BatchQueueValidator implements JobMetadataValidator { } } - public ValidationResults validate(ExperimentModel experiment, ProcessModel processModel, TaskModel taskModel) { + public ValidationResults validate(ExperimentModel experiment, ProcessModel processModel) { ValidationResults validationResults = new ValidationResults(); validationResults.setValidationState(true); try { http://git-wip-us.apache.org/repos/asf/airavata/blob/636a733b/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/validator/impl/ExperimentStatusValidator.java ---------------------------------------------------------------------- diff --git a/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/validator/impl/ExperimentStatusValidator.java b/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/validator/impl/ExperimentStatusValidator.java index faae1bf..9d7fa0e 100644 --- a/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/validator/impl/ExperimentStatusValidator.java +++ b/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/core/validator/impl/ExperimentStatusValidator.java @@ -36,7 +36,7 @@ import java.util.List; public class ExperimentStatusValidator implements JobMetadataValidator { private static Logger log = LoggerFactory.getLogger(ExperimentStatusValidator.class); - public ValidationResults validate(ExperimentModel experiment, ProcessModel processModel, TaskModel taskModel) { + public ValidationResults validate(ExperimentModel experiment, ProcessModel processModel) { String error = "During the validation step experiment status should be CREATED, But this experiment status is : "; ValidationResults validationResults = new ValidationResults(); validationResults.setValidationState(true); http://git-wip-us.apache.org/repos/asf/airavata/blob/636a733b/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/cpi/Orchestrator.java ---------------------------------------------------------------------- diff --git a/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/cpi/Orchestrator.java b/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/cpi/Orchestrator.java index 7bb9d5f..3671ad9 100644 --- a/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/cpi/Orchestrator.java +++ b/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/cpi/Orchestrator.java @@ -24,10 +24,8 @@ import org.apache.airavata.model.error.LaunchValidationException; import org.apache.airavata.model.error.ValidationResults; import org.apache.airavata.model.experiment.ExperimentModel; import org.apache.airavata.model.process.ProcessModel; -import org.apache.airavata.model.task.TaskModel; import org.apache.airavata.orchestrator.core.exception.OrchestratorException; -import javax.validation.Validation; import java.util.List; /* @@ -41,12 +39,10 @@ public interface Orchestrator { * sure the experiment is ready to launch and if its not this will return false * @param experiment * @param processModel - * @param taskModel * @return boolean if the experiments are valids after executing all the validators return true otherwise it iwll return false * @throws OrchestratorException */ - ValidationResults validateExperiment(ExperimentModel experiment, ProcessModel processModel, - TaskModel taskModel) throws OrchestratorException,LaunchValidationException; + ValidationResults validateExperiment(ExperimentModel experiment, ProcessModel processModel) throws OrchestratorException,LaunchValidationException; /** * After creating the experiment Data user have the * experimentID as the handler to the experiment, during the launchExperiment @@ -56,8 +52,7 @@ public interface Orchestrator { * @return launchExperiment status * @throws OrchestratorException */ - boolean launchExperiment(ExperimentModel experiment, ProcessModel processModel, - TaskModel taskModel,String tokenId) throws OrchestratorException; + boolean launchExperiment(ExperimentModel experiment, ProcessModel processModel,String tokenId) throws OrchestratorException; /** @@ -67,7 +62,7 @@ public interface Orchestrator { * @return * @throws OrchestratorException */ - public List<TaskModel> createTasks(String experimentId) throws OrchestratorException; + public List<ProcessModel> createProcesses(String experimentId) throws OrchestratorException; /** * After creating the experiment Data user have the @@ -76,11 +71,10 @@ public interface Orchestrator { * * @param experiment * @param processModel - * @param task * @param tokenId * @throws OrchestratorException */ - void cancelExperiment(ExperimentModel experiment, ProcessModel processModel, TaskModel task,String tokenId) throws OrchestratorException; + void cancelExperiment(ExperimentModel experiment, ProcessModel processModel, String tokenId) throws OrchestratorException; //todo have to add another method to handle failed or jobs to be recovered by orchestrator //todo if you don't add these this is not an orchestrator, its just an intemediate component which invoke gfac http://git-wip-us.apache.org/repos/asf/airavata/blob/636a733b/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/cpi/impl/SimpleOrchestratorImpl.java ---------------------------------------------------------------------- diff --git a/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/cpi/impl/SimpleOrchestratorImpl.java b/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/cpi/impl/SimpleOrchestratorImpl.java index bed31ce..ec942a1 100644 --- a/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/cpi/impl/SimpleOrchestratorImpl.java +++ b/modules/orchestrator/orchestrator-core/src/main/java/org/apache/airavata/orchestrator/cpi/impl/SimpleOrchestratorImpl.java @@ -25,8 +25,6 @@ import org.apache.airavata.model.error.LaunchValidationException; import org.apache.airavata.model.error.ValidationResults; import org.apache.airavata.model.error.ValidatorResult; import org.apache.airavata.model.process.ProcessModel; -import org.apache.airavata.model.task.TaskModel; -import org.apache.airavata.model.util.ExperimentModelUtil; import org.apache.airavata.model.experiment.*; import org.apache.airavata.orchestrator.core.exception.OrchestratorException; import org.apache.airavata.orchestrator.core.impl.GFACPassiveJobSubmitter; @@ -67,14 +65,14 @@ public class SimpleOrchestratorImpl extends AbstractOrchestrator{ } } - public boolean launchExperiment(ExperimentModel experiment, ProcessModel processModel, TaskModel task,String tokenId) throws OrchestratorException { + public boolean launchExperiment(ExperimentModel experiment, ProcessModel processModel, String tokenId) throws OrchestratorException { // we give higher priority to userExperimentID String experimentId = experiment.getExperimentId(); - String taskId = task.getTaskId(); + String processId = processModel.getProcessId(); // creating monitorID to register with monitoring queue // this is a special case because amqp has to be in place before submitting the job try { - return jobSubmitter.submit(experimentId, taskId,tokenId); + return jobSubmitter.submit(experimentId, processId, tokenId); } catch (Exception e) { throw new OrchestratorException("Error launching the job", e); } @@ -88,9 +86,9 @@ public class SimpleOrchestratorImpl extends AbstractOrchestrator{ * @return * @throws OrchestratorException */ - public List<TaskModel> createTasks(String experimentId) throws OrchestratorException { + public List<ProcessModel> createProcesses(String experimentId) throws OrchestratorException { ExperimentModel experiment = null; - List<TaskModel> tasks = new ArrayList<TaskModel>(); + List<ProcessModel> processModels = new ArrayList<ProcessModel>(); // FIXME : should change as create processes // try { // Registry newRegistry = orchestratorContext.getNewRegistry(); @@ -109,16 +107,16 @@ public class SimpleOrchestratorImpl extends AbstractOrchestrator{ // // TaskDetails taskDetails = ExperimentModelUtil.cloneTaskFromExperiment(experiment); // taskDetails.setTaskID((String) newRegistry.getExperimentCatalog().add(ExpCatChildDataType.TASK_DETAIL, taskDetails, nodeID)); -// tasks.add(taskDetails); +// processModels.add(taskDetails); // } // } catch (Exception e) { // throw new OrchestratorException("Error during creating a task"); // } - return tasks; + return processModels; } - public ValidationResults validateExperiment(ExperimentModel experiment, ProcessModel processModel, TaskModel taskModel) throws OrchestratorException,LaunchValidationException { + public ValidationResults validateExperiment(ExperimentModel experiment, ProcessModel processModel) throws OrchestratorException,LaunchValidationException { org.apache.airavata.model.error.ValidationResults validationResults = new org.apache.airavata.model.error.ValidationResults(); validationResults.setValidationState(true); // initially making it to success, if atleast one failed them simply mark it failed. String errorMsg = "Validation Errors : "; @@ -128,7 +126,7 @@ public class SimpleOrchestratorImpl extends AbstractOrchestrator{ try { Class<? extends JobMetadataValidator> vClass = Class.forName(validator.trim()).asSubclass(JobMetadataValidator.class); JobMetadataValidator jobMetadataValidator = vClass.newInstance(); - validationResults = jobMetadataValidator.validate(experiment, processModel, taskModel); + validationResults = jobMetadataValidator.validate(experiment, processModel); if (validationResults.isValidationState()) { logger.info("Validation of " + validator + " is SUCCESSFUL"); } else { @@ -147,8 +145,8 @@ public class SimpleOrchestratorImpl extends AbstractOrchestrator{ ErrorModel details = new ErrorModel(); details.setActualErrorMessage(errorMsg); details.setCreationTime(Calendar.getInstance().getTimeInMillis()); - orchestratorContext.getNewRegistry().getExperimentCatalog().add(ExpCatChildDataType.TASK_ERROR, details, - taskModel.getTaskId()); + orchestratorContext.getNewRegistry().getExperimentCatalog().add(ExpCatChildDataType.PROCESS_ERROR, details, + processModel.getProcessId()); } catch (RegistryException e) { logger.error("Error while saving error details to registry", e); } @@ -177,7 +175,7 @@ public class SimpleOrchestratorImpl extends AbstractOrchestrator{ } } - public void cancelExperiment(ExperimentModel experiment, ProcessModel processModel, TaskModel task, String tokenId) + public void cancelExperiment(ExperimentModel experiment, ProcessModel processModel, String tokenId) throws OrchestratorException { // FIXME // List<JobDetails> jobDetailsList = task.getJobDetailsList(); http://git-wip-us.apache.org/repos/asf/airavata/blob/636a733b/modules/orchestrator/orchestrator-service/src/main/java/org/apache/airavata/orchestrator/server/OrchestratorServerHandler.java ---------------------------------------------------------------------- diff --git a/modules/orchestrator/orchestrator-service/src/main/java/org/apache/airavata/orchestrator/server/OrchestratorServerHandler.java b/modules/orchestrator/orchestrator-service/src/main/java/org/apache/airavata/orchestrator/server/OrchestratorServerHandler.java index 38bdbc5..589b707 100644 --- a/modules/orchestrator/orchestrator-service/src/main/java/org/apache/airavata/orchestrator/server/OrchestratorServerHandler.java +++ b/modules/orchestrator/orchestrator-service/src/main/java/org/apache/airavata/orchestrator/server/OrchestratorServerHandler.java @@ -22,9 +22,12 @@ package org.apache.airavata.orchestrator.server; import org.apache.airavata.common.exception.AiravataException; +import org.apache.airavata.common.exception.ApplicationSettingsException; +import org.apache.airavata.common.utils.AiravataUtils; import org.apache.airavata.common.utils.ServerSettings; import org.apache.airavata.credential.store.store.CredentialReader; import org.apache.airavata.gfac.core.GFacUtils; +import org.apache.airavata.gfac.core.scheduler.HostScheduler; import org.apache.airavata.messaging.core.MessageContext; import org.apache.airavata.messaging.core.MessageHandler; import org.apache.airavata.messaging.core.MessagingConstants; @@ -32,38 +35,39 @@ import org.apache.airavata.messaging.core.Publisher; import org.apache.airavata.messaging.core.PublisherFactory; import org.apache.airavata.messaging.core.impl.RabbitMQProcessConsumer; import org.apache.airavata.messaging.core.impl.RabbitMQProcessPublisher; +import org.apache.airavata.model.appcatalog.appdeployment.ApplicationDeploymentDescription; import org.apache.airavata.model.appcatalog.appinterface.ApplicationInterfaceDescription; +import org.apache.airavata.model.appcatalog.computeresource.ComputeResourceDescription; import org.apache.airavata.model.error.LaunchValidationException; import org.apache.airavata.model.experiment.ExperimentModel; import org.apache.airavata.model.experiment.ExperimentType; +import org.apache.airavata.model.messaging.event.ExperimentStatusChangeEvent; +import org.apache.airavata.model.messaging.event.MessageType; import org.apache.airavata.model.messaging.event.ProcessSubmitEvent; import org.apache.airavata.model.process.ProcessModel; -import org.apache.airavata.model.task.TaskModel; +import org.apache.airavata.model.status.ExperimentState; +import org.apache.airavata.model.status.ExperimentStatus; import org.apache.airavata.orchestrator.core.exception.OrchestratorException; import org.apache.airavata.orchestrator.cpi.OrchestratorService; import org.apache.airavata.orchestrator.cpi.impl.SimpleOrchestratorImpl; import org.apache.airavata.orchestrator.cpi.orchestrator_cpi_serviceConstants; import org.apache.airavata.orchestrator.util.OrchestratorServerThreadPoolExecutor; +import org.apache.airavata.registry.core.app.catalog.resources.AppCatAbstractResource; import org.apache.airavata.registry.core.experiment.catalog.impl.RegistryFactory; import org.apache.airavata.registry.core.experiment.catalog.resources.AbstractExpCatResource; -import org.apache.airavata.registry.cpi.AppCatalog; -import org.apache.airavata.registry.cpi.AppCatalogException; -import org.apache.airavata.registry.cpi.ExperimentCatalog; -import org.apache.airavata.registry.cpi.ExperimentCatalogModelType; -import org.apache.airavata.registry.cpi.RegistryException; +import org.apache.airavata.registry.cpi.*; import org.apache.thrift.TBase; import org.apache.thrift.TException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.HashMap; -import java.util.List; -import java.util.Map; +import java.util.*; public class OrchestratorServerHandler implements OrchestratorService.Iface { private static Logger log = LoggerFactory.getLogger(OrchestratorServerHandler.class); private SimpleOrchestratorImpl orchestrator = null; private ExperimentCatalog experimentCatalog; + private AppCatalog appCatalog; private static Integer mutex = new Integer(-1); private String airavataUserName; private String gatewayName; @@ -79,7 +83,6 @@ public class OrchestratorServerHandler implements OrchestratorService.Iface { } public OrchestratorServerHandler() throws OrchestratorException{ - // registering with zk try { publisher = PublisherFactory.createActivityPublisher(); setAiravataUserName(ServerSettings.getDefaultUser()); @@ -93,6 +96,7 @@ public class OrchestratorServerHandler implements OrchestratorService.Iface { // the required properties orchestrator = new SimpleOrchestratorImpl(); experimentCatalog = RegistryFactory.getDefaultExpCatalog(); + appCatalog = RegistryFactory.getAppCatalog(); orchestrator.initialize(); orchestrator.getOrchestratorContext().setPublisher(this.publisher); startProcessConsumer(); @@ -102,8 +106,11 @@ public class OrchestratorServerHandler implements OrchestratorService.Iface { } catch (RegistryException e) { log.error(e.getMessage(), e); throw new OrchestratorException("Error while initializing orchestrator service", e); - } - } + } catch (AppCatalogException e) { + log.error(e.getMessage(), e); + throw new OrchestratorException("Error while initializing orchestrator service", e); + } + } private void startProcessConsumer() throws OrchestratorException { try { @@ -111,7 +118,6 @@ public class OrchestratorServerHandler implements OrchestratorService.Iface { ProcessConsumer processConsumer = new ProcessConsumer(); Thread thread = new Thread(processConsumer); thread.start(); - } catch (AiravataException e) { throw new OrchestratorException("Error while starting process consumer", e); } @@ -177,41 +183,16 @@ public class OrchestratorServerHandler implements OrchestratorService.Iface { * @return * @throws TException */ - public boolean validateExperiment(String experimentId) throws TException, - LaunchValidationException { - // TODO: Write the Orchestrator implementaion + public boolean validateExperiment(String experimentId) throws TException, LaunchValidationException { try { - List<TaskModel> tasks = orchestrator.createTasks(experimentId); - if (tasks.size() > 1) { - log.info("There are multiple tasks for this experiment, So Orchestrator will launch multiple Jobs"); - } - List<String> ids = experimentCatalog.getIds( - ExperimentCatalogModelType.PROCESS, - AbstractExpCatResource.ProcessConstants.EXPERIMENT_ID, experimentId); - for (String processId : ids) { - ProcessModel processModel = (ProcessModel) experimentCatalog - .get(ExperimentCatalogModelType.PROCESS, - processId); - // FIXME : no need to create tasks at orchestrator level -// List<Object> taskDetailList = experimentCatalog.get( -// ExperimentCatalogModelType.TASK_DETAIL, -//// TaskDetailConstants.NODE_ID, processId); -// for (Object o : taskDetailList) { -// TaskDetails taskID = (TaskDetails) o; -// // iterate through all the generated tasks and performs the -// // job submisssion+monitoring -// Experiment experiment = (Experiment) experimentCatalog.get( -// ExperimentCatalogModelType.EXPERIMENT, experimentId); -// if (experiment == null) { -// log.errorId(experimentId, "Error retrieving the Experiment by the given experimentID: {}.", -// experimentId); -// return false; -// } -// return orchestrator.validateExperiment(experiment, -// processModel, taskID).isSetValidationState(); -// } - } - + ExperimentModel experimentModel = (ExperimentModel)experimentCatalog.get(ExperimentCatalogModelType.EXPERIMENT, experimentId); + List<String> ids = experimentCatalog.getIds(ExperimentCatalogModelType.PROCESS,AbstractExpCatResource.ProcessConstants.EXPERIMENT_ID, experimentId); + if (ids != null && !ids.isEmpty()){ + for (String processId : ids) { + ProcessModel processModel = (ProcessModel) experimentCatalog.get(ExperimentCatalogModelType.PROCESS,processId); + return orchestrator.validateExperiment(experimentModel,processModel).isSetValidationState(); + } + } } catch (OrchestratorException e) { log.error(experimentId, "Error while validating experiment", e); throw new TException(e); @@ -252,79 +233,58 @@ public class OrchestratorServerHandler implements OrchestratorService.Iface { } @Override - public boolean launchTask(String taskId, String airavataCredStoreToken) throws TException { - // FIXME : should be launch process instead of the task -// try { -// TaskDetails taskData = (TaskDetails) experimentCatalog.get( -// ExperimentCatalogModelType.TASK_DETAIL, taskId); -// String applicationId = taskData.getApplicationId(); -// if (applicationId == null) { -// log.errorId(taskId, "Application id shouldn't be null."); -// throw new OrchestratorException("Error executing the job, application id shouldn't be null."); -// } -// ApplicationDeploymentDescription applicationDeploymentDescription = getAppDeployment(taskData, applicationId); -// taskData.setApplicationDeploymentId(applicationDeploymentDescription.getAppDeploymentId()); -// experimentCatalog.update(ExperimentCatalogModelType.TASK_DETAIL, taskData,taskData.getTaskID()); -// List<Object> workflowNodeDetailList = experimentCatalog.get(ExperimentCatalogModelType.WORKFLOW_NODE_DETAIL, -// org.apache.airavata.registry.cpi.utils.Constants.FieldConstants.WorkflowNodeConstants.TASK_LIST, taskData); -// if (workflowNodeDetailList != null -// && workflowNodeDetailList.size() > 0) { -// List<Object> experimentList = experimentCatalog.get(ExperimentCatalogModelType.EXPERIMENT, -// org.apache.airavata.registry.cpi.utils.Constants.FieldConstants.ExperimentConstants.WORKFLOW_NODE_LIST, -// (WorkflowNodeDetails) workflowNodeDetailList.get(0)); -// if (experimentList != null && experimentList.size() > 0) { -// return orchestrator -// .launchExperiment( -// (Experiment) experimentList.get(0), -// (WorkflowNodeDetails) workflowNodeDetailList -// .get(0), taskData,airavataCredStoreToken); -// } -// } -// } catch (Exception e) { -// log.errorId(taskId, "Error while launching task ", e); -// throw new TException(e); -// } -// log.infoId(taskId, "No experiment found associated in task {}", taskId); - return false; + public boolean launchProcess(String processId, String airavataCredStoreToken) throws TException { + try { + ProcessModel processModel = (ProcessModel) experimentCatalog.get( + ExperimentCatalogModelType.PROCESS, processId); + String experimentId = processModel.getExperimentId(); + ExperimentModel experimentModel = (ExperimentModel)experimentCatalog.get(ExperimentCatalogModelType.EXPERIMENT, experimentId); + String applicationId = processModel.getApplicationInterfaceId(); + if (applicationId == null) { + log.error(processId, "Application interface id shouldn't be null."); + throw new OrchestratorException("Error executing the job, application interface id shouldn't be null."); + } + ApplicationDeploymentDescription applicationDeploymentDescription = getAppDeployment(processModel, applicationId); + processModel.setApplicationDeploymentId(applicationDeploymentDescription.getAppDeploymentId()); + experimentCatalog.update(ExperimentCatalogModelType.PROCESS, processModel,processModel.getProcessId()); + return orchestrator.launchExperiment(experimentModel, processModel, airavataCredStoreToken); + } catch (Exception e) { + log.error(processId, "Error while launching process ", e); + throw new TException(e); + } } -// private ApplicationDeploymentDescription getAppDeployment( -// TaskDetails taskData, String applicationId) -// throws AppCatalogException, OrchestratorException, -// ClassNotFoundException, ApplicationSettingsException, -// InstantiationException, IllegalAccessException { -// AppCatalog appCatalog = RegistryFactory.getAppCatalog(); -// String selectedModuleId = getModuleId(appCatalog, applicationId); -// ApplicationDeploymentDescription applicationDeploymentDescription = getAppDeployment( -// appCatalog, taskData, selectedModuleId); -// return applicationDeploymentDescription; -// } - -// private ApplicationDeploymentDescription getAppDeployment( -// AppCatalog appCatalog, TaskDetails taskData, String selectedModuleId) -// throws AppCatalogException, ClassNotFoundException, -// ApplicationSettingsException, InstantiationException, -// IllegalAccessException { -// Map<String, String> moduleIdFilter = new HashMap<String, String>(); -// moduleIdFilter.put(AppCatAbstractResource.ApplicationDeploymentConstants.APP_MODULE_ID, selectedModuleId); -// if (taskData.getTaskScheduling()!=null && taskData.getTaskScheduling().getResourceHostId() != null) { -// moduleIdFilter.put(AppCatAbstractResource.ApplicationDeploymentConstants.COMPUTE_HOST_ID, taskData.getTaskScheduling().getResourceHostId()); -// } -// List<ApplicationDeploymentDescription> applicationDeployements = appCatalog.getApplicationDeployment().getApplicationDeployements(moduleIdFilter); -// Map<ComputeResourceDescription, ApplicationDeploymentDescription> deploymentMap = new HashMap<ComputeResourceDescription, ApplicationDeploymentDescription>(); -// ComputeResource computeResource = appCatalog.getComputeResource(); -// for (ApplicationDeploymentDescription deploymentDescription : applicationDeployements) { -// deploymentMap.put(computeResource.getComputeResource(deploymentDescription.getComputeHostId()),deploymentDescription); -// } -// List<ComputeResourceDescription> computeHostList = Arrays.asList(deploymentMap.keySet().toArray(new ComputeResourceDescription[]{})); -// Class<? extends HostScheduler> aClass = Class.forName( -// ServerSettings.getHostScheduler()).asSubclass( -// HostScheduler.class); -// HostScheduler hostScheduler = aClass.newInstance(); -// ComputeResourceDescription ComputeResourceDescription = hostScheduler.schedule(computeHostList); -// ApplicationDeploymentDescription applicationDeploymentDescription = deploymentMap.get(ComputeResourceDescription); -// return applicationDeploymentDescription; -// } + private ApplicationDeploymentDescription getAppDeployment(ProcessModel processModel, String applicationId) + throws AppCatalogException, OrchestratorException, + ClassNotFoundException, ApplicationSettingsException, + InstantiationException, IllegalAccessException { + String selectedModuleId = getModuleId(appCatalog, applicationId); + return getAppDeploymentForModule(processModel, selectedModuleId); + } + + private ApplicationDeploymentDescription getAppDeploymentForModule(ProcessModel processModel, String selectedModuleId) + throws AppCatalogException, ClassNotFoundException, + ApplicationSettingsException, InstantiationException, + IllegalAccessException { + Map<String, String> moduleIdFilter = new HashMap<String, String>(); + moduleIdFilter.put(AppCatAbstractResource.ApplicationDeploymentConstants.APP_MODULE_ID, selectedModuleId); + if (processModel.getResourceSchedule() != null && processModel.getResourceSchedule().getResourceHostId() != null) { + moduleIdFilter.put(AppCatAbstractResource.ApplicationDeploymentConstants.COMPUTE_HOST_ID, processModel.getResourceSchedule().getResourceHostId()); + } + List<ApplicationDeploymentDescription> applicationDeployements = appCatalog.getApplicationDeployment().getApplicationDeployements(moduleIdFilter); + Map<ComputeResourceDescription, ApplicationDeploymentDescription> deploymentMap = new HashMap<ComputeResourceDescription, ApplicationDeploymentDescription>(); + ComputeResource computeResource = appCatalog.getComputeResource(); + for (ApplicationDeploymentDescription deploymentDescription : applicationDeployements) { + deploymentMap.put(computeResource.getComputeResource(deploymentDescription.getComputeHostId()), deploymentDescription); + } + List<ComputeResourceDescription> computeHostList = Arrays.asList(deploymentMap.keySet().toArray(new ComputeResourceDescription[]{})); + Class<? extends HostScheduler> aClass = Class.forName( + ServerSettings.getHostScheduler()).asSubclass( + HostScheduler.class); + HostScheduler hostScheduler = aClass.newInstance(); + ComputeResourceDescription ComputeResourceDescription = hostScheduler.schedule(computeHostList); + return deploymentMap.get(ComputeResourceDescription); + } private String getModuleId(AppCatalog appCatalog, String applicationId) throws AppCatalogException, OrchestratorException { @@ -495,65 +455,45 @@ public class OrchestratorServerHandler implements OrchestratorService.Iface { } } - // FIXME private boolean launchSingleAppExperiment() throws TException { -// ExperimentModel experiment = null; -// try { -// List<String> ids = experimentCatalog.getIds(ExperimentCatalogModelType.WORKFLOW_NODE_DETAIL, WorkflowNodeConstants.EXPERIMENT_ID, experimentId); -// for (String workflowNodeId : ids) { -//// WorkflowNodeDetails workflowNodeDetail = (WorkflowNodeDetails) registry.get(RegistryModelType.WORKFLOW_NODE_DETAIL, workflowNodeId); -// List<Object> taskDetailList = experimentCatalog.get(ExperimentCatalogModelType.TASK_DETAIL, TaskDetailConstants.NODE_ID, workflowNodeId); -// for (Object o : taskDetailList) { -// TaskDetails taskData = (TaskDetails) o; -// //iterate through all the generated tasks and performs the job submisssion+monitoring -// experiment = (Experiment) experimentCatalog.get(ExperimentCatalogModelType.EXPERIMENT, experimentId); -// if (experiment == null) { -// log.errorId(experimentId, "Error retrieving the Experiment by the given experimentID: {}", experimentId); -// return false; -// } -// String gatewayId = null; -// CredentialReader credentialReader = GFacUtils.getCredentialReader(); -// if (credentialReader != null) { -// try { -// gatewayId = credentialReader.getGatewayID(airavataCredStoreToken); -// } catch (Exception e) { -// log.error(e.getLocalizedMessage()); -// } -// } -// if (gatewayId == null || gatewayId.isEmpty()) { -// gatewayId = ServerSettings.getDefaultUserGateway(); -// } -// ExperimentStatusChangeEvent event = new ExperimentStatusChangeEvent(ExperimentState.LAUNCHED, -// experimentId, -// gatewayId); -// String messageId = AiravataUtils.getId("EXPERIMENT"); -// MessageContext messageContext = new MessageContext(event, MessageType.EXPERIMENT, messageId, gatewayId); -// messageContext.setUpdatedTime(AiravataUtils.getCurrentTimestamp()); -// publisher.publish(messageContext); -// experimentCatalog.update(ExperimentCatalogModelType.TASK_DETAIL, taskData, taskData.getTaskID()); -// //launching the experiment -// launchTask(taskData.getTaskID(), airavataCredStoreToken); -// } -// } -// -// } catch (Exception e) { -// // Here we really do not have to do much because only potential failure can happen -// // is in gfac, if there are errors in gfac, it will handle the experiment/task/job statuses -// // We might get failures in registry access before submitting the jobs to gfac, in that case we -// // leave the status of these as created. -// ExperimentStatus status = new ExperimentStatus(); -// status.setExperimentState(ExperimentState.FAILED); -// status.setTimeOfStateChange(Calendar.getInstance().getTimeInMillis()); -// experiment.setExperimentStatus(status); -// try { -// experimentCatalog.update(ExperimentCatalogModelType.EXPERIMENT_STATUS, status, experimentId); -// } catch (RegistryException e1) { -// log.errorId(experimentId, "Error while updating experiment status to " + status.toString(), e); -// throw new TException(e); -// } -// log.errorId(experimentId, "Error while updating task status, hence updated experiment status to " + status.toString(), e); -// throw new TException(e); -// } + try { + List<String> processIds = experimentCatalog.getIds(ExperimentCatalogModelType.PROCESS, AbstractExpCatResource.ProcessConstants.EXPERIMENT_ID, experimentId); + for (String processId : processIds) { + String gatewayId = null; + CredentialReader credentialReader = GFacUtils.getCredentialReader(); + if (credentialReader != null) { + try { + gatewayId = credentialReader.getGatewayID(airavataCredStoreToken); + } catch (Exception e) { + log.error(e.getLocalizedMessage()); + } + } + if (gatewayId == null || gatewayId.isEmpty()) { + gatewayId = ServerSettings.getDefaultUserGateway(); + } + ExperimentStatusChangeEvent event = new ExperimentStatusChangeEvent(ExperimentState.LAUNCHED, + experimentId, + gatewayId); + String messageId = AiravataUtils.getId("EXPERIMENT"); + MessageContext messageContext = new MessageContext(event, MessageType.EXPERIMENT, messageId, gatewayId); + messageContext.setUpdatedTime(AiravataUtils.getCurrentTimestamp()); + publisher.publish(messageContext); + launchProcess(processId, airavataCredStoreToken); + } + + } catch (Exception e) { + ExperimentStatus status = new ExperimentStatus(); + status.setState(ExperimentState.FAILED); + status.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime()); + try { + experimentCatalog.update(ExperimentCatalogModelType.EXPERIMENT_STATUS, status, experimentId); + } catch (RegistryException e1) { + log.error(experimentId, "Error while updating experiment status to " + status.toString(), e); + throw new TException(e); + } + log.error(experimentId, "Error while updating task status, hence updated experiment status to " + status.toString(), e); + throw new TException(e); + } return true; } } @@ -584,9 +524,9 @@ public class OrchestratorServerHandler implements OrchestratorService.Iface { if (event instanceof ProcessSubmitEvent) { ProcessSubmitEvent processSubmitEvent = (ProcessSubmitEvent) event; try { - launchTask(processSubmitEvent.getTaskId(), processSubmitEvent.getCredentialToken()); + launchProcess(processSubmitEvent.getProcessId(), processSubmitEvent.getCredentialToken()); } catch (TException e) { - log.error("Error while launching task : " + processSubmitEvent.getTaskId()); + log.error("Error while launching task : " + processSubmitEvent.getProcessId()); } } } http://git-wip-us.apache.org/repos/asf/airavata/blob/636a733b/thrift-interface-descriptions/airavata-api/messaging_events.thrift ---------------------------------------------------------------------- diff --git a/thrift-interface-descriptions/airavata-api/messaging_events.thrift b/thrift-interface-descriptions/airavata-api/messaging_events.thrift index e20d8ba..a863edc 100644 --- a/thrift-interface-descriptions/airavata-api/messaging_events.thrift +++ b/thrift-interface-descriptions/airavata-api/messaging_events.thrift @@ -110,7 +110,7 @@ struct JobIdentifier { // } struct ProcessSubmitEvent{ - 1: required string taskId; + 1: required string processId; 2: required string credentialToken; } http://git-wip-us.apache.org/repos/asf/airavata/blob/636a733b/thrift-interface-descriptions/orchestrator-cpi/orchestrator.cpi.service.thrift ---------------------------------------------------------------------- diff --git a/thrift-interface-descriptions/orchestrator-cpi/orchestrator.cpi.service.thrift b/thrift-interface-descriptions/orchestrator-cpi/orchestrator.cpi.service.thrift index 62a6d8d..b24de58 100644 --- a/thrift-interface-descriptions/orchestrator-cpi/orchestrator.cpi.service.thrift +++ b/thrift-interface-descriptions/orchestrator-cpi/orchestrator.cpi.service.thrift @@ -26,7 +26,7 @@ include "../airavata-api/airavata_errors.thrift" namespace java org.apache.airavata.orchestrator.cpi -const string ORCHESTRATOR_CPI_VERSION = "0.13.0" +const string ORCHESTRATOR_CPI_VERSION = "0.16.0" service OrchestratorService { @@ -46,15 +46,15 @@ service OrchestratorService { /** * In order to run single applications users should create an associating - * WorkflowNodeDetails and a TaskDetails for it and hand it over for execution + * process and hand it over for execution * along with a credential store token for authentication * - * @param taskId + * @param processId * @param airavataCredStoreToken * @return sucess/failure * **/ - bool launchTask (1: required string taskId, 2: required string airavataCredStoreToken), + bool launchProcess (1: required string processId, 2: required string airavataCredStoreToken), /** *
