mirrors/velero
1
0
Fork 0
mirror of https://github.com/vmware-tanzu/velero.git synced 2026-09-13 11:34:54 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Update volume info in restore finalizing stage

Browse Source 
Update volume info in restore finalizing stage to record info from DataDownload result

Signed-off-by: Wenkai Yin (尹文开) <wenkai.yin@broadcom.com>
This commit is contained in:
Wenkai Yin (尹文开)
2026-09-10 15:04:54 +08:00
parent 0255c6b8bf
commit a3d744db6a
12 changed files with 775 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/controller/restore_finalizer_controller.go
+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)
pkg/controller/restore_finalizer_controller_test.go
+215 -18
View File
@@ -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)
}
})
}
}