Support collection level concurrent backup and restore

Signed-off-by: wayblink <[email protected]>

configs/backup.yaml

@@ -38,4 +38,16 @@ minio:
   backupRootPath: "backup" # Rootpath to store backup data. Backup data will store to backupBucketName/backupRootPath
 
 backup:
-  maxSegmentGroupSize: 2G
+  maxSegmentGroupSize: 2G
+  parallelism: 1 # collection level parallelism to backup, default 1
+  copydata:
+    # thread pool to copy data for each collection backup, default 100. 
+    # which means if you set backup.parallelism = 2 backup.copydata.parallelism = 100, there will be 200 copy executing at the same time.
+    # reduce it if blocks your storage's network bandwidth
+    parallelism: 100
+
+restore:
+  # Collection level parallelism to restore, default 1
+  # Only change it > 1 when you have more than one datanode.
+  # Because the max parallelism of Milvus bulkinsert is equal to datanodes' number.
+  parallelism: 1

core/backup_context.go

@@ -36,7 +36,7 @@ type BackupContext struct {
 	params  paramtable.BackupParams
 
 	// milvus client
-	milvusClient *gomilvus.Client
+	milvusClient *MilvusClient
 
 	// data storage client
 	storageClient    *storage.ChunkManager
@@ -50,7 +50,7 @@ type BackupContext struct {
 
 	restoreTasks map[string]*backuppb.RestoreBackupTask
 
-	//copyWorkerPool *common.WorkerPool
+	//backupWorkerPool *common.WorkerPool
 }
 
 func CreateMilvusClient(ctx context.Context, params paramtable.BackupParams) (gomilvus.Client, error) {
@@ -88,35 +88,9 @@ func CreateStorageClient(ctx context.Context, params paramtable.BackupParams) (s
 }
 
 func (b *BackupContext) Start() error {
-	// start milvus go SDK client
-	//milvusClient, err := CreateMilvusClient(b.ctx, b.params)
-	//if err != nil {
-	//	log.Error("failed to initial milvus client", zap.Error(err))
-	//	return err
-	//}
-	//b.milvusClient = milvusClient
-
-	// start milvus storage client
-	//minioClient, err := CreateStorageClient(b.ctx, b.params)
-	//if err != nil {
-	//	log.Error("failed to initial storage client", zap.Error(err))
-	//	return err
-	//}
-	//b.storageClient = minioClient
-
 	b.backupTasks = sync.Map{}
 	b.backupNameIdDict = sync.Map{}
 	b.restoreTasks = make(map[string]*backuppb.RestoreBackupTask)
-
-	// init worker pool
-	//wp, err := common.NewWorkerPool(b.ctx, WORKER_NUM, RPS)
-	//if err != nil {
-	//	log.Error("failed to initial copy data woker pool", zap.Error(err))
-	//	return err
-	//}
-	//b.copyWorkerPool = wp
-	//b.copyWorkerPool.Start()
-
 	b.started = true
 	return nil
 }
@@ -141,16 +115,18 @@ func CreateBackupContext(ctx context.Context, params paramtable.BackupParams) *B
 	}
 }
 
-func (b *BackupContext) getMilvusClient() gomilvus.Client {
+func (b *BackupContext) getMilvusClient() *MilvusClient {
 	if b.milvusClient == nil {
 		milvusClient, err := CreateMilvusClient(b.ctx, b.params)
 		if err != nil {
 			log.Error("failed to initial milvus client", zap.Error(err))
 			panic(err)
 		}
-		b.milvusClient = &milvusClient
+		b.milvusClient = &MilvusClient{
+			client: milvusClient,
+		}
 	}
-	return *b.milvusClient
+	return b.milvusClient
 }
 
 func (b *BackupContext) getStorageClient() storage.ChunkManager {
@@ -165,6 +141,19 @@ func (b *BackupContext) getStorageClient() storage.ChunkManager {
 	return *b.storageClient
 }
 
+//func (b *BackupContext) getBackupWorkerPool() *common.WorkerPool {
+//	if b.backupWorkerPool == nil {
+//		wp, err := common.NewWorkerPool(b.ctx, b.params.BackupCfg.BackupParallelism, RPS)
+//		if err != nil {
+//			log.Error("failed to initial copy data woker pool", zap.Error(err))
+//			panic(err)
+//		}
+//		b.backupWorkerPool = wp
+//		b.backupWorkerPool.Start()
+//	}
+//	return b.backupWorkerPool
+//}
+
 func (b *BackupContext) GetBackup(ctx context.Context, request *backuppb.GetBackupRequest) *backuppb.BackupInfoResponse {
 	if request.GetRequestId() == "" {
 		request.RequestId = utils.UUID()