Update protobuf

cmd/blast/main.go

@@ -234,14 +234,18 @@ func main() {
 							Action: managerNodeInfo,
 						},
 						{
-							Name:  "health",
+							Name:  "healthcheck",
 							Usage: "Health check the node",
 							Flags: []cli.Flag{
 								cli.StringFlag{
 									Name:  "grpc-address",
-									Value: "",
+									Value: ":5100",
 									Usage: "The gRPC listen address",
 								},
+								cli.BoolFlag{
+									Name:  "healthiness",
+									Usage: "healthiness probe",
+								},
 								cli.BoolFlag{
 									Name:  "liveness",
 									Usage: "Liveness probe",
@@ -251,7 +255,7 @@ func main() {
 									Usage: "Readiness probe",
 								},
 							},
-							Action: managerNodeHealth,
+							Action: managerNodeHealthCheck,
 						},
 					},
 				},

cmd/blast/manager_cluster_info.go

@@ -37,7 +37,7 @@ func managerClusterInfo(c *cli.Context) error {
 		}
 	}()
 
-	cluster, err := client.GetCluster()
+	cluster, err := client.ClusterInfo()
 	if err != nil {
 		return err
 	}

cmd/blast/manager_cluster_leave.go

@@ -43,7 +43,7 @@ func managerClusterLeave(c *cli.Context) error {
 		}
 	}()
 
-	err = client.DeleteNode(nodeId)
+	err = client.ClusterLeave(nodeId)
 	if err != nil {
 		return err
 	}

cmd/blast/manager_cluster_watch.go

@@ -46,7 +46,7 @@ func managerClusterWatch(c *cli.Context) error {
 		return err
 	}
 
-	watchClient, err := client.WatchCluster()
+	watchClient, err := client.ClusterWatch()
 	if err != nil {
 		return err
 	}

cmd/blast/manager_delete.go

@@ -43,7 +43,7 @@ func managerDelete(c *cli.Context) error {
 		}
 	}()
 
-	err = client.DeleteValue(key)
+	err = client.Delete(key)
 	if err != nil {
 		return err
 	}

cmd/blast/manager_get.go

@@ -39,7 +39,7 @@ func managerGet(c *cli.Context) error {
 		}
 	}()
 
-	value, err := client.GetValue(key)
+	value, err := client.Get(key)
 	if err != nil {
 		return err
 	}

cmd/blast/manager_node_health.go

@@ -19,11 +19,13 @@ import (
 	"os"
 
 	"github.com/mosuka/blast/manager"
+	"github.com/mosuka/blast/protobuf/management"
 	"github.com/urfave/cli"
 )
 
-func managerNodeHealth(c *cli.Context) error {
+func managerNodeHealthCheck(c *cli.Context) error {
 	grpcAddr := c.String("grpc-address")
+	healthiness := c.Bool("healthiness")
 	liveness := c.Bool("liveness")
 	readiness := c.Bool("readiness")
 
@@ -38,34 +40,30 @@ func managerNodeHealth(c *cli.Context) error {
 		}
 	}()
 
-	if !liveness && !readiness {
-		LivenessState, err := client.LivenessProbe()
+	var state string
+	if healthiness {
+		state, err = client.NodeHealthCheck(management.NodeHealthCheckRequest_HEALTHINESS.String())
 		if err != nil {
-			return err
+			state = management.NodeHealthCheckResponse_UNHEALTHY.String()
 		}
-		_, _ = fmt.Fprintln(os.Stdout, fmt.Sprintf("%v", LivenessState))
-
-		readinessState, err := client.ReadinessProbe()
+	} else if liveness {
+		state, err = client.NodeHealthCheck(management.NodeHealthCheckRequest_LIVENESS.String())
 		if err != nil {
-			return err
+			state = management.NodeHealthCheckResponse_DEAD.String()
 		}
-		_, _ = fmt.Fprintln(os.Stdout, fmt.Sprintf("%v", readinessState))
-	} else {
-		if liveness {
-			state, err := client.LivenessProbe()
-			if err != nil {
-				return err
-			}
-			_, _ = fmt.Fprintln(os.Stdout, fmt.Sprintf("%v", state))
+	} else if readiness {
+		state, err = client.NodeHealthCheck(management.NodeHealthCheckRequest_READINESS.String())
+		if err != nil {
+			state = management.NodeHealthCheckResponse_NOT_READY.String()
 		}
-		if readiness {
-			state, err := client.ReadinessProbe()
-			if err != nil {
-				return err
-			}
-			_, _ = fmt.Fprintln(os.Stdout, fmt.Sprintf("%v", state))
+	} else {
+		state, err = client.NodeHealthCheck(management.NodeHealthCheckRequest_HEALTHINESS.String())
+		if err != nil {
+			state = management.NodeHealthCheckResponse_UNHEALTHY.String()
 		}
 	}
 
+	_, _ = fmt.Fprintln(os.Stdout, fmt.Sprintf("%v", state))
+
 	return nil
 }

cmd/blast/manager_node_info.go

@@ -43,7 +43,7 @@ func managerNodeInfo(c *cli.Context) error {
 		}
 	}()
 
-	metadata, err := client.GetNode(nodeId)
+	metadata, err := client.NodeInfo(nodeId)
 	if err != nil {
 		return err
 	}

cmd/blast/manager_set.go

@@ -61,7 +61,7 @@ func managerSet(c *cli.Context) error {
 		}
 	}()
 
-	err = client.SetValue(key, value)
+	err = client.Set(key, value)
 	if err != nil {
 		return err
 	}

cmd/blast/manager_watch.go

@@ -43,7 +43,7 @@ func managerWatch(c *cli.Context) error {
 		}
 	}()
 
