Merge pull request #10 from uc-cdis/feat/job_status

Feat/job status

Author: giangbui

handlers/handler.go

@@ -6,18 +6,23 @@ import (
 	"errors"
 	"fmt"
 	"regexp"
+	"sync"
 	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/golang/glog"
 	mq "github.com/remind101/mq-go"
 )
 
+const MAX_RETRIES = 3
+
 type SQSHandler struct {
-	QueueURL   string
-	Start      bool
-	JobConfigs []JobConfig
-	Server     *mq.Server
+	QueueURL      string
+	Start         bool
+	JobConfigs    []JobConfig
+	Server        *mq.Server
+	MonitoredJobs []*JobInfo
+	Mu            sync.Mutex
 }
 
 type JobConfig struct {
@@ -44,21 +49,51 @@ func (handler *SQSHandler) StartServer() error {
 		return nil
 	}
 
-	fmt.Println("Start a new server")
 	newClient, err := NewSQSClient()
 	if err != nil {
 		return err
 	}
 
+	glog.Info("Starting a new server...")
 	handler.Server = mq.NewServer(handler.QueueURL, mq.HandlerFunc(func(m *mq.Message) error {
 		return handler.HandleSQSMessage(m)
 	}), mq.WithClient(newClient))
 	handler.Server.Start()
+	glog.Info("The server is started")
+
+	go handler.StartMonitoringProcess()
 
 	return nil
 
 }
 
+// StartMonitoringProcess starts the process to monitor the created job
+func (handler *SQSHandler) StartMonitoringProcess() {
+	for {
+		var nextMonitoredJobs []*JobInfo
+
+		for _, jobInfo := range handler.MonitoredJobs {
+			k8sJob, err := GetJobStatusByID(jobInfo.UID)
+			if err != nil {
+				glog.Errorf("Can not get k8s job %s. Detail %s", jobInfo.Name, err)
+			} else {
+				glog.Infof("%s: %s", k8sJob.Name, k8sJob.Status)
+				if k8sJob.Status == "Unknown" || k8sJob.Status == "Running" {
+					nextMonitoredJobs = append(nextMonitoredJobs, jobInfo)
+				}
+			}
+
+		}
+		handler.Mu.Lock()
+		handler.MonitoredJobs = nextMonitoredJobs
+		handler.Mu.Unlock()
+
+		RemoveCompletedJobs()
+
+		time.Sleep(30 * time.Second)
+	}
+}
+
 // ShutdownServer shutdowns a server
 func (handler *SQSHandler) ShutdownServer() error {
 	fmt.Println("Shutdown the server")
@@ -162,9 +197,6 @@ func (handler *SQSHandler) HandleSQSMessage(m *mq.Message) error {
 		jobNameList = append(jobNameList, jobConfig.Name)
 	}
 
-	// remove completed jobs
-	RemoveCompletedJobs(jobNameList)
-
 	jobMap := make(map[string]JobConfig)
 	for _, objectPath := range objectPaths {
 		for _, jobConfig := range handler.JobConfigs {
@@ -178,21 +210,24 @@ func (handler *SQSHandler) HandleSQSMessage(m *mq.Message) error {
 	glog.Infof("Start to run %d jobs", len(jobMap))
 
 	for objectPath, jobConfig := range jobMap {
-		for GetNumberRunningJobs(jobNameList) > GetMaxJobConfig() {
+		for GetNumberRunningJobs() > GetMaxJobConfig() {
 			time.Sleep(5 * time.Second)
 		}
 		glog.Info("Processing: ", objectPath)
-		result, err := CreateK8sJob(objectPath, jobConfig)
+		jobInfo, err := CreateK8sJob(objectPath, jobConfig)
 		if err != nil {
 			glog.Errorln(err)
 			return err
 		}
-		out, err := json.Marshal(result)
+		out, err := json.Marshal(jobInfo)
 		if err != nil {
 			glog.Errorln(err)
 			return err
 		}
 		glog.Info(string(out))
+		handler.Mu.Lock()
+		handler.MonitoredJobs = append(handler.MonitoredJobs, jobInfo)
+		handler.Mu.Unlock()
 	}
 
 	return nil

handlers/jobs.go

@@ -59,7 +59,8 @@ func getJobByID(jc batchtypev1.JobInterface, jobid string) (*batchv1.Job, error)
 	return nil, fmt.Errorf("job with jobid %s not found", jobid)
 }
 
-func getJobStatusByID(jobid string) (*JobInfo, error) {
+//GetJobStatusByID returns job status given job id
+func GetJobStatusByID(jobid string) (*JobInfo, error) {
 	job, err := getJobByID(getJobClient(), jobid)
 	if err != nil {
 		return nil, err
@@ -136,7 +137,7 @@ func jobStatusToString(status *batchv1.JobStatus) string {
 }
 
 // RemoveCompletedJobs removes all completed k8s jobs dispatched by the service
-func RemoveCompletedJobs(prefixList []string) {
+func RemoveCompletedJobs() {
 	jobs := listJobs(getJobClient())
 	for i := 0; i < len(jobs.JobInfo); i++ {
 		job := jobs.JobInfo[i]
@@ -147,7 +148,7 @@ func RemoveCompletedJobs(prefixList []string) {
 }
 
 // GetNumberRunningJobs returns number of k8s running jobs dispatched by the service
-func GetNumberRunningJobs(prefixList []string) int {
+func GetNumberRunningJobs() int {
 	jobs := listJobs(getJobClient())
 	nRunningJobs := 0
 	for i := 0; i < len(jobs.JobInfo); i++ {

handlers/jobs_api.go

@@ -19,7 +19,7 @@ func status(w http.ResponseWriter, r *http.Request) {
 	}
 	UID := r.URL.Query().Get("UID")
 	if UID != "" {
-		result, errUID := getJobStatusByID(UID)
+		result, errUID := GetJobStatusByID(UID)
 		if errUID != nil {
 			http.Error(w, errUID.Error(), 500)
 			return

main.go

@@ -68,7 +68,9 @@ func main() {
 
 	// start an SQSHandler instance
 	SQSHandler := handlers.NewSQSHandler(sqsURL)
-	SQSHandler.StartServer()
+	if err := SQSHandler.StartServer(); err != nil {
+		glog.Errorf("Can not start the server. Detail %s", err)
+	}
 	defer SQSHandler.Server.Shutdown(context.Background())
 
 	SQSHandler.JobConfigs = jobConfigs