support prefix for backup path

cmd/backup-manager/app/export/export.go

@@ -33,6 +33,7 @@ import (
 type Options struct {
 	util.GenericOptions
 	Bucket      string
+	Prefix      string
 	StorageType string
 }
 
@@ -41,8 +42,14 @@ func (bo *Options) getBackupFullPath() string {
 }
 
 func (bo *Options) getBackupRelativePath() string {
+	var backupRelativePath string
 	backupName := fmt.Sprintf("backup-%s", time.Now().UTC().Format(time.RFC3339))
-	return fmt.Sprintf("%s/%s", bo.Bucket, backupName)
+	if len(bo.Prefix) == 0 {
+		backupRelativePath = fmt.Sprintf("%s/%s", bo.Bucket, backupName)
+	} else {
+		backupRelativePath = fmt.Sprintf("%s/%s/%s", bo.Bucket, bo.Prefix, backupName)
+	}
+	return backupRelativePath
 }
 
 func (bo *Options) getDestBucketURI(remotePath string) string {

cmd/backup-manager/app/export/manager.go

@@ -22,6 +22,7 @@ import (
 	"github.com/pingcap/tidb-operator/cmd/backup-manager/app/util"
 	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
 	bkconstants "github.com/pingcap/tidb-operator/pkg/backup/constants"
+	backuputil "github.com/pingcap/tidb-operator/pkg/backup/util"
 	listers "github.com/pingcap/tidb-operator/pkg/client/listers/pingcap/v1alpha1"
 	"github.com/pingcap/tidb-operator/pkg/controller"
 	corev1 "k8s.io/api/core/v1"
@@ -49,7 +50,7 @@ func NewBackupManager(
 	}
 }
 
-func (bm *BackupManager) setOptions(backup *v1alpha1.Backup) {
+func (bm *BackupManager) setOptions(backup *v1alpha1.Backup) (string, error) {
 	bm.Options.Host = backup.Spec.From.Host
 
 	if backup.Spec.From.Port != 0 {
@@ -63,8 +64,14 @@ func (bm *BackupManager) setOptions(backup *v1alpha1.Backup) {
 	} else {
 		bm.Options.User = bkconstants.DefaultTidbUser
 	}
-
 	bm.Options.Password = util.GetOptionValueFromEnv(bkconstants.TidbPasswordKey, bkconstants.BackupManagerEnvVarPrefix)
+
+	prefix, reason, err := backuputil.GetBackupPrefixName(backup)
+	if err != nil {
+		return reason, err
+	}
+	bm.Options.Prefix = prefix
+	return "", nil
 }
 
 // ProcessBackup used to process the backup logic
@@ -80,7 +87,16 @@ func (bm *BackupManager) ProcessBackup() error {
 		})
 	}
 
-	bm.setOptions(backup)
+	reason, err := bm.setOptions(backup)
+	if err != nil {
+		klog.Errorf("set mydumper backup %s option for cluster %s failed, err: %v", bm.ResourceName, bm, err)
+		return bm.StatusUpdater.Update(backup, &v1alpha1.BackupCondition{
+			Type:    v1alpha1.BackupFailed,
+			Status:  corev1.ConditionTrue,
+			Reason:  reason,
+			Message: err.Error(),
+		})
+	}
 
 	var db *sql.DB
 	var dsn string

cmd/backup-manager/app/import/restore.go

@@ -33,9 +33,9 @@ type Options struct {
 }
 
 func (ro *Options) getRestoreDataPath() string {
-	backupName := filepath.Base(ro.BackupPath)
-	bucketName := filepath.Base(filepath.Dir(ro.BackupPath))
-	return filepath.Join(constants.BackupRootPath, bucketName, backupName)
+	// The backupPath format must be like this storageType://backup/to/path, so the split array must have two elements
+	backupSuffix := strings.Split(ro.BackupPath, "://")[1]
+	return filepath.Join(constants.BackupRootPath, backupSuffix)
 }
 
 func (ro *Options) downloadBackupData(localPath string, opts []string) error {

pkg/backup/util/util.go

@@ -275,6 +275,24 @@ func GetBackupBucketName(backup *v1alpha1.Backup) (string, string, error) {
 	return bucketName, "", nil
 }
 
+// GetBackupPrefixName return the prefix for remote storage
+func GetBackupPrefixName(backup *v1alpha1.Backup) (string, string, error) {
+	ns := backup.GetNamespace()
+	name := backup.GetName()
+	storageType := GetStorageType(backup.Spec.StorageProvider)
+	var prefix string
+
+	switch storageType {
+	case v1alpha1.BackupStorageTypeS3:
+		prefix = backup.Spec.S3.Prefix
+	case v1alpha1.BackupStorageTypeGcs:
+		prefix = backup.Spec.Gcs.Prefix
+	default:
+		return prefix, "UnsupportedStorageType", fmt.Errorf("backup %s/%s unsupported storage type %s", ns, name, storageType)
+	}
+	return prefix, "", nil
+}
+
 // GetStorageType return the backup storage type according to the specified StorageProvider
 func GetStorageType(provider v1alpha1.StorageProvider) v1alpha1.BackupStorageType {
 	// If there are multiple storages in the StorageProvider, the first one found is returned in the following order