Add LocalChunkManager support

Signed-off-by: wayblink <anyang.wang@zilliz.com>

core/backup_context.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"time"
 
 	gomilvus "github.com/milvus-io/milvus-sdk-go/v2/client"
 	"go.uber.org/zap"
@@ -514,7 +515,7 @@ func (b *BackupContext) Check(ctx context.Context) string {
 		return "Failed to connect to storage backup path " + info + err.Error()
 	}
 
-	CHECK_PATH := ".milvus_backup_check"
+	CHECK_PATH := "milvus_backup_check_" + time.Now().String()
 
 	err = b.getStorageClient().Write(ctx, b.milvusBucketName, b.milvusRootPath+SEPERATOR+CHECK_PATH, []byte{1})
 	if err != nil {

core/backup_context_test.go

@@ -28,6 +28,16 @@ func TestCreateBackup(t *testing.T) {
 	backup.CreateBackup(context, req)
 }
 
+func TestCheck(t *testing.T) {
+	var params paramtable.BackupParams
+	params.Init()
+	context := context.Background()
+	backup := CreateBackupContext(context, params)
+
+	res := backup.Check(context)
+	println(res)
+}
+
 func TestListBackups(t *testing.T) {
 	var params paramtable.BackupParams
 	params.Init()

core/paramtable/params.go

@@ -143,6 +143,7 @@ func (p *MilvusConfig) initTLSMode() {
 // /////////////////////////////////////////////////////////////////////////////
 // --- minio ---
 const (
+	Local               = "local"
 	Minio               = "minio"
 	CloudProviderAWS    = "aws"
 	CloudProviderGCP    = "gcp"
@@ -151,6 +152,7 @@ const (
 )
 
 var supportedCloudProvider = map[string]bool{
+	Local:               true,
 	Minio:               true,
 	CloudProviderAWS:    true,
 	CloudProviderGCP:    true,

core/storage/azure_chunk_manager.go

@@ -28,7 +28,6 @@ import (
 	"github.com/cockroachdb/errors"
 	"github.com/minio/minio-go/v7"
 	"go.uber.org/zap"
-	"golang.org/x/exp/mmap"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/zilliztech/milvus-backup/internal/log"
@@ -227,36 +226,6 @@ func (mcm *AzureChunkManager) ReadWithPrefix(ctx context.Context, bucketName str
 	return objectsKeys, objectsValues, nil
 }
 
-func (mcm *AzureChunkManager) Mmap(ctx context.Context, bucketName string, filePath string) (*mmap.ReaderAt, error) {
-	return nil, errors.New("this method has not been implemented")
-}
-
-// ReadAt reads specific position data of minio storage if exists.
-func (mcm *AzureChunkManager) ReadAt(ctx context.Context, bucketName string, filePath string, off int64, length int64) ([]byte, error) {
-	return nil, errors.New("this method has not been implemented")
-	//if off < 0 || length < 0 {
-	//	return nil, io.EOF
-	//}
-	//
-	//object, err := mcm.getObject(ctx, bucketName, filePath, off, length)
-	//if err != nil {
-	//	log.Warn("failed to get object", zap.String("bucket", bucketName), zap.String("path", filePath), zap.Error(err))
-	//	return nil, err
-	//}
-	//defer object.Close()
-	//
-	//data, err := Read(object, length)
-	//if err != nil {
-	//	errResponse := minio.ToErrorResponse(err)
-	//	if errResponse.Code == "NoSuchKey" {
-	//		return nil, WrapErrNoSuchKey(filePath)
-	//	}
-	//	log.Warn("failed to read object", zap.String("bucket", bucketName), zap.String("path", filePath), zap.Error(err))
-	//	return nil, err
-	//}
-	//return data, nil
-}
-
 // Remove deletes an object with @key.
 func (mcm *AzureChunkManager) Remove(ctx context.Context, bucketName string, filePath string) error {
 	err := mcm.removeObject(ctx, bucketName, filePath)

core/storage/chunk_manager.go

@@ -10,6 +10,8 @@ func NewChunkManager(ctx context.Context, params paramtable.BackupParams) (Chunk
 	engine := params.MinioCfg.CloudProvider
 	if engine == "azure" {
 		return newAzureChunkManagerWithParams(ctx, params)
+	} else if engine == "local" {
+		return newLocalChunkManagerWithParams(ctx, params)
 	} else {
 		return newMinioChunkManagerWithParams(ctx, params)
 	}
@@ -66,3 +68,13 @@ func newAzureChunkManagerWithParams(ctx context.Context, params paramtable.Backu
 
 	return NewAzureChunkManager(ctx, c)
 }
+
+func newLocalChunkManagerWithParams(ctx context.Context, params paramtable.BackupParams) (*LocalChunkManager, error) {
+	c := newDefaultConfig()
+	c.rootPath = params.MinioCfg.RootPath
+	c.cloudProvider = params.MinioCfg.CloudProvider
+	c.storageEngine = params.MinioCfg.StorageType
+	c.backupRootPath = params.MinioCfg.BackupRootPath
+
+	return NewLocalChunkManager(ctx, c)
+}