<commit_before>/*
 *
 * 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.task;

import org.apache.airavata.common.exception.ApplicationSettingsException;
import org.apache.airavata.common.utils.AiravataUtils;
import org.apache.airavata.gfac.core.*;
import org.apache.airavata.gfac.core.cluster.CommandInfo;
import org.apache.airavata.gfac.core.cluster.JobSubmissionOutput;
import org.apache.airavata.gfac.core.cluster.RawCommandInfo;
import org.apache.airavata.gfac.core.cluster.RemoteCluster;
import org.apache.airavata.gfac.core.context.ProcessContext;
import org.apache.airavata.gfac.core.context.TaskContext;
import org.apache.airavata.gfac.core.task.JobSubmissionTask;
import org.apache.airavata.gfac.core.task.TaskException;
import org.apache.airavata.gfac.impl.Factory;
import org.apache.airavata.model.appcatalog.computeresource.ComputeResourceDescription;
import org.apache.airavata.model.appcatalog.computeresource.ResourceJobManager;
import org.apache.airavata.model.commons.ErrorModel;
import org.apache.airavata.model.experiment.ExperimentModel;
import org.apache.airavata.model.job.JobModel;
import org.apache.airavata.model.status.*;
import org.apache.airavata.model.task.TaskTypes;
import org.apache.airavata.registry.cpi.AppCatalogException;
import org.apache.airavata.registry.cpi.ExperimentCatalogModelType;
import org.apache.airavata.registry.cpi.RegistryException;
import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.util.Map;

public class DefaultJobSubmissionTask implements JobSubmissionTask {
	private static final Logger log = LoggerFactory.getLogger(DefaultJobSubmissionTask.class);
	public static final String DEFAULT_JOB_ID = "DEFAULT_JOB_ID";
	private static int waitForProcessIdmillis = 5000;
	private static int pauseTimeInSec = waitForProcessIdmillis / 1000;

	@Override
    public void init(Map<String, String> propertyMap) throws TaskException {

    }

    @Override
    public TaskStatus execute(TaskContext taskContext){
	    TaskStatus taskStatus = new TaskStatus(TaskState.COMPLETED); // set to completed.
	    try {
		    ProcessContext processContext = taskContext.getParentProcessContext();
		    JobModel jobModel = processContext.getJobModel();
		    jobModel.setTaskId(taskContext.getTaskId());
		    RemoteCluster remoteCluster = processContext.getJobSubmissionRemoteCluster();
		    JobDescriptor jobDescriptor = GFacUtils.createJobDescriptor(processContext,taskContext);
		    jobModel.setJobName(jobDescriptor.getJobName());
		    ResourceJobManager resourceJobManager = GFacUtils.getResourceJobManager(processContext);
		    JobManagerConfiguration jConfig = null;
		    if (resourceJobManager != null) {
			    jConfig = Factory.getJobManagerConfiguration(resourceJobManager);
		    }
		    JobStatus jobStatus = new JobStatus();
		    File jobFile = GFacUtils.createJobFile(taskContext, jobDescriptor, jConfig);
		    if (jobFile != null && jobFile.exists()) {
			    jobModel.setJobDescription(FileUtils.readFileToString(jobFile));
			    JobSubmissionOutput jobSubmissionOutput = remoteCluster.submitBatchJob(jobFile.getPath(),
					    processContext.getWorkingDir());
				int exitCode = jobSubmissionOutput.getExitCode();
				jobModel.setExitCode(exitCode);
				jobModel.setStdErr(jobSubmissionOutput.getStdErr());
				jobModel.setStdOut(jobSubmissionOutput.getStdOut());
				String jobId = jobSubmissionOutput.getJobId();
                String experimentId = taskContext.getExperimentId();
                if (exitCode != 0 || jobSubmissionOutput.isJobSubmissionFailed()) {
					jobModel.setJobId(DEFAULT_JOB_ID);
					if (jobSubmissionOutput.isJobSubmissionFailed()) {
						jobModel.setJobStatus(new JobStatus(JobState.FAILED));
						jobModel.getJobStatus().setReason(jobSubmissionOutput.getFailureReason());
						log.error("expId: {}, processid: {}, taskId: {} :- Job submission failed for job name {}",
                                experimentId, taskContext.getProcessId(), taskContext.getTaskId(), jobModel.getJobName());
						ErrorModel errorModel = new ErrorModel();
						errorModel.setUserFriendlyMessage(jobSubmissionOutput.getFailureReason());
						errorModel.setActualErrorMessage(jobSubmissionOutput.getFailureReason());
						GFacUtils.saveExperimentError(processContext, errorModel);
						GFacUtils.saveProcessError(processContext, errorModel);
						GFacUtils.saveTaskError(taskContext, errorModel);
						taskStatus.setState(TaskState.FAILED);
						taskStatus.setReason("Job submission command exit with non zero exit code");
						taskStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime());
						taskContext.setTaskStatus(taskStatus);
					} else {
						String msg;
						GFacUtils.saveJobModel(processContext, jobModel);
						if (exitCode != Integer.MIN_VALUE) {
							msg = "expId:" + processContext.getProcessModel().getExperimentId() + ", processId:" +
									processContext.getProcessId() + ", taskId: " + taskContext.getTaskId() +
									" return non zero exit code:" + exitCode + "  for JobName:" + jobModel.getJobName() +
									", with failure reason :  " + jobSubmissionOutput.getStdOut() +  " Hence changing job state to Failed." ;
						} else {
							msg = "expId:" + processContext.getProcessModel().getExperimentId() + ", processId:" +
									processContext.getProcessId() + ", taskId: " + taskContext.getTaskId() +
									" doesn't  return valid job submission exit code for JobName:" + jobModel.getJobName() +
                                    ", with failure reason :  " + jobSubmissionOutput.getStdOut() +  " Hence changing job state to Failed." ;
						}
						log.error(msg);
						ErrorModel errorModel = new ErrorModel();
						errorModel.setUserFriendlyMessage(msg);
						errorModel.setActualErrorMessage(msg);
						GFacUtils.saveExperimentError(processContext, errorModel);
						GFacUtils.saveProcessError(processContext, errorModel);
						GFacUtils.saveTaskError(taskContext, errorModel);
						taskStatus.setState(TaskState.FAILED);
						taskStatus.setReason(msg);
						taskStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime());
						taskContext.setTaskStatus(taskStatus);
					}
					try {
						GFacUtils.saveAndPublishTaskStatus(taskContext);
					} catch (GFacException e) {
						log.error("Error while saving task status", e);
					}
					return taskStatus;
				} else if (jobId != null && !jobId.isEmpty()) {
				    jobModel.setJobId(jobId);
				    GFacUtils.saveJobModel(processContext, jobModel);
				    jobStatus.setJobState(JobState.SUBMITTED);
                    ComputeResourceDescription computeResourceDescription = taskContext.getParentProcessContext()
                            .getComputeResourceDescription();
                    jobStatus.setReason("Successfully Submitted to " + computeResourceDescription.getHostName());
                    jobStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime());
				    jobModel.setJobStatus(jobStatus);
				    GFacUtils.saveJobStatus(taskContext.getParentProcessContext(), jobModel);
				    if (verifyJobSubmissionByJobId(remoteCluster, jobId)) {
					    jobStatus.setJobState(JobState.QUEUED);
					    jobStatus.setReason("Verification step succeeded");
                        jobStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime());
					    jobModel.setJobStatus(jobStatus);
					    GFacUtils.saveJobStatus(taskContext.getParentProcessContext(), jobModel);
				    }
                    // doing gateway reporting
                    if (computeResourceDescription.isGatewayUsageReporting()){
                        String loadCommand = computeResourceDescription.getGatewayUsageModuleLoadCommand();
                        String usageExecutable = computeResourceDescription.getGatewayUsageExecutable();
                        ExperimentModel experiment = (ExperimentModel)taskContext.getParentProcessContext().getExperimentCatalog().get(ExperimentCatalogModelType.EXPERIMENT, experimentId);
                        RawCommandInfo rawCommandInfo = new RawCommandInfo(loadCommand + " && " + usageExecutable + " -gateway_user " +  experiment.getUserName()  +
                                                                           " -submit_time \"`date '+%F %T %:z'`\"  -jobid " + jobId );
                        remoteCluster.execute(rawCommandInfo);
                    }
				    taskStatus = new TaskStatus(TaskState.COMPLETED);
				    taskStatus.setReason("Submitted job to compute resource");
                    taskStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime());
				} else {
					int verificationTryCount = 0;
					while (verificationTryCount++ < 3) {<commit_msg>adding correct username for gateway usage<commit_after>/*
 *
 * 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.task;

import org.apache.airavata.common.exception.ApplicationSettingsException;
import org.apache.airavata.common.utils.AiravataUtils;
import org.apache.airavata.gfac.core.*;
import org.apache.airavata.gfac.core.cluster.CommandInfo;
import org.apache.airavata.gfac.core.cluster.JobSubmissionOutput;
import org.apache.airavata.gfac.core.cluster.RawCommandInfo;
import org.apache.airavata.gfac.core.cluster.RemoteCluster;
import org.apache.airavata.gfac.core.context.ProcessContext;
import org.apache.airavata.gfac.core.context.TaskContext;
import org.apache.airavata.gfac.core.task.JobSubmissionTask;
import org.apache.airavata.gfac.core.task.TaskException;
import org.apache.airavata.gfac.impl.Factory;
import org.apache.airavata.model.appcatalog.computeresource.ComputeResourceDescription;
import org.apache.airavata.model.appcatalog.computeresource.ResourceJobManager;
import org.apache.airavata.model.commons.ErrorModel;
import org.apache.airavata.model.experiment.ExperimentModel;
import org.apache.airavata.model.job.JobModel;
import org.apache.airavata.model.status.*;
import org.apache.airavata.model.task.TaskTypes;
import org.apache.airavata.registry.cpi.AppCatalogException;
import org.apache.airavata.registry.cpi.ExperimentCatalogModelType;
import org.apache.airavata.registry.cpi.RegistryException;
import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.util.Map;

public class DefaultJobSubmissionTask implements JobSubmissionTask {
	private static final Logger log = LoggerFactory.getLogger(DefaultJobSubmissionTask.class);
	public static final String DEFAULT_JOB_ID = "DEFAULT_JOB_ID";
	private static int waitForProcessIdmillis = 5000;
	private static int pauseTimeInSec = waitForProcessIdmillis / 1000;

	@Override
    public void init(Map<String, String> propertyMap) throws TaskException {

    }

    @Override
    public TaskStatus execute(TaskContext taskContext){
	    TaskStatus taskStatus = new TaskStatus(TaskState.COMPLETED); // set to completed.
	    try {
		    ProcessContext processContext = taskContext.getParentProcessContext();
		    JobModel jobModel = processContext.getJobModel();
		    jobModel.setTaskId(taskContext.getTaskId());
		    RemoteCluster remoteCluster = processContext.getJobSubmissionRemoteCluster();
		    JobDescriptor jobDescriptor = GFacUtils.createJobDescriptor(processContext,taskContext);
		    jobModel.setJobName(jobDescriptor.getJobName());
		    ResourceJobManager resourceJobManager = GFacUtils.getResourceJobManager(processContext);
		    JobManagerConfiguration jConfig = null;
		    if (resourceJobManager != null) {
			    jConfig = Factory.getJobManagerConfiguration(resourceJobManager);
		    }
		    JobStatus jobStatus = new JobStatus();
		    File jobFile = GFacUtils.createJobFile(taskContext, jobDescriptor, jConfig);
		    if (jobFile != null && jobFile.exists()) {
			    jobModel.setJobDescription(FileUtils.readFileToString(jobFile));
			    JobSubmissionOutput jobSubmissionOutput = remoteCluster.submitBatchJob(jobFile.getPath(),
					    processContext.getWorkingDir());
				int exitCode = jobSubmissionOutput.getExitCode();
				jobModel.setExitCode(exitCode);
				jobModel.setStdErr(jobSubmissionOutput.getStdErr());
				jobModel.setStdOut(jobSubmissionOutput.getStdOut());
				String jobId = jobSubmissionOutput.getJobId();
                String experimentId = taskContext.getExperimentId();
                if (exitCode != 0 || jobSubmissionOutput.isJobSubmissionFailed()) {
					jobModel.setJobId(DEFAULT_JOB_ID);
					if (jobSubmissionOutput.isJobSubmissionFailed()) {
						jobModel.setJobStatus(new JobStatus(JobState.FAILED));
						jobModel.getJobStatus().setReason(jobSubmissionOutput.getFailureReason());
						log.error("expId: {}, processid: {}, taskId: {} :- Job submission failed for job name {}",
                                experimentId, taskContext.getProcessId(), taskContext.getTaskId(), jobModel.getJobName());
						ErrorModel errorModel = new ErrorModel();
						errorModel.setUserFriendlyMessage(jobSubmissionOutput.getFailureReason());
						errorModel.setActualErrorMessage(jobSubmissionOutput.getFailureReason());
						GFacUtils.saveExperimentError(processContext, errorModel);
						GFacUtils.saveProcessError(processContext, errorModel);
						GFacUtils.saveTaskError(taskContext, errorModel);
						taskStatus.setState(TaskState.FAILED);
						taskStatus.setReason("Job submission command exit with non zero exit code");
						taskStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime());
						taskContext.setTaskStatus(taskStatus);
					} else {
						String msg;
						GFacUtils.saveJobModel(processContext, jobModel);
						if (exitCode != Integer.MIN_VALUE) {
							msg = "expId:" + processContext.getProcessModel().getExperimentId() + ", processId:" +
									processContext.getProcessId() + ", taskId: " + taskContext.getTaskId() +
									" return non zero exit code:" + exitCode + "  for JobName:" + jobModel.getJobName() +
									", with failure reason :  " + jobSubmissionOutput.getStdOut() +  " Hence changing job state to Failed." ;
						} else {
							msg = "expId:" + processContext.getProcessModel().getExperimentId() + ", processId:" +
									processContext.getProcessId() + ", taskId: " + taskContext.getTaskId() +
									" doesn't  return valid job submission exit code for JobName:" + jobModel.getJobName() +
                                    ", with failure reason :  " + jobSubmissionOutput.getStdOut() +  " Hence changing job state to Failed." ;
						}
						log.error(msg);
						ErrorModel errorModel = new ErrorModel();
						errorModel.setUserFriendlyMessage(msg);
						errorModel.setActualErrorMessage(msg);
						GFacUtils.saveExperimentError(processContext, errorModel);
						GFacUtils.saveProcessError(processContext, errorModel);
						GFacUtils.saveTaskError(taskContext, errorModel);
						taskStatus.setState(TaskState.FAILED);
						taskStatus.setReason(msg);
						taskStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime());
						taskContext.setTaskStatus(taskStatus);
					}
					try {
						GFacUtils.saveAndPublishTaskStatus(taskContext);
					} catch (GFacException e) {
						log.error("Error while saving task status", e);
					}
					return taskStatus;
				} else if (jobId != null && !jobId.isEmpty()) {
				    jobModel.setJobId(jobId);
				    GFacUtils.saveJobModel(processContext, jobModel);
				    jobStatus.setJobState(JobState.SUBMITTED);
                    ComputeResourceDescription computeResourceDescription = taskContext.getParentProcessContext()
                            .getComputeResourceDescription();
                    jobStatus.setReason("Successfully Submitted to " + computeResourceDescription.getHostName());
                    jobStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime());
				    jobModel.setJobStatus(jobStatus);
				    GFacUtils.saveJobStatus(taskContext.getParentProcessContext(), jobModel);
				    if (verifyJobSubmissionByJobId(remoteCluster, jobId)) {
					    jobStatus.setJobState(JobState.QUEUED);
					    jobStatus.setReason("Verification step succeeded");
                        jobStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime());
					    jobModel.setJobStatus(jobStatus);
					    GFacUtils.saveJobStatus(taskContext.getParentProcessContext(), jobModel);
				    }
                    // doing gateway reporting
                    if (computeResourceDescription.isGatewayUsageReporting()){
                        String loadCommand = computeResourceDescription.getGatewayUsageModuleLoadCommand();
                        String usageExecutable = computeResourceDescription.getGatewayUsageExecutable();
                        ExperimentModel experiment = (ExperimentModel)taskContext.getParentProcessContext().getExperimentCatalog().get(ExperimentCatalogModelType.EXPERIMENT, experimentId);
                        String username = experiment.getUserName() + "@" + taskContext.getParentProcessContext().getComputeResourcePreference().getUsageReportingGatewayId();
                        RawCommandInfo rawCommandInfo = new RawCommandInfo(loadCommand + " && " + usageExecutable + " -gateway_user " +  username  +
                                                                           " -submit_time \"`date '+%F %T %:z'`\"  -jobid " + jobId );
                        remoteCluster.execute(rawCommandInfo);
                    }
				    taskStatus = new TaskStatus(TaskState.COMPLETED);
				    taskStatus.setReason("Submitted job to compute resource");
                    taskStatus.setTimeOfStateChange(AiravataUtils.getCurrentTimestamp().getTime());
				} else {
					int verificationTryCount = 0;
					while (verificationTryCount++ < 3) {
{'fraction_non_alphanumeric': 0.09622524752475248, 'fraction_numerical': 0.0017533003300330033, 'mean_word_length': 3.733463509885282, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7255347', 'n_tokens_mistral': 5907, 'n_tokens_neox': 5133, 'n_words': 993}