Uploader Implementation: Restic backup and restore

Signed-off-by: Ming <mqiu@vmware.com>

changelogs/unreleased/5214-qiuming-best

@@ -0,0 +1 @@
+Uploader Implementation: Restic backup and restore

pkg/builder/backup_repository_builer.go

@@ -0,0 +1,59 @@
+/*
+Copyright the Velero contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package builder
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+)
+
+// BackupRepositoryBuilder builds BackupRepository objects.
+type BackupRepositoryBuilder struct {
+	object *velerov1api.BackupRepository
+}
+
+// ForBackupRepository is the constructor for a BackupRepositoryBuilder.
+func ForBackupRepository(ns, name string) *BackupRepositoryBuilder {
+	return &BackupRepositoryBuilder{
+		object: &velerov1api.BackupRepository{
+			Spec: velerov1api.BackupRepositorySpec{ResticIdentifier: ""},
+			TypeMeta: metav1.TypeMeta{
+				APIVersion: velerov1api.SchemeGroupVersion.String(),
+				Kind:       "BackupRepository",
+			},
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: ns,
+				Name:      name,
+			},
+		},
+	}
+}
+
+// Result returns the built BackupRepository.
+func (b *BackupRepositoryBuilder) Result() *velerov1api.BackupRepository {
+	return b.object
+}
+
+// ObjectMeta applies functional options to the BackupRepository's ObjectMeta.
+func (b *BackupRepositoryBuilder) ObjectMeta(opts ...ObjectMetaOpt) *BackupRepositoryBuilder {
+	for _, opt := range opts {
+		opt(b.object)
+	}
+
+	return b
+}

pkg/cmd/cli/restic/server.go

@@ -52,7 +52,6 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/signals"
 	"github.com/vmware-tanzu/velero/pkg/controller"
 	"github.com/vmware-tanzu/velero/pkg/metrics"
-	"github.com/vmware-tanzu/velero/pkg/restic"
 	"github.com/vmware-tanzu/velero/pkg/util/filesystem"
 	"github.com/vmware-tanzu/velero/pkg/util/logging"
 )
@@ -197,22 +196,27 @@ func (s *resticServer) run() {
 		s.logger.Fatalf("Failed to create credentials file store: %v", err)
 	}
 
+	credSecretStore, err := credentials.NewNamespacedSecretStore(s.mgr.GetClient(), s.namespace)
+	if err != nil {
+		s.logger.Fatalf("Failed to create secret file store: %v", err)
+	}
+
+	credentialGetter := &credentials.CredentialGetter{FromFile: credentialFileStore, FromSecret: credSecretStore}
 	pvbReconciler := controller.PodVolumeBackupReconciler{
-		Scheme:         s.mgr.GetScheme(),
-		Client:         s.mgr.GetClient(),
-		Clock:          clock.RealClock{},
-		Metrics:        s.metrics,
-		CredsFileStore: credentialFileStore,
-		NodeName:       s.nodeName,
-		FileSystem:     filesystem.NewFileSystem(),
-		ResticExec:     restic.BackupExec{},
-		Log:            s.logger,
+		Scheme:           s.mgr.GetScheme(),
+		Client:           s.mgr.GetClient(),
+		Clock:            clock.RealClock{},
+		Metrics:          s.metrics,
+		CredentialGetter: credentialGetter,
+		NodeName:         s.nodeName,
+		FileSystem:       filesystem.NewFileSystem(),
+		Log:              s.logger,
 	}
 	if err := pvbReconciler.SetupWithManager(s.mgr); err != nil {
 		s.logger.Fatal(err, "unable to create controller", "controller", controller.PodVolumeBackup)
 	}
 
-	if err = controller.NewPodVolumeRestoreReconciler(s.logger, s.mgr.GetClient(), credentialFileStore).SetupWithManager(s.mgr); err != nil {
+	if err = controller.NewPodVolumeRestoreReconciler(s.logger, s.mgr.GetClient(), credentialGetter).SetupWithManager(s.mgr); err != nil {
 		s.logger.WithError(err).Fatal("Unable to create the pod volume restore controller")
 	}
 

pkg/controller/pod_volume_backup_controller.go

@@ -19,7 +19,6 @@ package controller
 import (
 	"context"
 	"fmt"
-	"os"
 	"strings"
 	"time"
 
@@ -37,29 +36,25 @@ import (
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/metrics"
 	repokey "github.com/vmware-tanzu/velero/pkg/repository/keys"
-	"github.com/vmware-tanzu/velero/pkg/restic"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
+	"github.com/vmware-tanzu/velero/pkg/uploader/provider"
 	"github.com/vmware-tanzu/velero/pkg/util/filesystem"
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
 )
 
-// BackupExecuter runs backups.
-type BackupExecuter interface {
-	RunBackup(*restic.Command, logrus.FieldLogger, func(velerov1api.PodVolumeOperationProgress)) (string, string, error)
-	GetSnapshotID(*restic.Command) (string, error)
-}
+// For unit test to mock function
+var NewUploaderProviderFunc = provider.NewUploaderProvider
 
 // PodVolumeBackupReconciler reconciles a PodVolumeBackup object
 type PodVolumeBackupReconciler struct {
-	Scheme         *runtime.Scheme
-	Client         client.Client
-	Clock          clock.Clock
-	Metrics        *metrics.ServerMetrics
-	CredsFileStore credentials.FileStore
-	NodeName       string
-	FileSystem     filesystem.Interface
-	ResticExec     BackupExecuter
-	Log            logrus.FieldLogger
+	Scheme           *runtime.Scheme
+	Client           client.Client
+	Clock            clock.Clock
+	Metrics          *metrics.ServerMetrics
+	CredentialGetter *credentials.CredentialGetter
+	NodeName         string
+	FileSystem       filesystem.Interface
+	Log              logrus.FieldLogger
 }
 
 type BackupProgressUpdater struct {
@@ -85,7 +80,6 @@ func (r *PodVolumeBackupReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 		}
 		return ctrl.Result{}, errors.Wrap(err, "getting PodVolumeBackup")
 	}
-
 	if len(pvb.OwnerReferences) == 1 {
 		log = log.WithField(
 			"backup",
@@ -128,16 +122,19 @@ func (r *PodVolumeBackupReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 		return r.updateStatusToFailed(ctx, &pvb, err, fmt.Sprintf("getting pod %s/%s", pvb.Spec.Pod.Namespace, pvb.Spec.Pod.Name), log)
 	}
 
-	var resticDetails resticDetails
-	resticCmd, err := r.buildResticCommand(ctx, log, &pvb, &pod, &resticDetails)
+	volDir, err := kube.GetVolumeDirectory(ctx, log, &pod, pvb.Spec.Volume, r.Client)
 	if err != nil {
-		return r.updateStatusToFailed(ctx, &pvb, err, "building Restic command", log)
+		return r.updateStatusToFailed(ctx, &pvb, err, "getting volume directory name", log)
 	}
 
-	defer func() {
-		os.Remove(resticDetails.credsFile)
-		os.Remove(resticDetails.caCertFile)
-	}()
+	pathGlob := fmt.Sprintf("/host_pods/%s/volumes/*/%s", string(pvb.Spec.Pod.UID), volDir)
+	log.WithField("pathGlob", pathGlob).Debug("Looking for path matching glob")
+
+	path, err := kube.SinglePathMatch(pathGlob, r.FileSystem, log)
+	if err != nil {
+		return r.updateStatusToFailed(ctx, &pvb, err, "identifying unique volume path on host", log)
+	}
+	log.WithField("path", path).Debugf("Found path matching glob")
 
 	backupLocation := &velerov1api.BackupStorageLocation{}
 	if err := r.Client.Get(context.Background(), client.ObjectKey{
@@ -147,47 +144,72 @@ func (r *PodVolumeBackupReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 		return ctrl.Result{}, errors.Wrap(err, "error getting backup storage location")
 	}
 
-	// #4820: restrieve insecureSkipTLSVerify from BSL configuration for
-	// AWS plugin. If nothing is return, that means insecureSkipTLSVerify
-	// is not enable for Restic command.
-	skipTLSRet := restic.GetInsecureSkipTLSVerifyFromBSL(backupLocation, log)
-	if len(skipTLSRet) > 0 {
-		resticCmd.ExtraFlags = append(resticCmd.ExtraFlags, skipTLSRet)
+	// name of ResticRepository is generated with prefix volumeNamespace-backupLocation- and end with random characters
+	// it could not retrieve the ResticRepository CR with namespace + name. so first list all CRs with in the volumeNamespace
+	// then filtering the matched CR with prefix volumeNamespace-backupLocation-
+	backupRepos := &velerov1api.BackupRepositoryList{}
+	var backupRepo velerov1api.BackupRepository
+	isFoundRepo := false
+	if r.Client.List(ctx, backupRepos, &client.ListOptions{
+		Namespace: pvb.Namespace,
+	}); err != nil {
+		return ctrl.Result{}, errors.Wrap(err, "error getting backup repository")
+	} else if len(backupRepos.Items) == 0 {
+		return ctrl.Result{}, errors.Errorf("find empty BackupRepository found for workload namespace %s, backup storage location %s", pvb.Namespace, pvb.Spec.BackupStorageLocation)
+	} else {
+		for _, repo := range backupRepos.Items {
+			if strings.HasPrefix(repo.Name, fmt.Sprintf("%s-%s-", pvb.Spec.Pod.Namespace, pvb.Spec.BackupStorageLocation)) {
+				backupRepo = repo
+				isFoundRepo = true
+				break
+			}
+		}
+		if !isFoundRepo {
+			return ctrl.Result{}, errors.Errorf("could not found match BackupRepository for workload namespace %s, backup storage location %s", pvb.Namespace, pvb.Spec.BackupStorageLocation)
+		}
 	}
 
-	var stdout, stderr string
+	var uploaderProv provider.Provider
+	uploaderProv, err = NewUploaderProviderFunc(ctx, r.Client, pvb.Spec.UploaderType, pvb.Spec.RepoIdentifier,
+		backupLocation, &backupRepo, r.CredentialGetter, repokey.RepoKeySelector(), log)
+	if err != nil {
+		return r.updateStatusToFailed(ctx, &pvb, err, "error creating uploader", log)
+	}
+
+	// If this is a PVC, look for the most recent completed pod volume backup for it and get
+	// its restic snapshot ID to use as the value of the `--parent` flag. Without this,
+	// if the pod using the PVC (and therefore the directory path under /host_pods/) has
+	// changed since the PVC's last backup, restic will not be able to identify a suitable
+	// parent snapshot to use, and will have to do a full rescan of the contents of the PVC.
+	var parentSnapshotID string
+	if pvcUID, ok := pvb.Labels[velerov1api.PVCUIDLabel]; ok {
+		parentSnapshotID = r.getParentSnapshot(ctx, log, pvb.Namespace, pvcUID, pvb.Spec.BackupStorageLocation)
+		if parentSnapshotID == "" {
+			log.Info("No parent snapshot found for PVC, not using --parent flag for this backup")
+		} else {
+			log.WithField("parentSnapshotID", parentSnapshotID).Info("Setting --parent flag for this backup")
+		}
+	}
+
+	defer func() {
+		if err := uploaderProv.Close(ctx); err != nil {
+			log.Errorf("failed to close uploader provider with error %v", err)
+		}
+	}()
 
 	var emptySnapshot bool
-	stdout, stderr, err = r.ResticExec.RunBackup(resticCmd, log, r.updateBackupProgressFunc(&pvb, log))
+	snapshotID, err := uploaderProv.RunBackup(ctx, path, pvb.Spec.Tags, parentSnapshotID, r.NewBackupProgressUpdater(&pvb, log, ctx))
 	if err != nil {
-		if strings.Contains(stderr, "snapshot is empty") {
+		if strings.Contains(err.Error(), "snapshot is empty") {
 			emptySnapshot = true
 		} else {
-			return r.updateStatusToFailed(ctx, &pvb, err, fmt.Sprintf("running Restic backup, stderr=%s", stderr), log)
-		}
-	}
-	log.Debugf("Ran command=%s, stdout=%s, stderr=%s", resticCmd.String(), stdout, stderr)
-
-	var snapshotID string
-	if !emptySnapshot {
-		cmd := restic.GetSnapshotCommand(pvb.Spec.RepoIdentifier, resticDetails.credsFile, pvb.Spec.Tags)
-		cmd.Env = resticDetails.envs
-		cmd.CACertFile = resticDetails.caCertFile
-
-		// #4820: also apply the insecureTLS flag to Restic snapshots command
-		if len(skipTLSRet) > 0 {
-			cmd.ExtraFlags = append(cmd.ExtraFlags, skipTLSRet)
-		}
-
-		snapshotID, err = r.ResticExec.GetSnapshotID(cmd)
-		if err != nil {
-			return r.updateStatusToFailed(ctx, &pvb, err, "getting snapshot id", log)
+			return r.updateStatusToFailed(ctx, &pvb, err, fmt.Sprintf("running Restic backup, stderr=%v", err), log)
 		}
 	}
 
 	// Update status to Completed with path & snapshot ID.
 	original = pvb.DeepCopy()
-	pvb.Status.Path = resticDetails.path
+	pvb.Status.Path = path
 	pvb.Status.Phase = velerov1api.PodVolumeBackupPhaseCompleted
 	pvb.Status.SnapshotID = snapshotID
 	pvb.Status.CompletionTimestamp = &metav1.Time{Time: r.Clock.Now()}
@@ -202,8 +224,9 @@ func (r *PodVolumeBackupReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 	latencyDuration := pvb.Status.CompletionTimestamp.Time.Sub(pvb.Status.StartTimestamp.Time)
 	latencySeconds := float64(latencyDuration / time.Second)
 	backupName := fmt.Sprintf("%s/%s", req.Namespace, pvb.OwnerReferences[0].Name)
-	r.Metrics.ObserveResticOpLatency(r.NodeName, req.Name, resticCmd.Command, backupName, latencySeconds)
-	r.Metrics.RegisterResticOpLatencyGauge(r.NodeName, req.Name, resticCmd.Command, backupName, latencySeconds)
+	generateOpName := fmt.Sprintf("%s-%s-%s-%s-backup", pvb.Name, backupRepo.Name, pvb.Spec.BackupStorageLocation, pvb.Namespace)
+	r.Metrics.ObserveResticOpLatency(r.NodeName, req.Name, generateOpName, backupName, latencySeconds)
+	r.Metrics.RegisterResticOpLatencyGauge(r.NodeName, req.Name, generateOpName, backupName, latencySeconds)
 	r.Metrics.RegisterPodVolumeBackupDequeue(r.NodeName)
 
 	log.Info("PodVolumeBackup completed")
@@ -272,18 +295,6 @@ func (r *PodVolumeBackupReconciler) getParentSnapshot(ctx context.Context, log l
 	return mostRecentPVB.Status.SnapshotID
 }
 
-// updateBackupProgressFunc returns a func that takes progress info and patches
-// the PVB with the new progress.
-func (r *PodVolumeBackupReconciler) updateBackupProgressFunc(pvb *velerov1api.PodVolumeBackup, log logrus.FieldLogger) func(velerov1api.PodVolumeOperationProgress) {
-	return func(progress velerov1api.PodVolumeOperationProgress) {
-		original := pvb.DeepCopy()
-		pvb.Status.Progress = progress
-		if err := r.Client.Patch(context.Background(), pvb, client.MergeFrom(original)); err != nil {
-			log.WithError(err).Error("error update progress")
-		}
-	}
-}
-
 func (r *PodVolumeBackupReconciler) updateStatusToFailed(ctx context.Context, pvb *velerov1api.PodVolumeBackup, err error, msg string, log logrus.FieldLogger) (ctrl.Result, error) {
 	original := pvb.DeepCopy()
 	pvb.Status.Phase = velerov1api.PodVolumeBackupPhaseFailed
@@ -298,81 +309,6 @@ func (r *PodVolumeBackupReconciler) updateStatusToFailed(ctx context.Context, pv
 	return ctrl.Result{}, nil
 }
 
-type resticDetails struct {
-	credsFile, caCertFile string
-	envs                  []string
-	path                  string
-}
-
-func (r *PodVolumeBackupReconciler) buildResticCommand(ctx context.Context, log *logrus.Entry, pvb *velerov1api.PodVolumeBackup, pod *corev1.Pod, details *resticDetails) (*restic.Command, error) {
-	volDir, err := kube.GetVolumeDirectory(ctx, log, pod, pvb.Spec.Volume, r.Client)
-	if err != nil {
-		return nil, errors.Wrap(err, "getting volume directory name")
-	}
-
-	pathGlob := fmt.Sprintf("/host_pods/%s/volumes/*/%s", string(pvb.Spec.Pod.UID), volDir)
-	log.WithField("pathGlob", pathGlob).Debug("Looking for path matching glob")
-
-	path, err := kube.SinglePathMatch(pathGlob, r.FileSystem, log)
-	if err != nil {
-		return nil, errors.Wrap(err, "identifying unique volume path on host")
-	}
-	log.WithField("path", path).Debugf("Found path matching glob")
-
-	// Temporary credentials.
-	details.credsFile, err = r.CredsFileStore.Path(repokey.RepoKeySelector())
-	if err != nil {
-		return nil, errors.Wrap(err, "creating temporary Restic credentials file")
-	}
-
-	cmd := restic.BackupCommand(pvb.Spec.RepoIdentifier, details.credsFile, path, pvb.Spec.Tags)
-
-	backupLocation := &velerov1api.BackupStorageLocation{}
-	if err := r.Client.Get(context.Background(), client.ObjectKey{
-		Namespace: pvb.Namespace,
-		Name:      pvb.Spec.BackupStorageLocation,
-	}, backupLocation); err != nil {
-		return nil, errors.Wrap(err, "getting backup storage location")
-	}
-
-	// If there's a caCert on the ObjectStorage, write it to disk so that it can
-	// be passed to Restic.
-	if backupLocation.Spec.ObjectStorage != nil &&
-		backupLocation.Spec.ObjectStorage.CACert != nil {
-
-		details.caCertFile, err = restic.TempCACertFile(backupLocation.Spec.ObjectStorage.CACert, pvb.Spec.BackupStorageLocation, r.FileSystem)
-		if err != nil {
-			log.WithError(err).Error("creating temporary caCert file")
-		}
-	}
-	cmd.CACertFile = details.caCertFile
-
-	details.envs, err = restic.CmdEnv(backupLocation, r.CredsFileStore)
-	if err != nil {
-		return nil, errors.Wrap(err, "setting Restic command environment")
-	}
-	cmd.Env = details.envs
-
-	// If this is a PVC, look for the most recent completed PodVolumeBackup for
-	// it and get its Restic snapshot ID to use as the value of the `--parent`
-	// flag. Without this, if the pod using the PVC (and therefore the directory
-	// path under /host_pods/) has changed since the PVC's last backup, Restic
-	// will not be able to identify a suitable parent snapshot to use, and will
-	// have to do a full rescan of the contents of the PVC.
-	if pvcUID, ok := pvb.Labels[velerov1api.PVCUIDLabel]; ok {
-		parentSnapshotID := r.getParentSnapshot(ctx, log, pvb.Namespace, pvcUID, pvb.Spec.BackupStorageLocation)
-		if parentSnapshotID == "" {
-			log.Info("No parent snapshot found for PVC, not using --parent flag for this backup")
-		} else {
-			log.WithField("parentSnapshotID", parentSnapshotID).
-				Info("Setting --parent flag for this backup")
-			cmd.ExtraFlags = append(cmd.ExtraFlags, fmt.Sprintf("--parent=%s", parentSnapshotID))
-		}
-	}
-
-	return cmd, nil
-}
-
 func (r *PodVolumeBackupReconciler) NewBackupProgressUpdater(pvb *velerov1api.PodVolumeBackup, log logrus.FieldLogger, ctx context.Context) *BackupProgressUpdater {
 	return &BackupProgressUpdater{pvb, log, ctx, r.Client}
 }

pkg/controller/pod_volume_backup_controller_test.go

@@ -24,6 +24,7 @@ import (
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/ginkgo/extensions/table"
 	. "github.com/onsi/gomega"
+	"github.com/sirupsen/logrus"
 
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -35,11 +36,13 @@ import (
 	kbclient "sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 
+	"github.com/vmware-tanzu/velero/internal/credentials"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/builder"
 	"github.com/vmware-tanzu/velero/pkg/metrics"
-	"github.com/vmware-tanzu/velero/pkg/restic/mocks"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+	"github.com/vmware-tanzu/velero/pkg/uploader"
+	"github.com/vmware-tanzu/velero/pkg/uploader/provider"
 )
 
 const name = "pvb-1"
@@ -68,12 +71,16 @@ func bslBuilder() *builder.BackupStorageLocationBuilder {
 		ForBackupStorageLocation(velerov1api.DefaultNamespace, "bsl-loc")
 }
 
+func backupRepoBuilder() *builder.BackupRepositoryBuilder {
+	return builder.ForBackupRepository(velerov1api.DefaultNamespace, fmt.Sprintf("%s-bsl-loc-dn24h", velerov1api.DefaultNamespace))
+}
+
 var _ = Describe("PodVolumeBackup Reconciler", func() {
 	type request struct {
-		pvb *velerov1api.PodVolumeBackup
-		pod *corev1.Pod
-		bsl *velerov1api.BackupStorageLocation
-
+		pvb               *velerov1api.PodVolumeBackup
+		pod               *corev1.Pod
+		bsl               *velerov1api.BackupStorageLocation
+		backupRepo        *velerov1api.BackupRepository
 		expectedProcessed bool
 		expected          *velerov1api.PodVolumeBackup
 		expectedRequeue   ctrl.Result
@@ -100,31 +107,41 @@ var _ = Describe("PodVolumeBackup Reconciler", func() {
 			err = fakeClient.Create(ctx, test.bsl)
 			Expect(err).To(BeNil())
 
+			err = fakeClient.Create(ctx, test.backupRepo)
+			Expect(err).To(BeNil())
+
 			fakeFS := velerotest.NewFakeFileSystem()
 			pathGlob := fmt.Sprintf("/host_pods/%s/volumes/*/%s", "", "pvb-1-volume")
 			_, err = fakeFS.Create(pathGlob)
 			Expect(err).To(BeNil())
 
+			credentialFileStore, err := credentials.NewNamespacedFileStore(
+				fakeClient,
+				velerov1api.DefaultNamespace,
+				"/tmp/credentials",
+				fakeFS,
+			)
+
 			// Setup reconciler
 			Expect(velerov1api.AddToScheme(scheme.Scheme)).To(Succeed())
 			r := PodVolumeBackupReconciler{
-				Client:         fakeClient,
-				Clock:          clock.NewFakeClock(now),
-				Metrics:        metrics.NewResticServerMetrics(),
-				CredsFileStore: fakeCredsFileStore{},
-				NodeName:       "test_node",
-				FileSystem:     fakeFS,
-				ResticExec:     mocks.FakeResticBackupExec{},
-				Log:            velerotest.NewLogger(),
+				Client:           fakeClient,
+				Clock:            clock.NewFakeClock(now),
+				Metrics:          metrics.NewResticServerMetrics(),
+				CredentialGetter: &credentials.CredentialGetter{FromFile: credentialFileStore},
+				NodeName:         "test_node",
+				FileSystem:       fakeFS,
+				Log:              velerotest.NewLogger(),
+			}
+			NewUploaderProviderFunc = func(ctx context.Context, client kbclient.Client, uploaderType, repoIdentifier string, bsl *velerov1api.BackupStorageLocation, backupRepo *velerov1api.BackupRepository, credGetter *credentials.CredentialGetter, repoKeySelector *corev1.SecretKeySelector, log logrus.FieldLogger) (provider.Provider, error) {
+				return &fakeProvider{}, nil
 			}
-
 			actualResult, err := r.Reconcile(ctx, ctrl.Request{
 				NamespacedName: types.NamespacedName{
 					Namespace: velerov1api.DefaultNamespace,
 					Name:      test.pvb.Name,
 				},
 			})
-
 			Expect(actualResult).To(BeEquivalentTo(test.expectedRequeue))
 			if test.expectedErrMsg == "" {
 				Expect(err).To(BeNil())
@@ -137,7 +154,6 @@ var _ = Describe("PodVolumeBackup Reconciler", func() {
 				Name:      test.pvb.Name,
 				Namespace: test.pvb.Namespace,
 			}, &pvb)
-
 			// Assertions
 			if test.expected == nil {
 				Expect(apierrors.IsNotFound(err)).To(BeTrue())
@@ -160,6 +176,7 @@ var _ = Describe("PodVolumeBackup Reconciler", func() {
 			pvb:               pvbBuilder().Phase("").Node("test_node").Result(),
 			pod:               podBuilder().Result(),
 			bsl:               bslBuilder().Result(),
+			backupRepo:        backupRepoBuilder().Result(),
 			expectedProcessed: true,
 			expected: builder.ForPodVolumeBackup(velerov1api.DefaultNamespace, "pvb-1").
 				Phase(velerov1api.PodVolumeBackupPhaseCompleted).
@@ -173,6 +190,7 @@ var _ = Describe("PodVolumeBackup Reconciler", func() {
 				Result(),
 			pod:               podBuilder().Result(),
 			bsl:               bslBuilder().Result(),
+			backupRepo:        backupRepoBuilder().Result(),
 			expectedProcessed: true,
 			expected: builder.ForPodVolumeBackup(velerov1api.DefaultNamespace, "pvb-1").
 				Phase(velerov1api.PodVolumeBackupPhaseCompleted).
@@ -186,6 +204,7 @@ var _ = Describe("PodVolumeBackup Reconciler", func() {
 				Result(),
 			pod:               podBuilder().Result(),
 			bsl:               bslBuilder().Result(),
+			backupRepo:        backupRepoBuilder().Result(),
 			expectedProcessed: false,
 			expected: builder.ForPodVolumeBackup(velerov1api.DefaultNamespace, "pvb-1").
 				Phase(velerov1api.PodVolumeBackupPhaseInProgress).
@@ -199,6 +218,7 @@ var _ = Describe("PodVolumeBackup Reconciler", func() {
 				Result(),
 			pod:               podBuilder().Result(),
 			bsl:               bslBuilder().Result(),
+			backupRepo:        backupRepoBuilder().Result(),
 			expectedProcessed: false,
 			expected: builder.ForPodVolumeBackup(velerov1api.DefaultNamespace, "pvb-1").
 				Phase(velerov1api.PodVolumeBackupPhaseCompleted).
@@ -212,6 +232,7 @@ var _ = Describe("PodVolumeBackup Reconciler", func() {
 				Result(),
 			pod:               podBuilder().Result(),
 			bsl:               bslBuilder().Result(),
+			backupRepo:        backupRepoBuilder().Result(),
 			expectedProcessed: false,
 			expected: builder.ForPodVolumeBackup(velerov1api.DefaultNamespace, "pvb-1").
 				Phase(velerov1api.PodVolumeBackupPhaseFailed).
@@ -225,6 +246,7 @@ var _ = Describe("PodVolumeBackup Reconciler", func() {
 				Result(),
 			pod:               podBuilder().Result(),
 			bsl:               bslBuilder().Result(),
+			backupRepo:        backupRepoBuilder().Result(),
 			expectedProcessed: false,
 			expected: builder.ForPodVolumeBackup(velerov1api.DefaultNamespace, "pvb-1").
 				Phase(velerov1api.PodVolumeBackupPhaseFailed).
@@ -238,6 +260,7 @@ var _ = Describe("PodVolumeBackup Reconciler", func() {
 				Result(),
 			pod:               podBuilder().Result(),
 			bsl:               bslBuilder().Result(),
+			backupRepo:        backupRepoBuilder().Result(),
 			expectedProcessed: false,
 			expected: builder.ForPodVolumeBackup(velerov1api.DefaultNamespace, "pvb-1").
 				Phase(velerov1api.PodVolumeBackupPhaseNew).
@@ -251,6 +274,7 @@ var _ = Describe("PodVolumeBackup Reconciler", func() {
 				Result(),
 			pod:               podBuilder().Result(),
 			bsl:               bslBuilder().Result(),
+			backupRepo:        backupRepoBuilder().Result(),
 			expectedProcessed: false,
 			expected: builder.ForPodVolumeBackup(velerov1api.DefaultNamespace, "pvb-1").
 				Phase(velerov1api.PodVolumeBackupPhaseInProgress).
@@ -264,6 +288,7 @@ var _ = Describe("PodVolumeBackup Reconciler", func() {
 				Result(),
 			pod:               podBuilder().Result(),
 			bsl:               bslBuilder().Result(),
+			backupRepo:        backupRepoBuilder().Result(),
 			expectedProcessed: false,
 			expected: builder.ForPodVolumeBackup(velerov1api.DefaultNamespace, "pvb-1").
 				Phase(velerov1api.PodVolumeBackupPhaseCompleted).
@@ -277,6 +302,7 @@ var _ = Describe("PodVolumeBackup Reconciler", func() {
 				Result(),
 			pod:               podBuilder().Result(),
 			bsl:               bslBuilder().Result(),
+			backupRepo:        backupRepoBuilder().Result(),
 			expectedProcessed: false,
 			expected: builder.ForPodVolumeBackup(velerov1api.DefaultNamespace, "pvb-1").
 				Phase(velerov1api.PodVolumeBackupPhaseFailed).
@@ -286,8 +312,26 @@ var _ = Describe("PodVolumeBackup Reconciler", func() {
 	)
 })
 
-type fakeCredsFileStore struct{}
+type fakeProvider struct {
+}
 
-func (f fakeCredsFileStore) Path(selector *corev1.SecretKeySelector) (string, error) {
-	return "/fake/path", nil
+func (f *fakeProvider) RunBackup(
+	ctx context.Context,
+	path string,
+	tags map[string]string,
+	parentSnapshot string,
+	updater uploader.ProgressUpdater) (string, error) {
+	return "", nil
+}
+
+func (f *fakeProvider) RunRestore(
+	ctx context.Context,
+	snapshotID string,
+	volumePath string,
+	updater uploader.ProgressUpdater) error {
+	return nil
+}
+
+func (f *fakeProvider) Close(ctx context.Context) error {
+	return nil
 }

pkg/controller/pod_volume_restore_controller.go

@@ -22,6 +22,7 @@ import (
 	"io/ioutil"
 	"os"
 	"path/filepath"
+	"strings"
 
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
@@ -42,27 +43,28 @@ import (
 	repokey "github.com/vmware-tanzu/velero/pkg/repository/keys"
 	"github.com/vmware-tanzu/velero/pkg/restic"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
+	"github.com/vmware-tanzu/velero/pkg/uploader/provider"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
 	"github.com/vmware-tanzu/velero/pkg/util/filesystem"
 	"github.com/vmware-tanzu/velero/pkg/util/kube"
 )
 
-func NewPodVolumeRestoreReconciler(logger logrus.FieldLogger, client client.Client, credentialsFileStore credentials.FileStore) *PodVolumeRestoreReconciler {
+func NewPodVolumeRestoreReconciler(logger logrus.FieldLogger, client client.Client, credentialGetter *credentials.CredentialGetter) *PodVolumeRestoreReconciler {
 	return &PodVolumeRestoreReconciler{
-		Client:               client,
-		logger:               logger.WithField("controller", "PodVolumeRestore"),
-		credentialsFileStore: credentialsFileStore,
-		fileSystem:           filesystem.NewFileSystem(),
-		clock:                &clock.RealClock{},
+		Client:           client,
+		logger:           logger.WithField("controller", "PodVolumeRestore"),
+		credentialGetter: credentialGetter,
+		fileSystem:       filesystem.NewFileSystem(),
+		clock:            &clock.RealClock{},
 	}
 }
 
 type PodVolumeRestoreReconciler struct {
 	client.Client
-	logger               logrus.FieldLogger
-	credentialsFileStore credentials.FileStore
-	fileSystem           filesystem.Interface
-	clock                clock.Clock
+	logger           logrus.FieldLogger
+	credentialGetter *credentials.CredentialGetter
+	fileSystem       filesystem.Interface
+	clock            clock.Clock
 }
 
 type RestoreProgressUpdater struct {
@@ -239,20 +241,6 @@ func (c *PodVolumeRestoreReconciler) processRestore(ctx context.Context, req *ve
 		return errors.Wrap(err, "error identifying path of volume")
 	}
 
-	credsFile, err := c.credentialsFileStore.Path(repokey.RepoKeySelector())
-	if err != nil {
-		return errors.Wrap(err, "error creating temp restic credentials file")
-	}
-	// ignore error since there's nothing we can do and it's a temp file.
-	defer os.Remove(credsFile)
-
-	resticCmd := restic.RestoreCommand(
-		req.Spec.RepoIdentifier,
-		credsFile,
-		req.Spec.SnapshotID,
-		volumePath,
-	)
-
 	backupLocation := &velerov1api.BackupStorageLocation{}
 	if err := c.Get(ctx, client.ObjectKey{
 		Namespace: req.Namespace,
@@ -261,38 +249,46 @@ func (c *PodVolumeRestoreReconciler) processRestore(ctx context.Context, req *ve
 		return errors.Wrap(err, "error getting backup storage location")
 	}
 
-	// if there's a caCert on the ObjectStorage, write it to disk so that it can be passed to restic
-	var caCertFile string
-	if backupLocation.Spec.ObjectStorage != nil && backupLocation.Spec.ObjectStorage.CACert != nil {
-		caCertFile, err = restic.TempCACertFile(backupLocation.Spec.ObjectStorage.CACert, req.Spec.BackupStorageLocation, c.fileSystem)
-		if err != nil {
-			log.WithError(err).Error("Error creating temp cacert file")
+	// name of ResticRepository is generated with prefix volumeNamespace-backupLocation- and end with random characters
+	// it could not retrieve the ResticRepository CR with namespace + name. so first list all CRs with in the volumeNamespace
+	// then filtering the matched CR with prefix volumeNamespace-backupLocation-
+	backupRepos := &velerov1api.BackupRepositoryList{}
+	var backupRepo velerov1api.BackupRepository
+	isFoundRepo := false
+	if c.List(ctx, backupRepos, &client.ListOptions{
+		Namespace: req.Namespace,
+	}); err != nil {
+		return errors.Wrap(err, "error getting backup repository")
+	} else if len(backupRepos.Items) == 0 {
+		return errors.Errorf("find empty BackupRepository found for workload namespace %s, backup storage location %s", req.Namespace, req.Spec.BackupStorageLocation)
+	} else {
+		for _, repo := range backupRepos.Items {
+			if strings.HasPrefix(repo.Name, fmt.Sprintf("%s-%s-", req.Spec.Pod.Namespace, req.Spec.BackupStorageLocation)) {
+				backupRepo = repo
+				isFoundRepo = true
+				break
+			}
+		}
+		if !isFoundRepo {
+			return errors.Errorf("could not found match BackupRepository for workload namespace %s, backup storage location %s", req.Namespace, req.Spec.BackupStorageLocation)
 		}
-		// ignore error since there's nothing we can do and it's a temp file.
-		defer os.Remove(caCertFile)
 	}
-	resticCmd.CACertFile = caCertFile
 
-	env, err := restic.CmdEnv(backupLocation, c.credentialsFileStore)
+	uploaderProv, err := provider.NewUploaderProvider(ctx, c.Client, req.Spec.UploaderType,
+		req.Spec.RepoIdentifier, backupLocation, &backupRepo, c.credentialGetter, repokey.RepoKeySelector(), log)
 	if err != nil {
-		return errors.Wrap(err, "error setting restic cmd env")
-	}
-	resticCmd.Env = env
-
-	// #4820: restrieve insecureSkipTLSVerify from BSL configuration for
-	// AWS plugin. If nothing is return, that means insecureSkipTLSVerify
-	// is not enable for Restic command.
-	skipTLSRet := restic.GetInsecureSkipTLSVerifyFromBSL(backupLocation, log)
-	if len(skipTLSRet) > 0 {
-		resticCmd.ExtraFlags = append(resticCmd.ExtraFlags, skipTLSRet)
+		return errors.Wrap(err, "error creating uploader")
 	}
 
-	var stdout, stderr string
+	defer func() {
+		if err := uploaderProv.Close(ctx); err != nil {
+			log.Errorf("failed to close uploader provider with error %v", err)
+		}
+	}()
 
-	if stdout, stderr, err = restic.RunRestore(resticCmd, log, c.updateRestoreProgressFunc(req, log)); err != nil {
-		return errors.Wrapf(err, "error running restic restore, cmd=%s, stdout=%s, stderr=%s", resticCmd.String(), stdout, stderr)
+	if err = uploaderProv.RunRestore(ctx, req.Spec.SnapshotID, volumePath, c.NewRestoreProgressUpdater(req, log, ctx)); err != nil {
+		return errors.Wrapf(err, "error running restic restore err=%v", err)
 	}
-	log.Debugf("Ran command=%s, stdout=%s, stderr=%s", resticCmd.String(), stdout, stderr)
 
 	// Remove the .velero directory from the restored volume (it may contain done files from previous restores
 	// of this volume, which we don't want to carry over). If this fails for any reason, log and continue, since
@@ -326,18 +322,6 @@ func (c *PodVolumeRestoreReconciler) processRestore(ctx context.Context, req *ve
 	return nil
 }
 
-// updateRestoreProgressFunc returns a func that takes progress info and patches
-// the PVR with the new progress
-func (c *PodVolumeRestoreReconciler) updateRestoreProgressFunc(req *velerov1api.PodVolumeRestore, log logrus.FieldLogger) func(velerov1api.PodVolumeOperationProgress) {
-	return func(progress velerov1api.PodVolumeOperationProgress) {
-		original := req.DeepCopy()
-		req.Status.Progress = progress
-		if err := c.Patch(context.Background(), req, client.MergeFrom(original)); err != nil {
-			log.WithError(err).Error("Unable to update PodVolumeRestore progress")
-		}
-	}
-}
-
 func (r *PodVolumeRestoreReconciler) NewRestoreProgressUpdater(pvr *velerov1api.PodVolumeRestore, log logrus.FieldLogger, ctx context.Context) *RestoreProgressUpdater {
 	return &RestoreProgressUpdater{pvr, log, ctx, r.Client}
 }

pkg/restic/exec_commands.go

@@ -20,13 +20,10 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
-	"strings"
 	"time"
 
 	"github.com/pkg/errors"
-	"github.com/sirupsen/logrus"
 
-	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/util/exec"
 	"github.com/vmware-tanzu/velero/pkg/util/filesystem"
 )
@@ -69,82 +66,7 @@ func GetSnapshotID(snapshotIdCmd *Command) (string, error) {
 	return snapshots[0].ShortID, nil
 }
 
-// RunBackup runs a `restic backup` command and watches the output to provide
-// progress updates to the caller.
-func RunBackup(backupCmd *Command, log logrus.FieldLogger, updateFunc func(velerov1api.PodVolumeOperationProgress)) (string, string, error) {
-	// buffers for copying command stdout/err output into
-	stdoutBuf := new(bytes.Buffer)
-	stderrBuf := new(bytes.Buffer)
-
-	// create a channel to signal when to end the goroutine scanning for progress
-	// updates
-	quit := make(chan struct{})
-
-	cmd := backupCmd.Cmd()
-	cmd.Stdout = stdoutBuf
-	cmd.Stderr = stderrBuf
-
-	err := cmd.Start()
-	if err != nil {
-		return stdoutBuf.String(), stderrBuf.String(), err
-	}
-
-	go func() {
-		ticker := time.NewTicker(backupProgressCheckInterval)
-		for {
-			select {
-			case <-ticker.C:
-				lastLine := getLastLine(stdoutBuf.Bytes())
-				if len(lastLine) > 0 {
-					stat, err := decodeBackupStatusLine(lastLine)
-					if err != nil {
-						log.WithError(err).Errorf("error getting restic backup progress")
-					}
-
-					// if the line contains a non-empty bytes_done field, we can update the
-					// caller with the progress
-					if stat.BytesDone != 0 {
-						updateFunc(velerov1api.PodVolumeOperationProgress{
-							TotalBytes: stat.TotalBytes,
-							BytesDone:  stat.BytesDone,
-						})
-					}
-				}
-			case <-quit:
-				ticker.Stop()
-				return
-			}
-		}
-	}()
-
-	err = cmd.Wait()
-	if err != nil {
-		return stdoutBuf.String(), stderrBuf.String(), err
-	}
-	quit <- struct{}{}
-
-	summary, err := getSummaryLine(stdoutBuf.Bytes())
-	if err != nil {
-		return stdoutBuf.String(), stderrBuf.String(), err
-	}
-	stat, err := decodeBackupStatusLine(summary)
-	if err != nil {
-		return stdoutBuf.String(), stderrBuf.String(), err
-	}
-	if stat.MessageType != "summary" {
-		return stdoutBuf.String(), stderrBuf.String(), errors.WithStack(fmt.Errorf("error getting restic backup summary: %s", string(summary)))
-	}
-
-	// update progress to 100%
-	updateFunc(velerov1api.PodVolumeOperationProgress{
-		TotalBytes: stat.TotalBytesProcessed,
-		BytesDone:  stat.TotalBytesProcessed,
-	})
-
-	return string(summary), stderrBuf.String(), nil
-}
-
-func decodeBackupStatusLine(lastLine []byte) (backupStatusLine, error) {
+func DecodeBackupStatusLine(lastLine []byte) (backupStatusLine, error) {
 	var stat backupStatusLine
 	if err := json.Unmarshal(lastLine, &stat); err != nil {
 		return stat, errors.Wrapf(err, "unable to decode backup JSON line: %s", string(lastLine))
@@ -152,10 +74,10 @@ func decodeBackupStatusLine(lastLine []byte) (backupStatusLine, error) {
 	return stat, nil
 }
 
-// getLastLine returns the last line of a byte array. The string is assumed to
+// GetLastLine returns the last line of a byte array. The string is assumed to
 // have a newline at the end of it, so this returns the substring between the
 // last two newlines.
-func getLastLine(b []byte) []byte {
+func GetLastLine(b []byte) []byte {
 	if b == nil || len(b) == 0 {
 		return []byte("")
 	}
@@ -164,12 +86,12 @@ func getLastLine(b []byte) []byte {
 	return b[lastNewLineIdx+1 : len(b)-1]
 }
 
-// getSummaryLine looks for the summary JSON line
+// GetSummaryLine looks for the summary JSON line
 // (`{"message_type:"summary",...`) in the restic backup command output. Due to
 // an issue in Restic, this might not always be the last line
 // (https://github.com/restic/restic/issues/2389). It returns an error if it
 // can't be found.
-func getSummaryLine(b []byte) ([]byte, error) {
+func GetSummaryLine(b []byte) ([]byte, error) {
 	summaryLineIdx := bytes.LastIndex(b, []byte(`{"message_type":"summary"`))
 	if summaryLineIdx < 0 {
 		return nil, errors.New("unable to find summary in restic backup command output")
@@ -182,64 +104,7 @@ func getSummaryLine(b []byte) ([]byte, error) {
 	return b[summaryLineIdx : summaryLineIdx+newLineIdx], nil
 }
 
-// RunRestore runs a `restic restore` command and monitors the volume size to
-// provide progress updates to the caller.
-func RunRestore(restoreCmd *Command, log logrus.FieldLogger, updateFunc func(velerov1api.PodVolumeOperationProgress)) (string, string, error) {
-	insecureTLSFlag := ""
-
-	for _, extraFlag := range restoreCmd.ExtraFlags {
-		if strings.Contains(extraFlag, resticInsecureTLSFlag) {
-			insecureTLSFlag = extraFlag
-		}
-	}
-
-	snapshotSize, err := getSnapshotSize(restoreCmd.RepoIdentifier, restoreCmd.PasswordFile, restoreCmd.CACertFile, restoreCmd.Args[0], restoreCmd.Env, insecureTLSFlag)
-	if err != nil {
-		return "", "", errors.Wrap(err, "error getting snapshot size")
-	}
-
-	updateFunc(velerov1api.PodVolumeOperationProgress{
-		TotalBytes: snapshotSize,
-	})
-
-	// create a channel to signal when to end the goroutine scanning for progress
-	// updates
-	quit := make(chan struct{})
-
-	go func() {
-		ticker := time.NewTicker(restoreProgressCheckInterval)
-		for {
-			select {
-			case <-ticker.C:
-				volumeSize, err := getVolumeSize(restoreCmd.Dir)
-				if err != nil {
-					log.WithError(err).Errorf("error getting restic restore progress")
-				}
-
-				updateFunc(velerov1api.PodVolumeOperationProgress{
-					TotalBytes: snapshotSize,
-					BytesDone:  volumeSize,
-				})
-			case <-quit:
-				ticker.Stop()
-				return
-			}
-		}
-	}()
-
-	stdout, stderr, err := exec.RunCommand(restoreCmd.Cmd())
-	quit <- struct{}{}
-
-	// update progress to 100%
-	updateFunc(velerov1api.PodVolumeOperationProgress{
-		TotalBytes: snapshotSize,
-		BytesDone:  snapshotSize,
-	})
-
-	return stdout, stderr, err
-}
-
-func getSnapshotSize(repoIdentifier, passwordFile, caCertFile, snapshotID string, env []string, insecureTLS string) (int64, error) {
+func GetSnapshotSize(repoIdentifier, passwordFile, caCertFile, snapshotID string, env []string, insecureTLS string) (int64, error) {
 	cmd := StatsCommand(repoIdentifier, passwordFile, snapshotID)
 	cmd.Env = env
 	cmd.CACertFile = caCertFile
@@ -264,7 +129,7 @@ func getSnapshotSize(repoIdentifier, passwordFile, caCertFile, snapshotID string
 	return snapshotStats.TotalSize, nil
 }
 
-func getVolumeSize(path string) (int64, error) {
+func GetVolumeSize(path string) (int64, error) {
 	var size int64
 
 	files, err := fileSystem.ReadDir(path)
@@ -274,7 +139,7 @@ func getVolumeSize(path string) (int64, error) {
 
 	for _, file := range files {
 		if file.IsDir() {
-			s, err := getVolumeSize(fmt.Sprintf("%s/%s", path, file.Name()))
+			s, err := GetVolumeSize(fmt.Sprintf("%s/%s", path, file.Name()))
 			if err != nil {
 				return 0, err
 			}

pkg/restic/exec_commands_test.go

@@ -52,7 +52,7 @@ func Test_getSummaryLine(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			summary, err := getSummaryLine([]byte(tt.output))
+			summary, err := GetSummaryLine([]byte(tt.output))
 			if tt.wantErr {
 				assert.Error(t, err)
 			} else {
@@ -79,7 +79,7 @@ third line
 	}
 	for _, tt := range tests {
 		t.Run(tt.want, func(t *testing.T) {
-			assert.Equal(t, []byte(tt.want), getLastLine([]byte(tt.output)))
+			assert.Equal(t, []byte(tt.want), GetLastLine([]byte(tt.output)))
 		})
 	}
 }
@@ -103,7 +103,7 @@ func Test_getVolumeSize(t *testing.T) {
 	fileSystem = fakefs
 	defer func() { fileSystem = filesystem.NewFileSystem() }()
 
-	actualSize, err := getVolumeSize("/")
+	actualSize, err := GetVolumeSize("/")
 
 	assert.NoError(t, err)
 	assert.Equal(t, expectedSize, actualSize)

pkg/restic/executer.go

@@ -1,37 +0,0 @@
-/*
-Copyright The Velero Contributors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package restic
-
-import (
-	"github.com/sirupsen/logrus"
-
-	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
-)
-
-// BackupExec is able to run backups.
-type BackupExec struct{}
-
-// RunBackup is a wrapper for the restic.RunBackup function in order to be able
-// to use interfaces (and swap out objects for testing purposes).
-func (exec BackupExec) RunBackup(cmd *Command, log logrus.FieldLogger, updateFn func(velerov1api.PodVolumeOperationProgress)) (string, string, error) {
-	return RunBackup(cmd, log, updateFn)
-}
-
-// GetSnapshotID gets the Restic snapshot ID.
-func (exec BackupExec) GetSnapshotID(snapshotIdCmd *Command) (string, error) {
-	return GetSnapshotID(snapshotIdCmd)
-}

pkg/uploader/provider/kopia.go

@@ -97,8 +97,8 @@ func (kp *kopiaProvider) CheckContext(ctx context.Context, finishChan chan struc
 	}
 }
 
-func (kp *kopiaProvider) Close(ctx context.Context) {
-	kp.bkRepo.Close(ctx)
+func (kp *kopiaProvider) Close(ctx context.Context) error {
+	return kp.bkRepo.Close(ctx)
 }
 
 //RunBackup which will backup specific path and update backup progress

pkg/uploader/provider/kopia_test.go

@@ -25,10 +25,10 @@ import (
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
-	"github.com/vmware-tanzu/velero/pkg/controller"
 	"github.com/vmware-tanzu/velero/pkg/generated/clientset/versioned/scheme"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
 	"github.com/vmware-tanzu/velero/pkg/uploader/kopia"
@@ -37,7 +37,7 @@ import (
 func TestRunBackup(t *testing.T) {
 	var kp kopiaProvider
 	kp.log = logrus.New()
-	updater := controller.BackupProgressUpdater{PodVolumeBackup: &velerov1api.PodVolumeBackup{}, Log: kp.log, Ctx: context.Background(), Cli: fake.NewFakeClientWithScheme(scheme.Scheme)}
+	updater := FakeBackupProgressUpdater{PodVolumeBackup: &velerov1api.PodVolumeBackup{}, Log: kp.log, Ctx: context.Background(), Cli: fake.NewFakeClientWithScheme(scheme.Scheme)}
 	testCases := []struct {
 		name           string
 		hookBackupFunc func(ctx context.Context, fsUploader *snapshotfs.Uploader, repoWriter repo.RepositoryWriter, sourcePath, parentSnapshot string, log logrus.FieldLogger) (*uploader.SnapshotInfo, error)
@@ -81,7 +81,7 @@ func TestRunBackup(t *testing.T) {
 func TestRunRestore(t *testing.T) {
 	var kp kopiaProvider
 	kp.log = logrus.New()
-	updater := controller.RestoreProgressUpdater{PodVolumeRestore: &velerov1api.PodVolumeRestore{}, Log: kp.log, Ctx: context.Background(), Cli: fake.NewFakeClientWithScheme(scheme.Scheme)}
+	updater := FakeRestoreProgressUpdater{PodVolumeRestore: &velerov1api.PodVolumeRestore{}, Log: kp.log, Ctx: context.Background(), Cli: fake.NewFakeClientWithScheme(scheme.Scheme)}
 
 	testCases := []struct {
 		name            string
@@ -116,3 +116,21 @@ func TestRunRestore(t *testing.T) {
 		})
 	}
 }
+
+type FakeBackupProgressUpdater struct {
+	PodVolumeBackup *velerov1api.PodVolumeBackup
+	Log             logrus.FieldLogger
+	Ctx             context.Context
+	Cli             client.Client
+}
+
+func (f *FakeBackupProgressUpdater) UpdateProgress(p *uploader.UploaderProgress) {}
+
+type FakeRestoreProgressUpdater struct {
+	PodVolumeRestore *velerov1api.PodVolumeRestore
+	Log              logrus.FieldLogger
+	Ctx              context.Context
+	Cli              client.Client
+}
+
+func (f *FakeRestoreProgressUpdater) UpdateProgress(p *uploader.UploaderProgress) {}

pkg/uploader/provider/provider.go

@@ -20,11 +20,16 @@ import (
 	"context"
 	"time"
 
+	"github.com/pkg/errors"
+
 	"github.com/sirupsen/logrus"
 	v1 "k8s.io/api/core/v1"
 
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
 	"github.com/vmware-tanzu/velero/internal/credentials"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/repository/provider"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
 )
 
@@ -49,23 +54,34 @@ type Provider interface {
 		volumePath string,
 		updater uploader.ProgressUpdater) error
 	// Close which will close related repository
-	Close(ctx context.Context)
+	Close(ctx context.Context) error
 }
 
 // NewUploaderProvider initialize provider with specific uploaderType
 func NewUploaderProvider(
 	ctx context.Context,
+	client client.Client,
 	uploaderType string,
 	repoIdentifier string,
 	bsl *velerov1api.BackupStorageLocation,
-	backupReo *velerov1api.BackupRepository,
+	backupRepo *velerov1api.BackupRepository,
 	credGetter *credentials.CredentialGetter,
 	repoKeySelector *v1.SecretKeySelector,
 	log logrus.FieldLogger,
 ) (Provider, error) {
+	if credGetter.FromFile == nil {
+		return nil, errors.New("uninitialized FileStore credentail is not supported")
+	}
 	if uploaderType == uploader.KopiaType {
-		return NewResticUploaderProvider(repoIdentifier, bsl, credGetter, repoKeySelector, log)
+		if err := provider.NewUnifiedRepoProvider(*credGetter, log).ConnectToRepo(ctx, provider.RepoParam{BackupLocation: bsl, BackupRepo: backupRepo}); err != nil {
+			return nil, errors.Wrap(err, "failed to connect repository")
+		}
+		return NewKopiaUploaderProvider(ctx, credGetter, backupRepo, log)
 	} else {
-		return NewKopiaUploaderProvider(ctx, credGetter, backupReo, log)
+		err := provider.NewResticRepositoryProvider(credGetter.FromFile, nil, log).ConnectToRepo(ctx, provider.RepoParam{BackupLocation: bsl, BackupRepo: backupRepo})
+		if err != nil {
+			return nil, errors.Wrap(err, "failed to connect repository")
+		}
+		return NewResticUploaderProvider(repoIdentifier, bsl, credGetter, repoKeySelector, log)
 	}
 }

pkg/uploader/provider/restic.go

@@ -13,16 +13,45 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.
 */
+
 package provider
 
 import (
+	"bytes"
+	"context"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	v1 "k8s.io/api/core/v1"
 
 	"github.com/vmware-tanzu/velero/internal/credentials"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/restic"
+	"github.com/vmware-tanzu/velero/pkg/uploader"
+	"github.com/vmware-tanzu/velero/pkg/util/exec"
+	"github.com/vmware-tanzu/velero/pkg/util/filesystem"
 )
 
+// BackupFunc mainly used to make testing more convenient
+var ResticBackupFunc = restic.BackupCommand
+var ResticRunCMDFunc = exec.RunCommand
+var ResticGetSnapshotSizeFunc = restic.GetSnapshotSize
+var ResticGetVolumeSizeFunc = restic.GetVolumeSize
+var ResticRestoreFunc = restic.RestoreCommand
+
+type resticProvider struct {
+	repoIdentifier  string
+	credentialsFile string
+	caCertFile      string
+	repoEnv         []string
+	backupTags      map[string]string
+	log             logrus.FieldLogger
+	bsl             *velerov1api.BackupStorageLocation
+}
+
 func NewResticUploaderProvider(
 	repoIdentifier string,
 	bsl *velerov1api.BackupStorageLocation,
@@ -30,5 +59,234 @@ func NewResticUploaderProvider(
 	repoKeySelector *v1.SecretKeySelector,
 	log logrus.FieldLogger,
 ) (Provider, error) {
-	return nil, nil //TODO
+	provider := resticProvider{
+		repoIdentifier: repoIdentifier,
+		log:            log,
+		bsl:            bsl,
+	}
+
+	var err error
+	provider.credentialsFile, err = credGetter.FromFile.Path(repoKeySelector)
+	if err != nil {
+		return nil, errors.Wrap(err, "error creating temp restic credentials file")
+	}
+
+	// if there's a caCert on the ObjectStorage, write it to disk so that it can be passed to restic
+	if bsl.Spec.ObjectStorage != nil && bsl.Spec.ObjectStorage.CACert != nil {
+		provider.caCertFile, err = restic.TempCACertFile(bsl.Spec.ObjectStorage.CACert, bsl.Name, filesystem.NewFileSystem())
+		if err != nil {
+			return nil, errors.Wrap(err, "error create temp cert file")
+		}
+	}
+
+	provider.repoEnv, err = restic.CmdEnv(bsl, credGetter.FromFile)
+	if err != nil {
+		return nil, errors.Wrap(err, "error generating repository cmnd env")
+	}
+
+	return &provider, nil
+}
+
+// Not implement yet
+func (rp *resticProvider) Cancel() {
+}
+
+func (rp *resticProvider) Close(ctx context.Context) error {
+	if err := os.Remove(rp.credentialsFile); err != nil {
+		return errors.Wrapf(err, "failed to remove file %s", rp.credentialsFile)
+	}
+	if err := os.Remove(rp.caCertFile); err != nil {
+		return errors.Wrapf(err, "failed to remove file %s", rp.caCertFile)
+	}
+	return nil
+}
+
+// RunBackup runs a `backup` command and watches the output to provide
+// progress updates to the caller.
+func (rp *resticProvider) RunBackup(
+	ctx context.Context,
+	path string,
+	tags map[string]string,
+	parentSnapshot string,
+	updater uploader.ProgressUpdater) (string, error) {
+	if updater == nil {
+		return "", errors.New("Need to initial backup progress updater first")
+	}
+
+	log := rp.log.WithFields(logrus.Fields{
+		"path":           path,
+		"parentSnapshot": parentSnapshot,
+	})
+
+	// buffers for copying command stdout/err output into
+	stdoutBuf := new(bytes.Buffer)
+	stderrBuf := new(bytes.Buffer)
+
+	// create a channel to signal when to end the goroutine scanning for progress
+	// updates
+	quit := make(chan struct{})
+
+	rp.backupTags = tags
+	backupCmd := ResticBackupFunc(rp.repoIdentifier, rp.credentialsFile, path, tags)
+	backupCmd.Env = rp.repoEnv
+	backupCmd.CACertFile = rp.caCertFile
+
+	if parentSnapshot != "" {
+		backupCmd.ExtraFlags = append(backupCmd.ExtraFlags, fmt.Sprintf("--parent=%s", parentSnapshot))
+	}
+
+	// #4820: restrieve insecureSkipTLSVerify from BSL configuration for
+	// AWS plugin. If nothing is return, that means insecureSkipTLSVerify
+	// is not enable for Restic command.
+	skipTLSRet := restic.GetInsecureSkipTLSVerifyFromBSL(rp.bsl, rp.log)
+	if len(skipTLSRet) > 0 {
+		backupCmd.ExtraFlags = append(backupCmd.ExtraFlags, skipTLSRet)
+	}
+
+	cmd := backupCmd.Cmd()
+	cmd.Stdout = stdoutBuf
+	cmd.Stderr = stderrBuf
+
+	err := cmd.Start()
+	if err != nil {
+		log.Errorf("failed to execute %v with stderr %v error %v", cmd, stderrBuf.String(), err)
+		return "", err
+	}
+
+	go func() {
+		ticker := time.NewTicker(backupProgressCheckInterval)
+		for {
+			select {
+			case <-ticker.C:
+				lastLine := restic.GetLastLine(stdoutBuf.Bytes())
+				if len(lastLine) > 0 {
+					stat, err := restic.DecodeBackupStatusLine(lastLine)
+					if err != nil {
+						rp.log.WithError(err).Errorf("error getting backup progress")
+					}
+
+					// if the line contains a non-empty bytes_done field, we can update the
+					// caller with the progress
+					if stat.BytesDone != 0 {
+						updater.UpdateProgress(&uploader.UploaderProgress{
+							TotalBytes: stat.TotalBytesProcessed,
+							BytesDone:  stat.TotalBytesProcessed,
+						})
+					}
+				}
+			case <-quit:
+				ticker.Stop()
+				return
+			}
+		}
+	}()
+
+	err = cmd.Wait()
+	if err != nil {
+		return "", errors.WithStack(fmt.Errorf("failed to wait execute %v with stderr %v error %v", cmd, stderrBuf.String(), err))
+	}
+	quit <- struct{}{}
+
+	summary, err := restic.GetSummaryLine(stdoutBuf.Bytes())
+	if err != nil {
+		return "", errors.WithStack(fmt.Errorf("failed to get summary %v with error %v", stderrBuf.String(), err))
+	}
+	stat, err := restic.DecodeBackupStatusLine(summary)
+	if err != nil {
+		return "", errors.WithStack(fmt.Errorf("failed to decode summary %v with error %v", string(summary), err))
+	}
+	if stat.MessageType != "summary" {
+		return "", errors.WithStack(fmt.Errorf("error getting backup summary: %s", string(summary)))
+	}
+
+	// update progress to 100%
+	updater.UpdateProgress(&uploader.UploaderProgress{
+		TotalBytes: stat.TotalBytesProcessed,
+		BytesDone:  stat.TotalBytesProcessed,
+	})
+
+	//GetSnapshtotID
+	snapshotIdCmd := restic.GetSnapshotCommand(rp.repoIdentifier, rp.credentialsFile, rp.backupTags)
+	snapshotIdCmd.Env = rp.repoEnv
+	snapshotIdCmd.CACertFile = rp.caCertFile
+
+	snapshotID, err := restic.GetSnapshotID(snapshotIdCmd)
+	if err != nil {
+		return "", errors.WithStack(fmt.Errorf("error getting snapshot id with error: %s", err))
+	}
+	log.Debugf("Ran command=%s, stdout=%s, stderr=%s", cmd.String(), summary, stderrBuf.String())
+	return snapshotID, nil
+}
+
+// RunRestore runs a `restore` command and monitors the volume size to
+// provide progress updates to the caller.
+func (rp *resticProvider) RunRestore(
+	ctx context.Context,
+	snapshotID string,
+	volumePath string,
+	updater uploader.ProgressUpdater) error {
+	log := rp.log.WithFields(logrus.Fields{
+		"snapshotID": snapshotID,
+		"volumePath": volumePath,
+	})
+
+	restoreCmd := ResticRestoreFunc(rp.repoIdentifier, rp.credentialsFile, snapshotID, volumePath)
+	restoreCmd.Env = rp.repoEnv
+	restoreCmd.CACertFile = rp.caCertFile
+
+	// #4820: restrieve insecureSkipTLSVerify from BSL configuration for
+	// AWS plugin. If nothing is return, that means insecureSkipTLSVerify
+	// is not enable for Restic command.
+	skipTLSRet := restic.GetInsecureSkipTLSVerifyFromBSL(rp.bsl, log)
+	if len(skipTLSRet) > 0 {
+		restoreCmd.ExtraFlags = append(restoreCmd.ExtraFlags, skipTLSRet)
+	}
+	snapshotSize, err := ResticGetSnapshotSizeFunc(restoreCmd.RepoIdentifier, restoreCmd.PasswordFile, restoreCmd.CACertFile, restoreCmd.Args[0], restoreCmd.Env, skipTLSRet)
+	if err != nil {
+		return errors.Wrap(err, "error getting snapshot size")
+	}
+
+	updater.UpdateProgress(&uploader.UploaderProgress{
+		TotalBytes: snapshotSize,
+	})
+
+	// create a channel to signal when to end the goroutine scanning for progress
+	// updates
+	quit := make(chan struct{})
+
+	go func() {
+		ticker := time.NewTicker(restoreProgressCheckInterval)
+		for {
+			select {
+			case <-ticker.C:
+				volumeSize, err := ResticGetVolumeSizeFunc(restoreCmd.Dir)
+				if err != nil {
+					log.WithError(err).Errorf("error getting volume size for restore dir %v", restoreCmd.Dir)
+					return
+				}
+				if volumeSize != 0 {
+					updater.UpdateProgress(&uploader.UploaderProgress{
+						TotalBytes: snapshotSize,
+						BytesDone:  volumeSize,
+					})
+				}
+			case <-quit:
+				ticker.Stop()
+				return
+			}
+		}
+	}()
+
+	stdout, stderr, err := ResticRunCMDFunc(restoreCmd.Cmd())
+	quit <- struct{}{}
+	if err != nil {
+		return errors.Wrap(err, fmt.Sprintf("failed to execute restore command %v with stderr %v", restoreCmd.Cmd().String(), stderr))
+	}
+	// update progress to 100%
+	updater.UpdateProgress(&uploader.UploaderProgress{
+		TotalBytes: snapshotSize,
+		BytesDone:  snapshotSize,
+	})
+	log.Debugf("Ran command=%s, stdout=%s, stderr=%s", restoreCmd.Command, stdout, stderr)
+	return err
 }

pkg/uploader/provider/restic_test.go

@@ -0,0 +1,136 @@
+/*
+Copyright The Velero Contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package provider
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/require"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/generated/clientset/versioned/scheme"
+	"github.com/vmware-tanzu/velero/pkg/restic"
+)
+
+func TestResticRunBackup(t *testing.T) {
+	var rp resticProvider
+	rp.log = logrus.New()
+	updater := FakeBackupProgressUpdater{PodVolumeBackup: &velerov1api.PodVolumeBackup{}, Log: rp.log, Ctx: context.Background(), Cli: fake.NewFakeClientWithScheme(scheme.Scheme)}
+	testCases := []struct {
+		name            string
+		hookBackupFunc  func(repoIdentifier string, passwordFile string, path string, tags map[string]string) *restic.Command
+		errorHandleFunc func(err error) bool
+	}{
+		{
+			name: "wrong restic execute command",
+			hookBackupFunc: func(repoIdentifier string, passwordFile string, path string, tags map[string]string) *restic.Command {
+				return &restic.Command{Command: "date"}
+			},
+			errorHandleFunc: func(err error) bool {
+				return strings.Contains(err.Error(), "executable file not found in")
+			},
+		},
+		{
+			name: "wrong parsing json summary content",
+			hookBackupFunc: func(repoIdentifier string, passwordFile string, path string, tags map[string]string) *restic.Command {
+				return &restic.Command{Command: "version"}
+			},
+			errorHandleFunc: func(err error) bool {
+				return strings.Contains(err.Error(), "executable file not found in")
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			ResticBackupFunc = tc.hookBackupFunc
+			_, err := rp.RunBackup(context.Background(), "var", nil, "", &updater)
+			rp.log.Infof("test name %v error %v", tc.name, err)
+			require.Equal(t, true, tc.errorHandleFunc(err))
+		})
+	}
+}
+func TestResticRunRestore(t *testing.T) {
+	var rp resticProvider
+	rp.log = logrus.New()
+	updater := FakeBackupProgressUpdater{PodVolumeBackup: &velerov1api.PodVolumeBackup{}, Log: rp.log, Ctx: context.Background(), Cli: fake.NewFakeClientWithScheme(scheme.Scheme)}
+	ResticRestoreFunc = func(repoIdentifier, passwordFile, snapshotID, target string) *restic.Command {
+		return &restic.Command{Args: []string{""}}
+	}
+	testCases := []struct {
+		name                          string
+		hookResticGetVolumeSizeFunc   func(path string) (int64, error)
+		hookResticGetSnapshotSizeFunc func(repoIdentifier, passwordFile, caCertFile, snapshotID string, env []string, insecureTLS string) (int64, error)
+		hookResticRestoreFunc         func(repoIdentifier, passwordFile, snapshotID, target string) *restic.Command
+		errorHandleFunc               func(err error) bool
+	}{
+		{
+			name:                        "failed to get snapshot",
+			hookResticGetVolumeSizeFunc: func(path string) (int64, error) { return 100, nil },
+			hookResticGetSnapshotSizeFunc: func(repoIdentifier, passwordFile, caCertFile, snapshotID string, env []string, insecureTLS string) (int64, error) {
+				return 100, errors.New("failed to get snapshot")
+			},
+			hookResticRestoreFunc: func(repoIdentifier, passwordFile, snapshotID, target string) *restic.Command {
+				return &restic.Command{Args: []string{""}}
+			},
+			errorHandleFunc: func(err error) bool { return strings.Contains(err.Error(), "failed to get snapshot") },
+		},
+		{
+			name:                        "failed to get volume size",
+			hookResticGetVolumeSizeFunc: func(path string) (int64, error) { return 100, errors.New("failed to get volume size") },
+			hookResticGetSnapshotSizeFunc: func(repoIdentifier, passwordFile, caCertFile, snapshotID string, env []string, insecureTLS string) (int64, error) {
+				return 100, nil
+			},
+			hookResticRestoreFunc: func(repoIdentifier, passwordFile, snapshotID, target string) *restic.Command {
+				return &restic.Command{Args: []string{""}}
+			},
+			errorHandleFunc: func(err error) bool {
+				return strings.Contains(err.Error(), "failed to execute restore command restic")
+			},
+		},
+		{
+			name:                        "wrong restic execute command",
+			hookResticGetVolumeSizeFunc: func(path string) (int64, error) { return 100, nil },
+			hookResticGetSnapshotSizeFunc: func(repoIdentifier, passwordFile, caCertFile, snapshotID string, env []string, insecureTLS string) (int64, error) {
+				return 100, nil
+			},
+			hookResticRestoreFunc: func(repoIdentifier, passwordFile, snapshotID, target string) *restic.Command {
+				return &restic.Command{Args: []string{"date"}}
+			},
+			errorHandleFunc: func(err error) bool {
+				return strings.Contains(err.Error(), "failed to execute restore command restic")
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			ResticGetSnapshotSizeFunc = tc.hookResticGetSnapshotSizeFunc
+			ResticGetVolumeSizeFunc = tc.hookResticGetVolumeSizeFunc
+			ResticRestoreFunc = tc.hookResticRestoreFunc
+			err := rp.RunRestore(context.Background(), "", "var", &updater)
+			rp.log.Infof("test name %v error %v", tc.name, err)
+			require.Equal(t, true, tc.errorHandleFunc(err))
+		})
+	}
+
+}