feat(worker): Assign WorkerGroup via worker configuration (#332)

Author: Ethereal-O

vermeer/apps/master/bl/grpc_handlers.go

@@ -92,7 +92,7 @@ func (h *ServerHandler) SayHelloMaster(ctx context.Context, req *pb.HelloMasterR
 		}
 	}
 
-	reqWorker, err := workerMgr.CreateWorker(ip+":"+port, ip, req.Version)
+	reqWorker, err := workerMgr.CreateWorker(ip+":"+port, ip, req.Version, req.WorkerGroup)
 	if err != nil {
 		logrus.Errorf("failed to create a WorkerClient, error: %s", err)
 		return &pb.HelloMasterResp{WorkerId: -1, WorkerName: reqWorker.Name}, err
@@ -104,7 +104,7 @@ func (h *ServerHandler) SayHelloMaster(ctx context.Context, req *pb.HelloMasterR
 		return &pb.HelloMasterResp{}, err
 	}
 
-	logrus.Infof("worker say hello name: %s, client: %s", reqWorker.Name, p.Addr.String())
+	logrus.Infof("worker say hello name: %s and set to workgroup: %s, client: %s", reqWorker.Name, reqWorker.Group, p.Addr.String())
 
 	resp := pb.HelloMasterResp{
 		WorkerId:   reqWorker.Id,

vermeer/apps/master/workers/worker_manager.go

@@ -104,7 +104,7 @@ func (wm *workerManager) Init() {
 // CreateWorker Build a WorkerClient without an ID, and it'll receive one upon joining the WorkerManager.
 // The new WokerClient instance will be assigned a same name with the old one added to the WorkerManager,
 // which has the same workerPeer property.
-func (wm *workerManager) CreateWorker(workerPeer string, ipAddr string, version string) (*WorkerClient, error) {
+func (wm *workerManager) CreateWorker(workerPeer string, ipAddr string, version string, workerGroup string) (*WorkerClient, error) {
 	if workerPeer == "" {
 		return nil, fmt.Errorf("the argument 'workerPeer' is invalid")
 	}
@@ -115,12 +115,17 @@ func (wm *workerManager) CreateWorker(workerPeer string, ipAddr string, version
 		return nil, fmt.Errorf("the argument 'version' is invalid")
 	}
 
+	// must check if workerGroup is valid
+	if workerGroup == "" {
+		workerGroup = "$"
+	}
+
 	worker := &WorkerClient{
 		GrpcPeer:   workerPeer,
 		IpAddr:     ipAddr,
 		Version:    version,
 		LaunchTime: time.Now(),
-		Group:      "$",
+		Group:      workerGroup,
 	}
 
 	workerInDB := wm.retrieveWorker(workerPeer)
@@ -133,6 +138,11 @@ func (wm *workerManager) CreateWorker(workerPeer string, ipAddr string, version
 		worker.InitTime = worker.LaunchTime
 	}
 
+	// if workerGroup in workerInDB is different from the one in worker, give a warning to the user
+	if workerGroup != "$" && worker.Group != workerGroup {
+		logrus.Warnf("worker manager, worker group mismatch: given %s, but found %s in db for worker %s", workerGroup, worker.Group, worker.Name)
+	}
+
 	return worker, nil
 }