Remove job state update when calculating job state (#1718)

Author: vishalbollu

pkg/operator/resources/batchapi/job_state.go

@@ -17,7 +17,6 @@ limitations under the License.
 package batchapi
 
 import (
-	"fmt"
 	"path"
 	"path/filepath"
 	"time"
@@ -215,7 +214,12 @@ func setStatusForJob(jobKey spec.JobKey, jobStatus status.JobCode) error {
 }
 
 func setEnqueuingStatus(jobKey spec.JobKey) error {
-	err := config.AWS.UploadStringToS3("", config.Cluster.Bucket, path.Join(jobKey.Prefix(config.Cluster.ClusterName), status.JobEnqueuing.String()))
+	err := updateLiveness(jobKey)
+	if err != nil {
+		return err
+	}
+
+	err = config.AWS.UploadStringToS3("", config.Cluster.Bucket, path.Join(jobKey.Prefix(config.Cluster.ClusterName), status.JobEnqueuing.String()))
 	if err != nil {
 		return err
 	}
@@ -354,88 +358,49 @@ func setTimedOutStatus(jobKey spec.JobKey) error {
 	return nil
 }
 
-func getJobStatusFromJobState(initialJobState *JobState, k8sJob *kbatch.Job, pods []kcore.Pod) (*status.JobStatus, error) {
-	jobKey := initialJobState.JobKey
+func getJobStatusFromJobState(jobState *JobState, k8sJob *kbatch.Job, pods []kcore.Pod) (*status.JobStatus, error) {
+	jobKey := jobState.JobKey
 
 	jobSpec, err := downloadJobSpec(jobKey)
 	if err != nil {
 		return nil, err
 	}
 
-	latestJobState := initialJobState // Refetch the state of an in progress job in case the cron modifies the job state between the time the initial fetch and now
-
-	if initialJobState.Status.IsInProgress() {
-		queueURL, err := getJobQueueURL(jobKey)
-		if err != nil {
-			return nil, err
-		}
-
-		latestJobCode, message, err := reconcileInProgressJob(initialJobState, &queueURL, k8sJob)
-		if err != nil {
-			return nil, err
-		}
-
-		if latestJobCode != initialJobState.Status {
-			err := errors.FirstError(
-				writeToJobLogStream(jobKey, message),
-				setStatusForJob(jobKey, latestJobCode),
-			)
-			if err != nil {
-				return nil, err
-			}
-		}
-
-		latestJobState, err = getJobState(jobKey)
-		if err != nil {
-			return nil, err
-		}
-	}
-
 	jobStatus := status.JobStatus{
 		Job:     *jobSpec,
-		EndTime: latestJobState.EndTime,
-		Status:  latestJobState.Status,
+		EndTime: jobState.EndTime,
+		Status:  jobState.Status,
 	}
 
-	if latestJobState.Status.IsInProgress() {
+	if jobState.Status.IsInProgress() {
 		queueMetrics, err := getQueueMetrics(jobKey)
 		if err != nil {
 			return nil, err
 		}
 
 		jobStatus.BatchesInQueue = queueMetrics.TotalUserMessages()
 
-		if latestJobState.Status == status.JobEnqueuing {
+		if jobState.Status == status.JobEnqueuing {
 			jobStatus.TotalBatchCount = queueMetrics.TotalUserMessages()
 		}
 
-		if latestJobState.Status == status.JobRunning {
+		if jobState.Status == status.JobRunning {
 			metrics, err := getRealTimeBatchMetrics(jobKey)
 			if err != nil {
 				return nil, err
 			}
 			jobStatus.BatchMetrics = metrics
 
-			if k8sJob == nil {
-				err := setUnexpectedErrorStatus(jobKey)
-				if err != nil {
-					return nil, err
-				}
-
-				writeToJobLogStream(jobKey, fmt.Sprintf("unexpected: kubernetes job not found"))
-				deleteJobRuntimeResources(jobKey)
-
-				jobStatus.Status = status.JobUnexpectedError
-				return &jobStatus, nil
+			// There can be race conditions where the job state is temporarily out of sync with the cluster state
+			if k8sJob != nil {
+				workerCounts := getWorkerCountsForJob(*k8sJob, pods)
+				jobStatus.WorkerCounts = &workerCounts
 			}
-
-			workerCounts := getWorkerCountsForJob(*k8sJob, pods)
-			jobStatus.WorkerCounts = &workerCounts
 		}
 	}
 
-	if latestJobState.Status.IsCompleted() {
-		metrics, err := getCompletedBatchMetrics(jobKey, jobSpec.StartTime, *latestJobState.EndTime)
+	if jobState.Status.IsCompleted() {
+		metrics, err := getCompletedBatchMetrics(jobKey, jobSpec.StartTime, *jobState.EndTime)
 		if err != nil {
 			return nil, err
 		}

pkg/operator/resources/batchapi/manage_resources_cron.go

@@ -249,7 +249,7 @@ func reconcileInProgressJob(jobState *JobState, queueURL *string, k8sJob *kbatch
 		return status.JobUnexpectedError, fmt.Sprintf("terminating job %s; sqs queue with url %s was not found", jobKey.UserString(), expectedQueueURL), nil
 	}
 
-	if jobState.Status == status.JobEnqueuing && time.Now().Sub(jobState.LastUpdatedMap[_enqueuingLivenessFile]) >= _enqueuingLivenessPeriod+_enqueuingLivenessBuffer {
+	if jobState.Status == status.JobEnqueuing && time.Since(jobState.LastUpdatedMap[_enqueuingLivenessFile]) >= _enqueuingLivenessPeriod+_enqueuingLivenessBuffer {
 		return status.JobEnqueueFailed, fmt.Sprintf("terminating job %s; enqueuing liveness check failed", jobKey.UserString()), nil
 	}