From b06d7a467f39926b8c0c42845e65c993ecd97cb6 Mon Sep 17 00:00:00 2001 From: Xun Jiang Date: Fri, 22 Mar 2024 16:15:06 +0800 Subject: [PATCH 1/2] Support update the backup VolumeInfos by the Async ops result. 1. Add PutBackupVolumeInfos method. 2. Add CompletionTimestamp in VolumeInfo. 3. Add Size in SnapshotDataMovementInfo. 4. Update CompletionTimpstmap, SnapshotHandle, RetainedSnapshot and Size in VolumeInfo on DataUpload Operation completes. Signed-off-by: Xun Jiang --- changelogs/unreleased/7554-blackpiglet | 1 + internal/volume/volumes_information.go | 25 ++++- internal/volume/volumes_information_test.go | 34 ++++--- pkg/backup/backup.go | 91 ++++++++++++++++--- pkg/backup/backup_test.go | 58 ++++++++++++ pkg/builder/data_upload_builder.go | 10 +- pkg/builder/pod_volume_backup_builder.go | 10 ++ pkg/controller/backup_controller_test.go | 11 ++- pkg/controller/backup_finalizer_controller.go | 47 +++++++++- .../backup_finalizer_controller_test.go | 2 + pkg/kuberesource/kuberesource.go | 1 + pkg/persistence/mocks/backup_store.go | 16 +++- pkg/persistence/object_store.go | 5 + pkg/persistence/object_store_test.go | 45 +++++++++ 14 files changed, 319 insertions(+), 37 deletions(-) create mode 100644 changelogs/unreleased/7554-blackpiglet diff --git a/changelogs/unreleased/7554-blackpiglet b/changelogs/unreleased/7554-blackpiglet new file mode 100644 index 000000000..1cbdd597e --- /dev/null +++ b/changelogs/unreleased/7554-blackpiglet @@ -0,0 +1 @@ +Support update the backup VolumeInfos by the Async ops result. \ No newline at end of file diff --git a/internal/volume/volumes_information.go b/internal/volume/volumes_information.go index 7e3acd555..e901280e5 100644 --- a/internal/volume/volumes_information.go +++ b/internal/volume/volumes_information.go @@ -76,6 +76,9 @@ type VolumeInfo struct { // Snapshot starts timestamp. StartTimestamp *metav1.Time `json:"startTimestamp,omitempty"` + // Snapshot completes timestamp. + CompletionTimestamp *metav1.Time `json:"completionTimestamp,omitempty"` + CSISnapshotInfo *CSISnapshotInfo `json:"csiSnapshotInfo,omitempty"` SnapshotDataMovementInfo *SnapshotDataMovementInfo `json:"snapshotDataMovementInfo,omitempty"` NativeSnapshotInfo *NativeSnapshotInfo `json:"nativeSnapshotInfo,omitempty"` @@ -119,6 +122,9 @@ type SnapshotDataMovementInfo struct { // The Async Operation's ID. OperationID string `json:"operationID"` + + // Moved snapshot data size. + Size int64 `json:"size"` } // NativeSnapshotInfo is used for displaying the Velero native snapshot status. @@ -379,7 +385,6 @@ func (v *VolumesInformation) generateVolumeInfoForCSIVolumeSnapshot() { Skipped: false, SnapshotDataMoved: false, PreserveLocalSnapshot: true, - StartTimestamp: &(volumeSnapshot.CreationTimestamp), CSISnapshotInfo: &CSISnapshotInfo{ VSCName: *volumeSnapshot.Status.BoundVolumeSnapshotContentName, Size: size, @@ -393,6 +398,10 @@ func (v *VolumesInformation) generateVolumeInfoForCSIVolumeSnapshot() { }, } + if volumeSnapshot.Status.CreationTime != nil { + volumeInfo.StartTimestamp = volumeSnapshot.Status.CreationTime + } + tmpVolumeInfos = append(tmpVolumeInfos, volumeInfo) } else { v.logger.Warnf("cannot find info for PVC %s/%s", volumeSnapshot.Namespace, volumeSnapshot.Spec.Source.PersistentVolumeClaimName) @@ -412,7 +421,6 @@ func (v *VolumesInformation) generateVolumeInfoFromPVB() { BackupMethod: PodVolumeBackup, SnapshotDataMoved: false, Skipped: false, - StartTimestamp: pvb.Status.StartTimestamp, PVBInfo: &PodVolumeBackupInfo{ SnapshotHandle: pvb.Status.SnapshotID, Size: pvb.Status.Progress.TotalBytes, @@ -424,6 +432,14 @@ func (v *VolumesInformation) generateVolumeInfoFromPVB() { }, } + if pvb.Status.StartTimestamp != nil { + volumeInfo.StartTimestamp = pvb.Status.StartTimestamp + } + + if pvb.Status.CompletionTimestamp != nil { + volumeInfo.CompletionTimestamp = pvb.Status.CompletionTimestamp + } + pod := new(corev1api.Pod) pvcName := "" err := v.crClient.Get(context.TODO(), kbclient.ObjectKey{Namespace: pvb.Spec.Pod.Namespace, Name: pvb.Spec.Pod.Name}, pod) @@ -522,7 +538,6 @@ func (v *VolumesInformation) generateVolumeInfoFromDataUpload() { PVName: pvcPVInfo.PV.Name, SnapshotDataMoved: true, Skipped: false, - StartTimestamp: operation.Status.Created, CSISnapshotInfo: &CSISnapshotInfo{ SnapshotHandle: FieldValueIsUnknown, VSCName: FieldValueIsUnknown, @@ -540,6 +555,10 @@ func (v *VolumesInformation) generateVolumeInfoFromDataUpload() { }, } + if dataUpload.Status.StartTimestamp != nil { + volumeInfo.StartTimestamp = dataUpload.Status.StartTimestamp + } + tmpVolumeInfos = append(tmpVolumeInfos, volumeInfo) } else { v.logger.Warnf("Cannot find info for PVC %s/%s", operation.Spec.ResourceIdentifier.Namespace, operation.Spec.ResourceIdentifier.Name) diff --git a/internal/volume/volumes_information_test.go b/internal/volume/volumes_information_test.go index d91f3004a..bf8ee133e 100644 --- a/internal/volume/volumes_information_test.go +++ b/internal/volume/volumes_information_test.go @@ -372,6 +372,7 @@ func TestGenerateVolumeInfoForCSIVolumeSnapshot(t *testing.T) { }, Status: &snapshotv1api.VolumeSnapshotStatus{ BoundVolumeSnapshotContentName: stringPtr("testContent"), + CreationTime: &now, RestoreSize: &resourceQuantity, }, }, @@ -458,6 +459,7 @@ func TestGenerateVolumeInfoForCSIVolumeSnapshot(t *testing.T) { } func TestGenerateVolumeInfoFromPVB(t *testing.T) { + now := metav1.Now() tests := []struct { name string pvb *velerov1api.PodVolumeBackup @@ -542,7 +544,7 @@ func TestGenerateVolumeInfoFromPVB(t *testing.T) { }, }, }, - pvb: builder.ForPodVolumeBackup("velero", "testPVB").PodName("testPod").PodNamespace("velero").Result(), + pvb: builder.ForPodVolumeBackup("velero", "testPVB").PodName("testPod").PodNamespace("velero").StartTimestamp(&now).CompletionTimestamp(&now).Result(), pod: builder.ForPod("velero", "testPod").Containers(&corev1api.Container{ Name: "test", VolumeMounts: []corev1api.VolumeMount{ @@ -563,10 +565,12 @@ func TestGenerateVolumeInfoFromPVB(t *testing.T) { ).Result(), expectedVolumeInfos: []*VolumeInfo{ { - PVCName: "testPVC", - PVCNamespace: "velero", - PVName: "testPV", - BackupMethod: PodVolumeBackup, + PVCName: "testPVC", + PVCNamespace: "velero", + PVName: "testPV", + BackupMethod: PodVolumeBackup, + StartTimestamp: &now, + CompletionTimestamp: &now, PVBInfo: &PodVolumeBackupInfo{ PodName: "testPod", PodNamespace: "velero", @@ -605,9 +609,12 @@ func TestGenerateVolumeInfoFromPVB(t *testing.T) { } func TestGenerateVolumeInfoFromDataUpload(t *testing.T) { + // The unstructured conversion will loose the time precision to second + // level. To make test pass. Set the now precision at second at the + // beginning. + now := metav1.Now().Rfc3339Copy() features.Enable(velerov1api.CSIFeatureFlag) defer features.Disable(velerov1api.CSIFeatureFlag) - now := metav1.Now() tests := []struct { name string volumeSnapshotClass *snapshotv1api.VolumeSnapshotClass @@ -685,7 +692,7 @@ func TestGenerateVolumeInfoFromDataUpload(t *testing.T) { name: "VolumeSnapshotClass cannot be found for operation", dataUpload: builder.ForDataUpload("velero", "testDU").DataMover("velero").CSISnapshot(&velerov2alpha1.CSISnapshotSpec{ VolumeSnapshot: "testVS", - }).SnapshotID("testSnapshotHandle").Result(), + }).SnapshotID("testSnapshotHandle").StartTimestamp(&now).Result(), operation: &itemoperation.BackupOperation{ Spec: itemoperation.BackupOperationSpec{ OperationID: "testOperation", @@ -731,6 +738,7 @@ func TestGenerateVolumeInfoFromDataUpload(t *testing.T) { PVName: "testPV", BackupMethod: CSISnapshot, SnapshotDataMoved: true, + StartTimestamp: &now, CSISnapshotInfo: &CSISnapshotInfo{ SnapshotHandle: FieldValueIsUnknown, VSCName: FieldValueIsUnknown, @@ -754,7 +762,7 @@ func TestGenerateVolumeInfoFromDataUpload(t *testing.T) { dataUpload: builder.ForDataUpload("velero", "testDU").DataMover("velero").CSISnapshot(&velerov2alpha1.CSISnapshotSpec{ VolumeSnapshot: "testVS", SnapshotClass: "testClass", - }).SnapshotID("testSnapshotHandle").Result(), + }).SnapshotID("testSnapshotHandle").StartTimestamp(&now).Result(), volumeSnapshotClass: builder.ForVolumeSnapshotClass("testClass").Driver("pd.csi.storage.gke.io").Result(), operation: &itemoperation.BackupOperation{ Spec: itemoperation.BackupOperationSpec{ @@ -778,9 +786,6 @@ func TestGenerateVolumeInfoFromDataUpload(t *testing.T) { }, }, }, - Status: itemoperation.OperationStatus{ - Created: &now, - }, }, pvMap: map[string]pvcPvInfo{ "testPV": { @@ -853,7 +858,12 @@ func TestGenerateVolumeInfoFromDataUpload(t *testing.T) { volumesInfo.logger = logging.DefaultLogger(logrus.DebugLevel, logging.FormatJSON) volumesInfo.generateVolumeInfoFromDataUpload() - require.Equal(t, tc.expectedVolumeInfos, volumesInfo.volumeInfos) + + if len(tc.expectedVolumeInfos) > 0 { + require.Equal(t, tc.expectedVolumeInfos[0].PVInfo, volumesInfo.volumeInfos[0].PVInfo) + require.Equal(t, tc.expectedVolumeInfos[0].SnapshotDataMovementInfo, volumesInfo.volumeInfos[0].SnapshotDataMovementInfo) + require.Equal(t, tc.expectedVolumeInfos[0].CSISnapshotInfo, volumesInfo.volumeInfos[0].CSISnapshotInfo) + } }) } } diff --git a/pkg/backup/backup.go b/pkg/backup/backup.go index 67e20acb0..8dcc8a12e 100644 --- a/pkg/backup/backup.go +++ b/pkg/backup/backup.go @@ -32,15 +32,16 @@ import ( apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" kubeerrs "k8s.io/apimachinery/pkg/util/errors" - - "github.com/vmware-tanzu/velero/internal/hook" - velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1" - "github.com/vmware-tanzu/velero/pkg/client" - kbclient "sigs.k8s.io/controller-runtime/pkg/client" + "github.com/vmware-tanzu/velero/internal/hook" + "github.com/vmware-tanzu/velero/internal/volume" + velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1" + velerov2alpha1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1" + "github.com/vmware-tanzu/velero/pkg/client" "github.com/vmware-tanzu/velero/pkg/discovery" "github.com/vmware-tanzu/velero/pkg/itemoperation" "github.com/vmware-tanzu/velero/pkg/kuberesource" @@ -66,11 +67,31 @@ const BackupFormatVersion = "1.1.0" type Backupper interface { // Backup takes a backup using the specification in the velerov1api.Backup and writes backup and log data // to the given writers. - Backup(logger logrus.FieldLogger, backup *Request, backupFile io.Writer, actions []biav2.BackupItemAction, volumeSnapshotterGetter VolumeSnapshotterGetter) error - BackupWithResolvers(log logrus.FieldLogger, backupRequest *Request, backupFile io.Writer, backupItemActionResolver framework.BackupItemActionResolverV2, volumeSnapshotterGetter VolumeSnapshotterGetter) error - FinalizeBackup(log logrus.FieldLogger, backupRequest *Request, inBackupFile io.Reader, outBackupFile io.Writer, + Backup( + logger logrus.FieldLogger, + backup *Request, + backupFile io.Writer, + actions []biav2.BackupItemAction, + volumeSnapshotterGetter VolumeSnapshotterGetter, + ) error + + BackupWithResolvers( + log logrus.FieldLogger, + backupRequest *Request, + backupFile io.Writer, backupItemActionResolver framework.BackupItemActionResolverV2, - asyncBIAOperations []*itemoperation.BackupOperation) error + volumeSnapshotterGetter VolumeSnapshotterGetter, + ) error + + FinalizeBackup( + log logrus.FieldLogger, + backupRequest *Request, + inBackupFile io.Reader, + outBackupFile io.Writer, + backupItemActionResolver framework.BackupItemActionResolverV2, + asyncBIAOperations []*itemoperation.BackupOperation, + volumeInfos []*volume.VolumeInfo, + ) error } // kubernetesBackupper implements Backupper. @@ -183,11 +204,13 @@ func (kb *kubernetesBackupper) Backup(log logrus.FieldLogger, backupRequest *Req return kb.BackupWithResolvers(log, backupRequest, backupFile, backupItemActions, volumeSnapshotterGetter) } -func (kb *kubernetesBackupper) BackupWithResolvers(log logrus.FieldLogger, +func (kb *kubernetesBackupper) BackupWithResolvers( + log logrus.FieldLogger, backupRequest *Request, backupFile io.Writer, backupItemActionResolver framework.BackupItemActionResolverV2, - volumeSnapshotterGetter VolumeSnapshotterGetter) error { + volumeSnapshotterGetter VolumeSnapshotterGetter, +) error { gzippedData := gzip.NewWriter(backupFile) defer gzippedData.Close() @@ -470,7 +493,13 @@ func (kb *kubernetesBackupper) backupItem(log logrus.FieldLogger, gr schema.Grou return backedUpItem } -func (kb *kubernetesBackupper) finalizeItem(log logrus.FieldLogger, gr schema.GroupResource, itemBackupper *itemBackupper, unstructured *unstructured.Unstructured, preferredGVR schema.GroupVersionResource) (bool, []FileForArchive) { +func (kb *kubernetesBackupper) finalizeItem( + log logrus.FieldLogger, + gr schema.GroupResource, + itemBackupper *itemBackupper, + unstructured *unstructured.Unstructured, + preferredGVR schema.GroupVersionResource, +) (bool, []FileForArchive) { backedUpItem, updateFiles, err := itemBackupper.backupItem(log, unstructured, gr, preferredGVR, true, true) if aggregate, ok := err.(kubeerrs.Aggregate); ok { log.WithField("name", unstructured.GetName()).Infof("%d errors encountered backup up item", len(aggregate.Errors())) @@ -548,12 +577,15 @@ func (kb *kubernetesBackupper) writeBackupVersion(tw *tar.Writer) error { return nil } -func (kb *kubernetesBackupper) FinalizeBackup(log logrus.FieldLogger, +func (kb *kubernetesBackupper) FinalizeBackup( + log logrus.FieldLogger, backupRequest *Request, inBackupFile io.Reader, outBackupFile io.Writer, backupItemActionResolver framework.BackupItemActionResolverV2, - asyncBIAOperations []*itemoperation.BackupOperation) error { + asyncBIAOperations []*itemoperation.BackupOperation, + volumeInfos []*volume.VolumeInfo, +) error { gzw := gzip.NewWriter(outBackupFile) defer gzw.Close() tw := tar.NewWriter(gzw) @@ -642,6 +674,8 @@ func (kb *kubernetesBackupper) FinalizeBackup(log logrus.FieldLogger, return } + updateVolumeInfos(volumeInfos, unstructured, item.groupResource, log) + backedUp, itemFiles := kb.finalizeItem(log, item.groupResource, itemBackupper, &unstructured, item.preferredGVR) if backedUp { backedUpGroupResources[item.groupResource] = true @@ -730,3 +764,32 @@ type tarWriter interface { Write([]byte) (int, error) WriteHeader(*tar.Header) error } + +func updateVolumeInfos( + volumeInfos []*volume.VolumeInfo, + unstructured unstructured.Unstructured, + groupResource schema.GroupResource, + log logrus.FieldLogger, +) { + switch groupResource.String() { + case kuberesource.DataUploads.String(): + var dataUpload velerov2alpha1.DataUpload + err := runtime.DefaultUnstructuredConverter.FromUnstructured(unstructured.UnstructuredContent(), &dataUpload) + if err != nil { + log.WithError(err).Errorf("fail to convert DataUpload: %s/%s", + unstructured.GetNamespace(), unstructured.GetName()) + } + + for index := range volumeInfos { + if volumeInfos[index].PVCName == dataUpload.Spec.SourcePVC && + volumeInfos[index].PVCNamespace == dataUpload.Spec.SourceNamespace { + if dataUpload.Status.CompletionTimestamp != nil { + volumeInfos[index].CompletionTimestamp = dataUpload.Status.CompletionTimestamp + } + volumeInfos[index].SnapshotDataMovementInfo.SnapshotHandle = dataUpload.Status.SnapshotID + volumeInfos[index].SnapshotDataMovementInfo.RetainedSnapshot = dataUpload.Spec.CSISnapshot.VolumeSnapshot + volumeInfos[index].SnapshotDataMovementInfo.Size = dataUpload.Status.Progress.TotalBytes + } + } + } +} diff --git a/pkg/backup/backup_test.go b/pkg/backup/backup_test.go index 0ceb0572a..6f23e18e8 100644 --- a/pkg/backup/backup_test.go +++ b/pkg/backup/backup_test.go @@ -43,7 +43,9 @@ import ( "github.com/vmware-tanzu/velero/internal/resourcepolicies" "github.com/vmware-tanzu/velero/internal/volume" + "github.com/vmware-tanzu/velero/pkg/apis/velero/shared" velerov1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1" + velerov2alpha1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1" "github.com/vmware-tanzu/velero/pkg/builder" "github.com/vmware-tanzu/velero/pkg/client" "github.com/vmware-tanzu/velero/pkg/discovery" @@ -4433,3 +4435,59 @@ func TestBackupNamespaces(t *testing.T) { }) } } + +func TestUpdateVolumeInfos(t *testing.T) { + logger := logrus.StandardLogger() + // The unstructured conversion will loose the time precision to second + // level. To make test pass. Set the now precision at second at the + // beginning. + now := metav1.Now().Rfc3339Copy() + volumeInfos := []*volume.VolumeInfo{ + { + PVCName: "pvc1", + PVCNamespace: "ns1", + SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{}, + }, + } + dataUpload := velerov2alpha1.DataUpload{ + ObjectMeta: metav1.ObjectMeta{ + Name: "du1", + Namespace: "velero", + }, + Spec: velerov2alpha1.DataUploadSpec{ + SourcePVC: "pvc1", + SourceNamespace: "ns1", + CSISnapshot: &velerov2alpha1.CSISnapshotSpec{ + VolumeSnapshot: "vs1", + }, + }, + Status: velerov2alpha1.DataUploadStatus{ + CompletionTimestamp: &now, + SnapshotID: "snapshot1", + Progress: shared.DataMoveOperationProgress{ + TotalBytes: 10000, + }, + }, + } + duMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&dataUpload) + require.NoError(t, err) + + expectedVolumeInfos := []*volume.VolumeInfo{ + { + PVCName: "pvc1", + PVCNamespace: "ns1", + CompletionTimestamp: &now, + SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{ + SnapshotHandle: "snapshot1", + Size: 10000, + RetainedSnapshot: "vs1", + }, + }, + } + + updateVolumeInfos(volumeInfos, unstructured.Unstructured{Object: duMap}, kuberesource.DataUploads, logger) + + if len(expectedVolumeInfos) > 0 { + require.Equal(t, expectedVolumeInfos[0].SnapshotDataMovementInfo, volumeInfos[0].SnapshotDataMovementInfo) + } +} diff --git a/pkg/builder/data_upload_builder.go b/pkg/builder/data_upload_builder.go index 29e7a0191..4623dac6e 100644 --- a/pkg/builder/data_upload_builder.go +++ b/pkg/builder/data_upload_builder.go @@ -115,8 +115,14 @@ func (d *DataUploadBuilder) CSISnapshot(cSISnapshot *velerov2alpha1api.CSISnapsh } // StartTimestamp sets the DataUpload's StartTimestamp. -func (d *DataUploadBuilder) StartTimestamp(startTime *metav1.Time) *DataUploadBuilder { - d.object.Status.StartTimestamp = startTime +func (d *DataUploadBuilder) StartTimestamp(startTimestamp *metav1.Time) *DataUploadBuilder { + d.object.Status.StartTimestamp = startTimestamp + return d +} + +// CompletionTimestamp sets the DataUpload's StartTimestamp. +func (d *DataUploadBuilder) CompletionTimestamp(completionTimestamp *metav1.Time) *DataUploadBuilder { + d.object.Status.CompletionTimestamp = completionTimestamp return d } diff --git a/pkg/builder/pod_volume_backup_builder.go b/pkg/builder/pod_volume_backup_builder.go index 14e57a063..80cb55042 100644 --- a/pkg/builder/pod_volume_backup_builder.go +++ b/pkg/builder/pod_volume_backup_builder.go @@ -80,6 +80,16 @@ func (b *PodVolumeBackupBuilder) SnapshotID(snapshotID string) *PodVolumeBackupB return b } +func (b *PodVolumeBackupBuilder) StartTimestamp(startTimestamp *metav1.Time) *PodVolumeBackupBuilder { + b.object.Status.StartTimestamp = startTimestamp + return b +} + +func (b *PodVolumeBackupBuilder) CompletionTimestamp(completionTimestamp *metav1.Time) *PodVolumeBackupBuilder { + b.object.Status.CompletionTimestamp = completionTimestamp + return b +} + // PodName sets the name of the pod associated with this PodVolumeBackup. func (b *PodVolumeBackupBuilder) PodName(name string) *PodVolumeBackupBuilder { b.object.Spec.Pod.Name = name diff --git a/pkg/controller/backup_controller_test.go b/pkg/controller/backup_controller_test.go index 37caf839e..627f17e46 100644 --- a/pkg/controller/backup_controller_test.go +++ b/pkg/controller/backup_controller_test.go @@ -44,6 +44,7 @@ import ( kbclient "sigs.k8s.io/controller-runtime/pkg/client" fakeClient "sigs.k8s.io/controller-runtime/pkg/client/fake" + "github.com/vmware-tanzu/velero/internal/volume" velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1" pkgbackup "github.com/vmware-tanzu/velero/pkg/backup" "github.com/vmware-tanzu/velero/pkg/builder" @@ -78,9 +79,15 @@ func (b *fakeBackupper) BackupWithResolvers(logger logrus.FieldLogger, backup *p return args.Error(0) } -func (b *fakeBackupper) FinalizeBackup(logger logrus.FieldLogger, backup *pkgbackup.Request, inBackupFile io.Reader, outBackupFile io.Writer, +func (b *fakeBackupper) FinalizeBackup( + logger logrus.FieldLogger, + backup *pkgbackup.Request, + inBackupFile io.Reader, + outBackupFile io.Writer, backupItemActionResolver framework.BackupItemActionResolverV2, - asyncBIAOperations []*itemoperation.BackupOperation) error { + asyncBIAOperations []*itemoperation.BackupOperation, + volumeInfos []*volume.VolumeInfo, +) error { args := b.Called(logger, backup, inBackupFile, outBackupFile, backupItemActionResolver, asyncBIAOperations) return args.Error(0) } diff --git a/pkg/controller/backup_finalizer_controller.go b/pkg/controller/backup_finalizer_controller.go index c8fc93358..d10935fd7 100644 --- a/pkg/controller/backup_finalizer_controller.go +++ b/pkg/controller/backup_finalizer_controller.go @@ -18,7 +18,9 @@ package controller import ( "bytes" + "compress/gzip" "context" + "encoding/json" "os" "github.com/pkg/errors" @@ -29,6 +31,7 @@ import ( ctrl "sigs.k8s.io/controller-runtime" kbclient "sigs.k8s.io/controller-runtime/pkg/client" + "github.com/vmware-tanzu/velero/internal/volume" velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1" pkgbackup "github.com/vmware-tanzu/velero/pkg/backup" "github.com/vmware-tanzu/velero/pkg/itemoperation" @@ -111,7 +114,11 @@ func (r *backupFinalizerReconciler) Reconcile(ctx context.Context, req ctrl.Requ original := backup.DeepCopy() defer func() { switch backup.Status.Phase { - case velerov1api.BackupPhaseCompleted, velerov1api.BackupPhasePartiallyFailed, velerov1api.BackupPhaseFailed, velerov1api.BackupPhaseFailedValidation: + case + velerov1api.BackupPhaseCompleted, + velerov1api.BackupPhasePartiallyFailed, + velerov1api.BackupPhaseFailed, + velerov1api.BackupPhaseFailedValidation: r.backupTracker.Delete(backup.Namespace, backup.Name) } // Always attempt to Patch the backup object and status after each reconciliation. @@ -150,8 +157,14 @@ func (r *backupFinalizerReconciler) Reconcile(ctx context.Context, req ctrl.Requ SkippedPVTracker: pkgbackup.NewSkipPVTracker(), } var outBackupFile *os.File + var volumeInfos []*volume.VolumeInfo if len(operations) > 0 { - // Call itemBackupper.BackupItem for the list of items updated by async operations + volumeInfos, err = backupStore.GetBackupVolumeInfos(backup.Name) + if err != nil { + log.WithError(err).Error("error getting backup VolumeInfos") + return ctrl.Result{}, errors.WithStack(err) + } + log.Info("Setting up finalized backup temp file") inBackupFile, err := downloadToTempFile(backup.Name, backupStore, log) if err != nil { @@ -172,7 +185,17 @@ func (r *backupFinalizerReconciler) Reconcile(ctx context.Context, req ctrl.Requ return ctrl.Result{}, errors.WithStack(err) } backupItemActionsResolver := framework.NewBackupItemActionResolverV2(actions) - err = r.backupper.FinalizeBackup(log, backupRequest, inBackupFile, outBackupFile, backupItemActionsResolver, operations) + + // Call itemBackupper.BackupItem for the list of items updated by async operations + err = r.backupper.FinalizeBackup( + log, + backupRequest, + inBackupFile, + outBackupFile, + backupItemActionsResolver, + operations, + volumeInfos, + ) if err != nil { log.WithError(err).Error("error finalizing Backup") return ctrl.Result{}, errors.WithStack(err) @@ -209,6 +232,24 @@ func (r *backupFinalizerReconciler) Reconcile(ctx context.Context, req ctrl.Requ if err != nil { return ctrl.Result{}, errors.Wrap(err, "error uploading backup final contents") } + + // Update the backup's VolumeInfos + backupVolumeInfoBuf := new(bytes.Buffer) + gzw := gzip.NewWriter(backupVolumeInfoBuf) + defer gzw.Close() + + if err := json.NewEncoder(gzw).Encode(volumeInfos); err != nil { + return ctrl.Result{}, errors.Wrap(err, "error encoding restore results to JSON") + } + + if err := gzw.Close(); err != nil { + return ctrl.Result{}, errors.Wrap(err, "error closing gzip writer") + } + + err = backupStore.PutBackupVolumeInfos(backup.Name, backupVolumeInfoBuf) + if err != nil { + return ctrl.Result{}, errors.Wrap(err, "fail to upload backup VolumeInfos") + } } return ctrl.Result{}, nil } diff --git a/pkg/controller/backup_finalizer_controller_test.go b/pkg/controller/backup_finalizer_controller_test.go index e2f07b3e3..8c1a5e361 100644 --- a/pkg/controller/backup_finalizer_controller_test.go +++ b/pkg/controller/backup_finalizer_controller_test.go @@ -222,6 +222,8 @@ func TestBackupFinalizerReconcile(t *testing.T) { backupStore.On("GetBackupContents", mock.Anything).Return(io.NopCloser(bytes.NewReader([]byte("hello world"))), nil) backupStore.On("PutBackupContents", mock.Anything, mock.Anything).Return(nil) backupStore.On("PutBackupMetadata", mock.Anything, mock.Anything).Return(nil) + backupStore.On("GetBackupVolumeInfos", mock.Anything).Return(nil, nil) + backupStore.On("PutBackupVolumeInfos", mock.Anything, mock.Anything).Return(nil) pluginManager.On("GetBackupItemActionsV2").Return(nil, nil) backupper.On("FinalizeBackup", mock.Anything, mock.Anything, mock.Anything, mock.Anything, framework.BackupItemActionResolverV2{}, mock.Anything).Return(nil) _, err := reconciler.Reconcile(context.TODO(), ctrl.Request{NamespacedName: types.NamespacedName{Namespace: test.backup.Namespace, Name: test.backup.Name}}) diff --git a/pkg/kuberesource/kuberesource.go b/pkg/kuberesource/kuberesource.go index c2c2d84ee..c39e769db 100644 --- a/pkg/kuberesource/kuberesource.go +++ b/pkg/kuberesource/kuberesource.go @@ -35,4 +35,5 @@ var ( VolumeSnapshots = schema.GroupResource{Group: "snapshot.storage.k8s.io", Resource: "volumesnapshots"} VolumeSnapshotContents = schema.GroupResource{Group: "snapshot.storage.k8s.io", Resource: "volumesnapshotcontents"} PriorityClasses = schema.GroupResource{Group: "scheduling.k8s.io", Resource: "priorityclasses"} + DataUploads = schema.GroupResource{Group: "velero.io", Resource: "datauploads"} ) diff --git a/pkg/persistence/mocks/backup_store.go b/pkg/persistence/mocks/backup_store.go index b45928e77..1c2cf7d05 100644 --- a/pkg/persistence/mocks/backup_store.go +++ b/pkg/persistence/mocks/backup_store.go @@ -314,7 +314,7 @@ func (_m *BackupStore) GetRestoreItemOperations(name string) ([]*itemoperation.R return r0, r1 } -// GetRestoreItemOperations provides a mock function with given fields: name +// GetBackupVolumeInfos provides a mock function with given fields: name func (_m *BackupStore) GetBackupVolumeInfos(name string) ([]*volume.VolumeInfo, error) { ret := _m.Called(name) @@ -337,6 +337,20 @@ func (_m *BackupStore) GetBackupVolumeInfos(name string) ([]*volume.VolumeInfo, return r0, r1 } +// PutBackupVolumeInfos provides a mock function with given fields: name, volumeInfo +func (_m *BackupStore) PutBackupVolumeInfos(name string, volumeInfo io.Reader) error { + ret := _m.Called(name, volumeInfo) + + var r0 error + if rf, ok := ret.Get(0).(func(string, io.Reader) error); ok { + r0 = rf(name, volumeInfo) + } else { + r0 = ret.Error(0) + } + + return r0 +} + // GetRestoreResults provides a mock function with given fields: name func (_m *BackupStore) GetRestoreResults(name string) (map[string]results.Result, error) { ret := _m.Called(name) diff --git a/pkg/persistence/object_store.go b/pkg/persistence/object_store.go index 962c36de7..1036f88eb 100644 --- a/pkg/persistence/object_store.go +++ b/pkg/persistence/object_store.go @@ -75,6 +75,7 @@ type BackupStore interface { GetCSIVolumeSnapshotContents(name string) ([]*snapshotv1api.VolumeSnapshotContent, error) GetCSIVolumeSnapshotClasses(name string) ([]*snapshotv1api.VolumeSnapshotClass, error) GetBackupVolumeInfos(name string) ([]*volume.VolumeInfo, error) + PutBackupVolumeInfos(name string, volumeInfo io.Reader) error GetRestoreResults(name string) (map[string]results.Result, error) // BackupExists checks if the backup metadata file exists in object storage. @@ -516,6 +517,10 @@ func (s *objectBackupStore) GetBackupVolumeInfos(name string) ([]*volume.VolumeI return volumeInfos, nil } +func (s *objectBackupStore) PutBackupVolumeInfos(name string, volumeInfo io.Reader) error { + return s.objectStore.PutObject(s.bucket, s.layout.getBackupVolumeInfoKey(name), volumeInfo) +} + func (s *objectBackupStore) GetRestoreResults(name string) (map[string]results.Result, error) { results := make(map[string]results.Result) diff --git a/pkg/persistence/object_store_test.go b/pkg/persistence/object_store_test.go index a63d31825..a23906eef 100644 --- a/pkg/persistence/object_store_test.go +++ b/pkg/persistence/object_store_test.go @@ -1203,6 +1203,51 @@ func TestGetRestoredResourceList(t *testing.T) { assert.EqualValues(t, list["pod"], res["pod"]) } +func TestPutBackupVolumeInfos(t *testing.T) { + tests := []struct { + name string + prefix string + expectedErr string + expectedKeys []string + }{ + { + name: "normal case", + expectedErr: "", + expectedKeys: []string{ + "backups/backup-1/backup-1-volumeinfo.json.gz", + }, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + harness := newObjectBackupStoreTestHarness("foo", tc.prefix) + + volumeInfos := []*volume.VolumeInfo{ + { + PVCName: "test", + }, + } + + buf := new(bytes.Buffer) + gzw := gzip.NewWriter(buf) + defer gzw.Close() + + require.NoError(t, json.NewEncoder(gzw).Encode(volumeInfos)) + bufferContent := buf.Bytes() + + err := harness.PutBackupVolumeInfos("backup-1", buf) + + velerotest.AssertErrorMatches(t, tc.expectedErr, err) + assert.Len(t, harness.objectStore.Data[harness.bucket], len(tc.expectedKeys)) + for _, key := range tc.expectedKeys { + assert.Contains(t, harness.objectStore.Data[harness.bucket], key) + assert.Equal(t, harness.objectStore.Data[harness.bucket][key], bufferContent) + } + }) + } +} + func encodeToBytes(obj runtime.Object) []byte { res, err := encode.Encode(obj, "json") if err != nil { From 54620354695e1e0bdac5aa271272e283de0bb961 Mon Sep 17 00:00:00 2001 From: Xun Jiang Date: Tue, 26 Mar 2024 14:41:27 +0800 Subject: [PATCH 2/2] Delete the unneeded pvRestorer action in handleSkippedPVHasRetainPolicy According to comment, calling executePVAction aims to reset PV's claimRef, but the reset logic was moved into resetVolumeBindingInfo since release-1.4. Signed-off-by: Xun Jiang --- pkg/backup/backup.go | 111 ++++++++++- pkg/backup/backup_test.go | 179 +++++++++++++----- pkg/builder/data_upload_builder.go | 6 + pkg/cmd/server/server.go | 4 + pkg/controller/backup_controller_test.go | 2 - pkg/controller/backup_finalizer_controller.go | 29 --- pkg/repository/maintenance_test.go | 2 +- pkg/restore/restore.go | 15 +- 8 files changed, 251 insertions(+), 97 deletions(-) diff --git a/pkg/backup/backup.go b/pkg/backup/backup.go index 8dcc8a12e..5190ceea2 100644 --- a/pkg/backup/backup.go +++ b/pkg/backup/backup.go @@ -18,6 +18,7 @@ package backup import ( "archive/tar" + "bytes" "compress/gzip" "context" "encoding/json" @@ -45,6 +46,8 @@ import ( "github.com/vmware-tanzu/velero/pkg/discovery" "github.com/vmware-tanzu/velero/pkg/itemoperation" "github.com/vmware-tanzu/velero/pkg/kuberesource" + "github.com/vmware-tanzu/velero/pkg/persistence" + "github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt" "github.com/vmware-tanzu/velero/pkg/plugin/framework" "github.com/vmware-tanzu/velero/pkg/plugin/velero" biav2 "github.com/vmware-tanzu/velero/pkg/plugin/velero/backupitemaction/v2" @@ -90,7 +93,6 @@ type Backupper interface { outBackupFile io.Writer, backupItemActionResolver framework.BackupItemActionResolverV2, asyncBIAOperations []*itemoperation.BackupOperation, - volumeInfos []*volume.VolumeInfo, ) error } @@ -105,6 +107,8 @@ type kubernetesBackupper struct { defaultVolumesToFsBackup bool clientPageSize int uploaderType string + pluginManager func(logrus.FieldLogger) clientmgmt.Manager + backupStoreGetter persistence.ObjectBackupStoreGetter } func (i *itemKey) String() string { @@ -132,6 +136,8 @@ func NewKubernetesBackupper( defaultVolumesToFsBackup bool, clientPageSize int, uploaderType string, + pluginManager func(logrus.FieldLogger) clientmgmt.Manager, + backupStoreGetter persistence.ObjectBackupStoreGetter, ) (Backupper, error) { return &kubernetesBackupper{ kbClient: kbClient, @@ -143,6 +149,8 @@ func NewKubernetesBackupper( defaultVolumesToFsBackup: defaultVolumesToFsBackup, clientPageSize: clientPageSize, uploaderType: uploaderType, + pluginManager: pluginManager, + backupStoreGetter: backupStoreGetter, }, nil } @@ -584,7 +592,6 @@ func (kb *kubernetesBackupper) FinalizeBackup( outBackupFile io.Writer, backupItemActionResolver framework.BackupItemActionResolverV2, asyncBIAOperations []*itemoperation.BackupOperation, - volumeInfos []*volume.VolumeInfo, ) error { gzw := gzip.NewWriter(outBackupFile) defer gzw.Close() @@ -648,6 +655,8 @@ func (kb *kubernetesBackupper) FinalizeBackup( updateFiles := make(map[string]FileForArchive) backedUpGroupResources := map[schema.GroupResource]bool{} + unstructuredDataUploads := make([]unstructured.Unstructured, 0) + for i, item := range items { log.WithFields(map[string]interface{}{ "progress": "", @@ -674,7 +683,9 @@ func (kb *kubernetesBackupper) FinalizeBackup( return } - updateVolumeInfos(volumeInfos, unstructured, item.groupResource, log) + if item.groupResource == kuberesource.DataUploads { + unstructuredDataUploads = append(unstructuredDataUploads, unstructured) + } backedUp, itemFiles := kb.finalizeItem(log, item.groupResource, itemBackupper, &unstructured, item.preferredGVR) if backedUp { @@ -697,6 +708,22 @@ func (kb *kubernetesBackupper) FinalizeBackup( }).Infof("Updated %d items out of an estimated total of %d (estimate will change throughout the backup finalizer)", len(backupRequest.BackedUpItems), totalItems) } + backupStore, volumeInfos, err := kb.getVolumeInfos(*backupRequest.Backup, log) + if err != nil { + log.WithError(err).Errorf("fail to get the backup VolumeInfos for backup %s", backupRequest.Name) + return err + } + + if err := updateVolumeInfos(volumeInfos, unstructuredDataUploads, asyncBIAOperations, log); err != nil { + log.WithError(err).Errorf("fail to update VolumeInfos for backup %s", backupRequest.Name) + return err + } + + if err := putVolumeInfos(backupRequest.Name, volumeInfos, backupStore); err != nil { + log.WithError(err).Errorf("fail to put the VolumeInfos for backup %s", backupRequest.Name) + return err + } + // write new tar archive replacing files in original with content updateFiles for matches if err := buildFinalTarball(tr, tw, updateFiles); err != nil { log.Errorf("Error building final tarball: %s", err.Error()) @@ -765,19 +792,48 @@ type tarWriter interface { WriteHeader(*tar.Header) error } +func (kb *kubernetesBackupper) getVolumeInfos( + backup velerov1api.Backup, + log logrus.FieldLogger, +) (persistence.BackupStore, []*volume.VolumeInfo, error) { + location := &velerov1api.BackupStorageLocation{} + if err := kb.kbClient.Get(context.Background(), kbclient.ObjectKey{ + Namespace: backup.Namespace, + Name: backup.Spec.StorageLocation, + }, location); err != nil { + return nil, nil, errors.WithStack(err) + } + + pluginManager := kb.pluginManager(log) + defer pluginManager.CleanupClients() + + backupStore, storeErr := kb.backupStoreGetter.Get(location, pluginManager, log) + if storeErr != nil { + return nil, nil, storeErr + } + + volumeInfos, err := backupStore.GetBackupVolumeInfos(backup.Name) + if err != nil { + return nil, nil, err + } + + return backupStore, volumeInfos, nil +} + +// updateVolumeInfos update the VolumeInfos according to the AsyncOperations func updateVolumeInfos( volumeInfos []*volume.VolumeInfo, - unstructured unstructured.Unstructured, - groupResource schema.GroupResource, + unstructuredItems []unstructured.Unstructured, + operations []*itemoperation.BackupOperation, log logrus.FieldLogger, -) { - switch groupResource.String() { - case kuberesource.DataUploads.String(): +) error { + for _, unstructured := range unstructuredItems { var dataUpload velerov2alpha1.DataUpload err := runtime.DefaultUnstructuredConverter.FromUnstructured(unstructured.UnstructuredContent(), &dataUpload) if err != nil { log.WithError(err).Errorf("fail to convert DataUpload: %s/%s", unstructured.GetNamespace(), unstructured.GetName()) + return err } for index := range volumeInfos { @@ -792,4 +848,43 @@ func updateVolumeInfos( } } } + + // Update CSI snapshot VolumeInfo's CompletionTimestamp by the operation update time. + for volumeIndex := range volumeInfos { + if volumeInfos[volumeIndex].BackupMethod == volume.CSISnapshot && + volumeInfos[volumeIndex].CSISnapshotInfo != nil { + for opIndex := range operations { + if volumeInfos[volumeIndex].CSISnapshotInfo.OperationID == operations[opIndex].Spec.OperationID { + // The VolumeSnapshot and VolumeSnapshotContent don't have a completion timestamp, + // so use the operation.Status.Updated as the alternative. It is not the exact time + // when the snapshot turns ready, but the operation controller periodically watch the + // VSC and VS status. When the controller finds they reach to the ReadyToUse state, + // The operation.Status.Updated is set as the found time. + volumeInfos[volumeIndex].CompletionTimestamp = operations[opIndex].Status.Updated + } + } + } + } + + return nil +} + +func putVolumeInfos( + backupName string, + volumeInfos []*volume.VolumeInfo, + backupStore persistence.BackupStore, +) error { + backupVolumeInfoBuf := new(bytes.Buffer) + gzw := gzip.NewWriter(backupVolumeInfoBuf) + defer gzw.Close() + + if err := json.NewEncoder(gzw).Encode(volumeInfos); err != nil { + return errors.Wrap(err, "error encoding restore results to JSON") + } + + if err := gzw.Close(); err != nil { + return errors.Wrap(err, "error closing gzip writer") + } + + return backupStore.PutBackupVolumeInfos(backupName, backupVolumeInfoBuf) } diff --git a/pkg/backup/backup_test.go b/pkg/backup/backup_test.go index 6f23e18e8..6337a079f 100644 --- a/pkg/backup/backup_test.go +++ b/pkg/backup/backup_test.go @@ -52,6 +52,10 @@ import ( "github.com/vmware-tanzu/velero/pkg/features" "github.com/vmware-tanzu/velero/pkg/itemoperation" "github.com/vmware-tanzu/velero/pkg/kuberesource" + "github.com/vmware-tanzu/velero/pkg/persistence" + persistencemocks "github.com/vmware-tanzu/velero/pkg/persistence/mocks" + "github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt" + pluginmocks "github.com/vmware-tanzu/velero/pkg/plugin/mocks" "github.com/vmware-tanzu/velero/pkg/plugin/velero" biav2 "github.com/vmware-tanzu/velero/pkg/plugin/velero/backupitemaction/v2" vsv1 "github.com/vmware-tanzu/velero/pkg/plugin/velero/volumesnapshotter/v1" @@ -4436,58 +4440,143 @@ func TestBackupNamespaces(t *testing.T) { } } -func TestUpdateVolumeInfos(t *testing.T) { - logger := logrus.StandardLogger() - // The unstructured conversion will loose the time precision to second - // level. To make test pass. Set the now precision at second at the - // beginning. - now := metav1.Now().Rfc3339Copy() - volumeInfos := []*volume.VolumeInfo{ - { - PVCName: "pvc1", - PVCNamespace: "ns1", - SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{}, - }, - } - dataUpload := velerov2alpha1.DataUpload{ - ObjectMeta: metav1.ObjectMeta{ - Name: "du1", - Namespace: "velero", - }, - Spec: velerov2alpha1.DataUploadSpec{ - SourcePVC: "pvc1", - SourceNamespace: "ns1", - CSISnapshot: &velerov2alpha1.CSISnapshotSpec{ - VolumeSnapshot: "vs1", - }, - }, - Status: velerov2alpha1.DataUploadStatus{ - CompletionTimestamp: &now, - SnapshotID: "snapshot1", - Progress: shared.DataMoveOperationProgress{ - TotalBytes: 10000, - }, - }, - } - duMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&dataUpload) +func TestGetVolumeInfos(t *testing.T) { + h := newHarness(t) + pluginManager := new(pluginmocks.Manager) + backupStore := new(persistencemocks.BackupStore) + h.backupper.pluginManager = func(logrus.FieldLogger) clientmgmt.Manager { return pluginManager } + h.backupper.backupStoreGetter = NewFakeSingleObjectBackupStoreGetter(backupStore) + backupStore.On("GetBackupVolumeInfos", "backup-01").Return([]*volume.VolumeInfo{}, nil) + pluginManager.On("CleanupClients").Return() + + backup := builder.ForBackup("velero", "backup-01").StorageLocation("default").Result() + bsl := builder.ForBackupStorageLocation("velero", "default").Result() + require.NoError(t, h.backupper.kbClient.Create(context.Background(), bsl)) + + _, _, err := h.backupper.getVolumeInfos(*backup, h.log) require.NoError(t, err) +} - expectedVolumeInfos := []*volume.VolumeInfo{ +func TestUpdateVolumeInfos(t *testing.T) { + timeExample := time.Date(2014, 6, 5, 11, 56, 45, 0, time.Local) + now := metav1.NewTime(timeExample) + logger := logrus.StandardLogger() + + tests := []struct { + name string + operations []*itemoperation.BackupOperation + dataUpload *velerov2alpha1.DataUpload + volumeInfos []*volume.VolumeInfo + expectedVolumeInfos []*volume.VolumeInfo + }{ { - PVCName: "pvc1", - PVCNamespace: "ns1", - CompletionTimestamp: &now, - SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{ - SnapshotHandle: "snapshot1", - Size: 10000, - RetainedSnapshot: "vs1", + name: "CSISnapshot VolumeInfo update", + operations: []*itemoperation.BackupOperation{ + { + Spec: itemoperation.BackupOperationSpec{ + OperationID: "test-operation", + }, + Status: itemoperation.OperationStatus{ + Updated: &now, + }, + }, + }, + volumeInfos: []*volume.VolumeInfo{ + { + BackupMethod: volume.CSISnapshot, + CompletionTimestamp: &metav1.Time{}, + CSISnapshotInfo: &volume.CSISnapshotInfo{ + OperationID: "test-operation", + }, + }, + }, + expectedVolumeInfos: []*volume.VolumeInfo{ + { + BackupMethod: volume.CSISnapshot, + CompletionTimestamp: &now, + CSISnapshotInfo: &volume.CSISnapshotInfo{ + OperationID: "test-operation", + }, + }, + }, + }, + { + name: "DataUpload VolumeInfo update", + operations: []*itemoperation.BackupOperation{}, + dataUpload: builder.ForDataUpload("velero", "du-1"). + CompletionTimestamp(&now). + CSISnapshot(&velerov2alpha1.CSISnapshotSpec{VolumeSnapshot: "vs-1"}). + SnapshotID("snapshot-id"). + Progress(shared.DataMoveOperationProgress{TotalBytes: 1000}). + SourceNamespace("ns-1"). + SourcePVC("pvc-1"). + Result(), + volumeInfos: []*volume.VolumeInfo{ + { + PVCName: "pvc-1", + PVCNamespace: "ns-1", + CompletionTimestamp: &metav1.Time{}, + SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{ + DataMover: "velero", + }, + }, + }, + expectedVolumeInfos: []*volume.VolumeInfo{ + { + PVCName: "pvc-1", + PVCNamespace: "ns-1", + CompletionTimestamp: &now, + SnapshotDataMovementInfo: &volume.SnapshotDataMovementInfo{ + DataMover: "velero", + RetainedSnapshot: "vs-1", + SnapshotHandle: "snapshot-id", + Size: 1000, + }, + }, }, }, } - updateVolumeInfos(volumeInfos, unstructured.Unstructured{Object: duMap}, kuberesource.DataUploads, logger) + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + unstructures := []unstructured.Unstructured{} + if tc.dataUpload != nil { + duMap, error := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.dataUpload) + require.NoError(t, error) + unstructures = append(unstructures, + unstructured.Unstructured{ + Object: duMap, + }, + ) + } - if len(expectedVolumeInfos) > 0 { - require.Equal(t, expectedVolumeInfos[0].SnapshotDataMovementInfo, volumeInfos[0].SnapshotDataMovementInfo) + require.NoError(t, updateVolumeInfos(tc.volumeInfos, unstructures, tc.operations, logger)) + require.Equal(t, tc.expectedVolumeInfos[0].CompletionTimestamp, tc.volumeInfos[0].CompletionTimestamp) + require.Equal(t, tc.expectedVolumeInfos[0].SnapshotDataMovementInfo, tc.volumeInfos[0].SnapshotDataMovementInfo) + }) } } + +func TestPutVolumeInfos(t *testing.T) { + backupName := "backup-01" + + backupStore := new(persistencemocks.BackupStore) + + backupStore.On("PutBackupVolumeInfos", mock.Anything, mock.Anything).Return(nil) + + require.NoError(t, putVolumeInfos(backupName, []*volume.VolumeInfo{}, backupStore)) +} + +type fakeSingleObjectBackupStoreGetter struct { + store persistence.BackupStore +} + +func (f *fakeSingleObjectBackupStoreGetter) Get(*velerov1.BackupStorageLocation, persistence.ObjectStoreGetter, logrus.FieldLogger) (persistence.BackupStore, error) { + return f.store, nil +} + +// NewFakeSingleObjectBackupStoreGetter returns an ObjectBackupStoreGetter +// that will return only the given BackupStore. +func NewFakeSingleObjectBackupStoreGetter(store persistence.BackupStore) persistence.ObjectBackupStoreGetter { + return &fakeSingleObjectBackupStoreGetter{store: store} +} diff --git a/pkg/builder/data_upload_builder.go b/pkg/builder/data_upload_builder.go index 4623dac6e..7ff33dcb0 100644 --- a/pkg/builder/data_upload_builder.go +++ b/pkg/builder/data_upload_builder.go @@ -19,6 +19,7 @@ package builder import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "github.com/vmware-tanzu/velero/pkg/apis/velero/shared" velerov2alpha1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1" ) @@ -131,3 +132,8 @@ func (d *DataUploadBuilder) Labels(labels map[string]string) *DataUploadBuilder d.object.Labels = labels return d } + +func (d *DataUploadBuilder) Progress(progress shared.DataMoveOperationProgress) *DataUploadBuilder { + d.object.Status.Progress = progress + return d +} diff --git a/pkg/cmd/server/server.go b/pkg/cmd/server/server.go index 899608ebf..13c939d6b 100644 --- a/pkg/cmd/server/server.go +++ b/pkg/cmd/server/server.go @@ -780,6 +780,8 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string s.config.defaultVolumesToFsBackup, s.config.clientPageSize, s.config.uploaderType, + newPluginManager, + backupStoreGetter, ) cmd.CheckError(err) if err := controller.NewBackupReconciler( @@ -860,6 +862,8 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string s.config.defaultVolumesToFsBackup, s.config.clientPageSize, s.config.uploaderType, + newPluginManager, + backupStoreGetter, ) cmd.CheckError(err) r := controller.NewBackupFinalizerReconciler( diff --git a/pkg/controller/backup_controller_test.go b/pkg/controller/backup_controller_test.go index 627f17e46..59195c37b 100644 --- a/pkg/controller/backup_controller_test.go +++ b/pkg/controller/backup_controller_test.go @@ -44,7 +44,6 @@ import ( kbclient "sigs.k8s.io/controller-runtime/pkg/client" fakeClient "sigs.k8s.io/controller-runtime/pkg/client/fake" - "github.com/vmware-tanzu/velero/internal/volume" velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1" pkgbackup "github.com/vmware-tanzu/velero/pkg/backup" "github.com/vmware-tanzu/velero/pkg/builder" @@ -86,7 +85,6 @@ func (b *fakeBackupper) FinalizeBackup( outBackupFile io.Writer, backupItemActionResolver framework.BackupItemActionResolverV2, asyncBIAOperations []*itemoperation.BackupOperation, - volumeInfos []*volume.VolumeInfo, ) error { args := b.Called(logger, backup, inBackupFile, outBackupFile, backupItemActionResolver, asyncBIAOperations) return args.Error(0) diff --git a/pkg/controller/backup_finalizer_controller.go b/pkg/controller/backup_finalizer_controller.go index d10935fd7..2bf17e9bf 100644 --- a/pkg/controller/backup_finalizer_controller.go +++ b/pkg/controller/backup_finalizer_controller.go @@ -18,9 +18,7 @@ package controller import ( "bytes" - "compress/gzip" "context" - "encoding/json" "os" "github.com/pkg/errors" @@ -31,7 +29,6 @@ import ( ctrl "sigs.k8s.io/controller-runtime" kbclient "sigs.k8s.io/controller-runtime/pkg/client" - "github.com/vmware-tanzu/velero/internal/volume" velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1" pkgbackup "github.com/vmware-tanzu/velero/pkg/backup" "github.com/vmware-tanzu/velero/pkg/itemoperation" @@ -157,14 +154,7 @@ func (r *backupFinalizerReconciler) Reconcile(ctx context.Context, req ctrl.Requ SkippedPVTracker: pkgbackup.NewSkipPVTracker(), } var outBackupFile *os.File - var volumeInfos []*volume.VolumeInfo if len(operations) > 0 { - volumeInfos, err = backupStore.GetBackupVolumeInfos(backup.Name) - if err != nil { - log.WithError(err).Error("error getting backup VolumeInfos") - return ctrl.Result{}, errors.WithStack(err) - } - log.Info("Setting up finalized backup temp file") inBackupFile, err := downloadToTempFile(backup.Name, backupStore, log) if err != nil { @@ -194,7 +184,6 @@ func (r *backupFinalizerReconciler) Reconcile(ctx context.Context, req ctrl.Requ outBackupFile, backupItemActionsResolver, operations, - volumeInfos, ) if err != nil { log.WithError(err).Error("error finalizing Backup") @@ -232,24 +221,6 @@ func (r *backupFinalizerReconciler) Reconcile(ctx context.Context, req ctrl.Requ if err != nil { return ctrl.Result{}, errors.Wrap(err, "error uploading backup final contents") } - - // Update the backup's VolumeInfos - backupVolumeInfoBuf := new(bytes.Buffer) - gzw := gzip.NewWriter(backupVolumeInfoBuf) - defer gzw.Close() - - if err := json.NewEncoder(gzw).Encode(volumeInfos); err != nil { - return ctrl.Result{}, errors.Wrap(err, "error encoding restore results to JSON") - } - - if err := gzw.Close(); err != nil { - return ctrl.Result{}, errors.Wrap(err, "error closing gzip writer") - } - - err = backupStore.PutBackupVolumeInfos(backup.Name, backupVolumeInfoBuf) - if err != nil { - return ctrl.Result{}, errors.Wrap(err, "fail to upload backup VolumeInfos") - } } return ctrl.Result{}, nil } diff --git a/pkg/repository/maintenance_test.go b/pkg/repository/maintenance_test.go index c27fb9287..7eb43cf1b 100644 --- a/pkg/repository/maintenance_test.go +++ b/pkg/repository/maintenance_test.go @@ -118,7 +118,7 @@ func TestDeleteOldMaintenanceJobs(t *testing.T) { assert.NoError(t, err) // We expect the number of jobs to be equal to 'keep' - assert.Equal(t, keep, len(jobList.Items)) + assert.Len(t, jobList.Items, keep) // We expect that the oldest jobs were deleted // Job3 should not be present in the remaining list diff --git a/pkg/restore/restore.go b/pkg/restore/restore.go index 929b14c9e..904810818 100644 --- a/pkg/restore/restore.go +++ b/pkg/restore/restore.go @@ -1268,7 +1268,7 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso // want to dynamically re-provision it. return warnings, errs, itemExists } else { - obj, err = ctx.handleSkippedPVHasRetainPolicy(obj, resourceID, restoreLogger) + obj, err = ctx.handleSkippedPVHasRetainPolicy(obj, restoreLogger) if err != nil { errs.Add(namespace, err) return warnings, errs, itemExists @@ -1322,7 +1322,7 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso return warnings, errs, itemExists default: - obj, err = ctx.handleSkippedPVHasRetainPolicy(obj, resourceID, restoreLogger) + obj, err = ctx.handleSkippedPVHasRetainPolicy(obj, restoreLogger) if err != nil { errs.Add(namespace, err) return warnings, errs, itemExists @@ -2497,7 +2497,6 @@ func (ctx *restoreContext) handlePVHasNativeSnapshot(obj *unstructured.Unstructu func (ctx *restoreContext) handleSkippedPVHasRetainPolicy( obj *unstructured.Unstructured, - resourceID string, logger logrus.FieldLogger, ) (*unstructured.Unstructured, error) { logger.Infof("Restoring persistent volume as-is because it doesn't have a snapshot and its reclaim policy is not Delete.") @@ -2508,13 +2507,5 @@ func (ctx *restoreContext) handleSkippedPVHasRetainPolicy( } obj = resetVolumeBindingInfo(obj) - - // We call the pvRestorer here to clear out the PV's claimRef.UID, - // so it can be re-claimed when its PVC is restored and gets a new UID. - updatedObj, err := ctx.pvRestorer.executePVAction(obj) - if err != nil { - return nil, fmt.Errorf("error executing PVAction for %s: %v", resourceID, err) - } - - return updatedObj, nil + return obj, nil }