Add backup parallel&retry (#15828)

Add backup parallel&retry Approved by: @XuPeng-SH, @daviszhen
matrixorigin · May 3, 2024 · 6e94513 · 6e94513
1 parent 09d98cb
commit 6e94513
Show file tree

Hide file tree

Showing 6 changed files with 243 additions and 92 deletions.
diff --git a/pkg/backup/backup.go b/pkg/backup/backup.go
@@ -125,7 +125,7 @@ func backupConfigs(ctx context.Context, cfg *Config) error {
 
 var backupTae = func(ctx context.Context, config *Config) error {
 	fs := fileservice.SubPath(config.TaeDir, taeDir)
-	return BackupData(ctx, config.SharedFs, fs, "")
+	return BackupData(ctx, config.SharedFs, fs, "", int(config.Parallelism))
 }
 
 func backupHakeeper(ctx context.Context, config *Config) error {
@@ -218,6 +218,7 @@ func saveTaeFilesList(ctx context.Context, Fs fileservice.FileService, taeFiles
 	if err != nil {
 		return err
 	}
+
 	return writeFile(ctx, Fs, taeSum, []byte(metas))
 }
 

diff --git a/pkg/backup/backup_test.go b/pkg/backup/backup_test.go
@@ -123,7 +123,7 @@ func TestBackupData(t *testing.T) {
 	for _, location := range files {
 		locations = append(locations, location)
 	}
-	err = execBackup(ctx, db.Opts.Fs, service, locations)
+	err = execBackup(ctx, db.Opts.Fs, service, locations, 1)
 	assert.Nil(t, err)
 	db.Opts.Fs = service
 	db.Restart(ctx)

diff --git a/pkg/backup/fs.go b/pkg/backup/fs.go
@@ -117,16 +117,23 @@ func etlFSDir(filepath string) string {
 func writeFile(ctx context.Context, fs fileservice.FileService, path string, data []byte) error {
 	var err error
 	//write file
-	err = fs.Write(ctx, fileservice.IOVector{
-		FilePath: path,
-		Entries: []fileservice.IOEntry{
-			{
-				Offset: 0,
-				Size:   int64(len(data)),
-				Data:   data,
-			},
+	_, err = fileservice.DoWithRetry(
+		"BackupWrite",
+		func() (int, error) {
+			return 0, fs.Write(ctx, fileservice.IOVector{
+				FilePath: path,
+				Entries: []fileservice.IOEntry{
+					{
+						Offset: 0,
+						Size:   int64(len(data)),
+						Data:   data,
+					},
+				},
+			})
 		},
-	})
+		64,
+		fileservice.IsRetryableError,
+	)
 	if err != nil {
 		return err
 	}
@@ -135,16 +142,23 @@ func writeFile(ctx context.Context, fs fileservice.FileService, path string, dat
 
 	//write checksum file for the file
 	checksumFile := path + ".sha256"
-	err = fs.Write(ctx, fileservice.IOVector{
-		FilePath: checksumFile,
-		Entries: []fileservice.IOEntry{
-			{
-				Offset: 0,
-				Size:   int64(len(checksum)),
-				Data:   checksum[:],
-			},
+	_, err = fileservice.DoWithRetry(
+		"BackupWrite",
+		func() (int, error) {
+			return 0, fs.Write(ctx, fileservice.IOVector{
+				FilePath: checksumFile,
+				Entries: []fileservice.IOEntry{
+					{
+						Offset: 0,
+						Size:   int64(len(checksum)),
+						Data:   checksum[:],
+					},
+				},
+			})
 		},
-	})
+		64,
+		fileservice.IsRetryableError,
+	)
 	return err
 }