Merge pull request #34 from uc-cdis/chore/monitored_jobs

Chore/monitored jobs

Author: giangbui

Dockerfile

@@ -1,4 +1,4 @@
-FROM golang:1.14 as build-deps
+FROM quay.io/cdis/golang:1.15 as build-deps
 
 # Build static binary
 RUN mkdir -p /go/src/github.com/uc-cdis/ssjdispatcher

handlers/configuration.go

@@ -29,7 +29,7 @@ func GetMaxJobConfig() int {
 func GetCleanupTime() int {
 	cleanupTime, err := strconv.Atoi(os.Getenv("CLEANUP_TIME"))
 	if err != nil {
-		cleanupTime = 60
+		cleanupTime = 120
 	}
 	return cleanupTime
 }

handlers/handler.go

@@ -132,25 +132,26 @@ func (handler *SQSHandler) ResendSQSMessage(queueURL string, message *sqs.Messag
 func (handler *SQSHandler) StartMonitoringProcess() {
 	for {
 		var nextMonitoredJobs []*JobInfo
-
+		handler.Mu.Lock()
 		for _, jobInfo := range handler.MonitoredJobs {
 			if jobInfo.Status == "Completed" {
 				nextMonitoredJobs = append(nextMonitoredJobs, jobInfo)
 			} else {
 				k8sJob, err := GetJobStatusByID(jobInfo.UID)
 				if err != nil {
-					glog.Errorf("Can not get k8s job %s. Detail %s. Resend the message to the queue", jobInfo.Name, err)
-					handler.ResendSQSMessage(handler.QueueURL, jobInfo.SQSMessage)
+					jobInfo.Status = "Unknown"
 				} else {
 					glog.Infof("%s: %s", k8sJob.Name, k8sJob.Status)
 					if k8sJob.Status == "Unknown" || k8sJob.Status == "Running" || k8sJob.Status == "Completed" {
 						jobInfo.Status = k8sJob.Status
-						nextMonitoredJobs = append(nextMonitoredJobs, jobInfo)
+					} else if k8sJob.Status == "Failed" {
+						glog.Errorf("The k8s job %s failed. Detail %s. Resend the message to the queue", jobInfo.Name, err)
+						handler.ResendSQSMessage(handler.QueueURL, jobInfo.SQSMessage)
 					}
 				}
+				nextMonitoredJobs = append(nextMonitoredJobs, jobInfo)
 			}
 		}
-		handler.Mu.Lock()
 		handler.MonitoredJobs = nextMonitoredJobs
 		handler.Mu.Unlock()
 
@@ -163,7 +164,22 @@ func (handler *SQSHandler) RemoveCompletedJobsProcess() {
 	for {
 		time.Sleep(time.Duration(GetCleanupTime()) * time.Second)
 		glog.Info("Start to remove completed jobs")
-		RemoveCompletedJobs(handler.MonitoredJobs)
+		handler.Mu.Lock()
+		var tmp []*JobInfo
+		deletedJobs := RemoveCompletedJobs(handler.MonitoredJobs)
+		for _, jobInfo := range handler.MonitoredJobs {
+			var isDeleted = false
+			for _, jobid := range deletedJobs {
+				if jobInfo.UID == jobid {
+					isDeleted = true
+				}
+			}
+			if isDeleted == false {
+				tmp = append(tmp, jobInfo)
+			}
+		}
+		handler.MonitoredJobs = tmp
+		handler.Mu.Unlock()
 	}
 }
 

handlers/jobs.go

@@ -140,8 +140,9 @@ func jobStatusToString(status *batchv1.JobStatus) string {
 }
 
 // RemoveCompletedJobs removes all completed k8s jobs dispatched by the service
-func RemoveCompletedJobs(monitoredJobs []*JobInfo) {
+func RemoveCompletedJobs(monitoredJobs []*JobInfo) []string {
 	jobs := listJobs(getJobClient())
+	var deletedJobs []string
 	for i := 0; i < len(jobs.JobInfo); i++ {
 		job := jobs.JobInfo[i]
 		if job.Status == "Completed" {
@@ -154,9 +155,11 @@ func RemoveCompletedJobs(monitoredJobs []*JobInfo) {
 			}
 			if isMonitoredJob == true {
 				deleteJobByID(job.UID, GRACE_PERIOD)
+				deletedJobs = append(deletedJobs, job.UID)
 			}
 		}
 	}
+	return deletedJobs
 }
 
 // GetNumberRunningJobs returns number of k8s running jobs dispatched by the service