*: support migrate to zilliz

Signed-off-by: huanghaoyuanhhy <haoyuan.huang@zilliz.com>

Author: huanghaoyuanhhy

cmd/cmd.go

@@ -9,6 +9,7 @@ import (
 	"github.com/zilliztech/milvus-backup/cmd/del"
 	"github.com/zilliztech/milvus-backup/cmd/get"
 	"github.com/zilliztech/milvus-backup/cmd/list"
+	"github.com/zilliztech/milvus-backup/cmd/migrate"
 	"github.com/zilliztech/milvus-backup/cmd/restore"
 	"github.com/zilliztech/milvus-backup/cmd/root"
 	"github.com/zilliztech/milvus-backup/cmd/server"
@@ -26,6 +27,7 @@ func Execute() {
 	cmd.AddCommand(list.NewCmd(&opt))
 	cmd.AddCommand(restore.NewCmd(&opt))
 	cmd.AddCommand(server.NewCmd(&opt))
+	cmd.AddCommand(migrate.NewCmd(&opt))
 
 	cmd.SetOut(os.Stdout)
 	cmd.SetErr(os.Stderr)

cmd/migrate/migrate.go

@@ -0,0 +1,91 @@
+package migrate
+
+import (
+	"context"
+	"errors"
+
+	"github.com/google/uuid"
+	"github.com/spf13/cobra"
+
+	"github.com/zilliztech/milvus-backup/cmd/root"
+	"github.com/zilliztech/milvus-backup/core/meta/taskmgr"
+	"github.com/zilliztech/milvus-backup/core/migrate"
+	"github.com/zilliztech/milvus-backup/core/paramtable"
+)
+
+type options struct {
+	backupName string
+	clusterID  string
+}
+
+func (o *options) validate() error {
+	if o.clusterID == "" {
+		return errors.New("cluster id is required")
+	}
+
+	if o.backupName == "" {
+		return errors.New("backup name is required")
+	}
+
+	return nil
+}
+
+func (o *options) run(cmd *cobra.Command, params *paramtable.BackupParams) error {
+	taskID := uuid.NewString()
+	task, err := migrate.NewTask(taskID, o.backupName, o.clusterID, params)
+	if err != nil {
+		return err
+	}
+
+	if err := task.Prepare(context.TODO()); err != nil {
+		return err
+	}
+
+	if err := task.Execute(context.TODO()); err != nil {
+		return err
+	}
+
+	taskStat, err := taskmgr.DefaultMgr.GetMigrateTask(taskID)
+	if err != nil {
+		return err
+	}
+
+	cmd.Printf("Successfully triggered migration")
+	cmd.Printf("with backup name: %s target cluster: %s", o.backupName, o.clusterID)
+	cmd.Printf("migration job id: %s.", taskStat.MigrateJobID())
+	cmd.Printf("You can check the progress of the migration job in Zilliz Cloud console.")
+
+	return nil
+}
+
+func (o *options) addFlags(cmd *cobra.Command) {
+	cmd.Flags().StringVarP(&o.backupName, "name", "n", "", "need to migrate backup name")
+	cmd.Flags().StringVarP(&o.clusterID, "cluster_id", "c", "", "target cluster id")
+}
+
+func NewCmd(opt *root.Options) *cobra.Command {
+	var o options
+
+	cmd := &cobra.Command{
+		Use:   "migrate",
+		Short: "migrate the backup data to zilliz cloud cluster",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			var params paramtable.BackupParams
+			params.GlobalInitWithYaml(opt.Config)
+			params.Init()
+
+			if err := o.validate(); err != nil {
+				return err
+			}
+
+			err := o.run(cmd, &params)
+			cobra.CheckErr(err)
+
+			return nil
+		},
+	}
+
+	o.addFlags(cmd)
+
+	return cmd
+}

cmd/migrate/migrate_test.go

@@ -0,0 +1,31 @@
+package migrate
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestOptions_Validate(t *testing.T) {
+	t.Run("Valid", func(t *testing.T) {
+		var o options
+		o.backupName = "backup"
+		o.clusterID = "cluster"
+		err := o.validate()
+		assert.NoError(t, err)
+	})
+
+	t.Run("BackupNameEmpty", func(t *testing.T) {
+		var o options
+		o.clusterID = "cluster"
+		err := o.validate()
+		assert.Error(t, err)
+	})
+
+	t.Run("ClusterIDEmpty", func(t *testing.T) {
+		var o options
+		o.backupName = "backup"
+		err := o.validate()
+		assert.Error(t, err)
+	})
+}

configs/backup.yaml

@@ -8,6 +8,13 @@ log:
 http:
   simpleResponse: true
 
+# Zilliz Cloud config.
+# If you want to migrate data to Zilliz Cloud, you need to configure the following parameters.
+# Otherwise, you can ignore it.
+cloud:
+  address: https://api.cloud.zilliz.com
+  apikey: <your-api-key>
+
 # milvus proxy address, compatible to milvus.yaml
 milvus:
   address: localhost

core/backup/collection.go

@@ -15,7 +15,7 @@ import (
 	"golang.org/x/sync/semaphore"
 	"google.golang.org/protobuf/proto"
 
-	"github.com/zilliztech/milvus-backup/core/client"
+	"github.com/zilliztech/milvus-backup/core/client/milvus"
 	"github.com/zilliztech/milvus-backup/core/meta"
 	"github.com/zilliztech/milvus-backup/core/pbconv"
 	"github.com/zilliztech/milvus-backup/core/proto/backuppb"
@@ -44,8 +44,8 @@ type CollectionOpt struct {
 
 	Meta *meta.MetaManager
 
-	Grpc    client.Grpc
-	Restful client.Restful
+	Grpc    milvus.Grpc
+	Restful milvus.Restful
 }
 
 type CollectionTask struct {
@@ -70,8 +70,8 @@ type CollectionTask struct {
 
 	meta *meta.MetaManager
 
-	grpc    client.Grpc
-	restful client.Restful
+	grpc    milvus.Grpc
+	restful milvus.Restful
 
 	logger *zap.Logger
 }
@@ -419,7 +419,7 @@ func (ct *CollectionTask) getSegment(ctx context.Context, seg *milvuspb.Persiste
 	return bakSeg, nil
 }
 
-func (ct *CollectionTask) listInsertLogByAPI(ctx context.Context, binlogDir string, fieldsBinlog []client.BinlogInfo) ([]*backuppb.FieldBinlog, int64, error) {
+func (ct *CollectionTask) listInsertLogByAPI(ctx context.Context, binlogDir string, fieldsBinlog []milvus.BinlogInfo) ([]*backuppb.FieldBinlog, int64, error) {
 	keys, sizes, err := storage.ListPrefixFlat(ctx, ct.milvusStorage, binlogDir, true)
 	if err != nil {
 		return nil, 0, fmt.Errorf("backup: list insert logs %w", err)
@@ -449,7 +449,7 @@ func (ct *CollectionTask) listInsertLogByAPI(ctx context.Context, binlogDir stri
 	return bakFieldsBinlog, lo.Sum(sizes), nil
 }
 
-func (ct *CollectionTask) listDeltaLogByAPI(ctx context.Context, binlogDir string, fieldsBinlog []client.BinlogInfo) ([]*backuppb.FieldBinlog, int64, error) {
+func (ct *CollectionTask) listDeltaLogByAPI(ctx context.Context, binlogDir string, fieldsBinlog []milvus.BinlogInfo) ([]*backuppb.FieldBinlog, int64, error) {
 	keys, sizes, err := storage.ListPrefixFlat(ctx, ct.milvusStorage, binlogDir, true)
 	if err != nil {
 		return nil, 0, fmt.Errorf("backup: list insert logs %w", err)

core/backup/collection_test.go

@@ -10,7 +10,7 @@ import (
 	"github.com/stretchr/testify/mock"
 	"go.uber.org/zap"
 
-	"github.com/zilliztech/milvus-backup/core/client"
+	"github.com/zilliztech/milvus-backup/core/client/milvus"
 	"github.com/zilliztech/milvus-backup/core/proto/backuppb"
 	"github.com/zilliztech/milvus-backup/core/storage"
 	"github.com/zilliztech/milvus-backup/core/storage/mpath"
@@ -135,7 +135,7 @@ func TestCollectionTask_listInsertLogByAPI(t *testing.T) {
 	ct.milvusStorage = newMockStorage(t, dir, objs)
 	ct.milvusRootPath = "base"
 
-	binlogs := []client.BinlogInfo{{FieldID: 1, LogIDs: []int64{1, 2}}, {FieldID: 2, LogIDs: []int64{1, 2}}}
+	binlogs := []milvus.BinlogInfo{{FieldID: 1, LogIDs: []int64{1, 2}}, {FieldID: 2, LogIDs: []int64{1, 2}}}
 	fields, size, err := ct.listInsertLogByAPI(context.Background(), dir, binlogs)
 	assert.NoError(t, err)
 	assert.Equal(t, int64(10), size)
@@ -165,7 +165,7 @@ func TestCollectionTask_listDeltaLogByAPI(t *testing.T) {
 	ct.milvusStorage = newMockStorage(t, dir, objs)
 	ct.milvusRootPath = "base"
 
-	binlogs := []client.BinlogInfo{{FieldID: 1, LogIDs: []int64{1, 2}}}
+	binlogs := []milvus.BinlogInfo{{FieldID: 1, LogIDs: []int64{1, 2}}}
 	fields, size, err := ct.listDeltaLogByAPI(context.Background(), dir, binlogs)
 	assert.NoError(t, err)
 	assert.Equal(t, int64(3), size)

core/backup/database.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"fmt"
 
-	"github.com/zilliztech/milvus-backup/core/client"
+	"github.com/zilliztech/milvus-backup/core/client/milvus"
 	"github.com/zilliztech/milvus-backup/core/meta"
 	"github.com/zilliztech/milvus-backup/core/pbconv"
 	"github.com/zilliztech/milvus-backup/core/proto/backuppb"
@@ -16,10 +16,10 @@ type DatabaseTask struct {
 	dbName string
 
 	meta *meta.MetaManager
-	grpc client.Grpc
+	grpc milvus.Grpc
 }
 
-func NewDatabaseTask(backupID, dbName string, grpc client.Grpc, meta *meta.MetaManager) *DatabaseTask {
+func NewDatabaseTask(backupID, dbName string, grpc milvus.Grpc, meta *meta.MetaManager) *DatabaseTask {
 	return &DatabaseTask{backupID: backupID, dbName: dbName, meta: meta, grpc: grpc}
 }
 
@@ -28,7 +28,7 @@ func (dt *DatabaseTask) Execute(ctx context.Context) error {
 	var dbID int64
 
 	// if milvus does not support database, skip describe database
-	if dt.grpc.HasFeature(client.DescribeDatabase) {
+	if dt.grpc.HasFeature(milvus.DescribeDatabase) {
 		resp, err := dt.grpc.DescribeDatabase(ctx, dt.dbName)
 		if err != nil {
 			return fmt.Errorf("backup: describe database %s: %w", dt.dbName, err)

core/backup/database_test.go

@@ -9,17 +9,17 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 
-	"github.com/zilliztech/milvus-backup/core/client"
+	"github.com/zilliztech/milvus-backup/core/client/milvus"
 	"github.com/zilliztech/milvus-backup/core/meta"
 	"github.com/zilliztech/milvus-backup/core/namespace"
 	"github.com/zilliztech/milvus-backup/core/proto/backuppb"
 )
 
 func TestDatabaseTask_Execute(t *testing.T) {
 	t.Run("SupportDescribeDatabase", func(t *testing.T) {
-		cli := client.NewMockGrpc(t)
+		cli := milvus.NewMockGrpc(t)
 
-		cli.EXPECT().HasFeature(client.DescribeDatabase).Return(true).Once()
+		cli.EXPECT().HasFeature(milvus.DescribeDatabase).Return(true).Once()
 		cli.EXPECT().DescribeDatabase(mock.Anything, "db1").Return(&milvuspb.DescribeDatabaseResponse{
 			DbID: 1,
 			Properties: []*commonpb.KeyValuePair{
@@ -41,9 +41,9 @@ func TestDatabaseTask_Execute(t *testing.T) {
 	})
 
 	t.Run("NotSupportDescribeDatabase", func(t *testing.T) {
-		cli := client.NewMockGrpc(t)
+		cli := milvus.NewMockGrpc(t)
 
-		cli.EXPECT().HasFeature(client.DescribeDatabase).Return(false).Once()
+		cli.EXPECT().HasFeature(milvus.DescribeDatabase).Return(false).Once()
 
 		metaMgr := meta.NewMetaManager()
 		metaMgr.AddBackup(&backuppb.BackupInfo{Id: "backup1"})

core/backup/rbac.go

@@ -8,7 +8,7 @@ import (
 	"github.com/samber/lo"
 	"go.uber.org/zap"
 
-	"github.com/zilliztech/milvus-backup/core/client"
+	"github.com/zilliztech/milvus-backup/core/client/milvus"
 	"github.com/zilliztech/milvus-backup/core/meta"
 	"github.com/zilliztech/milvus-backup/core/proto/backuppb"
 	"github.com/zilliztech/milvus-backup/internal/log"
@@ -18,12 +18,12 @@ type RBACTask struct {
 	backupID string
 	meta     *meta.MetaManager
 
-	grpcCli client.Grpc
+	grpcCli milvus.Grpc
 
 	logger *zap.Logger
 }
 
-func NewRBACTask(backupID string, meta *meta.MetaManager, grpcCli client.Grpc) *RBACTask {
+func NewRBACTask(backupID string, meta *meta.MetaManager, grpcCli milvus.Grpc) *RBACTask {
 	logger := log.L().With(zap.String("backup_id", backupID))
 	return &RBACTask{
 		backupID: backupID,

core/backup/rpc_channel_pos.go

@@ -6,7 +6,7 @@ import (
 
 	"go.uber.org/zap"
 
-	"github.com/zilliztech/milvus-backup/core/client"
+	"github.com/zilliztech/milvus-backup/core/client/milvus"
 	"github.com/zilliztech/milvus-backup/core/meta"
 	"github.com/zilliztech/milvus-backup/internal/log"
 )
@@ -15,12 +15,12 @@ type RPCChannelPOSTask struct {
 	backupID       string
 	rpcChannelName string
 
-	grpc   client.Grpc
+	grpc   milvus.Grpc
 	meta   *meta.MetaManager
 	logger *zap.Logger
 }
 
-func NewRPCChannelPOSTask(backupID, rpcChannelName string, grpc client.Grpc, meta *meta.MetaManager) *RPCChannelPOSTask {
+func NewRPCChannelPOSTask(backupID, rpcChannelName string, grpc milvus.Grpc, meta *meta.MetaManager) *RPCChannelPOSTask {
 	return &RPCChannelPOSTask{
 		backupID: backupID,