Fix data service init

Signed-off-by: Nsunby <bingyi.sun@zilliz.com>

Makefile

@@ -80,14 +80,13 @@ verifiers: getdeps cppcheck fmt static-check ruleguard
 
 master: build-cpp
 	@echo "Building each component's binary to './bin'"
-	@echo "Building masterservice ..."
-	@mkdir -p $(INSTALL_PATH) && go env -w CGO_ENABLED="1" && GO111MODULE=on $(GO) build -o $(INSTALL_PATH)/masterservice $(PWD)/cmd/masterservice/main.go 1>/dev/null
-
+	@echo "Building master ..."
+	@mkdir -p $(INSTALL_PATH) && go env -w CGO_ENABLED="0" && GO111MODULE=on $(GO) build -o $(INSTALL_PATH)/master $(PWD)/cmd/master/main.go 1>/dev/null
 
 # Builds various components locally.
 proxynode: build-cpp
 	@echo "Building each component's binary to './bin'"
-	@echo "Building proxy node ..."
+	@echo "Building proxy ..."
 	@mkdir -p $(INSTALL_PATH) && go env -w CGO_ENABLED="0" && GO111MODULE=on $(GO) build -o $(INSTALL_PATH)/proxynode $(PWD)/cmd/proxy/node/proxy_node.go 1>/dev/null
 
 # Builds various components locally.
@@ -96,7 +95,6 @@ querynode: build-cpp
 	@echo "Building query node ..."
 	@mkdir -p $(INSTALL_PATH) && go env -w CGO_ENABLED="1" && GO111MODULE=on $(GO) build -o $(INSTALL_PATH)/querynode $(PWD)/cmd/querynode/query_node.go 1>/dev/null
 
-
 # Builds various components locally.
 writenode: build-cpp
 	@echo "Building each component's binary to './bin'"
@@ -112,13 +110,8 @@ datanode: build-cpp
 # Builds various components locally.
 indexnode: build-cpp
 	@echo "Building each component's binary to './bin'"
-	@echo "Building distributed indexnode ..."
-	@mkdir -p $(INSTALL_PATH) && go env -w CGO_ENABLED="1" && GO111MODULE=on $(GO) build -o $(INSTALL_PATH)/distributed/indexnode $(PWD)/cmd/distributed/indexnode/main.go 1>/dev/null
-
-indexservice: build-cpp
-	@echo "Building each component's binary to './bin'"
-	@echo "Building distributed indexservice ..."
-	@mkdir -p $(INSTALL_PATH) && go env -w CGO_ENABLED="1" && GO111MODULE=on $(GO) build -o $(INSTALL_PATH)/distributed/indexservice $(PWD)/cmd/distributed/indexservice/main.go 1>/dev/null
+	@echo "Building indexbuilder ..."
+	@mkdir -p $(INSTALL_PATH) && go env -w CGO_ENABLED="1" && GO111MODULE=on $(GO) build -o $(INSTALL_PATH)/indexbuilder $(PWD)/cmd/indexbuilder/indexbuilder.go 1>/dev/null
 
 
 # Builds various components locally.

cmd/dataservice/main.go

