Refactor kvPath param and unify the segment flush kvPath in write node

Signed-off-by: Nbigsheeper <yihao.dai@zilliz.com>

configs/milvus.yaml

@@ -21,7 +21,8 @@ etcd:
   rootPath: by-dev
   metaSubPath: meta # metaRootPath = rootPath + '/' + metaSubPath
   kvSubPath: kv # kvRootPath = rootPath + '/' + kvSubPath
-  writeNodeKvSubPath: writer
+  writeNodeSegKvSubPath: writer/segment
+  writeNodeDDLKvSubPath: writer/ddl
   segThreshold: 10000
 
 minio:

internal/master/param_table.go

@@ -16,10 +16,11 @@ type ParamTable struct {
 	Address string
 	Port    int
 
-	EtcdAddress   string
-	MetaRootPath  string
-	KvRootPath    string
-	PulsarAddress string
+	EtcdAddress           string
+	MetaRootPath          string
+	KvRootPath            string
+	WriteNodeSegKvSubPath string
+	PulsarAddress         string
 
 	// nodeID
 	ProxyIDList     []typeutil.UniqueID
@@ -68,6 +69,7 @@ func (p *ParamTable) Init() {
 	p.initEtcdAddress()
 	p.initMetaRootPath()
 	p.initKvRootPath()
+	p.initWriteNodeSegKvSubPath()
 	p.initPulsarAddress()
 
 	p.initProxyIDList()
@@ -147,6 +149,14 @@ func (p *ParamTable) initKvRootPath() {
 	p.KvRootPath = rootPath + "/" + subPath
 }
 
+func (p *ParamTable) initWriteNodeSegKvSubPath() {
+	subPath, err := p.Load("etcd.writeNodeSegKvSubPath")
+	if err != nil {
+		panic(err)
+	}
+	p.WriteNodeSegKvSubPath = subPath + "/"
+}
+
 func (p *ParamTable) initTopicNum() {
 	iRangeStr, err := p.Load("msgChannel.channelRange.insert")
 	if err != nil {

internal/writenode/client/client.go

@@ -25,7 +25,7 @@ type Client struct {
 	flushStream msgstream.MsgStream
 }
 
-func NewWriterClient(etcdAddress string, kvRootPath string, writeNodeKvSubPath string, flushStream msgstream.MsgStream) (*Client, error) {
+func NewWriterClient(etcdAddress string, kvRootPath string, writeNodeSegKvSubPath string, flushStream msgstream.MsgStream) (*Client, error) {
 	// init kv client
 	etcdClient, err := clientv3.New(clientv3.Config{Endpoints: []string{etcdAddress}})
 	if err != nil {
@@ -35,7 +35,7 @@ func NewWriterClient(etcdAddress string, kvRootPath string, writeNodeKvSubPath s
 
 	return &Client{
 		kvClient:    kvClient,
-		kvPrefix:    writeNodeKvSubPath,
+		kvPrefix:    writeNodeSegKvSubPath,
 		flushStream: flushStream,
 	}, nil
 }

internal/writenode/meta_table.go

@@ -105,12 +105,12 @@ func (mt *metaTable) saveDDLFlushMeta(meta *pb.DDLFlushMeta) error {
 
 	mt.collID2DdlMeta[meta.CollectionID] = meta
 
-	return mt.client.Save("/writer/ddl/"+strconv.FormatInt(meta.CollectionID, 10), value)
+	return mt.client.Save(Params.WriteNodeDDLKvSubPath+strconv.FormatInt(meta.CollectionID, 10), value)
 }
 
 func (mt *metaTable) reloadDdlMetaFromKV() error {
 	mt.collID2DdlMeta = make(map[UniqueID]*pb.DDLFlushMeta)
-	_, values, err := mt.client.LoadWithPrefix("writer/ddl")
+	_, values, err := mt.client.LoadWithPrefix(Params.WriteNodeDDLKvSubPath)
 	if err != nil {
 		return err
 	}
@@ -132,13 +132,13 @@ func (mt *metaTable) saveSegFlushMeta(meta *pb.SegmentFlushMeta) error {
 
 	mt.segID2FlushMeta[meta.SegmentID] = *meta
 
-	return mt.client.Save("/writer/segment/"+strconv.FormatInt(meta.SegmentID, 10), value)
+	return mt.client.Save(Params.WriteNodeSegKvSubPath+strconv.FormatInt(meta.SegmentID, 10), value)
 }
 
 func (mt *metaTable) reloadSegMetaFromKV() error {
 	mt.segID2FlushMeta = make(map[UniqueID]pb.SegmentFlushMeta)
 
-	_, values, err := mt.client.LoadWithPrefix("writer/segment")
+	_, values, err := mt.client.LoadWithPrefix(Params.WriteNodeSegKvSubPath)
 	if err != nil {
 		return err
 	}

internal/writenode/param_table.go

@@ -36,13 +36,15 @@ type ParamTable struct {
 	DefaultPartitionTag string
 	SliceIndex          int
 
-	EtcdAddress          string
-	MetaRootPath         string
-	MinioAddress         string
-	MinioAccessKeyID     string
-	MinioSecretAccessKey string
-	MinioUseSSL          bool
-	MinioBucketName      string
+	EtcdAddress           string
+	MetaRootPath          string
+	WriteNodeSegKvSubPath string
+	WriteNodeDDLKvSubPath string
+	MinioAddress          string
+	MinioAccessKeyID      string
+	MinioSecretAccessKey  string
+	MinioUseSSL           bool
+	MinioBucketName       string
 
 	FlushInsertBufSize int
 	FlushDdBufSize     int
@@ -78,6 +80,8 @@ func (p *ParamTable) Init() {
 	p.initPulsarAddress()
 	p.initEtcdAddress()
 	p.initMetaRootPath()
+	p.initWriteNodeSegKvSubPath()
+	p.initWriteNodeDDLKvSubPath()
 	p.initInsertLogRootPath()
 	p.initDdLogRootPath()
 
@@ -299,6 +303,22 @@ func (p *ParamTable) initMetaRootPath() {
 	p.MetaRootPath = rootPath + "/" + subPath
 }
 
+func (p *ParamTable) initWriteNodeSegKvSubPath() {
+	subPath, err := p.Load("etcd.writeNodeSegKvSubPath")
+	if err != nil {
+		panic(err)
+	}
+	p.WriteNodeSegKvSubPath = subPath + "/"
+}
+
+func (p *ParamTable) initWriteNodeDDLKvSubPath() {
+	subPath, err := p.Load("etcd.writeNodeDDLKvSubPath")
+	if err != nil {
+		panic(err)
+	}
+	p.WriteNodeDDLKvSubPath = subPath + "/"
+}
+
 func (p *ParamTable) initInsertLogRootPath() {
 	rootPath, err := p.Load("etcd.rootPath")
 	if err != nil {