Update status of workers in GetWorkers

cmd/manager/main.go

@@ -127,6 +127,13 @@ func (s *server) StopWorkers(ctx context.Context, in *pb.StopWorkersRequest) (*p
 func (s *server) GetWorkers(ctx context.Context, in *pb.GetWorkersRequest) (*pb.GetWorkersReply, error) {
 	var ws []*pb.Worker
 	var err error
+	if in.StudyId == "" {
+		return &pb.GetWorkersReply{Workers: ws}, errors.New("StudyId should be set")
+	}
+	err = s.wIF.UpdateWorkerStatus(in.StudyId)
+	if err != nil {
+		return &pb.GetWorkersReply{Workers: ws}, err
+	}
 	if in.WorkerId == "" {
 		ws, err = dbIf.GetWorkerList(in.StudyId, in.TrialId)
 	} else {
@@ -152,6 +159,9 @@ func (s *server) GetShouldStopWorkers(ctx context.Context, in *pb.GetShouldStopW
 
 func (s *server) GetMetrics(ctx context.Context, in *pb.GetMetricsRequest) (*pb.GetMetricsReply, error) {
 	var mNames []string
+	if in.StudyId == "" {
+		return &pb.GetMetricsReply{}, errors.New("StudyId should be set")
+	}
 	if len(in.MetricsNames) > 0 {
 		mNames = in.MetricsNames
 	} else {
@@ -167,7 +177,10 @@ func (s *server) GetMetrics(ctx context.Context, in *pb.GetMetricsRequest) (*pb.
 	}
 	mls := make([]*pb.MetricsLogSet, len(in.WorkerIds))
 	for i, w := range in.WorkerIds {
-		wr, err := s.GetWorkers(ctx, &pb.GetWorkersRequest{WorkerId: w})
+		wr, err := s.GetWorkers(ctx, &pb.GetWorkersRequest{
+			StudyId:  in.StudyId,
+			WorkerId: w,
+		})
 		if err != nil {
 			return &pb.GetMetricsReply{}, err
 		}

pkg/manager/worker/kubernetes/kubernetes.go

@@ -170,7 +170,16 @@ func (d *KubernetesWorkerInterface) UpdateWorkerStatus(studyId string) error {
 		return err
 	}
 	for _, w := range ws {
-		if w.Status == api.State_RUNNING {
+		if w.Status == api.State_PENDING {
+			err = d.StoreWorkerLog(w.WorkerId)
+			if err == nil {
+				err = d.db.UpdateWorker(w.WorkerId, api.State_RUNNING)
+				if err != nil {
+					log.Printf("Error updating status for %s: %v", w.WorkerId, err)
+					return err
+				}
+			}
+		} else if w.Status == api.State_RUNNING {
 			c, err := d.IsWorkerComplete(w.WorkerId)
 			if err != nil {
 				return err
@@ -205,11 +214,6 @@ func (d *KubernetesWorkerInterface) SpawnWorker(wid string, workerConf *api.Work
 	if err != nil {
 		return err
 	}
-	err = d.db.UpdateWorker(wid, api.State_RUNNING)
-	if err != nil {
-		log.Printf("Error updating status for %s: %v", job.ObjectMeta.Name, err)
-		return err
-	}
 	log.Printf("Created Job %q.", result.GetObjectMeta().GetName())
 	return nil
 }

pkg/suggestion/hyperband_service.go

@@ -337,10 +337,12 @@ func (h *HyperBandSuggestService) evalWorkers(ctx context.Context, c api.Manager
 	bracket := Bracket{}
 	for _, tid := range hbparam.evaluatingTrials {
 		gwreq := &api.GetWorkersRequest{
+			StudyId: studyId,
 			TrialId: tid,
 		}
 		gwrep, err := c.GetWorkers(ctx, gwreq)
 		if err != nil {
+			log.Printf("GetWorkers error %v", err)
 			return err, nil
 		}
 		wl := make([]string, len(gwrep.Workers))

test/e2e/test-client.go

@@ -13,7 +13,7 @@ import (
 )
 
 var managerAddr = flag.String("s", "127.0.0.1:6789", "Endpoint of manager default 127.0.0.1:6789")
-var suggestArgo = flag.String("a", "random", "Suggestion Algorithm (random, grid)")
+var suggestArgo = flag.String("a", "random", "Suggestion Algorithm (random, grid, hyperband)")
 var requestnum = flag.Int("r", 2, "Request number for random Suggestions (default: 2)")
 var suggestionConfFile = flag.String("c", "", "File path to suggestion config.")
 
@@ -53,6 +53,10 @@ func main() {
 			//RunTrials
 			workerIds := runTrials(c, studyId, getSuggestReply)
 
+			//GetWorkersCheck
+			getWorker(c, studyId, getSuggestReply, workerIds)
+			log.Println("GetWorkers checked!")
+
 			iter := 0
 
 			for !isCompletedAllWorker(c, studyId) {
@@ -83,8 +87,11 @@ func main() {
 		//RunTrials
 		workerIds := runTrials(c, studyId, getSuggestReply)
 
-		iter := 0
+		//GetWorkersCheck
+		getWorker(c, studyId, getSuggestReply, workerIds)
+		log.Println("GetWorkers checked!")
 
+		iter := 0
 		for !isCompletedAllWorker(c, studyId) {
 			if iter > TimeOut {
 				log.Fatal("GetMetrics Timeout.")
@@ -220,7 +227,7 @@ func getSuggestion(c api.ManagerClient, studyId string, paramId string) *api.Get
 
 	getSuggestReply, err := c.GetSuggestions(ctx, getSuggestRequest)
 	if err != nil {
-		log.Fatalf("GetSuggestion Error %v", err)
+		log.Fatalf("GetSuggestion Error %v \nRequest %v", err, getSuggestRequest)
 	}
 	log.Println("Get " + *suggestArgo + " Suggestions:")
 	for _, t := range getSuggestReply.Trials {
@@ -288,6 +295,44 @@ func runTrials(c api.ManagerClient, studyId string, getSuggestReply *api.GetSugg
 	return workerIds
 }
 
+func getWorker(c api.ManagerClient, studyId string, getSuggestReply *api.GetSuggestionsReply, workerIds []string) {
+	ctx := context.Background()
+	getByTID := []*api.Worker{}
+	getByWID := []*api.Worker{}
+	gwr := &api.GetWorkersRequest{
+		StudyId: studyId,
+	}
+	_, err := c.GetWorkers(ctx, gwr)
+	if err != nil {
+		log.Fatalf("GwtWorker by Study ID Error %v", err)
+	}
+	for _, t := range getSuggestReply.Trials {
+		gwr := &api.GetWorkersRequest{
+			StudyId: studyId,
+			TrialId: t.TrialId,
+		}
+		gwrep, err := c.GetWorkers(ctx, gwr)
+		if err != nil {
+			log.Fatalf("GwtWorker by Trial ID Error %v", err)
+		}
+		getByTID = append(getByTID, gwrep.Workers...)
+	}
+	for _, w := range workerIds {
+		gwr = &api.GetWorkersRequest{
+			StudyId:  studyId,
+			WorkerId: w,
+		}
+		gwrep, err := c.GetWorkers(ctx, gwr)
+		if err != nil {
+			log.Fatalf("GwtWorker by Worker ID Error %v", err)
+		}
+		getByWID = append(getByWID, gwrep.Workers...)
+	}
+	if len(getByTID) != len(getByWID) {
+		log.Fatalf("GwtWorker by Worker ID is mismatch with GwtWorker by Trial ID \n by Trial ID %v \n by Worker ID %v", getByTID, getByWID)
+	}
+}
+
 func SaveOrUpdateModel(c api.ManagerClient, getMetricsReply *api.GetMetricsReply) {
 	ctx := context.Background()
 	for _, mls := range getMetricsReply.MetricsLogSets {