Merge pull request #10513 from ywk253100/cli
Run the E2E test on kind / setup-test-matrix (push) Failing after 4s
Scorecard supply-chain security / Scorecard analysis (push) Skipped
e2e-test-kind.yaml / extract (push) Failing after 8s
Run the E2E test on kind / get-go-version (push) Failing after 9s
Run the E2E test on kind / build (push) Skipped
Run the E2E test on kind / run-e2e-test (push) Skipped
push.yml / extract (push) Failing after 6s
Main CI / get-go-version (push) Failing after 8s
Main CI / Build (push) Skipped

Update volume info in restore finalizing stage
This commit is contained in:
lyndon-li
2026-09-11 16:02:50 +08:00
committed by GitHub
12 changed files with 775 additions and 61 deletions
+1
View File
@@ -0,0 +1 @@
Update volume info in restore finalizing stage to record info from DataDownload result
+17 -15
View File
@@ -357,14 +357,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,
}
}
@@ -1002,13 +1003,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,
},
}
+150 -16
View File
@@ -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)
})
}
}
+12
View File
@@ -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
+19
View File
@@ -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
+25
View File
@@ -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
+83
View File
@@ -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)
}
+66 -12
View File
@@ -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)
@@ -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)
}
})
}
}
+30
View File
@@ -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)
+19
View File
@@ -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)
}
+138
View File
@@ -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 {