diff --git a/changelogs/unreleased/10513-ywk253100 b/changelogs/unreleased/10513-ywk253100 new file mode 100644 index 000000000..e5cbabbf7 --- /dev/null +++ b/changelogs/unreleased/10513-ywk253100 @@ -0,0 +1 @@ +Update volume info in restore finalizing stage to record info from DataDownload result \ No newline at end of file diff --git a/internal/volume/volumes_information.go b/internal/volume/volumes_information.go index 7f36168dd..559f813fe 100644 --- a/internal/volume/volumes_information.go +++ b/internal/volume/volumes_information.go @@ -350,14 +350,15 @@ func newPodVolumeInfoFromPVB(pvb *velerov1api.PodVolumeBackup) *PodVolumeBackupI func newPodVolumeInfoFromPVR(pvr *velerov1api.PodVolumeRestore) *PodVolumeRestoreInfo { return &PodVolumeRestoreInfo{ - SnapshotHandle: pvr.Spec.SnapshotID, - Size: pvr.Status.Progress.TotalBytes, - UploaderType: pvr.Spec.UploaderType, - VolumeName: pvr.Spec.Volume, - PodName: pvr.Spec.Pod.Name, - PodNamespace: pvr.Spec.Pod.Namespace, - Phase: pvr.Status.Phase, - RestoreType: pvr.Spec.RestoreType, + SnapshotHandle: pvr.Spec.SnapshotID, + Size: pvr.Status.Progress.TotalBytes, + UploaderType: pvr.Spec.UploaderType, + VolumeName: pvr.Spec.Volume, + PodName: pvr.Spec.Pod.Name, + PodNamespace: pvr.Spec.Pod.Namespace, + Phase: pvr.Status.Phase, + RestoreType: pvr.Spec.RestoreType, + IncrementalSize: pvr.Status.IncrementalBytes, } } @@ -995,13 +996,14 @@ func (t *RestoreVolumeInfoTracker) Result() []*RestoreVolumeInfo { // For the same reason, no CSI snapshot info will be populated into volumeInfo RestoreMethod: CSISnapshot, SnapshotDataMovementInfo: &RestoreSnapshotDataMovementInfo{ - DataMover: dataMover, - UploaderType: datamover.GetUploaderType(dataMover), - SnapshotHandle: dd.Spec.SnapshotID, - OperationID: operationID, - RestoreType: dd.Spec.RestoreType, - Size: dd.Status.Progress.TotalBytes, - Phase: dd.Status.Phase, + DataMover: dataMover, + UploaderType: datamover.GetUploaderType(dataMover), + SnapshotHandle: dd.Spec.SnapshotID, + OperationID: operationID, + RestoreType: dd.Spec.RestoreType, + Size: dd.Status.Progress.TotalBytes, + IncrementalSize: dd.Status.IncrementalBytes, + Phase: dd.Status.Phase, }, } diff --git a/internal/volume/volumes_information_test.go b/internal/volume/volumes_information_test.go index 4d31ad725..f5aa44163 100644 --- a/internal/volume/volumes_information_test.go +++ b/internal/volume/volumes_information_test.go @@ -685,6 +685,8 @@ func TestGenerateVolumeInfoFromPVB(t *testing.T) { StartTimestamp(&now). CompletionTimestamp(&now). Phase(velerov1api.PodVolumeBackupPhaseCompleted). + TotalBytes(1024). + IncrementalBytes(512). Result(), pod: builder.ForPod("velero", "testPod").Containers(&corev1api.Container{ Name: "test", @@ -715,9 +717,11 @@ func TestGenerateVolumeInfoFromPVB(t *testing.T) { CompletionTimestamp: &now, Result: VolumeResultSucceeded, PVBInfo: &PodVolumeBackupInfo{ - PodName: "testPod", - PodNamespace: "velero", - Phase: velerov1api.PodVolumeBackupPhaseCompleted, + PodName: "testPod", + PodNamespace: "velero", + Phase: velerov1api.PodVolumeBackupPhaseCompleted, + Size: 1024, + IncrementalSize: ptr.To(int64(512)), }, PVInfo: &PVInfo{ ReclaimPolicy: string(corev1api.PersistentVolumeReclaimDelete), @@ -1076,7 +1080,12 @@ func TestRestoreVolumeInfoResult(t *testing.T) { PodName("testPod"). Volume("data-volume-1"). UploaderType("kopia"). - SnapshotID("pvr-snap-001").Result(), + SnapshotID("pvr-snap-001"). + Phase(velerov1api.PodVolumeRestorePhaseCompleted). + RestoreType("Incremental"). + TotalBytes(1024). + IncrementalBytes(512). + Result(), }, }, expectResultValues: []RestoreVolumeInfo{ @@ -1087,11 +1096,15 @@ func TestRestoreVolumeInfoResult(t *testing.T) { RestoreMethod: PodVolumeRestore, SnapshotDataMoved: false, PVRInfo: &PodVolumeRestoreInfo{ - SnapshotHandle: "pvr-snap-001", - PodName: "testPod", - PodNamespace: "testNS", - UploaderType: "kopia", - VolumeName: "data-volume-1", + SnapshotHandle: "pvr-snap-001", + PodName: "testPod", + PodNamespace: "testNS", + UploaderType: "kopia", + VolumeName: "data-volume-1", + Phase: velerov1api.PodVolumeRestorePhaseCompleted, + RestoreType: "Incremental", + Size: 1024, + IncrementalSize: ptr.To(int64(512)), }, }, { @@ -1213,6 +1226,7 @@ func TestRestoreVolumeInfoResult(t *testing.T) { }). Phase(velerov2alpha1.DataDownloadPhaseCompleted). Progress(veleroshared.DataMoveOperationProgress{TotalBytes: 2048}). + IncrementalBytes(512). RestoreType("Incremental"). Result(), *builder.ForDataDownload("velero", "testDataDownload-2"). @@ -1238,13 +1252,14 @@ func TestRestoreVolumeInfoResult(t *testing.T) { RestoreMethod: CSISnapshot, SnapshotDataMoved: true, SnapshotDataMovementInfo: &RestoreSnapshotDataMovementInfo{ - DataMover: "velero", - UploaderType: velerov1api.BackupRepositoryTypeKopia, - SnapshotHandle: "dd-snap-001", - OperationID: "dd-operation-001", - Phase: velerov2alpha1.DataDownloadPhaseCompleted, - Size: 2048, - RestoreType: "Incremental", + DataMover: "velero", + UploaderType: velerov1api.BackupRepositoryTypeKopia, + SnapshotHandle: "dd-snap-001", + OperationID: "dd-operation-001", + Phase: velerov2alpha1.DataDownloadPhaseCompleted, + Size: 2048, + IncrementalSize: ptr.To(int64(512)), + RestoreType: "Incremental", }, }, { @@ -1466,6 +1481,7 @@ func TestRestoreVolumeInfoJSONRoundTrip(t *testing.T) { assert.Contains(t, jsonStr, `"podName":"pod-2"`) assert.Contains(t, jsonStr, `"podNamespace":"ns-2"`) assert.Contains(t, jsonStr, `"nodeName":"node-2"`) + assert.Contains(t, jsonStr, `"incrementalSize":60`) assert.Contains(t, jsonStr, `"csiSnapshotInfo":{`) assert.Contains(t, jsonStr, `"nativeSnapshotInfo":{`) @@ -1474,3 +1490,121 @@ func TestRestoreVolumeInfoJSONRoundTrip(t *testing.T) { require.NoError(t, err) assert.Equal(t, orig, unmarshaled) } + +func TestNewPodVolumeInfoFromPVR(t *testing.T) { + tests := []struct { + name string + pvr *velerov1api.PodVolumeRestore + expected *PodVolumeRestoreInfo + }{ + { + name: "all fields populated including incremental bytes and restore type", + pvr: builder.ForPodVolumeRestore("velero", "pvr-1"). + SnapshotID("snap-1"). + Volume("vol-1"). + PodName("pod-1"). + PodNamespace("ns-1"). + UploaderType("kopia"). + Phase(velerov1api.PodVolumeRestorePhaseCompleted). + RestoreType("Incremental"). + TotalBytes(2048). + IncrementalBytes(512). + Result(), + expected: &PodVolumeRestoreInfo{ + SnapshotHandle: "snap-1", + Size: 2048, + IncrementalSize: ptr.To(int64(512)), + UploaderType: "kopia", + VolumeName: "vol-1", + PodName: "pod-1", + PodNamespace: "ns-1", + Phase: velerov1api.PodVolumeRestorePhaseCompleted, + RestoreType: "Incremental", + }, + }, + { + name: "optional fields empty or nil", + pvr: builder.ForPodVolumeRestore("velero", "pvr-2"). + SnapshotID("snap-2"). + Volume("vol-2"). + PodName("pod-2"). + PodNamespace("ns-2"). + UploaderType("restic"). + Result(), + expected: &PodVolumeRestoreInfo{ + SnapshotHandle: "snap-2", + Size: 0, + UploaderType: "restic", + VolumeName: "vol-2", + PodName: "pod-2", + PodNamespace: "ns-2", + }, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + actual := newPodVolumeInfoFromPVR(tc.pvr) + assert.Equal(t, tc.expected, actual) + }) + } +} + +func TestNewPodVolumeInfoFromPVB(t *testing.T) { + tests := []struct { + name string + pvb *velerov1api.PodVolumeBackup + expected *PodVolumeBackupInfo + }{ + { + name: "all fields populated including incremental bytes", + pvb: builder.ForPodVolumeBackup("velero", "pvb-1"). + SnapshotID("snap-1"). + Volume("vol-1"). + PodName("pod-1"). + PodNamespace("ns-1"). + Node("node-1"). + UploaderType("kopia"). + Phase(velerov1api.PodVolumeBackupPhaseCompleted). + TotalBytes(2048). + IncrementalBytes(512). + Result(), + expected: &PodVolumeBackupInfo{ + SnapshotHandle: "snap-1", + Size: 2048, + IncrementalSize: ptr.To(int64(512)), + UploaderType: "kopia", + VolumeName: "vol-1", + PodName: "pod-1", + PodNamespace: "ns-1", + NodeName: "node-1", + Phase: velerov1api.PodVolumeBackupPhaseCompleted, + }, + }, + { + name: "optional fields empty or nil", + pvb: builder.ForPodVolumeBackup("velero", "pvb-2"). + SnapshotID("snap-2"). + Volume("vol-2"). + PodName("pod-2"). + PodNamespace("ns-2"). + UploaderType("restic"). + Result(), + expected: &PodVolumeBackupInfo{ + SnapshotHandle: "snap-2", + Size: 0, + UploaderType: "restic", + VolumeName: "vol-2", + PodName: "pod-2", + PodNamespace: "ns-2", + }, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + actual := newPodVolumeInfoFromPVB(tc.pvb) + assert.Equal(t, tc.expected, actual) + }) + } +} diff --git a/pkg/builder/data_download_builder.go b/pkg/builder/data_download_builder.go index cd0d15837..ed05303c6 100644 --- a/pkg/builder/data_download_builder.go +++ b/pkg/builder/data_download_builder.go @@ -148,6 +148,18 @@ func (d *DataDownloadBuilder) Progress(progress shared.DataMoveOperationProgress return d } +// TotalBytes sets the DataDownload's TotalBytes. +func (d *DataDownloadBuilder) TotalBytes(totalBytes int64) *DataDownloadBuilder { + d.object.Status.Progress.TotalBytes = totalBytes + return d +} + +// IncrementalBytes sets the DataDownload's IncrementalBytes. +func (d *DataDownloadBuilder) IncrementalBytes(incrementalBytes int64) *DataDownloadBuilder { + d.object.Status.IncrementalBytes = &incrementalBytes + return d +} + // Node sets the DataDownload's Node. func (d *DataDownloadBuilder) Node(node string) *DataDownloadBuilder { d.object.Status.Node = node diff --git a/pkg/builder/pod_volume_backup_builder.go b/pkg/builder/pod_volume_backup_builder.go index 9db8beacd..ae24da9a2 100644 --- a/pkg/builder/pod_volume_backup_builder.go +++ b/pkg/builder/pod_volume_backup_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" velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1" ) @@ -62,6 +63,24 @@ func (b *PodVolumeBackupBuilder) Phase(phase velerov1api.PodVolumeBackupPhase) * return b } +// Progress sets the PodVolumeBackup's Progress. +func (b *PodVolumeBackupBuilder) Progress(progress shared.DataMoveOperationProgress) *PodVolumeBackupBuilder { + b.object.Status.Progress = progress + return b +} + +// TotalBytes sets the PodVolumeBackup's TotalBytes. +func (b *PodVolumeBackupBuilder) TotalBytes(totalBytes int64) *PodVolumeBackupBuilder { + b.object.Status.Progress.TotalBytes = totalBytes + return b +} + +// IncrementalBytes sets the PodVolumeBackup's IncrementalBytes. +func (b *PodVolumeBackupBuilder) IncrementalBytes(incrementalBytes int64) *PodVolumeBackupBuilder { + b.object.Status.IncrementalBytes = &incrementalBytes + return b +} + // Node sets the PodVolumeBackup's node name. func (b *PodVolumeBackupBuilder) Node(name string) *PodVolumeBackupBuilder { b.object.Spec.Node = name diff --git a/pkg/builder/pod_volume_restore_builder.go b/pkg/builder/pod_volume_restore_builder.go index 50f277eac..1df3100f0 100644 --- a/pkg/builder/pod_volume_restore_builder.go +++ b/pkg/builder/pod_volume_restore_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" velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1" ) @@ -62,6 +63,30 @@ func (b *PodVolumeRestoreBuilder) Phase(phase velerov1api.PodVolumeRestorePhase) return b } +// RestoreType sets the PodVolumeRestore's RestoreType. +func (b *PodVolumeRestoreBuilder) RestoreType(restoreType string) *PodVolumeRestoreBuilder { + b.object.Spec.RestoreType = restoreType + return b +} + +// Progress sets the PodVolumeRestore's Progress. +func (b *PodVolumeRestoreBuilder) Progress(progress shared.DataMoveOperationProgress) *PodVolumeRestoreBuilder { + b.object.Status.Progress = progress + return b +} + +// TotalBytes sets the PodVolumeRestore's TotalBytes. +func (b *PodVolumeRestoreBuilder) TotalBytes(totalBytes int64) *PodVolumeRestoreBuilder { + b.object.Status.Progress.TotalBytes = totalBytes + return b +} + +// IncrementalBytes sets the PodVolumeRestore's IncrementalBytes. +func (b *PodVolumeRestoreBuilder) IncrementalBytes(incrementalBytes int64) *PodVolumeRestoreBuilder { + b.object.Status.IncrementalBytes = &incrementalBytes + return b +} + // BackupStorageLocation sets the PodVolumeRestore's backup storage location. func (b *PodVolumeRestoreBuilder) BackupStorageLocation(name string) *PodVolumeRestoreBuilder { b.object.Spec.BackupStorageLocation = name diff --git a/pkg/builder/volume_builder_test.go b/pkg/builder/volume_builder_test.go new file mode 100644 index 000000000..6b3291c2d --- /dev/null +++ b/pkg/builder/volume_builder_test.go @@ -0,0 +1,83 @@ +/* +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 ( + "testing" + + "github.com/stretchr/testify/assert" + + "github.com/vmware-tanzu/velero/pkg/apis/velero/shared" + velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1" +) + +func TestDataDownloadBuilder_Bytes(t *testing.T) { + dd := ForDataDownload("velero", "dd-1"). + TotalBytes(2048). + IncrementalBytes(512). + Result() + + assert.Equal(t, int64(2048), dd.Status.Progress.TotalBytes) + assert.NotNil(t, dd.Status.IncrementalBytes) + assert.Equal(t, int64(512), *dd.Status.IncrementalBytes) +} + +func TestPodVolumeBackupBuilder_ProgressAndBytes(t *testing.T) { + pvb1 := ForPodVolumeBackup("velero", "pvb-1"). + Progress(shared.DataMoveOperationProgress{ + TotalBytes: 4096, + BytesDone: 2048, + }). + Phase(velerov1api.PodVolumeBackupPhaseCompleted). + Result() + + assert.Equal(t, int64(4096), pvb1.Status.Progress.TotalBytes) + assert.Equal(t, int64(2048), pvb1.Status.Progress.BytesDone) + + pvb2 := ForPodVolumeBackup("velero", "pvb-2"). + TotalBytes(1024). + IncrementalBytes(256). + Result() + + assert.Equal(t, int64(1024), pvb2.Status.Progress.TotalBytes) + assert.NotNil(t, pvb2.Status.IncrementalBytes) + assert.Equal(t, int64(256), *pvb2.Status.IncrementalBytes) +} + +func TestPodVolumeRestoreBuilder_ProgressAndBytes(t *testing.T) { + pvr1 := ForPodVolumeRestore("velero", "pvr-1"). + RestoreType("Incremental"). + Progress(shared.DataMoveOperationProgress{ + TotalBytes: 8192, + BytesDone: 4096, + }). + Phase(velerov1api.PodVolumeRestorePhaseCompleted). + Result() + + assert.Equal(t, "Incremental", pvr1.Spec.RestoreType) + assert.Equal(t, int64(8192), pvr1.Status.Progress.TotalBytes) + assert.Equal(t, int64(4096), pvr1.Status.Progress.BytesDone) + + pvr2 := ForPodVolumeRestore("velero", "pvr-2"). + TotalBytes(2048). + IncrementalBytes(512). + Result() + + assert.Equal(t, int64(2048), pvr2.Status.Progress.TotalBytes) + assert.NotNil(t, pvr2.Status.IncrementalBytes) + assert.Equal(t, int64(512), *pvr2.Status.IncrementalBytes) +} diff --git a/pkg/controller/restore_finalizer_controller.go b/pkg/controller/restore_finalizer_controller.go index d9acd0a09..c423b42c4 100644 --- a/pkg/controller/restore_finalizer_controller.go +++ b/pkg/controller/restore_finalizer_controller.go @@ -17,7 +17,10 @@ limitations under the License. package controller import ( + "bytes" + "compress/gzip" "context" + "encoding/json" "fmt" "sync" "time" @@ -39,6 +42,7 @@ import ( "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" serverconfig "github.com/vmware-tanzu/velero/pkg/cmd/server/config" "github.com/vmware-tanzu/velero/pkg/constant" "github.com/vmware-tanzu/velero/pkg/itemoperation" @@ -148,10 +152,16 @@ func (r *restoreFinalizerReconciler) Reconcile(ctx context.Context, req ctrl.Req return ctrl.Result{}, errors.Wrap(err, "error getting backup store") } - volumeInfo, err := backupStore.GetBackupVolumeInfos(restore.Spec.BackupName) + backupVolumeInfos, err := backupStore.GetBackupVolumeInfos(restore.Spec.BackupName) if err != nil { log.WithError(err).Errorf("error getting volumeInfo for backup %s", restore.Spec.BackupName) - return ctrl.Result{}, errors.Wrap(err, "error getting volumeInfo") + return ctrl.Result{}, errors.Wrap(err, "error getting backup volumeInfos") + } + + restoreVolumeInfos, err := backupStore.GetRestoreVolumeInfos(restore.Name) + if err != nil { + log.WithError(err).Errorf("error getting volumeInfos for restore %s", restore.Name) + return ctrl.Result{}, errors.Wrap(err, "error getting restore volumeInfos") } restoredResourceList, err := backupStore.GetRestoredResourceList(restore.Name) @@ -169,13 +179,15 @@ func (r *restoreFinalizerReconciler) Reconcile(ctx context.Context, req ctrl.Req } finalizerCtx := &finalizerContext{ - logger: log, - restore: restore, - crClient: r.crClient, - volumeInfo: volumeInfo, - restoredPVCList: restoredPVCList, - multiHookTracker: r.multiHookTracker, - resourceTimeout: r.resourceTimeout, + logger: log, + backupStore: backupStore, + restore: restore, + crClient: r.crClient, + backupVolumeInfos: backupVolumeInfos, + restoreVolumeInfos: restoreVolumeInfos, + restoredPVCList: restoredPVCList, + multiHookTracker: r.multiHookTracker, + resourceTimeout: r.resourceTimeout, restoreItemOperationList: restoreItemOperationList{ items: restoreItemOperations, }, @@ -288,7 +300,9 @@ type finalizerContext struct { logger logrus.FieldLogger restore *velerov1api.Restore crClient client.Client - volumeInfo []*volume.BackupVolumeInfo + backupStore persistence.BackupStore + backupVolumeInfos []*volume.BackupVolumeInfo + restoreVolumeInfos []*volume.RestoreVolumeInfo restoredPVCList map[string]struct{} restoreItemOperationList restoreItemOperationList multiHookTracker *hook.MultiHookTracker @@ -307,6 +321,9 @@ func (ctx *finalizerContext) execute() (results.Result, results.Result) { warnings.Merge(&vgscWarnings) } + viErrs := ctx.updateVolumeInfos() + errs.Merge(&viErrs) + rehErrs := ctx.WaitRestoreExecHook() errs.Merge(&rehErrs) @@ -324,7 +341,7 @@ func (ctx *finalizerContext) patchDynamicPVWithVolumeInfo() (errs results.Result maxConcurrency := 3 semaphore := make(chan struct{}, maxConcurrency) - for _, volumeItem := range ctx.volumeInfo { + for _, volumeItem := range ctx.backupVolumeInfos { if (volumeItem.BackupMethod == volume.PodVolumeBackup || volumeItem.BackupMethod == volume.CSISnapshot) && volumeItem.PVInfo != nil { // Determine restored PVC namespace restoredNamespace := volumeItem.PVCNamespace @@ -463,7 +480,7 @@ func (ctx *finalizerContext) patchDynamicPVWithVolumeInfo() (errs results.Result } func (ctx *finalizerContext) hasVolumeGroupSnapshotHandles() bool { - for _, vi := range ctx.volumeInfo { + for _, vi := range ctx.backupVolumeInfos { if vi.CSISnapshotInfo != nil && vi.CSISnapshotInfo.VolumeGroupSnapshotHandle != "" { return true } @@ -573,6 +590,43 @@ func needPatch(newPV *corev1api.PersistentVolume, pvInfo *volume.PVInfo) bool { return false } +func (ctx *finalizerContext) updateVolumeInfos() (errs results.Result) { + dataDownloads := &velerov2alpha1.DataDownloadList{} + if err := ctx.crClient.List(context.Background(), dataDownloads, client.InNamespace(ctx.restore.Namespace), client.MatchingLabels{velerov1api.RestoreNameLabel: ctx.restore.Name}); err != nil { + errs.Add("cluster", errors.Wrapf(err, "failed to list data downloads of restore %s", ctx.restore.Name)) + return errs + } + for _, dataDownload := range dataDownloads.Items { + for index := range ctx.restoreVolumeInfos { + if ctx.restoreVolumeInfos[index].PVCName == dataDownload.Spec.TargetVolume.PVC && + ctx.restoreVolumeInfos[index].PVCNamespace == dataDownload.Spec.TargetVolume.Namespace && + ctx.restoreVolumeInfos[index].SnapshotDataMovementInfo != nil { + ctx.restoreVolumeInfos[index].SnapshotDataMovementInfo.Size = dataDownload.Status.Progress.TotalBytes + ctx.restoreVolumeInfos[index].SnapshotDataMovementInfo.IncrementalSize = dataDownload.Status.IncrementalBytes + ctx.restoreVolumeInfos[index].SnapshotDataMovementInfo.Phase = dataDownload.Status.Phase + } + } + } + + buffer := new(bytes.Buffer) + gzw := gzip.NewWriter(buffer) + defer gzw.Close() + if err := json.NewEncoder(gzw).Encode(ctx.restoreVolumeInfos); err != nil { + errs.Add("cluster", errors.Wrapf(err, "error encoding restore volume infos to JSON for restore %s", ctx.restore.Name)) + return errs + } + if err := gzw.Close(); err != nil { + errs.Add("cluster", errors.Wrapf(err, "error closing gzip writer for restore %s", ctx.restore.Name)) + return errs + } + if err := ctx.backupStore.PutRestoreVolumeInfo(ctx.restore.Name, buffer); err != nil { + errs.Add("cluster", errors.Wrapf(err, "failed to put restore volume info for restore %s", ctx.restore.Name)) + return errs + } + + return errs +} + // WaitRestoreExecHook waits for restore exec hooks to finish then update the hook execution results func (ctx *finalizerContext) WaitRestoreExecHook() (errs results.Result) { log := ctx.logger.WithField("restore", ctx.restore.Name) diff --git a/pkg/controller/restore_finalizer_controller_test.go b/pkg/controller/restore_finalizer_controller_test.go index 226a2283c..e2695b0bf 100644 --- a/pkg/controller/restore_finalizer_controller_test.go +++ b/pkg/controller/restore_finalizer_controller_test.go @@ -17,7 +17,13 @@ limitations under the License. package controller import ( + "bytes" + "compress/gzip" + "context" + "encoding/json" + "errors" "fmt" + "io" "syscall" "testing" "time" @@ -33,12 +39,15 @@ import ( "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" testclocks "k8s.io/utils/clock/testing" + "k8s.io/utils/ptr" ctrl "sigs.k8s.io/controller-runtime" crclient "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" "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/builder" "github.com/vmware-tanzu/velero/pkg/itemoperation" "github.com/vmware-tanzu/velero/pkg/metrics" @@ -61,16 +70,17 @@ func TestRestoreFinalizerReconcile(t *testing.T) { assert.NotNil(t, timestamp) rfrTests := []struct { - name string - restore *velerov1api.Restore - backup *velerov1api.Backup - location *velerov1api.BackupStorageLocation - expectError bool - expectPhase velerov1api.RestorePhase - expectWarningsCnt int - expectErrsCnt int - statusCompare bool - expectedCompletedTime *metav1.Time + name string + restore *velerov1api.Restore + backup *velerov1api.Backup + location *velerov1api.BackupStorageLocation + expectError bool + expectPhase velerov1api.RestorePhase + expectWarningsCnt int + expectErrsCnt int + statusCompare bool + expectedCompletedTime *metav1.Time + getRestoreVolumeInfosErr error }{ { name: "Restore is not awaiting finalization, skip", @@ -114,6 +124,15 @@ func TestRestoreFinalizerReconcile(t *testing.T) { expectError: false, statusCompare: false, }, + { + name: "Fail to get restore volume infos from backup store", + restore: builder.ForRestore(velerov1api.DefaultNamespace, "restore-1").Phase(velerov1api.RestorePhaseFinalizing).Backup("backup-1").Result(), + backup: defaultBackup().StorageLocation("default").Result(), + location: defaultStorageLocation, + expectError: true, + statusCompare: false, + getRestoreVolumeInfosErr: errors.New("failed to get restore volume infos"), + }, } for _, test := range rfrTests { @@ -149,8 +168,14 @@ func TestRestoreFinalizerReconcile(t *testing.T) { if test.restore != nil && test.restore.Namespace == velerov1api.DefaultNamespace { require.NoError(t, r.Client.Create(t.Context(), test.restore)) - backupStore.On("GetRestoredResourceList", test.restore.Name).Return(map[string][]string{}, nil) - backupStore.On("GetRestoreItemOperations", test.restore.Name).Return([]*itemoperation.RestoreOperation{}, nil) + if test.getRestoreVolumeInfosErr != nil { + backupStore.On("GetRestoreVolumeInfos", test.restore.Name).Return(nil, test.getRestoreVolumeInfosErr) + } else { + backupStore.On("GetRestoreVolumeInfos", test.restore.Name).Return([]*volume.RestoreVolumeInfo{}, nil) + backupStore.On("GetRestoredResourceList", test.restore.Name).Return(map[string][]string{}, nil) + backupStore.On("GetRestoreItemOperations", test.restore.Name).Return([]*itemoperation.RestoreOperation{}, nil) + backupStore.On("PutRestoreVolumeInfo", test.restore.Name, mock.Anything).Return(nil) + } } if test.backup != nil { require.NoError(t, r.Client.Create(t.Context(), test.backup)) @@ -437,11 +462,11 @@ func TestPatchDynamicPVWithVolumeInfo(t *testing.T) { logger = velerotest.NewLogger() ) ctx := &finalizerContext{ - logger: logger, - crClient: fakeClient, - restore: tc.restore, - restoredPVCList: tc.restoredPVCNames, - volumeInfo: tc.volumeInfo, + logger: logger, + crClient: fakeClient, + restore: tc.restore, + restoredPVCList: tc.restoredPVCNames, + backupVolumeInfos: tc.volumeInfo, } for _, pv := range tc.restoredPV { @@ -915,7 +940,7 @@ func TestHasVolumeGroupSnapshotHandles(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { ctx := &finalizerContext{ - volumeInfo: tc.volumeInfo, + backupVolumeInfos: tc.volumeInfo, } assert.Equal(t, tc.expected, ctx.hasVolumeGroupSnapshotHandles()) }) @@ -1171,3 +1196,175 @@ func TestCleanupStubVGSC(t *testing.T) { }) } } + +func TestUpdateVolumeInfos(t *testing.T) { + tests := []struct { + name string + restore *velerov1api.Restore + restoreVolumeInfos []*volume.RestoreVolumeInfo + dataDownloads []*velerov2alpha1.DataDownload + listErr error + putErr error + expectedSize int64 + expectedIncrSize *int64 + expectedPhase velerov2alpha1.DataDownloadPhase + expectErrs bool + expectErrMsg string + }{ + { + name: "successful update of restore volume infos from data downloads", + restore: builder.ForRestore("velero", "restore-1").Result(), + restoreVolumeInfos: []*volume.RestoreVolumeInfo{ + { + PVCName: "pvc-1", + PVCNamespace: "ns-1", + SnapshotDataMovementInfo: &volume.RestoreSnapshotDataMovementInfo{ + DataMover: "velero", + Size: 0, + Phase: "", + }, + }, + { + PVCName: "pvc-2", + PVCNamespace: "ns-2", + SnapshotDataMovementInfo: nil, + }, + { + PVCName: "pvc-3", + PVCNamespace: "ns-3", + SnapshotDataMovementInfo: &volume.RestoreSnapshotDataMovementInfo{ + DataMover: "velero", + Size: 100, + Phase: velerov2alpha1.DataDownloadPhaseCompleted, + }, + }, + }, + dataDownloads: []*velerov2alpha1.DataDownload{ + builder.ForDataDownload("velero", "dd-1"). + ObjectMeta(builder.WithLabelsMap(map[string]string{velerov1api.RestoreNameLabel: "restore-1"})). + TargetVolume(velerov2alpha1.TargetVolumeSpec{PVC: "pvc-1", Namespace: "ns-1"}). + TotalBytes(4096). + IncrementalBytes(1024). + Phase(velerov2alpha1.DataDownloadPhaseCompleted). + Result(), + builder.ForDataDownload("velero", "dd-2"). + ObjectMeta(builder.WithLabelsMap(map[string]string{velerov1api.RestoreNameLabel: "restore-1"})). + TargetVolume(velerov2alpha1.TargetVolumeSpec{PVC: "pvc-2", Namespace: "ns-2"}). + TotalBytes(2048). + IncrementalBytes(512). + Phase(velerov2alpha1.DataDownloadPhaseCompleted). + Result(), + builder.ForDataDownload("velero", "dd-other-restore"). + ObjectMeta(builder.WithLabelsMap(map[string]string{velerov1api.RestoreNameLabel: "restore-other"})). + TargetVolume(velerov2alpha1.TargetVolumeSpec{PVC: "pvc-3", Namespace: "ns-3"}). + TotalBytes(9999). + IncrementalBytes(8888). + Phase(velerov2alpha1.DataDownloadPhaseFailed). + Result(), + }, + expectedSize: 4096, + expectedIncrSize: ptr.To(int64(1024)), + expectedPhase: velerov2alpha1.DataDownloadPhaseCompleted, + expectErrs: false, + }, + { + name: "failed to list data downloads", + restore: builder.ForRestore("velero", "restore-1").Result(), + restoreVolumeInfos: []*volume.RestoreVolumeInfo{ + { + PVCName: "pvc-1", + PVCNamespace: "ns-1", + SnapshotDataMovementInfo: &volume.RestoreSnapshotDataMovementInfo{ + DataMover: "velero", + }, + }, + }, + listErr: errors.New("list error"), + expectErrs: true, + expectErrMsg: "failed to list data downloads of restore restore-1", + }, + { + name: "failed to put restore volume info to backup store", + restore: builder.ForRestore("velero", "restore-1").Result(), + restoreVolumeInfos: []*volume.RestoreVolumeInfo{ + { + PVCName: "pvc-1", + PVCNamespace: "ns-1", + SnapshotDataMovementInfo: &volume.RestoreSnapshotDataMovementInfo{ + DataMover: "velero", + }, + }, + }, + putErr: errors.New("put error"), + expectErrs: true, + expectErrMsg: "failed to put restore volume info for restore restore-1", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + clientBuilder := velerotest.NewFakeControllerRuntimeClientBuilder(t) + if tc.listErr != nil { + clientBuilder = clientBuilder.WithInterceptorFuncs(interceptor.Funcs{ + List: func(ctx context.Context, client crclient.WithWatch, list crclient.ObjectList, opts ...crclient.ListOption) error { + return tc.listErr + }, + }) + } + fakeClient := clientBuilder.Build() + + for _, dd := range tc.dataDownloads { + require.NoError(t, fakeClient.Create(t.Context(), dd)) + } + + backupStore := &persistencemocks.BackupStore{} + var uploadedData []byte + if tc.listErr == nil { + if tc.putErr != nil { + backupStore.On("PutRestoreVolumeInfo", tc.restore.Name, mock.Anything).Return(tc.putErr) + } else { + backupStore.On("PutRestoreVolumeInfo", tc.restore.Name, mock.Anything).Run(func(args mock.Arguments) { + reader, ok := args.Get(1).(io.Reader) + require.True(t, ok) + data, err := io.ReadAll(reader) + require.NoError(t, err) + uploadedData = data + }).Return(nil) + } + } + + ctx := &finalizerContext{ + logger: velerotest.NewLogger(), + restore: tc.restore, + crClient: fakeClient, + backupStore: backupStore, + restoreVolumeInfos: tc.restoreVolumeInfos, + } + + errs := ctx.updateVolumeInfos() + if tc.expectErrs { + assert.False(t, errs.IsEmpty()) + assert.Contains(t, errs.Namespaces["cluster"][0], tc.expectErrMsg) + } else { + assert.True(t, errs.IsEmpty()) + assert.Equal(t, tc.expectedSize, ctx.restoreVolumeInfos[0].SnapshotDataMovementInfo.Size) + assert.Equal(t, tc.expectedIncrSize, ctx.restoreVolumeInfos[0].SnapshotDataMovementInfo.IncrementalSize) + assert.Equal(t, tc.expectedPhase, ctx.restoreVolumeInfos[0].SnapshotDataMovementInfo.Phase) + // pvc-2 had nil SnapshotDataMovementInfo and should remain nil + assert.Nil(t, ctx.restoreVolumeInfos[1].SnapshotDataMovementInfo) + // pvc-3 belonged to another restore and should be untouched + assert.Equal(t, int64(100), ctx.restoreVolumeInfos[2].SnapshotDataMovementInfo.Size) + + // Verify the content uploaded to backup store can be decoded and matches + require.NotEmpty(t, uploadedData) + gzr, err := gzip.NewReader(bytes.NewReader(uploadedData)) + require.NoError(t, err) + defer gzr.Close() + + var decoded []*volume.RestoreVolumeInfo + require.NoError(t, json.NewDecoder(gzr).Decode(&decoded)) + assert.Equal(t, ctx.restoreVolumeInfos, decoded) + } + }) + } +} diff --git a/pkg/persistence/mocks/backup_store.go b/pkg/persistence/mocks/backup_store.go index b9c4184c3..f506b1d05 100644 --- a/pkg/persistence/mocks/backup_store.go +++ b/pkg/persistence/mocks/backup_store.go @@ -461,6 +461,36 @@ func (_m *BackupStore) GetRestoreResults(name string) (map[string]results.Result return r0, r1 } +// GetRestoreVolumeInfos provides a mock function with given fields: name +func (_m *BackupStore) GetRestoreVolumeInfos(name string) ([]*volume.RestoreVolumeInfo, error) { + ret := _m.Called(name) + + if len(ret) == 0 { + panic("no return value specified for GetRestoreVolumeInfos") + } + + var r0 []*volume.RestoreVolumeInfo + var r1 error + if rf, ok := ret.Get(0).(func(string) ([]*volume.RestoreVolumeInfo, error)); ok { + return rf(name) + } + if rf, ok := ret.Get(0).(func(string) []*volume.RestoreVolumeInfo); ok { + r0 = rf(name) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).([]*volume.RestoreVolumeInfo) + } + } + + if rf, ok := ret.Get(1).(func(string) error); ok { + r1 = rf(name) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + // GetRestoredResourceList provides a mock function with given fields: name func (_m *BackupStore) GetRestoredResourceList(name string) (map[string][]string, error) { ret := _m.Called(name) diff --git a/pkg/persistence/object_store.go b/pkg/persistence/object_store.go index 338ee016e..dfd86dc5d 100644 --- a/pkg/persistence/object_store.go +++ b/pkg/persistence/object_store.go @@ -85,6 +85,7 @@ type BackupStore interface { PutRestoredResourceList(restore string, results io.Reader) error PutRestoreItemOperations(restore string, restoreItemOperations io.Reader) error GetRestoreItemOperations(name string) ([]*itemoperation.RestoreOperation, error) + GetRestoreVolumeInfos(name string) ([]*volume.RestoreVolumeInfo, error) PutRestoreVolumeInfo(restore string, volumeInfo io.Reader) error DeleteRestore(name string) error GetRestoredResourceList(name string) (map[string][]string, error) @@ -628,6 +629,24 @@ func (s *objectBackupStore) PutRestoreVolumeInfo(restore string, volumeInfo io.R return seekAndPutObject(s.objectStore, s.bucket, s.layout.getRestoreVolumeInfoKey(restore), volumeInfo) } +func (s *objectBackupStore) GetRestoreVolumeInfos(name string) ([]*volume.RestoreVolumeInfo, error) { + volumeInfos := make([]*volume.RestoreVolumeInfo, 0) + + res, err := tryGet(s.objectStore, s.bucket, s.layout.getRestoreVolumeInfoKey(name)) + if err != nil { + return volumeInfos, err + } + if res == nil { + return volumeInfos, nil + } + defer res.Close() + + if err := decode(res, &volumeInfos); err != nil { + return volumeInfos, err + } + + return volumeInfos, nil +} func (s *objectBackupStore) PutBackupItemOperations(backup string, backupItemOperations io.Reader) error { return seekAndPutObject(s.objectStore, s.bucket, s.layout.getBackupItemOperationsKey(backup), backupItemOperations) } diff --git a/pkg/persistence/object_store_test.go b/pkg/persistence/object_store_test.go index e9a3bde36..ebe1c71c2 100644 --- a/pkg/persistence/object_store_test.go +++ b/pkg/persistence/object_store_test.go @@ -32,6 +32,7 @@ import ( "github.com/stretchr/testify/require" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/utils/ptr" "github.com/vmware-tanzu/velero/internal/credentials" "github.com/vmware-tanzu/velero/internal/volume" @@ -1155,6 +1156,98 @@ func TestGetBackupVolumeInfos(t *testing.T) { }) } } + +func TestGetRestoreVolumeInfos(t *testing.T) { + tests := []struct { + name string + volumeInfo []*volume.RestoreVolumeInfo + volumeInfoStr string + expectedErr string + expectedResult []*volume.RestoreVolumeInfo + }{ + { + name: "No VolumeInfos, expect no error.", + }, + { + name: "Valid RestoreVolumeInfo, should pass.", + volumeInfo: []*volume.RestoreVolumeInfo{ + { + PVCName: "pvcName", + PVCNamespace: "pvcNamespace", + RestoreMethod: volume.CSISnapshot, + SnapshotDataMovementInfo: &volume.RestoreSnapshotDataMovementInfo{ + DataMover: "velero", + Size: 1024, + IncrementalSize: ptr.To(int64(512)), + Phase: "Completed", + }, + }, + }, + expectedResult: []*volume.RestoreVolumeInfo{ + { + PVCName: "pvcName", + PVCNamespace: "pvcNamespace", + RestoreMethod: volume.CSISnapshot, + SnapshotDataMovementInfo: &volume.RestoreSnapshotDataMovementInfo{ + DataMover: "velero", + Size: 1024, + IncrementalSize: ptr.To(int64(512)), + Phase: "Completed", + }, + }, + }, + }, + { + name: "Invalid RestoreVolumeInfo string, should also pass.", + volumeInfoStr: `[{"abc": "123", "def": "456", "pvcName": "pvcName"}]`, + expectedResult: []*volume.RestoreVolumeInfo{ + { + PVCName: "pvcName", + }, + }, + }, + } + + harness := newObjectBackupStoreTestHarness("test-bucket", "") + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if tc.volumeInfo != nil { + obj := new(bytes.Buffer) + gzw := gzip.NewWriter(obj) + + require.NoError(t, json.NewEncoder(gzw).Encode(tc.volumeInfo)) + require.NoError(t, gzw.Close()) + harness.objectStore.PutObject(harness.bucket, "restores/test-restore/test-restore-volumeinfo.json.gz", obj) + } + + if tc.volumeInfoStr != "" { + obj := new(bytes.Buffer) + gzw := gzip.NewWriter(obj) + _, err := gzw.Write([]byte(tc.volumeInfoStr)) + require.NoError(t, err) + + require.NoError(t, gzw.Close()) + harness.objectStore.PutObject(harness.bucket, "restores/test-restore/test-restore-volumeinfo.json.gz", obj) + } + + result, err := harness.GetRestoreVolumeInfos("test-restore") + if tc.expectedErr != "" { + require.Equal(t, tc.expectedErr, err.Error()) + } else { + if err != nil { + fmt.Println(err.Error()) + } + require.NoError(t, err) + } + + if len(tc.expectedResult) > 0 { + require.Equal(t, tc.expectedResult, result) + } + }) + } +} + func TestGetRestoreResults(t *testing.T) { harness := newObjectBackupStoreTestHarness("test-bucket", "") @@ -1258,6 +1351,51 @@ func TestPutBackupVolumeInfos(t *testing.T) { } } +func TestPutRestoreVolumeInfo(t *testing.T) { + tests := []struct { + name string + prefix string + expectedErr string + expectedKeys []string + }{ + { + name: "normal case", + expectedErr: "", + expectedKeys: []string{ + "restores/restore-1/restore-1-volumeinfo.json.gz", + }, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + harness := newObjectBackupStoreTestHarness("foo", tc.prefix) + + volumeInfos := []*volume.RestoreVolumeInfo{ + { + 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.PutRestoreVolumeInfo("restore-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 {