support rclone options in backup and restore cr (#2318) (#2326)

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

@@ -63,9 +63,10 @@ func (bo *Options) cleanBRRemoteBackupData(backup *v1alpha1.Backup) error {
 	return nil
 }
 
-func (bo *Options) cleanRemoteBackupData(bucket string) error {
+func (bo *Options) cleanRemoteBackupData(bucket string, opts []string) error {
 	destBucket := util.NormalizeBucketURI(bucket)
-	output, err := exec.Command("rclone", constants.RcloneConfigArg, "deletefile", destBucket).CombinedOutput()
+	args := util.ConstructArgs(constants.RcloneConfigArg, opts, "deletefile", destBucket, "")
+	output, err := exec.Command("rclone", args...).CombinedOutput()
 	if err != nil {
 		return fmt.Errorf("cluster %s, execute rclone deletefile command failed, output: %s, err: %v", bo, string(output), err)
 	}

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

@@ -19,6 +19,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/klog"
 
+	"github.com/pingcap/tidb-operator/cmd/backup-manager/app/util"
 	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
 	listers "github.com/pingcap/tidb-operator/pkg/client/listers/pingcap/v1alpha1"
 	"github.com/pingcap/tidb-operator/pkg/controller"
@@ -68,7 +69,8 @@ func (bm *Manager) performCleanBackup(backup *v1alpha1.Backup) error {
 	if backup.Spec.BR != nil {
 		err = bm.cleanBRRemoteBackupData(backup)
 	} else {
-		err = bm.cleanRemoteBackupData(backup.Status.BackupPath)
+		opts := util.GetOptions(backup.Spec.StorageProvider)
+		err = bm.cleanRemoteBackupData(backup.Status.BackupPath, opts)
 	}
 
 	if err != nil {

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

@@ -74,11 +74,12 @@ func (bo *Options) dumpTidbClusterData() (string, error) {
 	return bfPath, nil
 }
 
-func (bo *Options) backupDataToRemote(source, bucketURI string) error {
+func (bo *Options) backupDataToRemote(source, bucketURI string, opts []string) error {
 	destBucket := util.NormalizeBucketURI(bucketURI)
 	tmpDestBucket := fmt.Sprintf("%s.tmp", destBucket)
+	args := util.ConstructArgs(constants.RcloneConfigArg, opts, "copyto", source, tmpDestBucket)
 	// TODO: We may need to use exec.CommandContext to control timeouts.
-	output, err := exec.Command("rclone", constants.RcloneConfigArg, "copyto", source, tmpDestBucket).CombinedOutput()
+	output, err := exec.Command("rclone", args...).CombinedOutput()
 	if err != nil {
 		return fmt.Errorf("cluster %s, execute rclone copyto command for upload backup data %s failed, output: %s, err: %v", bo, bucketURI, string(output), err)
 	}
@@ -87,7 +88,8 @@ func (bo *Options) backupDataToRemote(source, bucketURI string) error {
 
 	// the backup was a success
 	// remove .tmp extension
-	output, err = exec.Command("rclone", constants.RcloneConfigArg, "moveto", tmpDestBucket, destBucket).CombinedOutput()
+	args = util.ConstructArgs(constants.RcloneConfigArg, opts, "moveto", tmpDestBucket, destBucket)
+	output, err = exec.Command("rclone", args...).CombinedOutput()
 	if err != nil {
 		return fmt.Errorf("cluster %s, execute rclone moveto command failed, output: %s, err: %v", bo, string(output), err)
 	}
@@ -134,12 +136,13 @@ func getCommitTsFromMetadata(backupPath string) (string, error) {
 }
 
 // getBackupSize get the backup data size
-func getBackupSize(backupPath string) (int64, error) {
+func getBackupSize(backupPath string, opts []string) (int64, error) {
 	var size int64
 	if exist := util.IsFileExist(backupPath); !exist {
 		return size, fmt.Errorf("file %s does not exist or is not regular file", backupPath)
 	}
-	out, err := exec.Command("rclone", constants.RcloneConfigArg, "ls", backupPath).CombinedOutput()
+	args := util.ConstructArgs(constants.RcloneConfigArg, opts, "ls", backupPath, "")
+	out, err := exec.Command("rclone", args...).CombinedOutput()
 	if err != nil {
 		return size, fmt.Errorf("failed to get backup %s size, err: %v", backupPath, err)
 	}

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

@@ -229,7 +229,8 @@ func (bm *BackupManager) performBackup(backup *v1alpha1.Backup, db *sql.DB) erro
 	}
 	klog.Infof("archive cluster %s backup data %s success", bm, archiveBackupPath)
 
-	size, err := getBackupSize(archiveBackupPath)
+	opts := util.GetOptions(backup.Spec.StorageProvider)
+	size, err := getBackupSize(archiveBackupPath, opts)
 	if err != nil {
 		klog.Errorf("get cluster %s archived backup file %s size %d failed, err: %s", bm, archiveBackupPath, size, err)
 		return bm.StatusUpdater.Update(backup, &v1alpha1.BackupCondition{
@@ -255,7 +256,7 @@ func (bm *BackupManager) performBackup(backup *v1alpha1.Backup, db *sql.DB) erro
 
 	remotePath := strings.TrimPrefix(archiveBackupPath, constants.BackupRootPath+"/")
 	bucketURI := bm.getDestBucketURI(remotePath)
-	err = bm.backupDataToRemote(archiveBackupPath, bucketURI)
+	err = bm.backupDataToRemote(archiveBackupPath, bucketURI, opts)
 	if err != nil {
 		klog.Errorf("backup cluster %s data to %s failed, err: %s", bm, bm.StorageType, err)
 		return bm.StatusUpdater.Update(backup, &v1alpha1.BackupCondition{

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

@@ -127,7 +127,8 @@ func (rm *RestoreManager) performRestore(restore *v1alpha1.Restore) error {
 	}
 
 	restoreDataPath := rm.getRestoreDataPath()
-	if err := rm.downloadBackupData(restoreDataPath); err != nil {
+	opts := util.GetOptions(restore.Spec.StorageProvider)
+	if err := rm.downloadBackupData(restoreDataPath, opts); err != nil {
 		klog.Errorf("download cluster %s backup %s data failed, err: %s", rm, rm.BackupPath, err)
 		return rm.StatusUpdater.Update(restore, &v1alpha1.RestoreCondition{
 			Type:    v1alpha1.RestoreFailed,

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

@@ -15,13 +15,15 @@ package _import
 
 import (
 	"fmt"
+	"io/ioutil"
 	"os/exec"
 	"path/filepath"
 	"strings"
 
 	"github.com/mholt/archiver"
 	"github.com/pingcap/tidb-operator/cmd/backup-manager/app/constants"
 	"github.com/pingcap/tidb-operator/cmd/backup-manager/app/util"
+	"k8s.io/klog"
 )
 
 // Options contains the input arguments to the restore command
@@ -36,18 +38,41 @@ func (ro *Options) getRestoreDataPath() string {
 	return filepath.Join(constants.BackupRootPath, bucketName, backupName)
 }
 
-func (ro *Options) downloadBackupData(localPath string) error {
+func (ro *Options) downloadBackupData(localPath string, opts []string) error {
 	if err := util.EnsureDirectoryExist(filepath.Dir(localPath)); err != nil {
 		return err
 	}
 
 	remoteBucket := util.NormalizeBucketURI(ro.BackupPath)
-	rcCopy := exec.Command("rclone", constants.RcloneConfigArg, "copyto", remoteBucket, localPath)
+	args := util.ConstructArgs(constants.RcloneConfigArg, opts, "copyto", remoteBucket, localPath)
+	rcCopy := exec.Command("rclone", args...)
+
+	stdOut, err := rcCopy.StdoutPipe()
+	if err != nil {
+		return fmt.Errorf("cluster %s, create stdout pipe failed, err: %v", ro, err)
+	}
+	stdErr, err := rcCopy.StderrPipe()
+	if err != nil {
+		return fmt.Errorf("cluster %s, create stderr pipe failed, err: %v", ro, err)
+	}
+
 	if err := rcCopy.Start(); err != nil {
 		return fmt.Errorf("cluster %s, start rclone copyto command for download backup data %s falied, err: %v", ro, ro.BackupPath, err)
 	}
+
+	var errMsg string
+	tmpOut, _ := ioutil.ReadAll(stdOut)
+	if len(tmpOut) > 0 {
+		klog.Infof(string(tmpOut))
+	}
+	tmpErr, _ := ioutil.ReadAll(stdErr)
+	if len(tmpErr) > 0 {
+		klog.Infof(string(tmpErr))
+		errMsg = string(tmpErr)
+	}
+
 	if err := rcCopy.Wait(); err != nil {
-		return fmt.Errorf("cluster %s, execute rclone copyto command for download backup data %s failed, err: %v", ro, ro.BackupPath, err)
+		return fmt.Errorf("cluster %s, execute rclone copyto command for download backup data %s failed, errMsg: %v, err: %v", ro, ro.BackupPath, errMsg, err)
 	}
 
 	return nil

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

@@ -25,6 +25,7 @@ import (
 	cmdutil "k8s.io/kubectl/pkg/cmd/util"
 
 	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
+	"github.com/pingcap/tidb-operator/pkg/backup/util"
 )
 
 var (
@@ -187,3 +188,35 @@ func Suffix(version string) string {
 	}
 	return defaultSuffix
 }
+
+// GetOptions gets the rclone options
+func GetOptions(provider v1alpha1.StorageProvider) []string {
+	st := util.GetStorageType(provider)
+	switch st {
+	case v1alpha1.BackupStorageTypeS3:
+		return provider.S3.Options
+	default:
+		return nil
+	}
+}
+
+// ConstructArgs constructs the rclone args
+func ConstructArgs(conf string, opts []string, command, source, dest string) []string {
+	var args []string
+	if conf != "" {
+		args = append(args, conf)
+	}
+	if len(opts) > 0 {
+		args = append(args, opts...)
+	}
+	if command != "" {
+		args = append(args, command)
+	}
+	if source != "" {
+		args = append(args, source)
+	}
+	if dest != "" {
+		args = append(args, dest)
+	}
+	return args
+}

docs/api-references/docs.md

@@ -7277,6 +7277,17 @@ string
 <p>SSE Sever-Side Encryption.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>options</code></br>
+<em>
+[]string
+</em>
+</td>
+<td>
+<p>Options Rclone options for backup and restore with mydumper and lightning.</p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="s3storageprovidertype">S3StorageProviderType</h3>

manifests/crd.yaml

@@ -8509,6 +8509,12 @@ spec:
                 endpoint:
                   description: Endpoint of S3 compatible storage service
                   type: string
+                options:
+                  description: Options Rclone options for backup and restore with
+                    mydumper and lightning.
+                  items:
+                    type: string
+                  type: array
                 path:
                   description: 'Path is the full path where the backup is saved. The
                     format of the path must be: "<bucket-name>/<path-to-backup-file>"'
@@ -9330,6 +9336,12 @@ spec:
                 endpoint:
                   description: Endpoint of S3 compatible storage service
                   type: string
+                options:
+                  description: Options Rclone options for backup and restore with
+                    mydumper and lightning.
+                  items:
+                    type: string
+                  type: array
                 path:
                   description: 'Path is the full path where the backup is saved. The
                     format of the path must be: "<bucket-name>/<path-to-backup-file>"'
@@ -10242,6 +10254,12 @@ spec:
                     endpoint:
                       description: Endpoint of S3 compatible storage service
                       type: string
+                    options:
+                      description: Options Rclone options for backup and restore with
+                        mydumper and lightning.
+                      items:
+                        type: string
+                      type: array
                     path:
                       description: 'Path is the full path where the backup is saved.
                         The format of the path must be: "<bucket-name>/<path-to-backup-file>"'

pkg/apis/pingcap/v1alpha1/types.go

@@ -844,6 +844,8 @@ type S3StorageProvider struct {
 	Prefix string `json:"prefix,omitempty"`
 	// SSE Sever-Side Encryption.
 	SSE string `json:"sse,omitempty"`
+	// Options Rclone options for backup and restore with mydumper and lightning.
+	Options []string `json:"options,omitempty"`
 }
 
 // +k8s:openapi-gen=true