Add root path for index files in minio (#8212)

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

internal/indexnode/param_table.go

@@ -41,6 +41,7 @@ type ParamTable struct {
 
 	EtcdEndpoints []string
 	MetaRootPath  string
+	IndexRootPath string
 
 	MinIOAddress         string
 	MinIOAccessKeyID     string
@@ -94,6 +95,7 @@ func (pt *ParamTable) initParams() {
 	pt.initMinioBucketName()
 	pt.initEtcdEndpoints()
 	pt.initMetaRootPath()
+	pt.initIndexRootPath()
 }
 
 func (pt *ParamTable) initMinIOAddress() {
@@ -151,6 +153,14 @@ func (pt *ParamTable) initMetaRootPath() {
 	pt.MetaRootPath = path.Join(rootPath, subPath)
 }
 
+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) initMinioBucketName() {
 	bucketName, err := pt.Load("minio.bucketName")
 	if err != nil {

internal/indexnode/param_table_test.go

@@ -70,7 +70,6 @@ func TestParamTable(t *testing.T) {
 	t.Run("SimdType", func(t *testing.T) {
 		t.Logf("SimdType: %v", Params.SimdType)
 	})
-
 	// FIXME(dragondriver): how to cover panic case? we use `LoadWithDefault` to initialize `SimdType`
 
 	t.Run("CreatedTime", func(t *testing.T) {
@@ -82,6 +81,10 @@ func TestParamTable(t *testing.T) {
 		Params.UpdatedTime = time.Now()
 		t.Logf("UpdatedTime: %v", Params.UpdatedTime)
 	})
+
+	t.Run("IndexRootPath", func(t *testing.T) {
+		t.Logf("IndexRootPath: %v", Params.IndexRootPath)
+	})
 }
 
 //TODO: Params Load should be return error when key does not exist.

internal/indexnode/task.go

@@ -15,6 +15,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"path"
 	"runtime"
 	"strconv"
 
@@ -345,8 +346,9 @@ func (it *IndexBuildTask) Execute(ctx context.Context) error {
 		tr.Record("serialize index codec done")
 
 		getSavePathByKey := func(key string) string {
-			// TODO: fix me, use more reasonable method
-			return strconv.Itoa(int(it.req.IndexBuildID)) + "/" + strconv.Itoa(int(it.req.Version)) + "/" + strconv.Itoa(int(partitionID)) + "/" + strconv.Itoa(int(segmentID)) + "/" + key
+
+			return path.Join(Params.IndexRootPath, strconv.Itoa(int(it.req.IndexBuildID)), strconv.Itoa(int(it.req.Version)),
+				strconv.Itoa(int(partitionID)), strconv.Itoa(int(segmentID)), key)
 		}
 		saveBlob := func(path string, value []byte) error {
 			return it.kv.Save(path, string(value))