| /* |
| * |
| * 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.utils.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.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 org.apache.zookeeper.CreateMode; |
| import org.apache.zookeeper.KeeperException; |
| import org.apache.zookeeper.WatchedEvent; |
| import org.apache.zookeeper.Watcher; |
| import org.apache.zookeeper.ZooDefs; |
| import org.apache.zookeeper.ZooKeeper; |
| import org.apache.zookeeper.data.Stat; |
| |
| import java.io.File; |
| import java.io.IOException; |
| import java.util.Arrays; |
| import java.util.Calendar; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Random; |
| |
| public class OrchestratorServerHandler implements OrchestratorService.Iface, |
| Watcher { |
| private static AiravataLogger log = AiravataLoggerFactory |
| .getLogger(OrchestratorServerHandler.class); |
| |
| private SimpleOrchestratorImpl orchestrator = null; |
| |
| private Registry registry; |
| |
| private ZooKeeper zk; |
| |
| 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 { |
| if (ServerSettings.isRabbitMqPublishEnabled()) { |
| 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()); |
| try { |
| zk = new ZooKeeper(zkhostPort, 6000, this); // no watcher is |
| // required, this |
| // will only use to |
| // store some data |
| String OrchServer = ServerSettings |
| .getSetting(org.apache.airavata.common.utils.Constants.ZOOKEEPER_ORCHESTRATOR_SERVER_NODE); |
| synchronized (mutex) { |
| mutex.wait(); // waiting for the syncConnected event |
| } |
| registerOrchestratorService(airavataServerHostPort, OrchServer); |
| // creating a watch in orchestrator to monitor the gfac |
| // instances |
| zk.getChildren(ServerSettings.getSetting( |
| Constants.ZOOKEEPER_GFAC_SERVER_NODE, "/gfac-server"), |
| this); |
| log.info("Finished starting ZK: " + zk); |
| } catch (IOException e) { |
| log.error(e.getMessage(), e); |
| throw new OrchestratorException("Error while initializing orchestrator service", e); |
| } catch (InterruptedException e) { |
| log.error(e.getMessage(), e); |
| throw new OrchestratorException("Error while initializing orchestrator service", e); |
| } catch (KeeperException e) { |
| log.error(e.getMessage(), e); |
| throw new OrchestratorException("Error while initializing orchestrator service", e); |
| } |
| } catch (ApplicationSettingsException e) { |
| log.error(e.getMessage(), e); |
| throw new OrchestratorException("Error while initializing orchestrator service", e); |
| }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().setZk(this.zk); |
| 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); |
| } |
| |
| } |
| |
| private void registerOrchestratorService(String airavataServerHostPort, String orchServer) throws KeeperException, InterruptedException { |
| Stat zkStat = zk.exists(orchServer, false); |
| if (zkStat == null) { |
| zk.create(orchServer, new byte[0], |
| ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); |
| } |
| String instantNode = orchServer |
| + File.separator |
| + String.valueOf(new Random() |
| .nextInt(Integer.MAX_VALUE)); |
| zkStat = zk.exists(instantNode, false); |
| if (zkStat == null) { |
| zk.create(instantNode, airavataServerHostPort.getBytes(), |
| ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); |
| } |
| } |
| |
| /** |
| * * 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) { |
| 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.getFixedThreadPool().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()); |
| } |
| 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) throws TException { |
| log.infoId(experimentId, "Experiment: {} is cancelling !!!!!", experimentId); |
| return validateStatesAndCancel(experimentId); |
| } |
| |
| /** |
| * This method gracefully handler gfac node failures |
| */ |
| synchronized public void process(WatchedEvent watchedEvent) { |
| synchronized (mutex) { |
| try { |
| Event.KeeperState state = watchedEvent.getState(); |
| switch (state) { |
| case SyncConnected: |
| mutex.notify(); |
| break; |
| case Expired:case Disconnected: |
| try { |
| zk = new ZooKeeper(AiravataZKUtils.getZKhostPort(), 6000, this); |
| synchronized (mutex) { |
| mutex.wait(); // waiting for the syncConnected event |
| } |
| String airavataServerHostPort = ServerSettings |
| .getSetting(Constants.ORCHESTRATOR_SERVER_HOST) |
| + ":" |
| + ServerSettings |
| .getSetting(Constants.ORCHESTRATOR_SERVER_PORT); |
| String OrchServer = ServerSettings |
| .getSetting(org.apache.airavata.common.utils.Constants.ZOOKEEPER_ORCHESTRATOR_SERVER_NODE); |
| registerOrchestratorService(airavataServerHostPort, OrchServer); |
| } catch (IOException e) { |
| e.printStackTrace(); |
| } catch (ApplicationSettingsException e) { |
| e.printStackTrace(); |
| } catch (InterruptedException e) { |
| e.printStackTrace(); |
| } catch (KeeperException e) { |
| e.printStackTrace(); |
| } |
| break; |
| } |
| if (watchedEvent.getPath() != null |
| && watchedEvent.getPath().startsWith( |
| ServerSettings.getSetting( |
| Constants.ZOOKEEPER_GFAC_SERVER_NODE, |
| "/gfac-server"))) { |
| List<String> children = zk.getChildren(ServerSettings |
| .getSetting(Constants.ZOOKEEPER_GFAC_SERVER_NODE, |
| "/gfac-server"), true); |
| for (String gfacNodes : children) { |
| zk.exists( |
| ServerSettings.getSetting( |
| Constants.ZOOKEEPER_GFAC_SERVER_NODE, |
| "/gfac-server") |
| + File.separator + gfacNodes, this); |
| } |
| switch (watchedEvent.getType()) { |
| case NodeCreated: |
| mutex.notify(); |
| break; |
| case NodeDeleted: |
| // here we have to handle gfac node shutdown case |
| if (children.size() == 0) { |
| log.error("There are not gfac instances to route failed jobs"); |
| return; |
| } |
| // we recover one gfac node at a time |
| final WatchedEvent event = watchedEvent; |
| final OrchestratorServerHandler handler = this; |
| /*(new Thread() { // disabling ft implementation with zk |
| public void run() { |
| int retry = 0; |
| while (retry < 3) { |
| try { |
| (new OrchestratorRecoveryHandler( |
| handler, event.getPath())) |
| .recover(); |
| break; |
| } catch (Exception e) { |
| e.printStackTrace(); |
| log.error("error recovering the jobs for gfac-node: " |
| + event.getPath()); |
| log.error("Retrying again to recover jobs and retry attempt: " |
| + ++retry); |
| } |
| } |
| |
| } |
| }).start();*/ |
| break; |
| } |
| |
| |
| } |
| } catch (KeeperException e) { |
| e.printStackTrace(); |
| } catch (InterruptedException e) { |
| e.printStackTrace(); |
| } |
| } |
| } |
| |
| 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)throws TException{ |
| try { |
| Experiment experiment = (Experiment) registry.get( |
| RegistryModelType.EXPERIMENT, experimentId); |
| 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 (experimentState.getValue()> 5 && experimentState.getValue()<10) { |
| 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()); |
| }else if(experimentState.getValue()<3){ |
| // 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); |
| GFacUtils.setExperimentCancel(experimentId, taskDetails.getTaskID(), zk); |
| } |
| } |
| }else { |
| |
| 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()); |
| GFacUtils.setExperimentCancel(experimentId, taskDetails.getTaskID(), zk); |
| } |
| // iterate through all the generated tasks and performs the |
| // job submisssion+monitoring |
| // launching the experiment |
| orchestrator.cancelExperiment(experiment, |
| workflowNodeDetail, taskDetails, null); |
| |
| // after performing gfac level cancel operation |
| // mark task cancelled |
| taskStatus.setExecutionState(TaskState.CANCELED); |
| taskStatus.setTimeOfStateChange(Calendar.getInstance() |
| .getTimeInMillis()); |
| taskDetails.setTaskStatus(taskStatus); |
| registry.update(RegistryModelType.TASK_DETAIL, o, |
| taskDetails.getTaskID()); |
| } |
| // mark workflownode cancelled |
| WorkflowNodeStatus workflowNodeStatus = new WorkflowNodeStatus(); |
| workflowNodeStatus.setWorkflowNodeState(WorkflowNodeState.CANCELED); |
| workflowNodeStatus.setTimeOfStateChange(Calendar.getInstance() |
| .getTimeInMillis()); |
| workflowNodeDetail.setWorkflowNodeStatus(workflowNodeStatus); |
| registry.update(RegistryModelType.WORKFLOW_NODE_DETAIL, workflowNodeDetail, |
| workflowNodeId); |
| } |
| // mark experiment cancelled |
| status = new ExperimentStatus(); |
| status.setExperimentState(ExperimentState.CANCELED); |
| status.setTimeOfStateChange(Calendar.getInstance() |
| .getTimeInMillis()); |
| experiment.setExperimentStatus(status); |
| registry.update(RegistryModelType.EXPERIMENT, experiment, |
| experimentId); |
| } |
| log.info("Experiment: " + experimentId + " is cancelled !!!!!"); |
| |
| } catch (Exception e) { |
| throw new TException(e); |
| } |
| return true; |
| } |
| |
| 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; |
| } |
| ExperimentStatus status = new ExperimentStatus(); |
| status.setExperimentState(ExperimentState.LAUNCHED); |
| status.setTimeOfStateChange(Calendar.getInstance().getTimeInMillis()); |
| experiment.setExperimentStatus(status); |
| registry.update(RegistryModelType.EXPERIMENT_STATUS, status, experimentId); |
| if (ServerSettings.isRabbitMqPublishEnabled()) { |
| 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()); |
| } |
| } |
| } |
| } |
| |
| } |
