Add minio root path for indexcoord (#8673)

Signed-off-by: cai.zhang <cai.zhang@zilliz.com>

internal/indexcoord/index_coord.go

@@ -599,7 +599,7 @@ func (i *IndexCoord) recycleUnusedIndexFiles() {
 			log.Debug("IndexCoord recycleUnusedIndexFiles", zap.Int("Need recycle tasks num", len(metas)))
 			for _, meta := range metas {
 				if meta.indexMeta.MarkDeleted {
-					unusedIndexFilePathPrefix := strconv.Itoa(int(meta.indexMeta.IndexBuildID))
+					unusedIndexFilePathPrefix := Params.IndexRootPath + "/" + strconv.Itoa(int(meta.indexMeta.IndexBuildID))
 					log.Debug("IndexCoord recycleUnusedIndexFiles",
 						zap.Int64("Recycle the index files for deleted index with indexBuildID", meta.indexMeta.IndexBuildID))
 					if err := i.kv.RemoveWithPrefix(unusedIndexFilePathPrefix); err != nil {
@@ -613,7 +613,7 @@ func (i *IndexCoord) recycleUnusedIndexFiles() {
 					log.Debug("IndexCoord recycleUnusedIndexFiles",
 						zap.Int64("Recycle the low version index files of the index with indexBuildID", meta.indexMeta.IndexBuildID))
 					for j := 1; j < int(meta.indexMeta.Version); j++ {
-						unusedIndexFilePathPrefix := strconv.Itoa(int(meta.indexMeta.IndexBuildID)) + "/" + strconv.Itoa(j)
+						unusedIndexFilePathPrefix := Params.IndexRootPath + "/" + strconv.Itoa(int(meta.indexMeta.IndexBuildID)) + "/" + strconv.Itoa(j)
 						if err := i.kv.RemoveWithPrefix(unusedIndexFilePathPrefix); err != nil {
 							log.Error("IndexCoord recycleUnusedIndexFiles Remove index files failed",
 								zap.Bool("MarkDeleted", false), zap.Error(err))

internal/indexcoord/param_table.go

@@ -12,6 +12,7 @@
 package indexcoord
 
 import (
+	"path"
 	"strconv"
 	"strings"
 	"sync"
@@ -29,6 +30,7 @@ type ParamTable struct {
 	EtcdEndpoints []string
 	KvRootPath    string
 	MetaRootPath  string
+	IndexRootPath string
 
 	MinIOAddress         string
 	MinIOAccessKeyID     string
@@ -60,6 +62,7 @@ func (pt *ParamTable) Init() {
 	pt.initMinIOSecretAccessKey()
 	pt.initMinIOUseSSL()
 	pt.initMinioBucketName()
+	pt.initIndexRootPath()
 }
 
 func (pt *ParamTable) InitOnce() {
@@ -143,6 +146,14 @@ func (pt *ParamTable) initMinioBucketName() {
 	pt.MinioBucketName = bucketName
 }
 
+func (pt *ParamTable) initIndexRootPath() {
+	rootPath, err := pt.Load("minio.rootPath")
+	if err != nil {
+		panic(err)
+	}
+	pt.IndexRootPath = path.Join(rootPath, "index_files")
+}
+
 func (pt *ParamTable) initLogCfg() {
 	pt.InitLogCfg("indexcoord", 0)
 }

internal/indexcoord/param_table_test.go

@@ -68,6 +68,10 @@ func TestParamTable(t *testing.T) {
 		Params.UpdatedTime = time.Now()
 		t.Logf("UpdatedTime: %v", Params.UpdatedTime)
 	})
+
+	t.Run("initIndexRootPath", func(t *testing.T) {
+		t.Logf("IndexRootPath: %v", Params.IndexRootPath)
+	})
 }
 
 //TODO: Params Load should be return error when key does not exist.