Enchance datanode interface

Signed-off-by: XuanYang-cn <xuan.yang@zilliz.com>

configs/milvus.yaml

@@ -73,3 +73,7 @@ indexNode:
 indexServer:
   address: localhost
   port: 21118
+
+dataNode:
+  address: localhost
+  port: 21124

internal/datanode/allocator.go

+package datanode
+
+import (
+	"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
+	"github.com/zilliztech/milvus-distributed/internal/proto/masterpb"
+)
+
+type (
+	allocator interface {
+		allocID() (UniqueID, error)
+	}
+
+	allocatorImpl struct {
+		masterService MasterServiceInterface
+	}
+)
+
+func newAllocatorImpl(s MasterServiceInterface) *allocatorImpl {
+	return &allocatorImpl{
+		masterService: s,
+	}
+}
+
+func (alloc *allocatorImpl) allocID() (UniqueID, error) {
+	resp, err := alloc.masterService.AllocID(&masterpb.IDRequest{
+		Base: &commonpb.MsgBase{
+			MsgType:   commonpb.MsgType_kShowCollections,
+			MsgID:     1, // GOOSE TODO add msg id
+			Timestamp: 0, // GOOSE TODO
+			SourceID:  Params.NodeID,
+		},
+		Count: 1,
+	})
+	if err != nil {
+		return 0, err
+	}
+	return resp.ID, nil
+}

internal/datanode/collection_replica_test.go