-	watchClient, err := client.WatchStore(key)
+	watchClient, err := client.Watch(key)
 	if err != nil {
 		return err
 	}

dispatcher/grpc_service.go

@@ -137,7 +137,7 @@ func (s *GRPCService) getInitialManagers(managerAddr string) (map[string]interfa
 		return nil, err
 	}
 
-	managers, err := client.GetCluster()
+	managers, err := client.ClusterInfo()
 	if err != nil {
 		s.logger.Error(err.Error())
 		return nil, err
@@ -207,7 +207,7 @@ func (s *GRPCService) startUpdateManagers(checkInterval time.Duration) {
 			}
 
 			// create stream
-			stream, err := client.WatchCluster()
+			stream, err := client.ClusterWatch()
 			if err != nil {
 				s.logger.Error(err.Error())
 				continue
@@ -360,7 +360,7 @@ func (s *GRPCService) startUpdateIndexers(checkInterval time.Duration) {
 	}
 
 	// get initial indexers
-	clusters, err := client.GetValue("/cluster_config/clusters/")
+	clusters, err := client.Get("/cluster_config/clusters/")
 	if err != nil {
 		s.logger.Error(err.Error())
 	}
@@ -426,7 +426,7 @@ func (s *GRPCService) startUpdateIndexers(checkInterval time.Duration) {
 				continue
 			}
 
-			stream, err := client.WatchStore("/cluster_config/clusters/")
+			stream, err := client.Watch("/cluster_config/clusters/")
 			if err != nil {
 				s.logger.Error(err.Error())
 				continue
@@ -443,7 +443,7 @@ func (s *GRPCService) startUpdateIndexers(checkInterval time.Duration) {
 			}
 			s.logger.Debug("data has changed", zap.String("key", resp.Key))
 
-			cluster, err := client.GetValue("/cluster_config/clusters/")
+			cluster, err := client.Get("/cluster_config/clusters/")
 			if err != nil {
 				s.logger.Error(err.Error())
 				continue

dispatcher/server_test.go

@@ -127,7 +127,7 @@ func TestServer_Start(t *testing.T) {
 		t.Fatalf("%v", err)
 	}
 	// get cluster info from manager1
-	managerCluster1, err := managerClient1.GetCluster()
+	managerCluster1, err := managerClient1.ClusterInfo()
 	if err != nil {
 		t.Fatalf("%v", err)
 	}

indexer/grpc_service.go

@@ -147,7 +147,7 @@ func (s *GRPCService) getInitialManagers(managerAddr string) (map[string]interfa
 		return nil, err
 	}
 
-	managers, err := client.GetCluster()
+	managers, err := client.ClusterInfo()
 	if err != nil {
 		s.logger.Error(err.Error())
 		return nil, err
@@ -216,7 +216,7 @@ func (s *GRPCService) startUpdateManagers(checkInterval time.Duration) {
 				continue
 			}
 
-			stream, err := client.WatchCluster()
+			stream, err := client.ClusterWatch()
 			if err != nil {
 				s.logger.Error(err.Error())
 				continue
@@ -504,7 +504,7 @@ func (s *GRPCService) startUpdateCluster(checkInterval time.Duration) {
 					if err != nil {
 						s.logger.Error(err.Error())
 					}
-					err = client.SetValue(fmt.Sprintf("cluster_config/clusters/%s/nodes", s.clusterConfig.ClusterId), cluster)
+					err = client.Set(fmt.Sprintf("cluster_config/clusters/%s/nodes", s.clusterConfig.ClusterId), cluster)
 					if err != nil {
 						s.logger.Error(err.Error())
 					}

indexer/server.go

@@ -69,7 +69,7 @@ func (s *Server) Start() {
 			return
 		}
 
-		clusterIntr, err := mc.GetValue(fmt.Sprintf("cluster_config/clusters/%s/nodes", s.clusterConfig.ClusterId))
+		clusterIntr, err := mc.Get(fmt.Sprintf("cluster_config/clusters/%s/nodes", s.clusterConfig.ClusterId))
 		if err != nil && err != errors.ErrNotFound {
 			s.logger.Fatal(err.Error())
 			return
@@ -122,7 +122,7 @@ func (s *Server) Start() {
 		}
 
 		s.logger.Debug("pull index config from manager", zap.String("address", mc.GetAddress()))
-		value, err := mc.GetValue("/index_config")
+		value, err := mc.Get("/index_config")
 		if err != nil {
 			s.logger.Fatal(err.Error())
 			return