@@ -9,12 +9,13 @@ import (
 	"syscall"
 	"time"
 
+	ms "github.com/zilliztech/milvus-distributed/internal/distributed/masterservice"
+
+	"github.com/zilliztech/milvus-distributed/internal/masterservice"
+
 	"github.com/zilliztech/milvus-distributed/internal/distributed/dataservice"
 
 	"github.com/zilliztech/milvus-distributed/internal/proto/internalpb2"
-
-	"github.com/zilliztech/milvus-distributed/internal/distributed/masterservice"
-	"github.com/zilliztech/milvus-distributed/internal/master"
 )
 
 func main() {
@@ -22,8 +23,8 @@ func main() {
 
 	service := dataservice.NewGrpcService(ctx)
 
-	master.Params.Init()
-	client, err := masterservice.NewGrpcClient(fmt.Sprintf("%s:%d", master.Params.Address, master.Params.Port), 30*time.Second)
+	masterservice.Params.Init()
+	client, err := ms.NewGrpcClient(fmt.Sprintf("%s:%d", masterservice.Params.Address, masterservice.Params.Port), 30*time.Second)
 	if err != nil {
 		panic(err)
 	}

internal/distributed/dataservice/grpc_service.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"log"
 	"net"
+	"time"
 
 	"github.com/zilliztech/milvus-distributed/internal/distributed/masterservice"
 
@@ -43,6 +44,7 @@ func (s *Service) SetMasterClient(masterClient dataservice.MasterClient) {
 }
 
 func (s *Service) Init() error {
+	var err error
 	s.grpcServer = grpc.NewServer()
 	datapb.RegisterDataServiceServer(s.grpcServer, s)
 	lis, err := net.Listen("tcp", fmt.Sprintf("%s:%d", dataservice.Params.Address, dataservice.Params.Port))
@@ -50,9 +52,20 @@ func (s *Service) Init() error {
 		log.Fatal(err.Error())
 		return nil
 	}
-	if err = s.grpcServer.Serve(lis); err != nil {
-		log.Fatal(err.Error())
-		return nil
+	c := make(chan struct{})
+	go func() {
+		if err2 := s.grpcServer.Serve(lis); err2 != nil {
+			log.Println(err.Error())
+			close(c)
+			err = err2
+		}
+	}()
+	timer := time.NewTimer(1 * time.Second)
+	select {
+	case <-timer.C:
+		break
+	case <-c:
+		return err
 	}
 	return s.server.Init()
 }

internal/distributed/indexnode/client/client.go

@@ -2,13 +2,11 @@ package grpcindexnodeclient
 
 import (
 	"context"
-	"errors"
 	"log"
 	"time"
 
 	"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
 	"github.com/zilliztech/milvus-distributed/internal/proto/indexpb"
-	"github.com/zilliztech/milvus-distributed/internal/proto/internalpb2"
 	"google.golang.org/grpc"
 )
 
@@ -17,34 +15,6 @@ type Client struct {
 	nodeAddress string
 }
 
-func (c Client) GetComponentStates() (*internalpb2.ComponentStates, error) {
-	return c.grpcClient.GetComponentStates(context.Background(), &commonpb.Empty{})
-}
-
-func (c Client) GetTimeTickChannel() (string, error) {
-	resp, err := c.grpcClient.GetTimeTickChannel(context.Background(), &commonpb.Empty{})
-
-	if err != nil {
-		return "", err
-	}
-	if resp.Status.ErrorCode != commonpb.ErrorCode_SUCCESS {
-		return "", errors.New(resp.Status.Reason)
-	}
-	return resp.Value, nil
-}
-
-func (c Client) GetStatisticsChannel() (string, error) {
-	resp, err := c.grpcClient.GetStatisticsChannel(context.Background(), &commonpb.Empty{})
-
-	if err != nil {
-		return "", err
-	}
-	if resp.Status.ErrorCode != commonpb.ErrorCode_SUCCESS {
-		return "", errors.New(resp.Status.Reason)
-	}
-	return resp.Value, nil
-}
-
 func (c Client) Init() error {
 	return nil
 }

internal/distributed/indexnode/service.go

@@ -11,8 +11,6 @@ import (
 	"github.com/zilliztech/milvus-distributed/internal/indexnode"
 	"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
 	"github.com/zilliztech/milvus-distributed/internal/proto/indexpb"
-	"github.com/zilliztech/milvus-distributed/internal/proto/internalpb2"
-	"github.com/zilliztech/milvus-distributed/internal/proto/milvuspb"
 	"github.com/zilliztech/milvus-distributed/internal/util/typeutil"
 	"google.golang.org/grpc"
 )
@@ -27,42 +25,6 @@ type Server struct {
 	loopWg       sync.WaitGroup
 }
 
-func (s *Server) GetComponentStates(ctx context.Context, empty *commonpb.Empty) (*internalpb2.ComponentStates, error) {
-	return s.node.GetComponentStates()
-}
-
-func (s *Server) GetTimeTickChannel(ctx context.Context, empty *commonpb.Empty) (*milvuspb.StringResponse, error) {
-	ret, err := s.node.GetTimeTickChannel()
-	resp := &milvuspb.StringResponse{
-		Status: &commonpb.Status{
-			ErrorCode: commonpb.ErrorCode_SUCCESS,
-		},
-	}
-	if err != nil {
-		resp.Status.ErrorCode = commonpb.ErrorCode_UNEXPECTED_ERROR
-		resp.Status.Reason = err.Error()
-	} else {
-		resp.Value = ret
-	}
-	return resp, nil
-}
-
-func (s *Server) GetStatisticsChannel(ctx context.Context, empty *commonpb.Empty) (*milvuspb.StringResponse, error) {
-	ret, err := s.node.GetStatisticsChannel()
-	resp := &milvuspb.StringResponse{
-		Status: &commonpb.Status{
-			ErrorCode: commonpb.ErrorCode_SUCCESS,
-		},
-	}
-	if err != nil {
-		resp.Status.ErrorCode = commonpb.ErrorCode_UNEXPECTED_ERROR
-		resp.Status.Reason = err.Error()
-	} else {
-		resp.Value = ret
-	}
-	return resp, nil
-}
-
 func (s *Server) registerNode() error {
 
 	log.Printf("Registering node. IP = %s, Port = %d", indexnode.Params.NodeIP, indexnode.Params.NodePort)

internal/distributed/indexservice/service.go

@@ -7,12 +7,10 @@ import (
 	"strconv"
 	"sync"
 
-	"github.com/zilliztech/milvus-distributed/internal/proto/internalpb2"
-	"github.com/zilliztech/milvus-distributed/internal/proto/milvuspb"
-
 	"github.com/zilliztech/milvus-distributed/internal/indexservice"
 	"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
 	"github.com/zilliztech/milvus-distributed/internal/proto/indexpb"
+	"github.com/zilliztech/milvus-distributed/internal/proto/internalpb2"
 	"github.com/zilliztech/milvus-distributed/internal/util/typeutil"
 	"google.golang.org/grpc"
 )
@@ -30,43 +28,6 @@ type Server struct {
 	loopWg     sync.WaitGroup
 }
 
-func (s *Server) GetComponentStates(ctx context.Context, empty *commonpb.Empty) (*internalpb2.ComponentStates, error) {
-	return s.server.GetComponentStates()
-}
-
-func (s *Server) GetTimeTickChannel(ctx context.Context, empty *commonpb.Empty) (*milvuspb.StringResponse, error) {
-	resp := &milvuspb.StringResponse{
-		Status: &commonpb.Status{
-			ErrorCode: commonpb.ErrorCode_SUCCESS,
-		},
-	}
-	channel, err := s.server.GetTimeTickChannel()
-	if err != nil {
-		resp.Status.ErrorCode = commonpb.ErrorCode_UNEXPECTED_ERROR
-		resp.Status.Reason = err.Error()
-		return resp, nil
-	}
-
-	resp.Value = channel
-	return resp, nil
-}
-
-func (s *Server) GetStatisticsChannel(ctx context.Context, empty *commonpb.Empty) (*milvuspb.StringResponse, error) {
-	resp := &milvuspb.StringResponse{
-		Status: &commonpb.Status{
-			ErrorCode: commonpb.ErrorCode_SUCCESS,
-		},
-	}
-	channel, err := s.server.GetStatisticsChannel()
-	if err != nil {
-		resp.Status.ErrorCode = commonpb.ErrorCode_UNEXPECTED_ERROR
-		resp.Status.Reason = err.Error()
-		return resp, nil
-	}
-	resp.Value = channel
-	return resp, nil
-}
-
 func Init() error {
 	indexservice.Params.Init()
 	return nil
@@ -87,6 +48,18 @@ func (s *Server) Stop() error {
 	return nil
 }
 
+func (s *Server) GetComponentStates() (*internalpb2.ComponentStates, error) {
+	panic("implement me")
+}
+
+func (s *Server) GetTimeTickChannel() (string, error) {
+	panic("implement me")
+}
+
+func (s *Server) GetStatisticsChannel() (string, error) {
+	panic("implement me")
+}
+
 func (s *Server) RegisterNode(ctx context.Context, req *indexpb.RegisterNodeRequest) (*indexpb.RegisterNodeResponse, error) {
 
 	return s.server.RegisterNode(req)

internal/indexnode/indexnode.go

@@ -11,7 +11,6 @@ import (
 	miniokv "github.com/zilliztech/milvus-distributed/internal/kv/minio"
 	"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
 	"github.com/zilliztech/milvus-distributed/internal/proto/indexpb"
-	"github.com/zilliztech/milvus-distributed/internal/proto/internalpb2"
 	"github.com/zilliztech/milvus-distributed/internal/util/retry"
 	"github.com/zilliztech/milvus-distributed/internal/util/typeutil"
 )
@@ -24,8 +23,6 @@ type UniqueID = typeutil.UniqueID
 type Timestamp = typeutil.Timestamp
 
 type IndexNode struct {
-	state internalpb2.StateCode
-
 	loopCtx    context.Context
 	loopCancel func()
 
@@ -165,29 +162,3 @@ func (i *IndexNode) BuildIndex(request *indexpb.BuildIndexCmd) (*commonpb.Status
 	}
 	return ret, nil
 }
-
-func (i *IndexNode) GetComponentStates() (*internalpb2.ComponentStates, error) {
-
-	stateInfo := &internalpb2.ComponentInfo{
-		NodeID:    Params.NodeID,
-		Role:      "IndexNode",
-		StateCode: i.state,
-	}
-
-	ret := &internalpb2.ComponentStates{
-		State:              stateInfo,
-		SubcomponentStates: nil, // todo add subcomponents states
-		Status: &commonpb.Status{
-			ErrorCode: commonpb.ErrorCode_SUCCESS,
-		},
-	}
-	return ret, nil
-}
-
-func (i *IndexNode) GetTimeTickChannel() (string, error) {
-	return "", nil
-}
-
-func (i *IndexNode) GetStatisticsChannel() (string, error) {
-	return "", nil
-}

internal/indexservice/indexservice.go

@@ -26,11 +26,8 @@ const (
 
 type IndexService struct {
 	nodeClients *PriorityQueue
-	nodeStates  map[UniqueID]*internalpb2.ComponentStates
-	state       internalpb2.StateCode
-
-	ID UniqueID
 
+	//factory method
 	loopCtx    context.Context
 	loopCancel func()
 	loopWg     sync.WaitGroup
@@ -88,11 +85,6 @@ func CreateIndexService(ctx context.Context) (*IndexService, error) {
 		return nil, err
 	}
 
-	i.ID, err = i.idAllocator.AllocOne()
-	if err != nil {
-		return nil, err
-	}
-
 	connectMinIOFn := func() error {
 		option := &miniokv.Option{
 			Address:           Params.MinIOAddress,
@@ -147,21 +139,7 @@ func (i *IndexService) Stop() error {
 }
 
 func (i *IndexService) GetComponentStates() (*internalpb2.ComponentStates, error) {
-
-	stateInfo := &internalpb2.ComponentInfo{
-		NodeID:    i.ID,
-		Role:      "IndexService",
-		StateCode: i.state,
-	}
-
-	ret := &internalpb2.ComponentStates{
-		State:              stateInfo,
-		SubcomponentStates: nil, // todo add subcomponents states
-		Status: &commonpb.Status{
-			ErrorCode: commonpb.ErrorCode_SUCCESS,
-		},
-	}
-	return ret, nil
+	panic("implement me")
 }
 
 func (i *IndexService) GetTimeTickChannel() (string, error) {
@@ -169,7 +147,7 @@ func (i *IndexService) GetTimeTickChannel() (string, error) {
 }
 
 func (i *IndexService) GetStatisticsChannel() (string, error) {
-	return "", nil
+	panic("implement me")
 }
 
 func (i *IndexService) BuildIndex(req *indexpb.BuildIndexRequest) (*indexpb.BuildIndexResponse, error) {

internal/proto/index_service.proto

@@ -6,7 +6,6 @@ option go_package = "github.com/zilliztech/milvus-distributed/internal/proto/ind
 
 import "common.proto";
 import "internal.proto";
-import "milvus.proto";
 
 
 message RegisterNodeRequest {
@@ -94,10 +93,6 @@ service IndexService {
   rpc GetIndexStates(IndexStatesRequest) returns (IndexStatesResponse) {}
   rpc GetIndexFilePaths(IndexFilePathsRequest) returns (IndexFilePathsResponse){}
   rpc NotifyBuildIndex(BuildIndexNotification) returns (common.Status) {}
-
-  rpc GetComponentStates(common.Empty) returns (internal.ComponentStates) {}
-  rpc GetTimeTickChannel(common.Empty) returns(milvus.StringResponse) {}
-  rpc GetStatisticsChannel(common.Empty) returns(milvus.StringResponse){}
 }
 
 
@@ -111,7 +106,4 @@ service IndexNode {
    */
   rpc BuildIndex(BuildIndexCmd) returns (common.Status){}
 
-  rpc GetComponentStates(common.Empty) returns (internal.ComponentStates) {}
-  rpc GetTimeTickChannel(common.Empty) returns(milvus.StringResponse) {}
-  rpc GetStatisticsChannel(common.Empty) returns(milvus.StringResponse){}
 }

internal/util/typeutil/interface.go

@@ -21,7 +21,6 @@ type Component interface {
 
 type IndexNodeInterface interface {
 	Service
-	Component
 	BuildIndex(req *indexpb.BuildIndexCmd) (*commonpb.Status, error)
 }