| /* |
| * |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| * |
| */ |
| |
| package org.apache.airavata.gfac.impl; |
| |
| import org.apache.airavata.common.exception.AiravataException; |
| import org.apache.airavata.common.utils.AiravataUtils; |
| import org.apache.airavata.common.utils.ServerSettings; |
| import org.apache.airavata.common.utils.ThriftUtils; |
| import org.apache.airavata.common.utils.ZkConstants; |
| import org.apache.airavata.gfac.core.GFacConstants; |
| import org.apache.airavata.gfac.core.GFacEngine; |
| import org.apache.airavata.gfac.core.GFacException; |
| import org.apache.airavata.gfac.core.GFacUtils; |
| import org.apache.airavata.gfac.core.cluster.ServerInfo; |
| import org.apache.airavata.gfac.core.context.ProcessContext; |
| import org.apache.airavata.gfac.core.context.TaskContext; |
| import org.apache.airavata.gfac.core.monitor.JobMonitor; |
| import org.apache.airavata.gfac.core.task.JobSubmissionTask; |
| import org.apache.airavata.gfac.core.task.Task; |
| import org.apache.airavata.gfac.core.task.TaskException; |
| import org.apache.airavata.gfac.impl.task.DataStageTask; |
| import org.apache.airavata.gfac.impl.task.DataStreamingTask; |
| import org.apache.airavata.gfac.impl.task.EnvironmentSetupTask; |
| import org.apache.airavata.model.appcatalog.appinterface.ApplicationInterfaceDescription; |
| import org.apache.airavata.model.appcatalog.computeresource.*; |
| import org.apache.airavata.model.appcatalog.gatewayprofile.ComputeResourcePreference; |
| import org.apache.airavata.model.appcatalog.gatewayprofile.GatewayResourceProfile; |
| import org.apache.airavata.model.appcatalog.gatewayprofile.StoragePreference; |
| import org.apache.airavata.model.appcatalog.storageresource.StorageResourceDescription; |
| import org.apache.airavata.model.application.io.DataType; |
| import org.apache.airavata.model.application.io.InputDataObjectType; |
| import org.apache.airavata.model.application.io.OutputDataObjectType; |
| import org.apache.airavata.model.commons.ErrorModel; |
| import org.apache.airavata.model.data.movement.SecurityProtocol; |
| import org.apache.airavata.model.job.JobModel; |
| import org.apache.airavata.model.process.ProcessModel; |
| import org.apache.airavata.model.status.JobState; |
| import org.apache.airavata.model.status.JobStatus; |
| import org.apache.airavata.model.status.ProcessState; |
| import org.apache.airavata.model.status.ProcessStatus; |
| import org.apache.airavata.model.status.TaskState; |
| import org.apache.airavata.model.status.TaskStatus; |
| import org.apache.airavata.model.task.*; |
| import org.apache.airavata.registry.cpi.AppCatalog; |
| import org.apache.airavata.registry.cpi.AppCatalogException; |
| import org.apache.airavata.registry.cpi.ExpCatChildDataType; |
| 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.utils.Constants; |
| import org.apache.curator.framework.CuratorFramework; |
| import org.apache.curator.utils.ZKPaths; |
| import org.apache.thrift.TException; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import java.io.File; |
| import java.net.URI; |
| import java.net.URISyntaxException; |
| import java.util.*; |
| |
| public class GFacEngineImpl implements GFacEngine { |
| |
| private static final Logger log = LoggerFactory.getLogger(GFacEngineImpl.class); |
| |
| public GFacEngineImpl() throws GFacException { |
| |
| } |
| |
| @Override |
| public ProcessContext populateProcessContext(String processId, String gatewayId, String |
| tokenId) throws GFacException { |
| ProcessContext processContext = new ProcessContext(processId, gatewayId, tokenId); |
| try { |
| AppCatalog appCatalog = Factory.getDefaultAppCatalog(); |
| processContext.setAppCatalog(appCatalog); |
| ExperimentCatalog expCatalog = Factory.getDefaultExpCatalog(); |
| processContext.setExperimentCatalog(expCatalog); |
| processContext.setCuratorClient(Factory.getCuratorClient()); |
| processContext.setStatusPublisher(Factory.getStatusPublisher()); |
| |
| ProcessModel processModel = (ProcessModel) expCatalog.get(ExperimentCatalogModelType.PROCESS, processId); |
| processContext.setProcessModel(processModel); |
| |
| try { |
| checkRecoveryWithCancel(processContext); |
| } catch (Exception e) { |
| log.error("expId: {}, processId: {}, Error while checking process cancel data in zookeeper", |
| processContext.getExperimentId(), processContext.getProcessId()); |
| } |
| |
| GatewayResourceProfile gatewayProfile = appCatalog.getGatewayProfile().getGatewayProfile(gatewayId); |
| processContext.setGatewayResourceProfile(gatewayProfile); |
| ComputeResourcePreference computeResourcePreference = appCatalog.getGatewayProfile().getComputeResourcePreference |
| (gatewayId, processModel.getComputeResourceId()); |
| //FIXME: Temporary revert, this needs a proper fix. |
| // String scratchLocation = Factory.getScratchLocation(processContext); |
| String scratchLocation = computeResourcePreference.getScratchLocation(); |
| scratchLocation = scratchLocation + File.separator + processId + File.separator; |
| processContext.setComputeResourcePreference(computeResourcePreference); |
| StoragePreference storagePreference = appCatalog.getGatewayProfile().getStoragePreference(gatewayId, processModel.getStorageResourceId()); |
| StorageResourceDescription storageResource = appCatalog.getStorageResource().getStorageResource(processModel.getStorageResourceId()); |
| if (storageResource != null){ |
| processContext.setStorageResource(storageResource); |
| }else { |
| // we need to fail the process which will fail the experiment |
| processContext.setProcessStatus(new ProcessStatus(ProcessState.FAILED)); |
| GFacUtils.saveAndPublishProcessStatus(processContext); |
| throw new GFacException("expId: " + processModel.getExperimentId() + ", processId: " + processId + |
| ":- Couldn't find storage resource for storage resource id :" + processModel.getStorageResourceId()); |
| } |
| if (storagePreference != null) { |
| processContext.setStoragePreference(storagePreference); |
| } else { |
| // we need to fail the process which will fail the experiment |
| processContext.setProcessStatus(new ProcessStatus(ProcessState.FAILED)); |
| GFacUtils.saveAndPublishProcessStatus(processContext); |
| throw new GFacException("expId: " + processModel.getExperimentId() + ", processId: " + processId + |
| ":- Couldn't find storage preference for storage resource id :" + processModel.getStorageResourceId()); |
| |
| } |
| |
| |
| /* StorageResourceDescription storageResource = appCatalog.getStorageResource().getStorageResource(processModel.getStorageResourceId()); |
| if (storageResource != null){ |
| processContext.setStorageResource(storageResource); |
| }*/ |
| processContext.setComputeResourceDescription(appCatalog.getComputeResource().getComputeResource |
| (processContext.getComputeResourcePreference().getComputeResourceId())); |
| processContext.setApplicationDeploymentDescription(appCatalog.getApplicationDeployment() |
| .getApplicationDeployement(processModel.getApplicationDeploymentId())); |
| ApplicationInterfaceDescription applicationInterface = appCatalog.getApplicationInterface() |
| .getApplicationInterface(processModel.getApplicationInterfaceId()); |
| processContext.setApplicationInterfaceDescription(applicationInterface); |
| String computeResourceId = processContext.getComputeResourceDescription().getComputeResourceId(); |
| String hostName = Factory.getDefaultAppCatalog().getComputeResource().getComputeResource(computeResourceId).getHostName(); |
| ServerInfo serverInfo = new ServerInfo(Factory.getLoginUserName(processContext), hostName); |
| processContext.setServerInfo(serverInfo); |
| List<OutputDataObjectType> applicationOutputs = applicationInterface.getApplicationOutputs(); |
| if (applicationOutputs != null && !applicationOutputs.isEmpty()) { |
| for (OutputDataObjectType outputDataObjectType : applicationOutputs) { |
| if (outputDataObjectType.getType().equals(DataType.STDOUT)) { |
| if (outputDataObjectType.getValue() == null || outputDataObjectType.getValue().equals("")) { |
| outputDataObjectType.setValue(scratchLocation + applicationInterface.getApplicationName() + ".stdout"); |
| processContext.setStdoutLocation(scratchLocation + applicationInterface.getApplicationName() + ".stdout"); |
| } else { |
| processContext.setStdoutLocation(outputDataObjectType.getValue()); |
| } |
| } |
| if (outputDataObjectType.getType().equals(DataType.STDERR)) { |
| if (outputDataObjectType.getValue() == null || outputDataObjectType.getValue().equals("")) { |
| String stderrLocation = scratchLocation + applicationInterface.getApplicationName() + ".stderr"; |
| outputDataObjectType.setValue(stderrLocation); |
| processContext.setStderrLocation(stderrLocation); |
| } else { |
| processContext.setStderrLocation(outputDataObjectType.getValue()); |
| } |
| } |
| } |
| } |
| expCatalog.update(ExperimentCatalogModelType.PROCESS, processModel, processId); |
| processModel.setProcessOutputs(applicationOutputs); |
| |
| processContext.setSshKeyAuthentication(Factory.getComputerResourceSSHKeyAuthentication(processContext)); |
| if (processContext.getJobSubmissionProtocol() == JobSubmissionProtocol.UNICORE) { |
| // process monitor mode set in getResourceJobManager method, but unicore doesn't have resource job manager. |
| // hence we set process monitor mode here. |
| processContext.setMonitorMode(MonitorMode.FORK); |
| } else { |
| processContext.setResourceJobManager(getResourceJobManager(processContext)); |
| processContext.setJobSubmissionRemoteCluster(Factory.getJobSubmissionRemoteCluster(processContext)); |
| processContext.setDataMovementRemoteCluster(Factory.getDataMovementRemoteCluster(processContext)); |
| } |
| |
| String inputPath = ServerSettings.getLocalDataLocation(); |
| if (inputPath != null) { |
| processContext.setLocalWorkingDir((inputPath.endsWith("/") ? inputPath : inputPath + "/") + |
| processContext.getProcessId()); |
| } |
| |
| List<Object> jobModels = expCatalog.get(ExperimentCatalogModelType.JOB, "processId", processId); |
| if (jobModels != null && !jobModels.isEmpty()) { |
| if (jobModels.size() > 1) { |
| log.warn("Process has more than one job model, take first one"); |
| } |
| processContext.setJobModel(((JobModel) jobModels.get(0))); |
| } |
| return processContext; |
| } catch (AppCatalogException e) { |
| String msg = "App catalog access exception "; |
| updateProcessFailure(processContext, msg); |
| throw new GFacException(msg, e); |
| } catch (RegistryException e) { |
| String msg = "Registry access exception"; |
| updateProcessFailure(processContext, msg); |
| throw new GFacException(msg, e); |
| } catch (AiravataException e) { |
| String msg = "Remote cluster initialization error"; |
| updateProcessFailure(processContext, msg); |
| throw new GFacException(msg, e); |
| } |
| |
| } |
| |
| private void checkRecoveryWithCancel(ProcessContext processContext) throws Exception { |
| CuratorFramework curatorClient = processContext.getCuratorClient(); |
| String experimentId = processContext.getExperimentId(); |
| String processId = processContext.getProcessId(); |
| String processCancelNodePath = ZKPaths.makePath(ZKPaths.makePath(ZKPaths.makePath( |
| ZkConstants.ZOOKEEPER_EXPERIMENT_NODE, experimentId), processId), ZkConstants.ZOOKEEPER_CANCEL_LISTENER_NODE); |
| log.info("expId: {}, processId: {}, get process cancel data from zookeeper node {}", experimentId, processId, processCancelNodePath); |
| byte[] bytes = curatorClient.getData().forPath(processCancelNodePath); |
| if (bytes != null && new String(bytes).equalsIgnoreCase(ZkConstants.ZOOKEEPER_CANCEL_REQEUST)) { |
| processContext.setRecoveryWithCancel(true); |
| } |
| } |
| |
| @Override |
| public void executeProcess(ProcessContext processContext) throws GFacException { |
| if (processContext.isInterrupted()) { |
| GFacUtils.handleProcessInterrupt(processContext); |
| return; |
| } |
| String taskDag = processContext.getTaskDag(); |
| List<String> taskIds = GFacUtils.parseTaskDag(taskDag); |
| processContext.setTaskExecutionOrder(taskIds); |
| executeTaskListFrom(processContext, taskIds.get(0)); |
| } |
| |
| private void executeTaskListFrom(ProcessContext processContext, String startingTaskId) throws GFacException { |
| // checkpoint |
| if (processContext.isInterrupted() && processContext.getProcessState() != ProcessState.MONITORING) { |
| GFacUtils.handleProcessInterrupt(processContext); |
| return; |
| } |
| List<TaskModel> taskList = processContext.getTaskList(); |
| Map<String, TaskModel> taskMap = processContext.getTaskMap(); |
| boolean fastForward = true; |
| for (String taskId : processContext.getTaskExecutionOrder()) { |
| if (fastForward) { |
| if (taskId.equalsIgnoreCase(startingTaskId)) { |
| fastForward = false; |
| } else { |
| continue; |
| } |
| } |
| |
| TaskModel taskModel = taskMap.get(taskId); |
| processContext.setCurrentExecutingTaskModel(taskModel); |
| TaskTypes taskType = taskModel.getTaskType(); |
| TaskContext taskContext = getTaskContext(processContext); |
| taskContext.setTaskModel(taskModel); |
| ProcessStatus status = null; |
| switch (taskType) { |
| case ENV_SETUP: |
| status = new ProcessStatus(ProcessState.CONFIGURING_WORKSPACE); |
| status.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime()); |
| processContext.setProcessStatus(status); |
| GFacUtils.saveAndPublishProcessStatus(processContext); |
| // checkpoint |
| if (processContext.isInterrupted()) { |
| GFacUtils.handleProcessInterrupt(processContext); |
| return; |
| } |
| configureWorkspace(taskContext, processContext.isRecovery()); |
| // checkpoint |
| if (processContext.isInterrupted()) { |
| GFacUtils.handleProcessInterrupt(processContext); |
| return; |
| } |
| break; |
| case DATA_STAGING: |
| try { |
| // checkpoint |
| if (processContext.isInterrupted()) { |
| GFacUtils.handleProcessInterrupt(processContext); |
| return; |
| } |
| DataStagingTaskModel subTaskModel = (DataStagingTaskModel) taskContext.getSubTaskModel(); |
| DataStageType type = subTaskModel.getType(); |
| switch (type) { |
| case INPUT: |
| status = new ProcessStatus(ProcessState.INPUT_DATA_STAGING); |
| status.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime()); |
| processContext.setProcessStatus(status); |
| GFacUtils.saveAndPublishProcessStatus(processContext); |
| taskContext.setProcessInput(subTaskModel.getProcessInput()); |
| inputDataStaging(taskContext, processContext.isRecovery()); |
| break; |
| case OUPUT: |
| status = new ProcessStatus(ProcessState.OUTPUT_DATA_STAGING); |
| status.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime()); |
| processContext.setProcessStatus(status); |
| GFacUtils.saveAndPublishProcessStatus(processContext); |
| taskContext.setProcessOutput(subTaskModel.getProcessOutput()); |
| outputDataStaging(taskContext, processContext.isRecovery(), false); |
| break; |
| case ARCHIVE_OUTPUT: |
| status = new ProcessStatus(ProcessState.OUTPUT_DATA_STAGING); |
| status.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime()); |
| processContext.setProcessStatus(status); |
| GFacUtils.saveAndPublishProcessStatus(processContext); |
| outputDataStaging(taskContext, processContext.isRecovery(), true); |
| break; |
| |
| } |
| // checkpoint |
| if (processContext.isInterrupted()) { |
| GFacUtils.handleProcessInterrupt(processContext); |
| return; |
| } |
| } catch (TException e) { |
| throw new GFacException(e); |
| } |
| break; |
| |
| case JOB_SUBMISSION: |
| // checkpoint |
| if (processContext.isInterrupted()) { |
| GFacUtils.handleProcessInterrupt(processContext); |
| return; |
| } |
| status = new ProcessStatus(ProcessState.EXECUTING); |
| status.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime()); |
| processContext.setProcessStatus(status); |
| GFacUtils.saveAndPublishProcessStatus(processContext); |
| executeJobSubmission(taskContext, processContext.isRecovery()); |
| // Don't put any checkpoint in between JobSubmission and Monitoring tasks |
| |
| JobStatus jobStatus = processContext.getJobModel().getJobStatus(); |
| if (jobStatus != null && (jobStatus.getJobState() == JobState.SUBMITTED |
| || jobStatus.getJobState() == JobState.QUEUED || jobStatus.getJobState() == JobState.ACTIVE)) { |
| |
| List<OutputDataObjectType> processOutputs = processContext.getProcessModel().getProcessOutputs(); |
| if (processOutputs != null && !processOutputs.isEmpty()){ |
| for (OutputDataObjectType output : processOutputs){ |
| try { |
| if (output.isOutputStreaming()){ |
| TaskModel streamingTaskModel = new TaskModel(); |
| streamingTaskModel.setTaskType(TaskTypes.OUTPUT_FETCHING); |
| streamingTaskModel.setTaskStatus(new TaskStatus(TaskState.CREATED)); |
| streamingTaskModel.setCreationTime(AiravataUtils.getCurrentTimestamp().getTime()); |
| streamingTaskModel.setParentProcessId(processContext.getProcessId()); |
| TaskContext streamingTaskContext = getTaskContext(processContext); |
| DataStagingTaskModel submodel = new DataStagingTaskModel(); |
| submodel.setType(DataStageType.OUPUT); |
| submodel.setProcessOutput(output); |
| URI source = new URI(processContext.getDataMovementProtocol().name(), |
| Factory.getLoginUserName(processContext), |
| processContext.getComputeResourceDescription().getHostName(), |
| 22, |
| processContext.getWorkingDir() + output.getValue(), null, null); |
| submodel.setSource(source.getPath()); |
| submodel.setDestination("dummy://temp/file/location"); |
| streamingTaskModel.setSubTaskModel(ThriftUtils.serializeThriftObject(submodel)); |
| String streamTaskId = (String) processContext.getExperimentCatalog() |
| .add(ExpCatChildDataType.TASK, streamingTaskModel, processContext.getProcessId()); |
| streamingTaskModel.setTaskId(streamTaskId); |
| streamingTaskContext.setTaskModel(streamingTaskModel); |
| executeDataStreaming(streamingTaskContext, processContext.isRecovery()); |
| } |
| } catch (URISyntaxException | TException | RegistryException e) { |
| log.error("Error while streaming output " + output.getValue()); |
| } |
| } |
| } |
| } |
| break; |
| |
| case MONITORING: |
| status = new ProcessStatus(ProcessState.MONITORING); |
| status.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime()); |
| processContext.setProcessStatus(status); |
| GFacUtils.saveAndPublishProcessStatus(processContext); |
| executeJobMonitoring(taskContext, processContext.isRecovery()); |
| break; |
| |
| case ENV_CLEANUP: |
| // TODO implement environment clean up task logic |
| break; |
| |
| default: |
| throw new GFacException("Unsupported Task type"); |
| |
| } |
| |
| |
| if (processContext.isPauseTaskExecution()) { |
| return; // If any task put processContext to wait, the same task must continue processContext execution. |
| } |
| |
| } |
| processContext.setComplete(true); |
| } |
| |
| private void executeJobMonitoring(TaskContext taskContext, boolean recovery) throws GFacException { |
| ProcessContext processContext = taskContext.getParentProcessContext(); |
| TaskStatus taskStatus; |
| JobMonitor monitorService = null; |
| try { |
| taskStatus = new TaskStatus(TaskState.EXECUTING); |
| taskStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime()); |
| taskContext.setTaskStatus(taskStatus); |
| GFacUtils.saveAndPublishTaskStatus(taskContext); |
| |
| MonitorTaskModel monitorTaskModel = ((MonitorTaskModel) taskContext.getSubTaskModel()); |
| monitorService = Factory.getMonitorService(monitorTaskModel.getMonitorMode()); |
| if (!monitorService.isMonitoring(processContext.getJobModel().getJobId())) { |
| monitorService.monitor(processContext.getJobModel().getJobId(), taskContext); |
| } else { |
| log.warn("Jobid: {}, already in monitoring map", processContext.getJobModel().getJobId()); |
| } |
| } catch (AiravataException | TException e) { |
| taskStatus = new TaskStatus(TaskState.FAILED); |
| taskStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime()); |
| taskStatus.setReason("Couldn't handover jobId {} to monitor service, monitor service type {}"); |
| taskContext.setTaskStatus(taskStatus); |
| GFacUtils.saveAndPublishTaskStatus(taskContext); |
| |
| String errorMsg = new StringBuilder("expId: ").append(processContext.getExperimentId()).append(", processId: ") |
| .append(processContext.getProcessId()).append(", taskId: ").append(taskContext.getTaskId()) |
| .append(", type: ").append(taskContext.getTaskType().name()).append(" :- Input staging failed. Reason: ") |
| .append(taskStatus.getReason()).toString(); |
| ErrorModel errorModel = new ErrorModel(); |
| errorModel.setUserFriendlyMessage("Error while staging output data"); |
| errorModel.setActualErrorMessage(errorMsg); |
| GFacUtils.saveTaskError(taskContext, errorModel); |
| throw new GFacException(e); |
| } |
| if (processContext.isPauseTaskExecution()) { |
| // we won't update task status to complete, job monitor will update task status to complete after it complete monitoring for this job id. |
| return; |
| } |
| taskStatus = new TaskStatus(TaskState.COMPLETED); |
| taskStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime()); |
| taskStatus.setReason("Successfully handed over job id to job monitor service."); |
| taskContext.setTaskStatus(taskStatus); |
| GFacUtils.saveAndPublishTaskStatus(taskContext); |
| } |
| |
| private boolean executeJobSubmission(TaskContext taskContext, boolean recovery) throws GFacException { |
| TaskStatus taskStatus = new TaskStatus(TaskState.EXECUTING); |
| taskStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime()); |
| taskContext.setTaskStatus(taskStatus); |
| GFacUtils.saveAndPublishTaskStatus(taskContext); |
| try { |
| JobSubmissionTaskModel jobSubmissionTaskModel = ((JobSubmissionTaskModel) taskContext.getSubTaskModel()); |
| JobSubmissionTask jobSubmissionTask = Factory.getJobSubmissionTask(jobSubmissionTaskModel.getJobSubmissionProtocol()); |
| |
| ProcessContext processContext = taskContext.getParentProcessContext(); |
| taskStatus = executeTask(taskContext, jobSubmissionTask, recovery); |
| taskStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime()); |
| taskContext.setTaskStatus(taskStatus); |
| GFacUtils.saveAndPublishTaskStatus(taskContext); |
| checkFailures(taskContext, taskStatus, jobSubmissionTask); |
| return false; |
| } catch (TException e) { |
| throw new GFacException(e); |
| } |
| } |
| |
| private void executeDataStreaming(TaskContext taskContext, boolean recovery) throws GFacException { |
| TaskStatus taskStatus = new TaskStatus(TaskState.EXECUTING); |
| taskStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime()); |
| taskContext.setTaskStatus(taskStatus); |
| GFacUtils.saveAndPublishTaskStatus(taskContext); |
| try { |
| DataStreamingTask dataStreamingTask = new DataStreamingTask(); |
| taskStatus = executeTask(taskContext, dataStreamingTask, recovery); |
| taskStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime()); |
| taskContext.setTaskStatus(taskStatus); |
| GFacUtils.saveAndPublishTaskStatus(taskContext); |
| } catch (Exception e) { |
| throw new GFacException(e); |
| } |
| } |
| |
| private boolean configureWorkspace(TaskContext taskContext, boolean recover) throws GFacException { |
| |
| try { |
| EnvironmentSetupTaskModel subTaskModel = (EnvironmentSetupTaskModel) taskContext.getSubTaskModel(); |
| Task envSetupTask = null; |
| if (subTaskModel.getProtocol() == SecurityProtocol.SSH_KEYS) { |
| envSetupTask = new EnvironmentSetupTask(); |
| } else { |
| throw new GFacException("Unsupported security protocol, Airavata doesn't support " + |
| subTaskModel.getProtocol().name() + " protocol yet."); |
| } |
| TaskStatus status = new TaskStatus(TaskState.EXECUTING); |
| status.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime()); |
| taskContext.setTaskStatus(status); |
| GFacUtils.saveAndPublishTaskStatus(taskContext); |
| TaskStatus taskStatus = executeTask(taskContext, envSetupTask, recover); |
| taskStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime()); |
| taskContext.setTaskStatus(taskStatus); |
| GFacUtils.saveAndPublishTaskStatus(taskContext); |
| |
| if (taskStatus.getState() == TaskState.FAILED) { |
| log.error("expId: {}, processId: {}, taskId: {} type: {},:- Input staging failed, " + |
| "reason:" + " {}", taskContext.getParentProcessContext().getExperimentId(), taskContext |
| .getParentProcessContext().getProcessId(), taskContext.getTaskId(), envSetupTask.getType |
| ().name(), taskStatus.getReason()); |
| ProcessContext processContext = taskContext.getParentProcessContext(); |
| String errorMsg = new StringBuilder("expId: ").append(processContext.getExperimentId()).append(", processId: ") |
| .append(processContext.getProcessId()).append(", taskId: ").append(taskContext.getTaskId()) |
| .append(", type: ").append(taskContext.getTaskType().name()).append(" :- Environment Setup failed. Reason: ") |
| .append(taskStatus.getReason()).toString(); |
| ErrorModel errorModel = new ErrorModel(); |
| errorModel.setUserFriendlyMessage("Error while environment setup"); |
| errorModel.setActualErrorMessage(errorMsg); |
| GFacUtils.saveTaskError(taskContext, errorModel); |
| throw new GFacException("Error while environment setup"); |
| } |
| } catch (TException e) { |
| throw new GFacException("Couldn't get environment setup task model", e); |
| } |
| return false; |
| } |
| |
| private boolean inputDataStaging(TaskContext taskContext, boolean recover) throws GFacException, TException { |
| TaskStatus taskStatus = new TaskStatus(TaskState.EXECUTING); |
| taskStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime()); |
| taskContext.setTaskStatus(taskStatus); |
| GFacUtils.saveAndPublishTaskStatus(taskContext); |
| |
| ProcessContext processContext = taskContext.getParentProcessContext(); |
| // handle URI_COLLECTION input data type |
| Task dMoveTask = Factory.getDataMovementTask(processContext.getDataMovementProtocol()); |
| if (taskContext.getProcessInput().getType() == DataType.URI_COLLECTION) { |
| String values = taskContext.getProcessInput().getValue(); |
| String[] multiple_inputs = values.split(GFacConstants.MULTIPLE_INPUTS_SPLITTER); |
| DataStagingTaskModel subTaskModel = (DataStagingTaskModel) taskContext.getSubTaskModel(); |
| for (String input : multiple_inputs) { |
| taskContext.getProcessInput().setValue(input); |
| subTaskModel.setSource(input); |
| taskStatus = executeTask(taskContext, dMoveTask, false); |
| } |
| taskContext.getProcessInput().setValue(values); |
| } else { |
| taskStatus = executeTask(taskContext, dMoveTask, false); |
| } |
| taskStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime()); |
| taskContext.setTaskStatus(taskStatus); |
| GFacUtils.saveAndPublishTaskStatus(taskContext); |
| checkFailures(taskContext, taskStatus, dMoveTask); |
| return false; |
| } |
| |
| private void checkFailures(TaskContext taskContext, TaskStatus taskStatus, Task task) throws GFacException { |
| if (taskStatus.getState() == TaskState.FAILED) { |
| log.error("expId: {}, processId: {}, taskId: {} type: {},:- " + task.getType().toString() + " failed, " + |
| "reason:" + " {}", taskContext.getParentProcessContext().getExperimentId(), taskContext |
| .getParentProcessContext().getProcessId(), taskContext.getTaskId(), task.getType |
| ().name(), taskStatus.getReason()); |
| String errorMsg = new StringBuilder("expId: ").append(taskContext.getParentProcessContext().getExperimentId()).append(", processId: ") |
| .append(taskContext.getParentProcessContext().getProcessId()).append(", taskId: ").append(taskContext.getTaskId()) |
| .append(", type: ").append(taskContext.getTaskType().name()).append(" :- " + task.getType().toString() + " failed. Reason: ") |
| .append(taskStatus.getReason()).toString(); |
| ErrorModel errorModel = new ErrorModel(); |
| errorModel.setUserFriendlyMessage("Error while executing " + task.getType() + " task" ); |
| errorModel.setActualErrorMessage(errorMsg); |
| GFacUtils.saveTaskError(taskContext, errorModel); |
| throw new GFacException("Error: userFriendly msg :" + errorModel.getUserFriendlyMessage() + ", actual msg :" |
| + errorModel.getActualErrorMessage()); |
| } |
| } |
| |
| @Override |
| public void recoverProcess(ProcessContext processContext) throws GFacException { |
| processContext.setRecovery(true); |
| String taskDag = processContext.getProcessModel().getTaskDag(); |
| List<String> taskExecutionOrder = GFacUtils.parseTaskDag(taskDag); |
| processContext.setTaskExecutionOrder(taskExecutionOrder); |
| Map<String, TaskModel> taskMap = processContext.getTaskMap(); |
| String recoverTaskId = null; |
| String previousTaskId = null; |
| TaskModel taskModel = null; |
| for (String taskId : taskExecutionOrder) { |
| taskModel = taskMap.get(taskId); |
| TaskState state = taskModel.getTaskStatus().getState(); |
| if (state == TaskState.CREATED || state == TaskState.EXECUTING) { |
| recoverTaskId = taskId; |
| break; |
| } |
| previousTaskId = taskId; |
| } |
| final String rTaskId = recoverTaskId; |
| final String pTaskId = previousTaskId; |
| if (recoverTaskId != null) { |
| if (processContext.isRecoveryWithCancel()) { |
| cancelJobSubmission(processContext, rTaskId, pTaskId); |
| } |
| continueProcess(processContext, recoverTaskId); |
| } else { |
| log.error("expId: {}, processId: {}, Error while recovering process, couldn't find recovery task", |
| processContext.getExperimentId(), processContext.getProcessId()); |
| } |
| |
| |
| } |
| |
| private void cancelJobSubmission(ProcessContext processContext, String rTaskId, String pTaskId) { |
| new Thread(() -> { |
| try { |
| processContext.setCancel(true); |
| ProcessState processState = processContext.getProcessState(); |
| List<Object> jobModels = null; |
| switch (processState) { |
| case EXECUTING: |
| jobModels = processContext.getExperimentCatalog().get( |
| ExperimentCatalogModelType.JOB, Constants.FieldConstants.TaskConstants.TASK_ID, |
| rTaskId); |
| break; |
| case MONITORING: |
| if (pTaskId != null) { |
| jobModels = processContext.getExperimentCatalog().get( |
| ExperimentCatalogModelType.JOB, Constants.FieldConstants.TaskConstants.TASK_ID, |
| pTaskId); |
| } |
| } |
| |
| if (jobModels != null && !jobModels.isEmpty()) { |
| JobModel jobModel = (JobModel) jobModels.get(jobModels.size() - 1); |
| if (jobModel.getJobId() != null) { |
| processContext.setJobModel(jobModel); |
| log.info("expId: {}, processId: {}, Canceling jobId {}", processContext.getExperimentId(), |
| processContext.getProcessId(), jobModel.getJobId()); |
| cancelProcess(processContext); |
| log.info("expId: {}, processId: {}, Canceled jobId {}", processContext.getExperimentId(), |
| processContext.getProcessId(), jobModel.getJobId()); |
| } else { |
| log.error("expId: {}, processId: {}, Couldn't find jobId in jobModel, aborting process recovery", |
| processContext.getExperimentId(), processContext.getProcessId()); |
| } |
| } |
| } catch (GFacException e) { |
| log.error("expId: {}, processId: {}, Error while canceling process which is in recovery mode", |
| processContext.getExperimentId(), processContext.getProcessId()); |
| } catch (RegistryException e) { |
| log.error("expId: {}, processId: {}, Error while getting job model for taskId {}, " + |
| "couldn't cancel process which is in recovery mode", processContext.getExperimentId(), |
| processContext.getProcessId(), rTaskId); |
| } |
| }).start(); |
| } |
| |
| private JobModel getJobModel(ProcessContext processContext) { |
| try { |
| return GFacUtils.getJobModel(processContext); |
| } catch (RegistryException e) { |
| log.error("Error while retrieving jobId,", e); |
| return null; |
| } |
| } |
| |
| @Override |
| public void continueProcess(ProcessContext processContext, String taskId) throws GFacException { |
| executeTaskListFrom(processContext, taskId); |
| } |
| |
| /** |
| * @param processContext |
| * @param recovery |
| * @return <code>true</code> if you need to interrupt processing <code>false</code> otherwise. |
| * @throws GFacException |
| */ |
| private boolean postProcessing(ProcessContext processContext, boolean recovery) throws GFacException { |
| ProcessStatus status = new ProcessStatus(ProcessState.POST_PROCESSING); |
| status.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime()); |
| processContext.setProcessStatus(status); |
| GFacUtils.saveAndPublishProcessStatus(processContext); |
| // taskCtx = getEnvCleanupTaskContext(processContext); |
| if (processContext.isInterrupted()) { |
| GFacUtils.handleProcessInterrupt(processContext); |
| return true; |
| } |
| return false; |
| } |
| |
| /** |
| * @param taskContext |
| * @param recovery |
| * @return <code>true</code> if process execution interrupted , <code>false</code> otherwise. |
| * @throws GFacException |
| */ |
| private boolean outputDataStaging(TaskContext taskContext, boolean recovery, boolean isArchive) throws GFacException { |
| TaskStatus taskStatus = new TaskStatus(TaskState.EXECUTING); |
| taskStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime()); |
| taskContext.setTaskStatus(taskStatus); |
| GFacUtils.saveAndPublishTaskStatus(taskContext); |
| |
| ProcessContext processContext = taskContext.getParentProcessContext(); |
| Task dMoveTask = null; |
| if (isArchive) { |
| dMoveTask = Factory.getArchiveTask(); |
| } else { |
| dMoveTask = Factory.getDataMovementTask(processContext.getDataMovementProtocol()); |
| } |
| taskStatus = executeTask(taskContext, dMoveTask, recovery); |
| taskStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime()); |
| taskContext.setTaskStatus(taskStatus); |
| GFacUtils.saveAndPublishTaskStatus(taskContext); |
| |
| if (taskStatus.getState() == TaskState.FAILED) { |
| log.error("expId: {}, processId: {}, taskId: {} type: {},:- output staging failed, " + |
| "reason:" + " {}", taskContext.getParentProcessContext().getExperimentId(), taskContext |
| .getParentProcessContext().getProcessId(), taskContext.getTaskId(), dMoveTask.getType |
| ().name(), taskStatus.getReason()); |
| |
| String errorMsg = new StringBuilder("expId: ").append(processContext.getExperimentId()).append(", processId: ") |
| .append(processContext.getProcessId()).append(", taskId: ").append(taskContext.getTaskId()) |
| .append(", type: ").append(taskContext.getTaskType().name()).append(" :- Output staging failed. Reason: ") |
| .append(taskStatus.getReason()).toString(); |
| ErrorModel errorModel = new ErrorModel(); |
| errorModel.setUserFriendlyMessage("Error while staging output data"); |
| errorModel.setActualErrorMessage(errorMsg); |
| GFacUtils.saveTaskError(taskContext, errorModel); |
| } |
| return false; |
| } |
| |
| |
| @Override |
| public void cancelProcess(ProcessContext processContext) throws GFacException { |
| if (processContext != null) { |
| switch (processContext.getProcessState()) { |
| case MONITORING: case EXECUTING: |
| // get job submission task and invoke cancel |
| JobSubmissionTask jobSubmissionTask = Factory.getJobSubmissionTask(processContext.getJobSubmissionProtocol()); |
| TaskContext taskCtx = getJobSubmissionTaskContext(processContext); |
| executeCancel(taskCtx, jobSubmissionTask); |
| break; |
| case COMPLETED: case FAILED: case CANCELED : case CANCELLING: |
| log.warn("Process cancel trigger for already {} process", processContext.getProcessState().name()); |
| break; |
| default: |
| break; |
| } |
| } |
| } |
| |
| private TaskStatus executeTask(TaskContext taskCtx, Task task, boolean recover) throws GFacException { |
| TaskStatus taskStatus = null; |
| if (recover) { |
| taskStatus = task.recover(taskCtx); |
| } else { |
| taskStatus = task.execute(taskCtx); |
| } |
| return taskStatus; |
| } |
| |
| private void executeCancel(TaskContext taskContext, JobSubmissionTask jSTask) throws GFacException { |
| try { |
| JobStatus oldJobStatus = jSTask.cancel(taskContext); |
| // If Job was in Queued state when cancel command runs, then there won't be any email from this job. |
| ProcessContext pc = taskContext.getParentProcessContext(); |
| JobMonitor monitorService = Factory.getMonitorService(pc.getMonitorMode()); |
| monitorService.canceledJob(pc.getJobModel().getJobId()); |
| |
| } catch (TaskException e) { |
| throw new GFacException("Error while cancelling job"); |
| } catch (AiravataException e) { |
| throw new GFacException("Error wile getting monitoring service"); |
| } |
| } |
| |
| private TaskContext getJobSubmissionTaskContext(ProcessContext processContext) throws GFacException { |
| TaskContext taskCtx = new TaskContext(); |
| taskCtx.setParentProcessContext(processContext); |
| |
| TaskModel taskModel = new TaskModel(); |
| taskModel.setParentProcessId(processContext.getProcessId()); |
| taskModel.setCreationTime(new Date().getTime()); |
| taskModel.setLastUpdateTime(taskModel.getCreationTime()); |
| TaskStatus taskStatus = new TaskStatus(TaskState.CREATED); |
| taskStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime()); |
| taskModel.setTaskStatus(taskStatus); |
| taskModel.setTaskType(TaskTypes.JOB_SUBMISSION); |
| taskCtx.setTaskModel(taskModel); |
| return taskCtx; |
| } |
| |
| private TaskContext getDataStagingTaskContext(ProcessContext processContext, OutputDataObjectType processOutput) |
| throws TException, TaskException { |
| TaskContext taskCtx = new TaskContext(); |
| taskCtx.setParentProcessContext(processContext); |
| // create new task model for this task |
| TaskModel taskModel = new TaskModel(); |
| taskModel.setParentProcessId(processContext.getProcessId()); |
| taskModel.setCreationTime(AiravataUtils.getCurrentTimestamp().getTime()); |
| taskModel.setLastUpdateTime(taskModel.getCreationTime()); |
| TaskStatus taskStatus = new TaskStatus(TaskState.CREATED); |
| taskStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime()); |
| taskModel.setTaskStatus(taskStatus); |
| taskModel.setTaskType(TaskTypes.DATA_STAGING); |
| // create data staging sub task model |
| String remoteOutputDir = processContext.getOutputDir(); |
| remoteOutputDir = remoteOutputDir.endsWith("/") ? remoteOutputDir : remoteOutputDir + "/"; |
| DataStagingTaskModel submodel = new DataStagingTaskModel(); |
| ServerInfo serverInfo = processContext.getServerInfo(); |
| URI source = null; |
| try { |
| source = new URI(processContext.getDataMovementProtocol().name(), serverInfo.getHost(), |
| serverInfo.getUserName(), serverInfo.getPort(), remoteOutputDir + processOutput.getValue(), null, null); |
| } catch (URISyntaxException e) { |
| throw new TaskException("Error while constructing source file URI"); |
| } |
| submodel.setSource(source.toString()); |
| // We don't know destination location at this time, data staging task will set this. |
| // because destination is required field we set dummy destination |
| submodel.setDestination("dummy://temp/file/location"); |
| taskModel.setSubTaskModel(ThriftUtils.serializeThriftObject(submodel)); |
| taskCtx.setTaskModel(taskModel); |
| taskCtx.setProcessOutput(processOutput); |
| return taskCtx; |
| } |
| |
| /** |
| * Persist task model |
| */ |
| private void saveTaskModel(TaskContext taskContext) throws GFacException { |
| try { |
| TaskModel taskModel = taskContext.getTaskModel(); |
| taskContext.getParentProcessContext().getExperimentCatalog().add(ExpCatChildDataType.TASK, taskModel, |
| taskModel.getParentProcessId()); |
| } catch (RegistryException e) { |
| throw new GFacException("Error while saving task model", e); |
| } |
| } |
| |
| private TaskContext getTaskContext(ProcessContext processContext) { |
| TaskContext taskCtx = new TaskContext(); |
| taskCtx.setParentProcessContext(processContext); |
| return taskCtx; |
| } |
| |
| |
| /** |
| * Sort input data type by input order. |
| */ |
| private void sortByInputOrder(List<InputDataObjectType> processInputs) { |
| Collections.sort(processInputs, new Comparator<InputDataObjectType>() { |
| @Override |
| public int compare(InputDataObjectType inputDT_1, InputDataObjectType inputDT_2) { |
| return inputDT_1.getInputOrder() - inputDT_2.getInputOrder(); |
| } |
| }); |
| } |
| |
| private void updateProcessFailure(ProcessContext pc, String reason){ |
| ProcessStatus status = new ProcessStatus(ProcessState.FAILED); |
| status.setReason(reason); |
| pc.setProcessStatus(status); |
| try { |
| GFacUtils.saveAndPublishProcessStatus(pc); |
| } catch (GFacException e) { |
| log.error("Error while save and publishing process failed status event"); |
| } |
| } |
| |
| public static ResourceJobManager getResourceJobManager(ProcessContext processCtx) throws AppCatalogException, GFacException { |
| List<JobSubmissionInterface> jobSubmissionInterfaces = Factory.getDefaultAppCatalog().getComputeResource() |
| .getComputeResource(processCtx.getComputeResourceId()).getJobSubmissionInterfaces(); |
| |
| ResourceJobManager resourceJobManager = null; |
| JobSubmissionInterface jsInterface = null; |
| for (JobSubmissionInterface jobSubmissionInterface : jobSubmissionInterfaces) { |
| if (jobSubmissionInterface.getJobSubmissionProtocol() == processCtx.getJobSubmissionProtocol()) { |
| jsInterface = jobSubmissionInterface; |
| break; |
| } |
| } |
| if (jsInterface == null) { |
| throw new GFacException("Job Submission interface cannot be empty at this point"); |
| } else if (jsInterface.getJobSubmissionProtocol() == JobSubmissionProtocol.SSH) { |
| SSHJobSubmission sshJobSubmission = Factory.getDefaultAppCatalog().getComputeResource().getSSHJobSubmission |
| (jsInterface.getJobSubmissionInterfaceId()); |
| processCtx.setMonitorMode(sshJobSubmission.getMonitorMode()); // fixme - Move this to populate process |
| // context method. |
| resourceJobManager = sshJobSubmission.getResourceJobManager(); |
| } else if (jsInterface.getJobSubmissionProtocol() == JobSubmissionProtocol.LOCAL) { |
| LOCALSubmission localSubmission = Factory.getDefaultAppCatalog().getComputeResource().getLocalJobSubmission |
| (jsInterface.getJobSubmissionInterfaceId()); |
| resourceJobManager = localSubmission.getResourceJobManager(); |
| } else if (jsInterface.getJobSubmissionProtocol() == JobSubmissionProtocol.SSH_FORK) { |
| SSHJobSubmission sshJobSubmission = Factory.getDefaultAppCatalog().getComputeResource().getSSHJobSubmission |
| (jsInterface.getJobSubmissionInterfaceId()); |
| processCtx.setMonitorMode(sshJobSubmission.getMonitorMode()); // fixme - Move this to populate process |
| resourceJobManager = sshJobSubmission.getResourceJobManager(); |
| } else { |
| throw new GFacException("Unsupported JobSubmissionProtocol - " + jsInterface.getJobSubmissionProtocol() |
| .name()); |
| } |
| |
| if (resourceJobManager == null) { |
| throw new GFacException("Resource Job Manager is empty."); |
| } |
| return resourceJobManager; |
| } |
| } |
