You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by sh...@apache.org on 2015/06/04 22:16:31 UTC
[78/81] [abbrv] airavata git commit: Merge moduleRefactor branch
http://git-wip-us.apache.org/repos/asf/airavata/blob/b4ede9cb/modules/orchestrator/orchestrator-service/src/main/java/org/apache/airavata/orchestrator/server/OrchestratorServerHandler.java
----------------------------------------------------------------------
diff --cc modules/orchestrator/orchestrator-service/src/main/java/org/apache/airavata/orchestrator/server/OrchestratorServerHandler.java
index 0000000,389913b..8f94633
mode 000000,100644..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
@@@ -1,0 -1,643 +1,643 @@@
+ /*
+ *
+ * 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.orchestrator.server;
+
+ import org.airavata.appcatalog.cpi.AppCatalog;
+ import org.airavata.appcatalog.cpi.AppCatalogException;
+ import org.airavata.appcatalog.cpi.ComputeResource;
+ import org.apache.aiaravata.application.catalog.data.impl.AppCatalogFactory;
+ import org.apache.aiaravata.application.catalog.data.resources.AbstractResource;
+ import org.apache.airavata.common.exception.AiravataException;
+ import org.apache.airavata.common.exception.ApplicationSettingsException;
+ import org.apache.airavata.common.logger.AiravataLogger;
+ import org.apache.airavata.common.logger.AiravataLoggerFactory;
+ import org.apache.airavata.common.utils.AiravataUtils;
+ import org.apache.airavata.common.utils.AiravataZKUtils;
+ import org.apache.airavata.common.utils.Constants;
+ import org.apache.airavata.common.utils.ServerSettings;
+ import org.apache.airavata.credential.store.store.CredentialReader;
+ import org.apache.airavata.gfac.core.scheduler.HostScheduler;
+ import org.apache.airavata.gfac.core.GFacUtils;
+ import org.apache.airavata.messaging.core.MessageContext;
+ import org.apache.airavata.messaging.core.MessageHandler;
+ import org.apache.airavata.messaging.core.MessagingConstants;
+ 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.messaging.event.ExperimentStatusChangeEvent;
+ import org.apache.airavata.model.messaging.event.MessageType;
+ import org.apache.airavata.model.messaging.event.ProcessSubmitEvent;
+ import org.apache.airavata.model.util.ExecutionType;
+ import org.apache.airavata.model.workspace.experiment.Experiment;
+ import org.apache.airavata.model.workspace.experiment.ExperimentState;
+ import org.apache.airavata.model.workspace.experiment.ExperimentStatus;
+ import org.apache.airavata.model.workspace.experiment.TaskDetails;
+ import org.apache.airavata.model.workspace.experiment.TaskState;
+ import org.apache.airavata.model.workspace.experiment.TaskStatus;
+ import org.apache.airavata.model.workspace.experiment.WorkflowNodeDetails;
+ import org.apache.airavata.model.workspace.experiment.WorkflowNodeState;
+ import org.apache.airavata.model.workspace.experiment.WorkflowNodeStatus;
+ 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.DataModelUtils;
+ import org.apache.airavata.orchestrator.util.OrchestratorServerThreadPoolExecutor;
-import org.apache.airavata.persistance.registry.jpa.impl.RegistryFactory;
++import org.apache.airavata.experiment.catalog.impl.RegistryFactory;
+ import org.apache.airavata.registry.cpi.Registry;
+ import org.apache.airavata.registry.cpi.RegistryException;
+ import org.apache.airavata.registry.cpi.RegistryModelType;
+ import org.apache.airavata.registry.cpi.utils.Constants.FieldConstants.TaskDetailConstants;
+ import org.apache.airavata.registry.cpi.utils.Constants.FieldConstants.WorkflowNodeConstants;
+ import org.apache.airavata.workflow.core.WorkflowEnactmentService;
+ import org.apache.thrift.TBase;
+ import org.apache.thrift.TException;
+
+ import java.util.Arrays;
+ import java.util.Calendar;
+ import java.util.HashMap;
+ import java.util.List;
+ import java.util.Map;
+
+ public class OrchestratorServerHandler implements OrchestratorService.Iface {
+ private static AiravataLogger log = AiravataLoggerFactory .getLogger(OrchestratorServerHandler.class);
+ private SimpleOrchestratorImpl orchestrator = null;
+ private Registry registry;
+ private static Integer mutex = new Integer(-1);
+ private String airavataUserName;
+ private String gatewayName;
+ private Publisher publisher;
+ private RabbitMQProcessConsumer rabbitMQProcessConsumer;
+ private RabbitMQProcessPublisher rabbitMQProcessPublisher;
+
+ /**
+ * Query orchestrator server to fetch the CPI version
+ */
+ public String getOrchestratorCPIVersion() throws TException {
+ return orchestrator_cpi_serviceConstants.ORCHESTRATOR_CPI_VERSION;
+ }
+
+ public OrchestratorServerHandler() throws OrchestratorException{
+ // registering with zk
+ try {
+ publisher = PublisherFactory.createActivityPublisher();
+ String zkhostPort = AiravataZKUtils.getZKhostPort();
+ String airavataServerHostPort = ServerSettings
+ .getSetting(Constants.ORCHESTRATOR_SERVER_HOST)
+ + ":"
+ + ServerSettings
+ .getSetting(Constants.ORCHESTRATOR_SERVER_PORT);
+
+ // setGatewayName(ServerSettings.getDefaultUserGateway());
+ setAiravataUserName(ServerSettings.getDefaultUser());
+ } catch (AiravataException e) {
+ log.error(e.getMessage(), e);
+ throw new OrchestratorException("Error while initializing orchestrator service", e);
+ }
+ // orchestrator init
+ try {
+ // first constructing the monitorManager and orchestrator, then fill
+ // the required properties
+ orchestrator = new SimpleOrchestratorImpl();
+ registry = RegistryFactory.getDefaultRegistry();
+ orchestrator.initialize();
+ orchestrator.getOrchestratorContext().setPublisher(this.publisher);
+ startProcessConsumer();
+ } catch (OrchestratorException e) {
+ log.error(e.getMessage(), e);
+ throw new OrchestratorException("Error while initializing orchestrator service", e);
+ } catch (RegistryException e) {
+ log.error(e.getMessage(), e);
+ throw new OrchestratorException("Error while initializing orchestrator service", e);
+ }
+ }
+
+ private void startProcessConsumer() throws OrchestratorException {
+ try {
+ rabbitMQProcessConsumer = new RabbitMQProcessConsumer();
+ ProcessConsumer processConsumer = new ProcessConsumer();
+ Thread thread = new Thread(processConsumer);
+ thread.start();
+
+ } catch (AiravataException e) {
+ throw new OrchestratorException("Error while starting process consumer", e);
+ }
+
+ }
+
+ /**
+ * * After creating the experiment Data user have the * experimentID as the
+ * handler to the experiment, during the launchExperiment * We just have to
+ * give the experimentID * * @param experimentID * @return sucess/failure *
+ * *
+ *
+ * @param experimentId
+ */
+ public boolean launchExperiment(String experimentId, String token) throws TException {
+ Experiment experiment = null; // this will inside the bottom catch statement
+ try {
+ experiment = (Experiment) registry.get(
+ RegistryModelType.EXPERIMENT, experimentId);
+ if (experiment == null) {
+ log.errorId(experimentId, "Error retrieving the Experiment by the given experimentID: {} ", experimentId);
+ return false;
+ }
+ CredentialReader credentialReader = GFacUtils.getCredentialReader();
+ String gatewayId = null;
+ if (credentialReader != null) {
+ try {
+ gatewayId = credentialReader.getGatewayID(token);
+ } catch (Exception e) {
+ log.error(e.getLocalizedMessage());
+ }
+ }
+ if (gatewayId == null) {
+ gatewayId = ServerSettings.getDefaultUserGateway();
+ log.info("Couldn't identify the gateway Id using the credential token, Use default gateway Id");
+ // throw new AiravataException("Couldn't identify the gateway Id using the credential token");
+ }
+ ExecutionType executionType = DataModelUtils.getExecutionType(gatewayId, experiment);
+ if (executionType == ExecutionType.SINGLE_APP) {
+ //its an single application execution experiment
+ log.debugId(experimentId, "Launching single application experiment {}.", experimentId);
+ OrchestratorServerThreadPoolExecutor.getCachedThreadPool().execute(new SingleAppExperimentRunner(experimentId, token));
+ } else if (executionType == ExecutionType.WORKFLOW) {
+ //its a workflow execution experiment
+ log.debugId(experimentId, "Launching workflow experiment {}.", experimentId);
+ launchWorkflowExperiment(experimentId, token);
+ } else {
+ log.errorId(experimentId, "Couldn't identify experiment type, experiment {} is neither single application nor workflow.", experimentId);
+ throw new TException("Experiment '" + experimentId + "' launch failed. Unable to figureout execution type for application " + experiment.getApplicationId());
+ }
+ } catch (Exception e) {
+ throw new TException("Experiment '" + experimentId + "' launch failed. Unable to figureout execution type for application " + experiment.getApplicationId(), e);
+ }
+ return true;
+ }
+
+ /**
+ * This method will validate the experiment before launching, if is failed
+ * we do not run the launch in airavata thrift service (only if validation
+ * is enabled
+ *
+ * @param experimentId
+ * @return
+ * @throws TException
+ */
+ public boolean validateExperiment(String experimentId) throws TException,
+ LaunchValidationException {
+ // TODO: Write the Orchestrator implementaion
+ try {
+ List<TaskDetails> 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 = registry.getIds(
+ RegistryModelType.WORKFLOW_NODE_DETAIL,
+ WorkflowNodeConstants.EXPERIMENT_ID, experimentId);
+ for (String workflowNodeId : ids) {
+ WorkflowNodeDetails workflowNodeDetail = (WorkflowNodeDetails) registry
+ .get(RegistryModelType.WORKFLOW_NODE_DETAIL,
+ workflowNodeId);
+ List<Object> taskDetailList = registry.get(
+ RegistryModelType.TASK_DETAIL,
+ TaskDetailConstants.NODE_ID, workflowNodeId);
+ for (Object o : taskDetailList) {
+ TaskDetails taskID = (TaskDetails) o;
+ // iterate through all the generated tasks and performs the
+ // job submisssion+monitoring
+ Experiment experiment = (Experiment) registry.get(
+ RegistryModelType.EXPERIMENT, experimentId);
+ if (experiment == null) {
+ log.errorId(experimentId, "Error retrieving the Experiment by the given experimentID: {}.",
+ experimentId);
+ return false;
+ }
+ return orchestrator.validateExperiment(experiment,
+ workflowNodeDetail, taskID).isSetValidationState();
+ }
+ }
+
+ } catch (OrchestratorException e) {
+ log.errorId(experimentId, "Error while validating experiment", e);
+ throw new TException(e);
+ } catch (RegistryException e) {
+ log.errorId(experimentId, "Error while validating experiment", e);
+ throw new TException(e);
+ }
+ return false;
+ }
+
+ /**
+ * This can be used to cancel a running experiment and store the status to
+ * terminated in registry
+ *
+ * @param experimentId
+ * @return
+ * @throws TException
+ */
+ public boolean terminateExperiment(String experimentId, String tokenId) throws TException {
+ log.infoId(experimentId, "Experiment: {} is cancelling !!!!!", experimentId);
+ return validateStatesAndCancel(experimentId, tokenId);
+ }
+
+ private String getAiravataUserName() {
+ return airavataUserName;
+ }
+
+ private String getGatewayName() {
+ return gatewayName;
+ }
+
+ public void setAiravataUserName(String airavataUserName) {
+ this.airavataUserName = airavataUserName;
+ }
+
+ public void setGatewayName(String gatewayName) {
+ this.gatewayName = gatewayName;
+ }
+
+ @Override
+ public boolean launchTask(String taskId, String airavataCredStoreToken) throws TException {
+ try {
+ TaskDetails taskData = (TaskDetails) registry.get(
+ RegistryModelType.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());
+ registry.update(RegistryModelType.TASK_DETAIL, taskData,taskData.getTaskID());
+ List<Object> workflowNodeDetailList = registry.get(RegistryModelType.WORKFLOW_NODE_DETAIL,
+ org.apache.airavata.registry.cpi.utils.Constants.FieldConstants.WorkflowNodeConstants.TASK_LIST, taskData);
+ if (workflowNodeDetailList != null
+ && workflowNodeDetailList.size() > 0) {
+ List<Object> experimentList = registry.get(RegistryModelType.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;
+ }
+
+ private ApplicationDeploymentDescription getAppDeployment(
+ TaskDetails taskData, String applicationId)
+ throws AppCatalogException, OrchestratorException,
+ ClassNotFoundException, ApplicationSettingsException,
+ InstantiationException, IllegalAccessException {
+ AppCatalog appCatalog = AppCatalogFactory.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(AbstractResource.ApplicationDeploymentConstants.APP_MODULE_ID, selectedModuleId);
+ if (taskData.getTaskScheduling()!=null && taskData.getTaskScheduling().getResourceHostId() != null) {
+ moduleIdFilter.put(AbstractResource.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 String getModuleId(AppCatalog appCatalog, String applicationId)
+ throws AppCatalogException, OrchestratorException {
+ ApplicationInterfaceDescription applicationInterface = appCatalog.getApplicationInterface().getApplicationInterface(applicationId);
+ List<String> applicationModules = applicationInterface.getApplicationModules();
+ if (applicationModules.size()==0){
+ throw new OrchestratorException(
+ "No modules defined for application "
+ + applicationId);
+ }
+ // AiravataAPI airavataAPI = getAiravataAPI();
+ String selectedModuleId=applicationModules.get(0);
+ return selectedModuleId;
+ }
+
+ private boolean validateStatesAndCancel(String experimentId, String tokenId)throws TException{
+ try {
+ Experiment experiment = (Experiment) registry.get(
+ RegistryModelType.EXPERIMENT, experimentId);
+ log.info("Waiting for zookeeper to connect to the server");
+ synchronized (mutex){
+ mutex.wait(5000);
+ }
+ if (experiment == null) {
+ log.errorId(experimentId, "Error retrieving the Experiment by the given experimentID: {}.", experimentId);
+ throw new OrchestratorException("Error retrieving the Experiment by the given experimentID: " + experimentId);
+ }
+ ExperimentState experimentState = experiment.getExperimentStatus().getExperimentState();
+ if (isCancelValid(experimentState)){
+ ExperimentStatus status = new ExperimentStatus();
+ status.setExperimentState(ExperimentState.CANCELING);
+ status.setTimeOfStateChange(Calendar.getInstance()
+ .getTimeInMillis());
+ experiment.setExperimentStatus(status);
+ registry.update(RegistryModelType.EXPERIMENT, experiment,
+ experimentId);
+
+ List<String> ids = registry.getIds(
+ RegistryModelType.WORKFLOW_NODE_DETAIL,
+ WorkflowNodeConstants.EXPERIMENT_ID, experimentId);
+ for (String workflowNodeId : ids) {
+ WorkflowNodeDetails workflowNodeDetail = (WorkflowNodeDetails) registry
+ .get(RegistryModelType.WORKFLOW_NODE_DETAIL,
+ workflowNodeId);
+ int value = workflowNodeDetail.getWorkflowNodeStatus().getWorkflowNodeState().getValue();
+ if ( value> 1 && value < 7) { // we skip the unknown state
+ log.error(workflowNodeDetail.getNodeName() + " Workflow Node status cannot mark as cancelled, because " +
+ "current status is " + workflowNodeDetail.getWorkflowNodeStatus().getWorkflowNodeState().toString());
+ continue; // this continue is very useful not to process deeper loops if the upper layers have non-cancel states
+ } else {
+ WorkflowNodeStatus workflowNodeStatus = new WorkflowNodeStatus();
+ workflowNodeStatus.setWorkflowNodeState(WorkflowNodeState.CANCELING);
+ workflowNodeStatus.setTimeOfStateChange(Calendar.getInstance()
+ .getTimeInMillis());
+ workflowNodeDetail.setWorkflowNodeStatus(workflowNodeStatus);
+ registry.update(RegistryModelType.WORKFLOW_NODE_DETAIL, workflowNodeDetail,
+ workflowNodeId);
+ }
+ List<Object> taskDetailList = registry.get(
+ RegistryModelType.TASK_DETAIL,
+ TaskDetailConstants.NODE_ID, workflowNodeId);
+ for (Object o : taskDetailList) {
+ TaskDetails taskDetails = (TaskDetails) o;
+ TaskStatus taskStatus = ((TaskDetails) o).getTaskStatus();
+ if (taskStatus.getExecutionState().getValue() > 7 && taskStatus.getExecutionState().getValue()<12) {
+ log.error(((TaskDetails) o).getTaskID() + " Task status cannot mark as cancelled, because " +
+ "current task state is " + ((TaskDetails) o).getTaskStatus().getExecutionState().toString());
+ continue;// this continue is very useful not to process deeper loops if the upper layers have non-cancel states
+ } else {
+ taskStatus.setExecutionState(TaskState.CANCELING);
+ taskStatus.setTimeOfStateChange(Calendar.getInstance()
+ .getTimeInMillis());
+ taskDetails.setTaskStatus(taskStatus);
+ registry.update(RegistryModelType.TASK_DETAIL, o,
+ taskDetails.getTaskID());
+ }
+ orchestrator.cancelExperiment(experiment,
+ workflowNodeDetail, taskDetails, tokenId);
+ // Status update should be done at the monitor
+ }
+ }
+ }else {
+ if (isCancelAllowed(experimentState)){
+ // when experiment status is < 3 no jobDetails object is created,
+ // so we don't have to worry, we simply have to change the status and stop the execution
+ ExperimentStatus status = new ExperimentStatus();
+ status.setExperimentState(ExperimentState.CANCELED);
+ status.setTimeOfStateChange(Calendar.getInstance()
+ .getTimeInMillis());
+ experiment.setExperimentStatus(status);
+ registry.update(RegistryModelType.EXPERIMENT, experiment,
+ experimentId);
+ List<String> ids = registry.getIds(
+ RegistryModelType.WORKFLOW_NODE_DETAIL,
+ WorkflowNodeConstants.EXPERIMENT_ID, experimentId);
+ for (String workflowNodeId : ids) {
+ WorkflowNodeDetails workflowNodeDetail = (WorkflowNodeDetails) registry
+ .get(RegistryModelType.WORKFLOW_NODE_DETAIL,
+ workflowNodeId);
+ WorkflowNodeStatus workflowNodeStatus = new WorkflowNodeStatus();
+ workflowNodeStatus.setWorkflowNodeState(WorkflowNodeState.CANCELED);
+ workflowNodeStatus.setTimeOfStateChange(Calendar.getInstance()
+ .getTimeInMillis());
+ workflowNodeDetail.setWorkflowNodeStatus(workflowNodeStatus);
+ registry.update(RegistryModelType.WORKFLOW_NODE_DETAIL, workflowNodeDetail,
+ workflowNodeId);
+ List<Object> taskDetailList = registry.get(
+ RegistryModelType.TASK_DETAIL,
+ TaskDetailConstants.NODE_ID, workflowNodeId);
+ for (Object o : taskDetailList) {
+ TaskDetails taskDetails = (TaskDetails) o;
+ TaskStatus taskStatus = ((TaskDetails) o).getTaskStatus();
+ taskStatus.setExecutionState(TaskState.CANCELED);
+ taskStatus.setTimeOfStateChange(Calendar.getInstance()
+ .getTimeInMillis());
+ taskDetails.setTaskStatus(taskStatus);
+ registry.update(RegistryModelType.TASK_DETAIL, o,
+ taskDetails);
+ }
+ }
+ }else {
+ log.errorId(experimentId, "Unable to mark experiment as Cancelled, current state {} doesn't allow to cancel the experiment {}.",
+ experiment.getExperimentStatus().getExperimentState().toString(), experimentId);
+ throw new OrchestratorException("Unable to mark experiment as Cancelled, because current state is: "
+ + experiment.getExperimentStatus().getExperimentState().toString());
+ }
+ }
+ log.info("Experiment: " + experimentId + " is cancelled !!!!!");
+ } catch (Exception e) {
+ throw new TException(e);
+ }
+ return true;
+ }
+
+ private boolean isCancelValid(ExperimentState state){
+ switch (state) {
+ case LAUNCHED:
+ case EXECUTING:
+ case CANCELING:
+ return true;
+ default:
+ return false;
+ }
+ }
+
+ private boolean isCancelAllowed(ExperimentState state){
+ switch (state) {
+ case CREATED:
+ case VALIDATED:
+ case SCHEDULED:
+ return true;
+ default:
+ return false;
+ }
+ }
+
+ private void launchWorkflowExperiment(String experimentId, String airavataCredStoreToken) throws TException {
+ try {
+ WorkflowEnactmentService.getInstance().
+ submitWorkflow(experimentId, airavataCredStoreToken, getGatewayName(), getRabbitMQProcessPublisher());
+ } catch (Exception e) {
+ log.error("Error while launching workflow", e);
+ }
+ }
+
+ public synchronized RabbitMQProcessPublisher getRabbitMQProcessPublisher() throws Exception {
+ if (rabbitMQProcessPublisher == null) {
+ rabbitMQProcessPublisher = new RabbitMQProcessPublisher();
+ }
+ return rabbitMQProcessPublisher;
+ }
+
+
+ private class SingleAppExperimentRunner implements Runnable {
+
+ String experimentId;
+ String airavataCredStoreToken;
+ public SingleAppExperimentRunner(String experimentId,String airavataCredStoreToken){
+ this.experimentId = experimentId;
+ this.airavataCredStoreToken = airavataCredStoreToken;
+ }
+ @Override
+ public void run() {
+ try {
+ launchSingleAppExperiment();
+ } catch (TException e) {
+ e.printStackTrace();
+ }
+ }
+
+ private boolean launchSingleAppExperiment() throws TException {
+ Experiment experiment = null;
+ try {
+ List<String> ids = registry.getIds(RegistryModelType.WORKFLOW_NODE_DETAIL, WorkflowNodeConstants.EXPERIMENT_ID, experimentId);
+ for (String workflowNodeId : ids) {
+ // WorkflowNodeDetails workflowNodeDetail = (WorkflowNodeDetails) registry.get(RegistryModelType.WORKFLOW_NODE_DETAIL, workflowNodeId);
+ List<Object> taskDetailList = registry.get(RegistryModelType.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) registry.get(RegistryModelType.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);
+ registry.update(RegistryModelType.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 {
+ registry.update(RegistryModelType.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);
+ }
+ return true;
+ }
+ }
+
+ private class ProcessConsumer implements Runnable, MessageHandler{
+
+
+ @Override
+ public void run() {
+ try {
+ rabbitMQProcessConsumer.listen(this);
+ } catch (AiravataException e) {
+ log.error("Error while listen to the RabbitMQProcessConsumer");
+ }
+ }
+
+ @Override
+ public Map<String, Object> getProperties() {
+ Map<String, Object> props = new HashMap<String, Object>();
+ props.put(MessagingConstants.RABBIT_QUEUE, RabbitMQProcessPublisher.PROCESS);
+ props.put(MessagingConstants.RABBIT_ROUTING_KEY, RabbitMQProcessPublisher.PROCESS);
+ return props;
+ }
+
+ @Override
+ public void onMessage(MessageContext msgCtx) {
+ TBase event = msgCtx.getEvent();
+ if (event instanceof ProcessSubmitEvent) {
+ ProcessSubmitEvent processSubmitEvent = (ProcessSubmitEvent) event;
+ try {
+ launchTask(processSubmitEvent.getTaskId(), processSubmitEvent.getCredentialToken());
+ } catch (TException e) {
+ log.error("Error while launching task : " + processSubmitEvent.getTaskId());
+ }
+ }
+ }
+ }
+
+ }
http://git-wip-us.apache.org/repos/asf/airavata/blob/b4ede9cb/modules/orchestrator/orchestrator-service/src/test/java/org/apache/airavata/orchestrator/client/OrchestratorClientFactoryTest.java
----------------------------------------------------------------------
diff --cc modules/orchestrator/orchestrator-service/src/test/java/org/apache/airavata/orchestrator/client/OrchestratorClientFactoryTest.java
index 0000000,18168c7..3902c9d
mode 000000,100644..100644
--- a/modules/orchestrator/orchestrator-service/src/test/java/org/apache/airavata/orchestrator/client/OrchestratorClientFactoryTest.java
+++ b/modules/orchestrator/orchestrator-service/src/test/java/org/apache/airavata/orchestrator/client/OrchestratorClientFactoryTest.java
@@@ -1,0 -1,90 +1,77 @@@
+ /*
+ *
+ * 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.orchestrator.client;
+
+ //import org.apache.airavata.client.AiravataAPIFactory;
+ //import org.apache.airavata.client.api.exception.AiravataAPIInvocationException;
+ //import org.apache.airavata.client.tools.DocumentCreator;
+ //import org.apache.airavata.client.tools.DocumentCreatorNew;
-import org.apache.airavata.common.exception.ApplicationSettingsException;
-import org.apache.airavata.common.utils.AiravataUtils;
-import org.apache.airavata.common.utils.AiravataZKUtils;
-import org.apache.airavata.common.utils.Constants;
-import org.apache.airavata.common.utils.ServerSettings;
-import org.apache.airavata.model.error.AiravataClientConnectException;
-import org.apache.airavata.orchestrator.client.util.Initialize;
-import org.apache.airavata.orchestrator.cpi.OrchestratorService;
-import org.apache.airavata.orchestrator.server.OrchestratorServer;
-import org.apache.airavata.persistance.registry.jpa.impl.RegistryFactory;
-import org.apache.airavata.registry.cpi.Registry;
-import org.apache.zookeeper.server.ServerCnxnFactory;
-import org.junit.Test;
+
+ public class OrchestratorClientFactoryTest {
+ /* private DocumentCreatorNew documentCreator;
+ private OrchestratorService.Client orchestratorClient;
+ private Registry registry;
+ private int NUM_CONCURRENT_REQUESTS = 1;
+ Initialize initialize;
+ OrchestratorServer service;
+ private static ServerCnxnFactory cnxnFactory;
+
+ @Test
+ public void setUp() {
+ AiravataUtils.setExecutionAsServer();
+ initialize = new Initialize("registry-derby.sql");
+ initialize.initializeDB();
+ System.setProperty(Constants.ZOOKEEPER_SERVER_PORT,"2185");
+ AiravataZKUtils.startEmbeddedZK(cnxnFactory);
+
+ try {
+ service = (new OrchestratorServer());
+ service.start();
+ registry = RegistryFactory.getDefaultRegistry();
+ documentCreator = new DocumentCreatorNew(getAiravataClient());
+ documentCreator.createLocalHostDocs();
+ } catch (Exception e) {
+ e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
+ }
+ AiravataUtils.setExecutionAsServer();
+ try {
+ service.stop();
+ } catch (Exception e) {
+ e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
+ }
+
+ }
+
+ private Airavata.Client getAiravataClient() {
+ Airavata.Client client = null;
+ try {
+ client = AiravataClientFactory.createAiravataClient("localhost", 8930);
+ } catch (AiravataClientConnectException e) {
+ e.printStackTrace();
+ }
+ return client;
+ }
+
+ private void storeDescriptors() {
+
+ }*/
+ }
http://git-wip-us.apache.org/repos/asf/airavata/blob/b4ede9cb/modules/orchestrator/orchestrator-service/src/test/java/org/apache/airavata/orchestrator/client/util/Initialize.java
----------------------------------------------------------------------
diff --cc modules/orchestrator/orchestrator-service/src/test/java/org/apache/airavata/orchestrator/client/util/Initialize.java
index 0000000,c827fc4..7fe8ff3
mode 000000,100644..100644
--- a/modules/orchestrator/orchestrator-service/src/test/java/org/apache/airavata/orchestrator/client/util/Initialize.java
+++ b/modules/orchestrator/orchestrator-service/src/test/java/org/apache/airavata/orchestrator/client/util/Initialize.java
@@@ -1,0 -1,330 +1,329 @@@
+ /*
+ *
+ * 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.orchestrator.client.util;
+
+ 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.persistance.registry.jpa.ResourceType;
-import org.apache.airavata.persistance.registry.jpa.resources.*;
++import org.apache.airavata.experiment.catalog.ResourceType;
++import org.apache.airavata.experiment.catalog.resources.*;
+ import org.apache.airavata.registry.cpi.RegistryException;
+ import org.apache.derby.drda.NetworkServerControl;
+ import org.slf4j.Logger;
+ import org.slf4j.LoggerFactory;
+
+ import java.io.BufferedReader;
+ import java.io.IOException;
+ import java.io.InputStream;
+ import java.io.InputStreamReader;
+ import java.net.InetAddress;
+ import java.sql.*;
+ import java.util.StringTokenizer;
+
+ public class Initialize {
+ private static final Logger logger = LoggerFactory.getLogger(Initialize.class);
+ public static final String DERBY_SERVER_MODE_SYS_PROPERTY = "derby.drda.startNetworkServer";
+ public String scriptName = "registry-derby.sql";
+ private NetworkServerControl server;
+ private static final String delimiter = ";";
+ public static final String PERSISTANT_DATA = "Configuration";
+
+ public Initialize(String scriptName) {
+ this.scriptName = scriptName;
+ }
+
+ public static boolean checkStringBufferEndsWith(StringBuffer buffer, String suffix) {
+ if (suffix.length() > buffer.length()) {
+ return false;
+ }
+ // this loop is done on purpose to avoid memory allocation performance
+ // problems on various JDKs
+ // StringBuffer.lastIndexOf() was introduced in jdk 1.4 and
+ // implementation is ok though does allocation/copying
+ // StringBuffer.toString().endsWith() does massive memory
+ // allocation/copying on JDK 1.5
+ // See http://issues.apache.org/bugzilla/show_bug.cgi?id=37169
+ int endIndex = suffix.length() - 1;
+ int bufferIndex = buffer.length() - 1;
+ while (endIndex >= 0) {
+ if (buffer.charAt(bufferIndex) != suffix.charAt(endIndex)) {
+ return false;
+ }
+ bufferIndex--;
+ endIndex--;
+ }
+ return true;
+ }
+
+ private static boolean isServerStarted(NetworkServerControl server, int ntries)
+ {
+ for (int i = 1; i <= ntries; i ++)
+ {
+ try {
+ Thread.sleep(500);
+ server.ping();
+ return true;
+ }
+ catch (Exception e) {
+ if (i == ntries)
+ return false;
+ }
+ }
+ return false;
+ }
+
+ public void initializeDB() throws SQLException{
+ String jdbcUrl = null;
+ String jdbcUser = null;
+ String jdbcPassword = null;
+ try{
+ jdbcUrl = ServerSettings.getSetting("registry.jdbc.url");
+ jdbcUser = ServerSettings.getSetting("registry.jdbc.user");
+ jdbcPassword = ServerSettings.getSetting("registry.jdbc.password");
+ jdbcUrl = jdbcUrl + "?" + "user=" + jdbcUser + "&" + "password=" + jdbcPassword;
+ } catch (ApplicationSettingsException e) {
+ logger.error("Unable to read properties", e);
+ }
+ startDerbyInServerMode();
+ if(!isServerStarted(server, 20)){
+ throw new RuntimeException("Derby server cound not started within five seconds...");
+ }
+
+ Connection conn = null;
+ try {
+ Class.forName(Utils.getJDBCDriver()).newInstance();
+ conn = DriverManager.getConnection(jdbcUrl, jdbcUser, jdbcPassword);
+ if (!isDatabaseStructureCreated(PERSISTANT_DATA, conn)) {
+ executeSQLScript(conn);
+ logger.info("New Database created for Registry");
+ } else {
+ logger.debug("Database already created for Registry!");
+ }
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ throw new RuntimeException("Database failure", e);
+ } finally {
+ try {
+ if (conn != null){
+ if (!conn.getAutoCommit()) {
+ conn.commit();
+ }
+ conn.close();
+ }
+ } catch (SQLException e) {
+ logger.error(e.getMessage(), e);
+ }
+ }
+
+ try{
+ GatewayResource gatewayResource = new GatewayResource();
+ gatewayResource.setGatewayId(ServerSettings.getSetting("default.registry.gateway"));
+ gatewayResource.setGatewayName(ServerSettings.getSetting("default.registry.gateway"));
+ gatewayResource.setDomain("test-domain");
+ gatewayResource.setEmailAddress("test-email");
+ gatewayResource.save();
+
+ UserResource userResource = new UserResource();
+ userResource.setUserName(ServerSettings.getSetting("default.registry.user"));
+ userResource.setPassword(ServerSettings.getSetting("default.registry.password"));
+ userResource.save();
+
+ WorkerResource workerResource = (WorkerResource) gatewayResource.create(ResourceType.GATEWAY_WORKER);
+ workerResource.setUser(userResource.getUserName());
+ workerResource.save();
+
+ ProjectResource projectResource = (ProjectResource)workerResource.create(ResourceType.PROJECT);
+ projectResource.setGatewayId(gatewayResource.getGatewayId());
+ projectResource.setId("default");
+ projectResource.setName("default");
+ projectResource.setWorker(workerResource);
+ projectResource.save();
+
+
+ } catch (ApplicationSettingsException e) {
+ logger.error("Unable to read properties", e);
+ throw new SQLException(e.getMessage(), e);
+ } catch (RegistryException e) {
+ logger.error("Unable to save data to registry", e);
+ throw new SQLException(e.getMessage(), e);
+ }
+ }
+
+ public static boolean isDatabaseStructureCreated(String tableName, Connection conn) {
+ try {
+ System.out.println("Running a query to test the database tables existence.");
+ // check whether the tables are already created with a query
+ Statement statement = null;
+ try {
+ statement = conn.createStatement();
+ ResultSet rs = statement.executeQuery("select * from " + tableName);
+ if (rs != null) {
+ rs.close();
+ }
+ } finally {
+ try {
+ if (statement != null) {
+ statement.close();
+ }
+ } catch (SQLException e) {
+ return false;
+ }
+ }
+ } catch (SQLException e) {
+ return false;
+ }
+
+ return true;
+ }
+
+ private void executeSQLScript(Connection conn) throws Exception {
+ StringBuffer sql = new StringBuffer();
+ BufferedReader reader = null;
+ try{
+
+ InputStream inputStream = this.getClass().getClassLoader().getResourceAsStream(scriptName);
+ reader = new BufferedReader(new InputStreamReader(inputStream));
+ String line;
+ while ((line = reader.readLine()) != null) {
+ line = line.trim();
+ if (line.startsWith("//")) {
+ continue;
+ }
+ if (line.startsWith("--")) {
+ continue;
+ }
+ StringTokenizer st = new StringTokenizer(line);
+ if (st.hasMoreTokens()) {
+ String token = st.nextToken();
+ if ("REM".equalsIgnoreCase(token)) {
+ continue;
+ }
+ }
+ sql.append(" ").append(line);
+
+ // SQL defines "--" as a comment to EOL
+ // and in Oracle it may contain a hint
+ // so we cannot just remove it, instead we must end it
+ if (line.indexOf("--") >= 0) {
+ sql.append("\n");
+ }
+ if ((checkStringBufferEndsWith(sql, delimiter))) {
+ executeSQL(sql.substring(0, sql.length() - delimiter.length()), conn);
+ sql.replace(0, sql.length(), "");
+ }
+ }
+ // Catch any statements not followed by ;
+ if (sql.length() > 0) {
+ executeSQL(sql.toString(), conn);
+ }
+ }catch (IOException e){
+ logger.error("Error occurred while executing SQL script for creating Airavata database", e);
+ throw new Exception("Error occurred while executing SQL script for creating Airavata database", e);
+ }finally {
+ if (reader != null) {
+ reader.close();
+ }
+
+ }
+
+ }
+
+ private static void executeSQL(String sql, Connection conn) throws Exception {
+ // Check and ignore empty statements
+ if ("".equals(sql.trim())) {
+ return;
+ }
+
+ Statement statement = null;
+ try {
+ logger.debug("SQL : " + sql);
+
+ boolean ret;
+ int updateCount = 0, updateCountTotal = 0;
+ statement = conn.createStatement();
+ ret = statement.execute(sql);
+ updateCount = statement.getUpdateCount();
+ do {
+ if (!ret) {
+ if (updateCount != -1) {
+ updateCountTotal += updateCount;
+ }
+ }
+ ret = statement.getMoreResults();
+ if (ret) {
+ updateCount = statement.getUpdateCount();
+ }
+ } while (ret);
+
+ logger.debug(sql + " : " + updateCountTotal + " rows affected");
+
+ SQLWarning warning = conn.getWarnings();
+ while (warning != null) {
+ logger.warn(warning + " sql warning");
+ warning = warning.getNextWarning();
+ }
+ conn.clearWarnings();
+ } catch (SQLException e) {
+ if (e.getSQLState().equals("X0Y32")) {
+ // eliminating the table already exception for the derby
+ // database
+ logger.info("Table Already Exists", e);
+ } else {
+ throw new Exception("Error occurred while executing : " + sql, e);
+ }
+ } finally {
+ if (statement != null) {
+ try {
+ statement.close();
+ } catch (SQLException e) {
+ logger.error("Error occurred while closing result set.", e);
+ }
+ }
+ }
+ }
+
+ private void startDerbyInServerMode() {
+ try {
+ System.setProperty(DERBY_SERVER_MODE_SYS_PROPERTY, "true");
+ server = new NetworkServerControl(InetAddress.getByName(Utils.getHost()),
+ 20000,
+ Utils.getJDBCUser(), Utils.getJDBCPassword());
+ java.io.PrintWriter consoleWriter = new java.io.PrintWriter(System.out, true);
+ server.start(consoleWriter);
+ } catch (IOException e) {
+ logger.error("Unable to start Apache derby in the server mode! Check whether " +
+ "specified port is available");
+ } catch (Exception e) {
+ logger.error("Unable to start Apache derby in the server mode! Check whether " +
+ "specified port is available");
+ }
+
+ }
+
+ public void stopDerbyServer() throws SQLException{
+ try {
+ server.shutdown();
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ throw new SQLException("Error while stopping derby server", e);
+ }
+ }
+ }
http://git-wip-us.apache.org/repos/asf/airavata/blob/b4ede9cb/modules/workflow-model/workflow-engine/pom.xml
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/airavata/blob/b4ede9cb/modules/xbaya-gui/pom.xml
----------------------------------------------------------------------