@@ -4,24 +4,23 @@ import (
 	"testing"
 
 	"github.com/golang/protobuf/proto"
-
-	"github.com/zilliztech/milvus-distributed/internal/datanode/factory"
-
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
 func newReplica() collectionReplica {
 	collections := make([]*Collection, 0)
+	segments := make([]*Segment, 0)
 
 	var replica collectionReplica = &collectionReplicaImpl{
 		collections: collections,
+		segments:    segments,
 	}
 	return replica
 }
 
 func initTestReplicaMeta(t *testing.T, replica collectionReplica, collectionName string, collectionID UniqueID, segmentID UniqueID) {
-	Factory := &factory.MetaFactory{}
+	Factory := &MetaFactory{}
 	collectionMeta := Factory.CollectionMetaFactory(collectionID, collectionName)
 
 	schemaBlob := proto.MarshalTextString(collectionMeta.Schema)

internal/datanode/collection_test.go

@@ -5,14 +5,12 @@ import (
 
 	"github.com/golang/protobuf/proto"
 	"github.com/stretchr/testify/assert"
-
-	"github.com/zilliztech/milvus-distributed/internal/datanode/factory"
 )
 
 func TestCollection_newCollection(t *testing.T) {
 	collectionName := "collection0"
 	collectionID := UniqueID(1)
-	Factory := &factory.MetaFactory{}
+	Factory := &MetaFactory{}
 	collectionMeta := Factory.CollectionMetaFactory(collectionID, collectionName)
 
 	schemaBlob := proto.MarshalTextString(collectionMeta.Schema)
@@ -26,7 +24,7 @@ func TestCollection_newCollection(t *testing.T) {
 func TestCollection_deleteCollection(t *testing.T) {
 	collectionName := "collection0"
 	collectionID := UniqueID(1)
-	Factory := &factory.MetaFactory{}
+	Factory := &MetaFactory{}
 	collectionMeta := Factory.CollectionMetaFactory(collectionID, collectionName)
 
 	schemaBlob := proto.MarshalTextString(collectionMeta.Schema)

internal/datanode/data_node.go

@@ -4,51 +4,130 @@ import (
 	"context"
 	"io"
 	"log"
+	"time"
 
 	"github.com/opentracing/opentracing-go"
 	"github.com/uber/jaeger-client-go"
 	"github.com/uber/jaeger-client-go/config"
+
+	"github.com/zilliztech/milvus-distributed/internal/errors"
+	"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
 	"github.com/zilliztech/milvus-distributed/internal/proto/datapb"
 	"github.com/zilliztech/milvus-distributed/internal/proto/internalpb2"
+	"github.com/zilliztech/milvus-distributed/internal/proto/masterpb"
+	"github.com/zilliztech/milvus-distributed/internal/proto/milvuspb"
+	"github.com/zilliztech/milvus-distributed/internal/util/typeutil"
 )
 
-type DataNode struct {
-	ctx             context.Context
-	DataNodeID      uint64
-	dataSyncService *dataSyncService
-	metaService     *metaService
+const (
+	RPCConnectionTimeout = 30 * time.Second
+)
 
-	replica collectionReplica
+type (
+	Inteface interface {
+		typeutil.Service
 
-	tracer opentracing.Tracer
-	closer io.Closer
-}
+		GetComponentStates() (*internalpb2.ComponentStates, error)
+		WatchDmChannels(in *datapb.WatchDmChannelRequest) (*commonpb.Status, error)
+		FlushSegments(in *datapb.FlushSegRequest) (*commonpb.Status, error)
+	}
 
-func NewDataNode(ctx context.Context, dataNodeID uint64) *DataNode {
+	DataServiceInterface interface {
+		RegisterNode(req *datapb.RegisterNodeRequest) (*datapb.RegisterNodeResponse, error)
+	}
 
-	collections := make([]*Collection, 0)
+	MasterServiceInterface interface {
+		AllocID(in *masterpb.IDRequest) (*masterpb.IDResponse, error)
+		ShowCollections(in *milvuspb.ShowCollectionRequest) (*milvuspb.ShowCollectionResponse, error)
+		DescribeCollection(in *milvuspb.DescribeCollectionRequest) (*milvuspb.DescribeCollectionResponse, error)
+	}
 
-	var replica collectionReplica = &collectionReplicaImpl{
-		collections: collections,
+	DataNode struct {
+		ctx    context.Context
+		NodeID UniqueID
+		Role   string
+		State  internalpb2.StateCode
+
+		dataSyncService *dataSyncService
+		metaService     *metaService
+
+		masterService MasterServiceInterface
+		dataService   DataServiceInterface
+
+		replica collectionReplica
+
+		tracer opentracing.Tracer
+		closer io.Closer
 	}
+)
+
+func NewDataNode(ctx context.Context, nodeID UniqueID, masterService MasterServiceInterface,
+	dataService DataServiceInterface) *DataNode {
 
+	Params.Init()
 	node := &DataNode{
 		ctx:             ctx,
-		DataNodeID:      dataNodeID,
+		NodeID:          nodeID,     // GOOSE TODO
+		Role:            "DataNode", // GOOSE TODO
+		State:           internalpb2.StateCode_INITIALIZING,
 		dataSyncService: nil,
-		// metaService:     nil,
-		replica: replica,
+		metaService:     nil,
+		masterService:   masterService,
+		dataService:     dataService,
+		replica:         nil,
 	}
 
 	return node
 }
 
 func (node *DataNode) Init() error {
-	Params.Init()
-	return nil
-}
 
-func (node *DataNode) Start() error {
+	req := &datapb.RegisterNodeRequest{
+		Base: &commonpb.MsgBase{
+			MsgType:  commonpb.MsgType_kNone, //GOOSE TODO
+			SourceID: node.NodeID,
+		},
+		Address: &commonpb.Address{
+			Ip:   Params.IP,
+			Port: Params.Port,
+		},
+	}
+
+	resp, err := node.dataService.RegisterNode(req)
+	if err != nil {
+		return errors.Errorf("Init failed: %v", err)
+	}
+
+	for _, kv := range resp.InitParams.StartParams {
+		log.Println(kv)
+		switch kv.Key {
+		case "DDChannelName":
+			Params.DDChannelNames = []string{kv.Value}
+		case "SegmentStatisticsChannelName":
+			Params.SegmentStatisticsChannelName = kv.Value
+		case "TimeTickChannelName":
+			Params.TimeTickChannelName = kv.Value
+		case "CompleteFlushChannelName":
+			Params.CompleteFlushChannelName = kv.Value
+		default:
+			return errors.Errorf("Invalid key: %v", kv.Key)
+		}
+
+	}
+
+	var replica collectionReplica = &collectionReplicaImpl{
+		collections: make([]*Collection, 0),
+		segments:    make([]*Segment, 0),
+	}
+
+	var alloc allocator = newAllocatorImpl(node.masterService)
+	chanSize := 100
+	flushChan := make(chan *flushMsg, chanSize)
+	node.dataSyncService = newDataSyncService(node.ctx, flushChan, replica, alloc)
+	node.metaService = newMetaService(node.ctx, replica)
+	node.replica = replica
+
+	// Opentracing
 	cfg := &config.Configuration{
 		ServiceName: "data_node",
 		Sampler: &config.SamplerConfig{
@@ -59,24 +138,22 @@ func (node *DataNode) Start() error {
 			LogSpans: true,
 		},
 	}
-
-	var err error
-	node.tracer, node.closer, err = cfg.NewTracer(config.Logger(jaeger.StdLogger))
+	tracer, closer, err := cfg.NewTracer(config.Logger(jaeger.StdLogger))
 	if err != nil {
-		log.Printf("ERROR: cannot init Jaeger: %v\n", err)
-	} else {
-		opentracing.SetGlobalTracer(node.tracer)
+		return errors.Errorf("ERROR: cannot init Jaeger: %v\n", err)
 	}
+	node.tracer = tracer
+	node.closer = closer
 
-	// TODO GOOSE Init Size??
-	chanSize := 100
-	flushChan := make(chan *flushMsg, chanSize)
+	opentracing.SetGlobalTracer(node.tracer)
 
-	node.dataSyncService = newDataSyncService(node.ctx, flushChan, node.replica)
-	node.metaService = newMetaService(node.ctx, node.replica)
+	node.State = internalpb2.StateCode_HEALTHY
+	return nil
+}
+
+func (node *DataNode) Start() error {
 
 	go node.dataSyncService.start()
-	// go node.flushSyncService.start()
 	node.metaService.start()
 
 	return nil
@@ -88,8 +165,16 @@ func (node *DataNode) WatchDmChannels(in *datapb.WatchDmChannelRequest) error {
 }
 
 func (node *DataNode) GetComponentStates() (*internalpb2.ComponentStates, error) {
-	// GOOSE TODO: Implement me
-	return nil, nil
+	states := &internalpb2.ComponentStates{
+		State: &internalpb2.ComponentInfo{
+			NodeID:    Params.NodeID,
+			Role:      node.Role,
+			StateCode: node.State,
+		},
+		SubcomponentStates: make([]*internalpb2.ComponentInfo, 0),
+		Status:             &commonpb.Status{},
+	}
+	return states, nil
 }
 
 func (node *DataNode) FlushSegments(in *datapb.FlushSegRequest) error {

internal/datanode/data_node_test.go

@@ -26,6 +26,10 @@ func makeNewChannelNames(names []string, suffix string) []string {
 }
 
 func refreshChannelNames() {
+	Params.DDChannelNames = []string{"datanode-test"}
+	Params.SegmentStatisticsChannelName = "segtment-statistics"
+	Params.CompleteFlushChannelName = "flush-completed"
+	Params.TimeTickChannelName = "hard-timetick"
 	suffix := "-test-data-node" + strconv.FormatInt(rand.Int63n(100), 10)
 	Params.DDChannelNames = makeNewChannelNames(Params.DDChannelNames, suffix)
 	Params.InsertChannelNames = makeNewChannelNames(Params.InsertChannelNames, suffix)
@@ -81,28 +85,6 @@ func TestMain(m *testing.M) {
 	os.Exit(exitCode)
 }
 
-func newDataNode() *DataNode {
-
-	const ctxTimeInMillisecond = 2000
-	const closeWithDeadline = true
-	var ctx context.Context
-
-	if closeWithDeadline {
-		var cancel context.CancelFunc
-		d := time.Now().Add(ctxTimeInMillisecond * time.Millisecond)
-		ctx, cancel = context.WithDeadline(context.Background(), d)
-		go func() {
-			<-ctx.Done()
-			cancel()
-		}()
-	} else {
-		ctx = context.Background()
-	}
-
-	svr := NewDataNode(ctx, 0)
-	return svr
-}
-
 func newMetaTable() *metaTable {
 	etcdClient, _ := clientv3.New(clientv3.Config{Endpoints: []string{Params.EtcdAddress}})
 

internal/datanode/data_sync_service.go

@@ -10,20 +10,22 @@ import (
 )
 
 type dataSyncService struct {
-	ctx       context.Context
-	fg        *flowgraph.TimeTickedFlowGraph
-	flushChan chan *flushMsg
-	replica   collectionReplica
+	ctx         context.Context
+	fg          *flowgraph.TimeTickedFlowGraph
+	flushChan   chan *flushMsg
+	replica     collectionReplica
+	idAllocator allocator
 }
 
 func newDataSyncService(ctx context.Context, flushChan chan *flushMsg,
-	replica collectionReplica) *dataSyncService {
+	replica collectionReplica, alloc allocator) *dataSyncService {
 
 	return &dataSyncService{
-		ctx:       ctx,
-		fg:        nil,
-		flushChan: flushChan,
-		replica:   replica,
+		ctx:         ctx,
+		fg:          nil,
+		flushChan:   flushChan,
+		replica:     replica,
+		idAllocator: alloc,
 	}
 }
 
@@ -59,8 +61,8 @@ func (dsService *dataSyncService) initNodes() {
 
 	var filterDmNode Node = newFilteredDmNode()
 
-	var ddNode Node = newDDNode(dsService.ctx, mt, dsService.flushChan, dsService.replica)
-	var insertBufferNode Node = newInsertBufferNode(dsService.ctx, mt, dsService.replica)
+	var ddNode Node = newDDNode(dsService.ctx, mt, dsService.flushChan, dsService.replica, dsService.idAllocator)
+	var insertBufferNode Node = newInsertBufferNode(dsService.ctx, mt, dsService.replica, dsService.idAllocator)
 	var gcNode Node = newGCNode(dsService.replica)
 
 	dsService.fg.AddNode(&dmStreamNode)

internal/datanode/data_sync_service_test.go

@@ -10,7 +10,6 @@ import (
 	"github.com/golang/protobuf/proto"
 	"github.com/stretchr/testify/assert"
 
-	"github.com/zilliztech/milvus-distributed/internal/datanode/factory"
 	"github.com/zilliztech/milvus-distributed/internal/msgstream"
 	"github.com/zilliztech/milvus-distributed/internal/msgstream/pulsarms"
 	"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
@@ -35,10 +34,16 @@ func TestDataSyncService_Start(t *testing.T) {
 	// init data node
 	pulsarURL := Params.PulsarAddress
 
-	Factory := &factory.MetaFactory{}
+	Factory := &MetaFactory{}
 	collMeta := Factory.CollectionMetaFactory(UniqueID(0), "coll1")
-	node := NewDataNode(ctx, 0)
-	node.replica.addCollection(collMeta.ID, proto.MarshalTextString(collMeta.Schema))
+
+	chanSize := 100
+	flushChan := make(chan *flushMsg, chanSize)
+	replica := newReplica()
+	allocFactory := AllocatorFactory{}
+	sync := newDataSyncService(ctx, flushChan, replica, allocFactory)
+	sync.replica.addCollection(collMeta.ID, proto.MarshalTextString(collMeta.Schema))
+	go sync.start()
 
 	// test data generate
 	// GOOSE TODO orgnize
@@ -204,10 +209,6 @@ func TestDataSyncService_Start(t *testing.T) {
 
 	// dataSync
 	Params.FlushInsertBufferSize = 1
-	node.dataSyncService = newDataSyncService(node.ctx, nil, node.replica)
-	go node.dataSyncService.start()
-
-	node.Stop()
 
-	<-ctx.Done()
+	sync.close()
 }

internal/datanode/factory/factory.go → internal/datanode/factory.go

-package factory
+package datanode
 
 import (
 	"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
 	"github.com/zilliztech/milvus-distributed/internal/proto/etcdpb"
 	"github.com/zilliztech/milvus-distributed/internal/proto/schemapb"
-	"github.com/zilliztech/milvus-distributed/internal/util/typeutil"
 )
 
 type (
-	UniqueID  = typeutil.UniqueID
-	Timestamp = typeutil.Timestamp
-
 	Factory interface {
 	}
 
 	MetaFactory struct {
 	}
+
+	AllocatorFactory struct {
+	}
 )
 
 func (mf *MetaFactory) CollectionMetaFactory(collectionID UniqueID, collectionName string) *etcdpb.CollectionMeta {
@@ -152,3 +151,8 @@ func (mf *MetaFactory) CollectionMetaFactory(collectionID UniqueID, collectionNa
 	}
 	return &collection
 }
+
+func (alloc AllocatorFactory) allocID() (UniqueID, error) {
+	// GOOSE TODO: random ID generate
+	return UniqueID(0), nil
+}

internal/datanode/flow_graph_dd_node.go

@@ -9,7 +9,6 @@ import (
 	"strconv"
 
 	"github.com/golang/protobuf/proto"
-	"github.com/zilliztech/milvus-distributed/internal/allocator"
 	"github.com/zilliztech/milvus-distributed/internal/kv"
 	miniokv "github.com/zilliztech/milvus-distributed/internal/kv/minio"
 	"github.com/zilliztech/milvus-distributed/internal/msgstream"
@@ -25,7 +24,7 @@ type ddNode struct {
 	ddBuffer  *ddBuffer
 	inFlushCh chan *flushMsg
 
-	idAllocator *allocator.IDAllocator
+	idAllocator allocator
 	kv          kv.Base
 	replica     collectionReplica
 	flushMeta   *metaTable
@@ -174,7 +173,7 @@ func (ddNode *ddNode) flush() {
 			keyCommon := path.Join(Params.DdBinlogRootPath, strconv.FormatInt(collectionID, 10))
 
 			// save ts binlog
-			timestampLogIdx, err := ddNode.idAllocator.AllocOne()
+			timestampLogIdx, err := ddNode.idAllocator.allocID()
 			if err != nil {
 				log.Println(err)
 			}
@@ -186,7 +185,7 @@ func (ddNode *ddNode) flush() {
 			log.Println("save ts binlog, key = ", timestampKey)
 
 			// save dd binlog
-			ddLogIdx, err := ddNode.idAllocator.AllocOne()
+			ddLogIdx, err := ddNode.idAllocator.allocID()
 			if err != nil {
 				log.Println(err)
 			}
@@ -370,7 +369,7 @@ func (ddNode *ddNode) dropPartition(msg *msgstream.DropPartitionMsg) {
 }
 
 func newDDNode(ctx context.Context, flushMeta *metaTable,
-	inFlushCh chan *flushMsg, replica collectionReplica) *ddNode {
+	inFlushCh chan *flushMsg, replica collectionReplica, alloc allocator) *ddNode {
 	maxQueueLength := Params.FlowGraphMaxQueueLength
 	maxParallelism := Params.FlowGraphMaxParallelism
 
@@ -397,15 +396,6 @@ func newDDNode(ctx context.Context, flushMeta *metaTable,
 		panic(err)
 	}
 
-	idAllocator, err := allocator.NewIDAllocator(ctx, Params.MasterAddress)
-	if err != nil {
-		panic(err)
-	}
-	err = idAllocator.Start()
-	if err != nil {
-		panic(err)
-	}
-
 	return &ddNode{
 		BaseNode:  baseNode,
 		ddRecords: ddRecords,
@@ -413,10 +403,9 @@ func newDDNode(ctx context.Context, flushMeta *metaTable,
 			ddData:  make(map[UniqueID]*ddData),
 			maxSize: Params.FlushDdBufferSize,
 		},
-		// outCh:     outCh,
 		inFlushCh: inFlushCh,
 
-		idAllocator: idAllocator,
+		idAllocator: alloc,
 		kv:          minioKV,
 		replica:     replica,
 		flushMeta:   flushMeta,

internal/datanode/flow_graph_dd_node_test.go

@@ -15,7 +15,7 @@ import (
 
 func TestFlowGraphDDNode_Operate(t *testing.T) {
 	const ctxTimeInMillisecond = 2000
-	const closeWithDeadline = false
+	const closeWithDeadline = true
 	var ctx context.Context
 
 	if closeWithDeadline {
@@ -37,7 +37,8 @@ func TestFlowGraphDDNode_Operate(t *testing.T) {
 
 	Params.FlushDdBufferSize = 4
 	replica := newReplica()
-	ddNode := newDDNode(ctx, newMetaTable(), inFlushCh, replica)
+	idFactory := AllocatorFactory{}
+	ddNode := newDDNode(ctx, newMetaTable(), inFlushCh, replica, idFactory)
 
 	colID := UniqueID(0)
 	colName := "col-test-0"

internal/datanode/flow_graph_insert_buffer_node.go

@@ -15,7 +15,6 @@ import (
 	"github.com/opentracing/opentracing-go"
 	oplog "github.com/opentracing/opentracing-go/log"
 
-	"github.com/zilliztech/milvus-distributed/internal/allocator"
 	"github.com/zilliztech/milvus-distributed/internal/errors"
 	"github.com/zilliztech/milvus-distributed/internal/kv"
 	miniokv "github.com/zilliztech/milvus-distributed/internal/kv/minio"
@@ -45,7 +44,7 @@ type (
 		minIOKV     kv.Base
 		minioPrefix string
 
-		idAllocator *allocator.IDAllocator
+		idAllocator allocator
 
 		timeTickStream          msgstream.MsgStream
 		segmentStatisticsStream msgstream.MsgStream
@@ -514,7 +513,7 @@ func (ibNode *insertBufferNode) flushSegment(segID UniqueID, partitionID UniqueI
 
 	log.Printf(".. Saving (%v) binlogs to MinIO ...", len(binLogs))
 	for index, blob := range binLogs {
-		uid, err := ibNode.idAllocator.AllocOne()
+		uid, err := ibNode.idAllocator.allocID()
 		if err != nil {
 			return errors.Errorf("Allocate Id failed, %v", err)
 		}
@@ -543,7 +542,7 @@ func (ibNode *insertBufferNode) completeFlush(segID UniqueID) error {
 			MsgType:   commonpb.MsgType_kSegmentFlushDone,
 			MsgID:     0, // GOOSE TODO
 			Timestamp: 0, // GOOSE TODO
-			SourceID:  Params.DataNodeID,
+			SourceID:  Params.NodeID,
 		},
 		SegmentID: segID,
 	}
@@ -572,7 +571,7 @@ func (ibNode *insertBufferNode) writeHardTimeTick(ts Timestamp) error {
 				MsgType:   commonpb.MsgType_kTimeTick,
 				MsgID:     0,  // GOOSE TODO
 				Timestamp: ts, // GOOSE TODO
-				SourceID:  Params.DataNodeID,
+				SourceID:  Params.NodeID,
 			},
 		},
 	}
@@ -597,7 +596,7 @@ func (ibNode *insertBufferNode) updateSegStatistics(segIDs []UniqueID) error {
 			MsgType:   commonpb.MsgType_kSegmentStatistics,
 			MsgID:     UniqueID(0),  // GOOSE TODO
 			Timestamp: Timestamp(0), // GOOSE TODO
-			SourceID:  Params.DataNodeID,
+			SourceID:  Params.NodeID,
 		},
 		SegStats: statsUpdates,
 	}
@@ -623,8 +622,8 @@ func (ibNode *insertBufferNode) getCollectionSchemaByID(collectionID UniqueID) (
 	return ret.schema, nil
 }
 
-func newInsertBufferNode(ctx context.Context,
-	flushMeta *metaTable, replica collectionReplica) *insertBufferNode {
+func newInsertBufferNode(ctx context.Context, flushMeta *metaTable,
+	replica collectionReplica, alloc allocator) *insertBufferNode {
 	maxQueueLength := Params.FlowGraphMaxQueueLength
 	maxParallelism := Params.FlowGraphMaxParallelism
 
@@ -654,42 +653,33 @@ func newInsertBufferNode(ctx context.Context,
 	}
 	minioPrefix := Params.InsertBinlogRootPath
 
-	idAllocator, err := allocator.NewIDAllocator(ctx, Params.MasterAddress)
-	if err != nil {
-		panic(err)
-	}
-	err = idAllocator.Start()
-	if err != nil {
-		panic(err)
-	}
-
 	//input stream, data node time tick
 	wTt := pulsarms.NewPulsarMsgStream(ctx, 1024)
 	wTt.SetPulsarClient(Params.PulsarAddress)
 	wTt.CreatePulsarProducers([]string{Params.TimeTickChannelName})
 	var wTtMsgStream msgstream.MsgStream = wTt
-	wTtMsgStream.Start() // GOOSE TODO remove
+	wTtMsgStream.Start()
 
 	// update statistics channel
-	segS := pulsarms.NewPulsarMsgStream(ctx, Params.SegmentStatisticsBufSize)
+	segS := pulsarms.NewPulsarMsgStream(ctx, 1024)
 	segS.SetPulsarClient(Params.PulsarAddress)
 	segS.CreatePulsarProducers([]string{Params.SegmentStatisticsChannelName})
 	var segStatisticsMsgStream msgstream.MsgStream = segS
-	segStatisticsMsgStream.Start() // GOOSE TODO remove
+	segStatisticsMsgStream.Start()
 
 	// segment flush completed channel
 	cf := pulsarms.NewPulsarMsgStream(ctx, 1024)
 	cf.SetPulsarClient(Params.PulsarAddress)
 	cf.CreatePulsarProducers([]string{Params.CompleteFlushChannelName})
 	var completeFlushStream msgstream.MsgStream = cf
-	completeFlushStream.Start() // GOOSE TODO remove
+	completeFlushStream.Start()
 
 	return &insertBufferNode{
 		BaseNode:                baseNode,
 		insertBuffer:            iBuffer,
 		minIOKV:                 minIOKV,
 		minioPrefix:             minioPrefix,
-		idAllocator:             idAllocator,
+		idAllocator:             alloc,
 		timeTickStream:          wTtMsgStream,
 		segmentStatisticsStream: segStatisticsMsgStream,
 		completeFlushStream:     completeFlushStream,

internal/datanode/flow_graph_insert_buffer_node_test.go

@@ -12,7 +12,6 @@ import (
 	"github.com/golang/protobuf/proto"
 	"github.com/stretchr/testify/require"
 
-	"github.com/zilliztech/milvus-distributed/internal/datanode/factory"
 	"github.com/zilliztech/milvus-distributed/internal/msgstream"
 	"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
 	"github.com/zilliztech/milvus-distributed/internal/proto/internalpb2"
@@ -38,7 +37,7 @@ func TestFlowGraphInsertBufferNode_Operate(t *testing.T) {
 	require.NoError(t, err)
 	Params.MetaRootPath = testPath
 
-	Factory := &factory.MetaFactory{}
+	Factory := &MetaFactory{}
 	collMeta := Factory.CollectionMetaFactory(UniqueID(0), "coll1")
 	schemaBlob := proto.MarshalTextString(collMeta.Schema)
 	require.NotEqual(t, "", schemaBlob)
@@ -48,7 +47,8 @@ func TestFlowGraphInsertBufferNode_Operate(t *testing.T) {
 	require.NoError(t, err)
 
 	// Params.FlushInsertBufSize = 2
-	iBNode := newInsertBufferNode(ctx, newMetaTable(), replica)
+	idFactory := AllocatorFactory{}
+	iBNode := newInsertBufferNode(ctx, newMetaTable(), replica, idFactory)
 	inMsg := genInsertMsg()
 	var iMsg flowgraph.Msg = &inMsg
 	iBNode.Operate([]*flowgraph.Msg{&iMsg})

internal/datanode/flow_graph_msg_stream_input_node.go

@@ -35,18 +35,14 @@ func newDmInputNode(ctx context.Context) *flowgraph.InputNode {
 }
 
 func newDDInputNode(ctx context.Context) *flowgraph.InputNode {
-	receiveBufSize := Params.DDReceiveBufSize
-	pulsarBufSize := Params.DDPulsarBufSize
-
-	msgStreamURL := Params.PulsarAddress
 
 	consumeChannels := Params.DDChannelNames
 	consumeSubName := Params.MsgChannelSubName
 
-	ddStream := pulsarms.NewPulsarTtMsgStream(ctx, receiveBufSize)
-	ddStream.SetPulsarClient(msgStreamURL)
+	ddStream := pulsarms.NewPulsarTtMsgStream(ctx, 1024)
+	ddStream.SetPulsarClient(Params.PulsarAddress)
 	unmarshalDispatcher := util.NewUnmarshalDispatcher()
-	ddStream.CreatePulsarConsumers(consumeChannels, consumeSubName, unmarshalDispatcher, pulsarBufSize)
+	ddStream.CreatePulsarConsumers(consumeChannels, consumeSubName, unmarshalDispatcher, 1024)
 
 	var stream msgstream.MsgStream = ddStream
 

internal/datanode/interface.go

-package datanode
-
-import (
-	"github.com/zilliztech/milvus-distributed/internal/proto/datapb"
-)
-
-type Interface interface {
-	Init() error
-	Start() error
-	Stop() error
-
-	WatchDmChannels(in *datapb.WatchDmChannelRequest) error
-	FlushSegments(req *datapb.FlushSegRequest) error
-}

internal/datanode/param_table.go

@@ -16,7 +16,9 @@ type ParamTable struct {
 	paramtable.BaseTable
 
 	// === DataNode Internal Components Configs ===
-	DataNodeID              UniqueID
+	NodeID                  UniqueID
+	IP                      string // GOOSE TODO load from config file
+	Port                    int64
 	FlowGraphMaxQueueLength int32
 	FlowGraphMaxParallelism int32
 	FlushInsertBufferSize   int32
@@ -25,8 +27,9 @@ type ParamTable struct {
 	DdBinlogRootPath        string
 
 	// === DataNode External Components Configs ===
-	// --- Master ---
-	MasterAddress string
+	// --- External Client Address ---
+	MasterAddress  string
+	ServiceAddress string // GOOSE TODO: init from config file
 
 	// --- Pulsar ---
 	PulsarAddress string
@@ -38,20 +41,20 @@ type ParamTable struct {
 	InsertPulsarBufSize  int64
 
 	// - dd channel -
-	DDChannelNames   []string
-	DDReceiveBufSize int64
-	DDPulsarBufSize  int64
+	DDChannelNames []string // GOOSE TODO, set after Init
+	// DDReceiveBufSize int64
+	// DDPulsarBufSize  int64
 
 	// - seg statistics channel -
-	SegmentStatisticsChannelName    string
-	SegmentStatisticsBufSize        int64
-	SegmentStatisticsUpdateInterval int //  GOOSE TODO remove
+	SegmentStatisticsChannelName string // GOOSE TODO, set after init
+	// SegmentStatisticsBufSize        int64
+	// SegmentStatisticsUpdateInterval int //  GOOSE TODO remove
 
 	// - timetick channel -
-	TimeTickChannelName string
+	TimeTickChannelName string // GOOSE TODO: set after init
 
 	// - complete flush channel -
-	CompleteFlushChannelName string
+	CompleteFlushChannelName string // GOOSE TODO: set after init
 
 	// - channel subname -
 	MsgChannelSubName string
@@ -82,7 +85,9 @@ func (p *ParamTable) Init() {
 	}
 
 	// === DataNode Internal Components Configs ===
-	p.initDataNodeID()
+	p.initNodeID()
+	p.initIP()
+	p.initPort()
 	p.initFlowGraphMaxQueueLength()
 	p.initFlowGraphMaxParallelism()
 	p.initFlushInsertBufferSize()
@@ -104,20 +109,20 @@ func (p *ParamTable) Init() {
 	p.initInsertPulsarBufSize()
 
 	// - dd channel -
-	p.initDDChannelNames()
-	p.initDDReceiveBufSize()
-	p.initDDPulsarBufSize()
+	// p.initDDChannelNames()
+	// p.initDDReceiveBufSize()
+	// p.initDDPulsarBufSize()
 
 	// - seg statistics channel -
-	p.initSegmentStatisticsChannelName()
-	p.initSegmentStatisticsBufSize()
-	p.initSegmentStatisticsUpdateInterval()
+	// p.initSegmentStatisticsChannelName()
+	// p.initSegmentStatisticsBufSize()
+	// p.initSegmentStatisticsUpdateInterval()
 
 	// - timetick channel -
-	p.initTimeTickChannelName()
+	// p.initTimeTickChannelName()
 
 	// - flush completed channel -
-	p.initCompleteFlushChannelName()
+	// p.initCompleteFlushChannelName()
 
 	// - channel subname -
 	p.initMsgChannelSubName()
@@ -141,7 +146,7 @@ func (p *ParamTable) Init() {
 }
 
 // ==== DataNode internal components configs ====
-func (p *ParamTable) initDataNodeID() {
+func (p *ParamTable) initNodeID() {
 	p.dataNodeIDList = p.DataNodeIDList()
 	dataNodeIDStr := os.Getenv("DATA_NODE_ID")
 	if dataNodeIDStr == "" {
@@ -156,7 +161,20 @@ func (p *ParamTable) initDataNodeID() {
 		panic(err)
 	}
 
-	p.DataNodeID = p.ParseInt64("_dataNodeID")
+	p.NodeID = p.ParseInt64("_dataNodeID")
+}
+
+func (p *ParamTable) initIP() {
+	addr, err := p.Load("dataNode.address")
+	if err != nil {
+		panic(err)
+	}
+	p.IP = addr
+}
+
+func (p *ParamTable) initPort() {
+	port := p.ParseInt64("dataNode.port")
+	p.Port = port
 }
 
 // ---- flowgraph configs ----
@@ -257,7 +275,7 @@ func (p *ParamTable) initInsertPulsarBufSize() {
 	p.InsertPulsarBufSize = p.ParseInt64("dataNode.msgStream.insert.pulsarBufSize")
 }
 
-// - dd channel -
+// - dd channel - GOOSE TODO: remove
 func (p *ParamTable) initDDChannelNames() {
 	prefix, err := p.Load("msgChannel.chanNamePrefix.dataDefinition")
 	if err != nil {
@@ -276,31 +294,31 @@ func (p *ParamTable) initDDChannelNames() {
 	p.DDChannelNames = ret
 }
 
-func (p *ParamTable) initDDReceiveBufSize() {
-	revBufSize, err := p.Load("dataNode.msgStream.dataDefinition.recvBufSize")
-	if err != nil {
-		panic(err)
-	}
-	bufSize, err := strconv.Atoi(revBufSize)
-	if err != nil {
-		panic(err)
-	}
-	p.DDReceiveBufSize = int64(bufSize)
-}
-
-func (p *ParamTable) initDDPulsarBufSize() {
-	pulsarBufSize, err := p.Load("dataNode.msgStream.dataDefinition.pulsarBufSize")
-	if err != nil {
-		panic(err)
-	}
-	bufSize, err := strconv.Atoi(pulsarBufSize)
-	if err != nil {
-		panic(err)
-	}
-	p.DDPulsarBufSize = int64(bufSize)
-}
-
-// - seg statistics channel -
+// func (p *ParamTable) initDDReceiveBufSize() {
+//     revBufSize, err := p.Load("dataNode.msgStream.dataDefinition.recvBufSize")
+//     if err != nil {
+//         panic(err)
+//     }
+//     bufSize, err := strconv.Atoi(revBufSize)
+//     if err != nil {
+//         panic(err)
+//     }
+//     p.DDReceiveBufSize = int64(bufSize)
+// }
+
+// func (p *ParamTable) initDDPulsarBufSize() {
+//     pulsarBufSize, err := p.Load("dataNode.msgStream.dataDefinition.pulsarBufSize")
+//     if err != nil {
+//         panic(err)
+//     }
+//     bufSize, err := strconv.Atoi(pulsarBufSize)
+//     if err != nil {
+//         panic(err)
+//     }
+//     p.DDPulsarBufSize = int64(bufSize)
+// }
+
+// - seg statistics channel - GOOSE TODO: remove
 func (p *ParamTable) initSegmentStatisticsChannelName() {
 
 	channelName, err := p.Load("msgChannel.chanNamePrefix.dataNodeSegStatistics")
@@ -311,28 +329,26 @@ func (p *ParamTable) initSegmentStatisticsChannelName() {
 	p.SegmentStatisticsChannelName = channelName
 }
 
-func (p *ParamTable) initSegmentStatisticsBufSize() {
-	p.SegmentStatisticsBufSize = p.ParseInt64("dataNode.msgStream.segStatistics.recvBufSize")
-}
-
-func (p *ParamTable) initSegmentStatisticsUpdateInterval() {
-	p.SegmentStatisticsUpdateInterval = p.ParseInt("dataNode.msgStream.segStatistics.updateInterval")
-}
-
-// - flush completed channel -
+// func (p *ParamTable) initSegmentStatisticsBufSize() {
+//     p.SegmentStatisticsBufSize = p.ParseInt64("dataNode.msgStream.segStatistics.recvBufSize")
+// }
+//
+// func (p *ParamTable) initSegmentStatisticsUpdateInterval() {
+//     p.SegmentStatisticsUpdateInterval = p.ParseInt("dataNode.msgStream.segStatistics.updateInterval")
+// }
 
+// - flush completed channel - GOOSE TODO: remove
 func (p *ParamTable) initCompleteFlushChannelName() {
-	// GOOSE TODO
 	p.CompleteFlushChannelName = "flush-completed"
 }
 
-// - Timetick channel -
+// - Timetick channel - GOOSE TODO: remove
 func (p *ParamTable) initTimeTickChannelName() {
 	channels, err := p.Load("msgChannel.chanNamePrefix.dataNodeTimeTick")
 	if err != nil {
 		panic(err)
 	}
-	p.TimeTickChannelName = channels + "-" + strconv.FormatInt(p.DataNodeID, 10)
+	p.TimeTickChannelName = channels + "-" + strconv.FormatInt(p.NodeID, 10)
 }
 
 // - msg channel subname -
@@ -341,7 +357,7 @@ func (p *ParamTable) initMsgChannelSubName() {
 	if err != nil {
 		log.Panic(err)
 	}
-	p.MsgChannelSubName = name + "-" + strconv.FormatInt(p.DataNodeID, 10)
+	p.MsgChannelSubName = name + "-" + strconv.FormatInt(p.NodeID, 10)
 }
 
 func (p *ParamTable) initDefaultPartitionName() {
@@ -431,8 +447,8 @@ func (p *ParamTable) initMinioBucketName() {
 }
 
 func (p *ParamTable) sliceIndex() int {
-	dataNodeID := p.DataNodeID
-	dataNodeIDList := p.DataNodeIDList()
+	dataNodeID := p.NodeID
+	dataNodeIDList := p.dataNodeIDList
 	for i := 0; i < len(dataNodeIDList); i++ {
 		if dataNodeID == dataNodeIDList[i] {
 			return i

internal/datanode/param_table_test.go

@@ -9,9 +9,9 @@ func TestParamTable_DataNode(t *testing.T) {
 
 	Params.Init()
 
-	t.Run("Test DataNodeID", func(t *testing.T) {
-		id := Params.DataNodeID
-		log.Println("DataNodeID:", id)
+	t.Run("Test NodeID", func(t *testing.T) {
+		id := Params.NodeID
+		log.Println("NodeID:", id)
 	})
 
 	t.Run("Test flowGraphMaxQueueLength", func(t *testing.T) {
@@ -79,31 +79,11 @@ func TestParamTable_DataNode(t *testing.T) {
 		log.Println("DDChannelNames:", names)
 	})
 
-	t.Run("Test DdMsgStreamReceiveBufSize", func(t *testing.T) {
-		bufSize := Params.DDReceiveBufSize
-		log.Println("DDReceiveBufSize:", bufSize)
-	})
-
-	t.Run("Test DdPulsarBufSize", func(t *testing.T) {
-		bufSize := Params.DDPulsarBufSize
-		log.Println("DDPulsarBufSize:", bufSize)
-	})
-
 	t.Run("Test SegmentStatisticsChannelName", func(t *testing.T) {
 		name := Params.SegmentStatisticsChannelName
 		log.Println("SegmentStatisticsChannelName:", name)
 	})
 
-	t.Run("Test SegmentStatisticsBufSize", func(t *testing.T) {
-		size := Params.SegmentStatisticsBufSize
-		log.Println("SegmentStatisticsBufSize:", size)
-	})
-
-	t.Run("Test SegmentStatisticsUpdateInterval", func(t *testing.T) {
-		interval := Params.SegmentStatisticsUpdateInterval
-		log.Println("SegmentStatisticsUpdateInterval:", interval)
-	})
-
 	t.Run("Test timeTickChannelName", func(t *testing.T) {
 		name := Params.TimeTickChannelName
 		log.Println("TimeTickChannelName:", name)

internal/dataservice/cluster.go

@@ -84,7 +84,7 @@ func (c *dataNodeCluster) WatchInsertChannels(groups []channelGroup) {
 	defer c.mu.Unlock()
 	sort.Slice(c.nodes, func(i, j int) bool { return c.nodes[i].channelNum < c.nodes[j].channelNum })
 	for i, group := range groups {
-		err := c.nodes[i%len(c.nodes)].client.WatchDmChannels(&datapb.WatchDmChannelRequest{
+		_, err := c.nodes[i%len(c.nodes)].client.WatchDmChannels(&datapb.WatchDmChannelRequest{
 			Base: &commonpb.MsgBase{
 				MsgType:   commonpb.MsgType_kDescribeCollection,
 				MsgID:     -1, // todo
@@ -119,7 +119,7 @@ func (c *dataNodeCluster) FlushSegment(request *datapb.FlushSegRequest) {
 	c.mu.RLock()
 	defer c.mu.RUnlock()
 	for _, node := range c.nodes {
-		if err := node.client.FlushSegments(request); err != nil {
+		if _, err := node.client.FlushSegments(request); err != nil {
 			log.Println(err.Error())
 			continue
 		}

internal/distributed/datanode/client.go

@@ -6,34 +6,55 @@ import (
 	"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
 	"github.com/zilliztech/milvus-distributed/internal/proto/datapb"
 	"github.com/zilliztech/milvus-distributed/internal/proto/internalpb2"
+
+	"google.golang.org/grpc"
 )
 
 type Client struct {
-	ctx context.Context
+	ctx     context.Context
+	grpc    datapb.DataNodeClient
+	conn    *grpc.ClientConn
+	address string
+}
 
-	// GOOSE TODO: add DataNodeClient
+func NewClient(address string) *Client {
+	return &Client{
+		address: address,
+	}
 }
 
 func (c *Client) Init() error {
-	panic("implement me")
+	ctx, cancel := context.WithTimeout(context.Background(), RPCConnectionTimeout)
+	defer cancel()
+	var err error
+	for i := 0; i < Retry; i++ {
+		if c.conn, err = grpc.DialContext(ctx, c.address, grpc.WithInsecure(), grpc.WithBlock()); err == nil {
+			break
+		}
+	}
+	if err != nil {
+		return err
+	}
+	c.grpc = datapb.NewDataNodeClient(c.conn)
+	return nil
 }
 
 func (c *Client) Start() error {
-	panic("implement me")
+	return nil
 }
 
 func (c *Client) Stop() error {
-	panic("implement me")
+	return c.conn.Close()
 }
 
 func (c *Client) GetComponentStates(empty *commonpb.Empty) (*internalpb2.ComponentStates, error) {
-	panic("implement me")
+	return c.grpc.GetComponentStates(context.Background(), empty)
 }
 
-func (c *Client) WatchDmChannels(in *datapb.WatchDmChannelRequest) error {
-	panic("implement me")
+func (c *Client) WatchDmChannels(in *datapb.WatchDmChannelRequest) (*commonpb.Status, error) {
+	return c.grpc.WatchDmChannels(context.Background(), in)
 }
 
-func (c *Client) FlushSegments(in *datapb.FlushSegRequest) error {
-	panic("implement me")
+func (c *Client) FlushSegments(in *datapb.FlushSegRequest) (*commonpb.Status, error) {
+	return c.grpc.FlushSegments(context.Background(), in)
 }

internal/distributed/datanode/service.go

@@ -2,9 +2,13 @@ package datanode
 
 import (
 	"context"
+	"net"
+	"strconv"
 	"sync"
+	"time"
 
-	"github.com/zilliztech/milvus-distributed/internal/datanode"
+	dn "github.com/zilliztech/milvus-distributed/internal/datanode"
+	"github.com/zilliztech/milvus-distributed/internal/errors"
 	"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
 	"github.com/zilliztech/milvus-distributed/internal/proto/datapb"
 	"github.com/zilliztech/milvus-distributed/internal/proto/internalpb2"
@@ -12,9 +16,13 @@ import (
 	"google.golang.org/grpc"
 )
 
+const (
+	RPCConnectionTimeout = 30 * time.Second
+	Retry                = 3
+)
+
 type Server struct {
-	node datanode.Interface
-	core datanode.DataNode
+	core *dn.DataNode
 
 	grpcServer *grpc.Server
 	grpcError  error
@@ -24,11 +32,43 @@ type Server struct {
 	cancel context.CancelFunc
 }
 
-func NewGrpcServer() (*Server, error) {
-	panic("implement me")
+func New(masterService dn.MasterServiceInterface, dataService dn.DataServiceInterface) (*Server, error) {
+	var s = &Server{}
+
+	s.ctx, s.cancel = context.WithCancel(context.Background())
+	s.core = dn.NewDataNode(s.ctx, 0, masterService, dataService)
+	s.grpcServer = grpc.NewServer()
+	datapb.RegisterDataNodeServer(s.grpcServer, s)
+	addr := dn.Params.IP + ":" + strconv.FormatInt(dn.Params.Port, 10)
+	lis, err := net.Listen("tcp", addr)
+	if err != nil {
+		return nil, err
+	}
+
+	go func() {
+		if err = s.grpcServer.Serve(lis); err != nil {
+			s.grpcErrMux.Lock()
+			defer s.grpcErrMux.Unlock()
+			s.grpcError = err
+		}
+	}()
+
+	s.grpcErrMux.Lock()
+	err = s.grpcError
+	s.grpcErrMux.Unlock()
+
+	if err != nil {
+		return nil, err
+	}
+	return s, nil
 }
 
 func (s *Server) Init() error {
+	err := s.core.Init()
+	if err != nil {
+		return errors.Errorf("Init failed: %v", err)
+	}
+
 	return s.core.Init()
 }
 
@@ -41,13 +81,17 @@ func (s *Server) Stop() error {
 }
 
 func (s *Server) GetComponentStates(ctx context.Context, empty *commonpb.Empty) (*internalpb2.ComponentStates, error) {
-	return nil, nil
+	return s.core.GetComponentStates()
 }
 
-func (s *Server) WatchDmChannels(ctx context.Context, in *datapb.WatchDmChannelRequest) error {
-	return s.core.WatchDmChannels(in)
+func (s *Server) WatchDmChannels(ctx context.Context, in *datapb.WatchDmChannelRequest) (*commonpb.Status, error) {
+	return &commonpb.Status{
+		ErrorCode: commonpb.ErrorCode_SUCCESS,
+	}, s.core.WatchDmChannels(in)
 }
 
-func (s *Server) FlushSegments(ctx context.Context, in *datapb.FlushSegRequest) error {
-	return s.core.FlushSegments(in)
+func (s *Server) FlushSegments(ctx context.Context, in *datapb.FlushSegRequest) (*commonpb.Status, error) {
+	return &commonpb.Status{
+		ErrorCode: commonpb.ErrorCode_SUCCESS,
+	}, s.core.FlushSegments(in)
 }