From f15757a3d820a49050069901d146edc8b744a2d4 Mon Sep 17 00:00:00 2001 From: Ming Date: Mon, 15 Aug 2022 10:34:08 +0000 Subject: [PATCH] Uploader Implementation: Restic backup and restore Signed-off-by: Ming --- changelogs/unreleased/5214-qiuming-best | 1 + pkg/builder/backup_repository_builer.go | 59 ++++ pkg/cmd/cli/restic/server.go | 26 +- .../pod_volume_backup_controller.go | 220 ++++++--------- .../pod_volume_backup_controller_test.go | 82 ++++-- .../pod_volume_restore_controller.go | 104 +++---- pkg/restic/exec_commands.go | 151 +--------- pkg/restic/exec_commands_test.go | 6 +- pkg/restic/executer.go | 37 --- pkg/uploader/provider/kopia.go | 4 +- pkg/uploader/provider/kopia_test.go | 24 +- pkg/uploader/provider/provider.go | 24 +- pkg/uploader/provider/restic.go | 260 +++++++++++++++++- pkg/uploader/provider/restic_test.go | 136 +++++++++ 14 files changed, 709 insertions(+), 425 deletions(-) create mode 100644 changelogs/unreleased/5214-qiuming-best create mode 100644 pkg/builder/backup_repository_builer.go delete mode 100644 pkg/restic/executer.go create mode 100644 pkg/uploader/provider/restic_test.go diff --git a/changelogs/unreleased/5214-qiuming-best b/changelogs/unreleased/5214-qiuming-best new file mode 100644 index 000000000..9c2ddd1de --- /dev/null +++ b/changelogs/unreleased/5214-qiuming-best @@ -0,0 +1 @@ +Uploader Implementation: Restic backup and restore diff --git a/pkg/builder/backup_repository_builer.go b/pkg/builder/backup_repository_builer.go new file mode 100644 index 000000000..a78f3238a --- /dev/null +++ b/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 +} diff --git a/pkg/cmd/cli/restic/server.go b/pkg/cmd/cli/restic/server.go index ae593306b..73d7cdec5 100644 --- a/pkg/cmd/cli/restic/server.go +++ 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") } diff --git a/pkg/controller/pod_volume_backup_controller.go b/pkg/controller/pod_volume_backup_controller.go index 635ef00a7..97b099341 100644 --- a/pkg/controller/pod_volume_backup_controller.go +++ b/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} } diff --git a/pkg/controller/pod_volume_backup_controller_test.go b/pkg/controller/pod_volume_backup_controller_test.go index ffc5f662c..794606cb0 100644 --- a/pkg/controller/pod_volume_backup_controller_test.go +++ b/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 } diff --git a/pkg/controller/pod_volume_restore_controller.go b/pkg/controller/pod_volume_restore_controller.go index 00853710f..7e09bd5a4 100644 --- a/pkg/controller/pod_volume_restore_controller.go +++ b/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} } diff --git a/pkg/restic/exec_commands.go b/pkg/restic/exec_commands.go index 7dd0057c0..1a4b055bf 100644 --- a/pkg/restic/exec_commands.go +++ b/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 } diff --git a/pkg/restic/exec_commands_test.go b/pkg/restic/exec_commands_test.go index 0353f3e2c..e8ca9076f 100644 --- a/pkg/restic/exec_commands_test.go +++ b/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) diff --git a/pkg/restic/executer.go b/pkg/restic/executer.go deleted file mode 100644 index e89883e76..000000000 --- a/pkg/restic/executer.go +++ /dev/null @@ -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) -} diff --git a/pkg/uploader/provider/kopia.go b/pkg/uploader/provider/kopia.go index 6384142ef..73798d9ac 100644 --- a/pkg/uploader/provider/kopia.go +++ b/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 diff --git a/pkg/uploader/provider/kopia_test.go b/pkg/uploader/provider/kopia_test.go index 747acbfb4..ac061d6c7 100644 --- a/pkg/uploader/provider/kopia_test.go +++ b/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) {} diff --git a/pkg/uploader/provider/provider.go b/pkg/uploader/provider/provider.go index b435708f6..cac542798 100644 --- a/pkg/uploader/provider/provider.go +++ b/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) } } diff --git a/pkg/uploader/provider/restic.go b/pkg/uploader/provider/restic.go index f908077a3..1f9e18eba 100644 --- a/pkg/uploader/provider/restic.go +++ b/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 } diff --git a/pkg/uploader/provider/restic_test.go b/pkg/uploader/provider/restic_test.go new file mode 100644 index 000000000..22b44c2ac --- /dev/null +++ b/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)) + }) + } + +}