Merge pull request #8684 from blackpiglet/7979_fix

7979 fix

pkg/restore/actions/csi/pvc_action.go

@@ -220,8 +220,10 @@ func (p *pvcRestoreItemAction) Execute(
 			logger.Infof("DataDownload %s/%s is created successfully.",
 				dataDownload.Namespace, dataDownload.Name)
 		} else {
-			volumeSnapshotName, ok := pvcFromBackup.Annotations[velerov1api.VolumeSnapshotLabel]
-			if !ok {
+			targetVSName := ""
+			if vsName, nameOK := pvcFromBackup.Annotations[velerov1api.VolumeSnapshotLabel]; nameOK {
+				targetVSName = util.GenerateSha256FromRestoreUIDAndVsName(string(input.Restore.UID), vsName)
+			} else {
 				logger.Info("Skipping PVCRestoreItemAction for PVC,",
 					"PVC does not have a CSI VolumeSnapshot.")
 				// Make no change in the input PVC.
@@ -229,8 +231,9 @@ func (p *pvcRestoreItemAction) Execute(
 					UpdatedItem: input.Item,
 				}, nil
 			}
+
 			if err := restoreFromVolumeSnapshot(
-				&pvc, newNamespace, p.crClient, volumeSnapshotName, logger,
+				&pvc, newNamespace, p.crClient, targetVSName, logger,
 			); err != nil {
 				logger.Errorf("Failed to restore PVC from VolumeSnapshot.")
 				return nil, errors.WithStack(err)
@@ -502,19 +505,17 @@ func restoreFromVolumeSnapshot(
 		},
 		vs,
 	); err != nil {
-		return errors.Wrapf(err,
-			fmt.Sprintf("Failed to get Volumesnapshot %s/%s to restore PVC %s/%s",
-				newNamespace, volumeSnapshotName, newNamespace, pvc.Name),
-		)
+		return errors.Wrapf(err, "Failed to get Volumesnapshot %s/%s to restore PVC %s/%s",
+			newNamespace, volumeSnapshotName, newNamespace, pvc.Name)
 	}
 
 	if _, exists := vs.Annotations[velerov1api.VolumeSnapshotRestoreSize]; exists {
 		restoreSize, err := resource.ParseQuantity(
 			vs.Annotations[velerov1api.VolumeSnapshotRestoreSize])
 		if err != nil {
-			return errors.Wrapf(err, fmt.Sprintf(
+			return errors.Wrapf(err,
 				"Failed to parse %s from annotation on Volumesnapshot %s/%s into restore size",
-				vs.Annotations[velerov1api.VolumeSnapshotRestoreSize], vs.Namespace, vs.Name))
+				vs.Annotations[velerov1api.VolumeSnapshotRestoreSize], vs.Namespace, vs.Name)
 		}
 		// It is possible that the volume provider allocated a larger
 		// capacity volume than what was requested in the backed up PVC.

pkg/restore/actions/csi/pvc_action_test.go

@@ -45,6 +45,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/label"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+	"github.com/vmware-tanzu/velero/pkg/util"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
 )
 
@@ -546,6 +547,7 @@ func TestCancel(t *testing.T) {
 }
 
 func TestExecute(t *testing.T) {
+	vsName := util.GenerateSha256FromRestoreUIDAndVsName("restoreUID", "vsName")
 	tests := []struct {
 		name                 string
 		backup               *velerov1api.Backup
@@ -573,20 +575,22 @@ func TestExecute(t *testing.T) {
 		{
 			name:        "VolumeSnapshot cannot be found",
 			backup:      builder.ForBackup("velero", "testBackup").Result(),
-			restore:     builder.ForRestore("velero", "testRestore").Backup("testBackup").Result(),
-			pvc:         builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "testVS")).Result(),
-			expectedErr: "Failed to get Volumesnapshot velero/testVS to restore PVC velero/testPVC: volumesnapshots.snapshot.storage.k8s.io \"testVS\" not found",
+			restore:     builder.ForRestore("velero", "testRestore").ObjectMeta(builder.WithUID("restoreUID")).Backup("testBackup").Result(),
+			pvc:         builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName")).Result(),
+			expectedErr: fmt.Sprintf("Failed to get Volumesnapshot velero/%s to restore PVC velero/testPVC: volumesnapshots.snapshot.storage.k8s.io \"%s\" not found", vsName, vsName),
 		},
 		{
 			name:    "Restore from VolumeSnapshot",
 			backup:  builder.ForBackup("velero", "testBackup").Result(),
-			restore: builder.ForRestore("velero", "testRestore").Backup("testBackup").Result(),
-			pvc: builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "testVS")).
+			restore: builder.ForRestore("velero", "testRestore").ObjectMeta(builder.WithUID("restoreUID")).Backup("testBackup").Result(),
+			pvc: builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName")).
 				RequestResource(map[corev1api.ResourceName]resource.Quantity{corev1api.ResourceStorage: resource.MustParse("10Gi")}).
 				DataSource(&corev1api.TypedLocalObjectReference{APIGroup: &snapshotv1api.SchemeGroupVersion.Group, Kind: "VolumeSnapshot", Name: "testVS"}).
 				DataSourceRef(&corev1api.TypedObjectReference{APIGroup: &snapshotv1api.SchemeGroupVersion.Group, Kind: "VolumeSnapshot", Name: "testVS"}).
 				Result(),
-			vs:          builder.ForVolumeSnapshot("velero", "testVS").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotRestoreSize, "10Gi")).Result(),
+			vs: builder.ForVolumeSnapshot("velero", vsName).ObjectMeta(
+				builder.WithAnnotations(velerov1api.VolumeSnapshotRestoreSize, "10Gi"),
+			).Result(),
 			expectedPVC: builder.ForPersistentVolumeClaim("velero", "testPVC").Result(),
 		},
 		{

pkg/restore/actions/csi/volumesnapshot_action.go

@@ -17,24 +17,19 @@ limitations under the License.
 package csi
 
 import (
-	"context"
-
 	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
-	core_v1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	crclient "sigs.k8s.io/controller-runtime/pkg/client"
 
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/client"
-	"github.com/vmware-tanzu/velero/pkg/label"
 	plugincommon "github.com/vmware-tanzu/velero/pkg/plugin/framework/common"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	"github.com/vmware-tanzu/velero/pkg/util"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
-	"github.com/vmware-tanzu/velero/pkg/util/csi"
 )
 
 // volumeSnapshotRestoreItemAction is a Velero restore item
@@ -54,8 +49,7 @@ func (p *volumeSnapshotRestoreItemAction) AppliesTo() (
 	}, nil
 }
 
-func resetVolumeSnapshotSpecForRestore(
-	vs *snapshotv1api.VolumeSnapshot, vscName *string) {
+func resetVolumeSnapshotSpecForRestore(vs *snapshotv1api.VolumeSnapshot, vscName *string) {
 	// Spec of the backed-up object used the PVC as the source
 	// of the volumeSnapshot.  Restore operation will however,
 	// restore the VolumeSnapshot from the VolumeSnapshotContent
@@ -68,10 +62,6 @@ func resetVolumeSnapshotAnnotation(vs *snapshotv1api.VolumeSnapshot) {
 		string(snapshotv1api.VolumeSnapshotContentRetain)
 }
 
-// Execute uses the data such as CSI driver name, storage
-// snapshot handle, snapshot deletion secret (if any) from
-// the annotations to recreate a VolumeSnapshotContent object
-// and statically bind the VolumeSnapshot object being restored.
 func (p *volumeSnapshotRestoreItemAction) Execute(
 	input *velero.RestoreItemActionExecuteInput,
 ) (*velero.RestoreItemActionExecuteOutput, error) {
@@ -90,84 +80,29 @@ func (p *volumeSnapshotRestoreItemAction) Execute(
 			errors.Wrapf(err, "failed to convert input.Item from unstructured")
 	}
 
-	// If cross-namespace restore is configured, change the namespace
-	// for VolumeSnapshot object to be restored
-	newNamespace, ok := input.Restore.Spec.NamespaceMapping[vs.GetNamespace()]
-	if !ok {
-		// Use original namespace
-		newNamespace = vs.Namespace
+	var vsFromBackup snapshotv1api.VolumeSnapshot
+	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(
+		input.ItemFromBackup.UnstructuredContent(), &vsFromBackup); err != nil {
+		return &velero.RestoreItemActionExecuteOutput{},
+			errors.Wrapf(err, "failed to convert input.Item from unstructured")
 	}
 
-	if !csi.IsVolumeSnapshotExists(newNamespace, vs.Name, p.crClient) {
-		snapHandle, exists := vs.Annotations[velerov1api.VolumeSnapshotHandleAnnotation]
-		if !exists {
-			return nil, errors.Errorf(
-				"VolumeSnapshot %s/%s does not have a %s annotation",
-				vs.Namespace,
-				vs.Name,
-				velerov1api.VolumeSnapshotHandleAnnotation,
-			)
-		}
+	generatedName := util.GenerateSha256FromRestoreUIDAndVsName(string(input.Restore.UID), vsFromBackup.Name)
 
-		csiDriverName, exists := vs.Annotations[velerov1api.DriverNameAnnotation]
-		if !exists {
-			return nil, errors.Errorf(
-				"VolumeSnapshot %s/%s does not have a %s annotation",
-				vs.Namespace, vs.Name, velerov1api.DriverNameAnnotation)
-		}
+	// Reset Spec to convert the VolumeSnapshot from using
+	// the dynamic VolumeSnapshotContent to the static one.
+	resetVolumeSnapshotSpecForRestore(&vs, &generatedName)
+	// Also reset the VS name to avoid potential conflict caused by multiple restores of the same backup.
+	// Both VS and VSC share the same generated name.
+	vs.Name = generatedName
 
-		p.log.Debugf("Set VolumeSnapshotContent %s/%s DeletionPolicy to Retain to make sure VS deletion in namespace will not delete Snapshot on cloud provider.",
-			newNamespace, vs.Name)
-
-		vsc := snapshotv1api.VolumeSnapshotContent{
-			ObjectMeta: metav1.ObjectMeta{
-				GenerateName: vs.Name + "-",
-				Labels: map[string]string{
-					velerov1api.RestoreNameLabel: label.GetValidName(input.Restore.Name),
-				},
-			},
-			Spec: snapshotv1api.VolumeSnapshotContentSpec{
-				DeletionPolicy: snapshotv1api.VolumeSnapshotContentRetain,
-				Driver:         csiDriverName,
-				VolumeSnapshotRef: core_v1.ObjectReference{
-					APIVersion: "snapshot.storage.k8s.io/v1",
-					Kind:       "VolumeSnapshot",
-					Namespace:  newNamespace,
-					Name:       vs.Name,
-				},
-				Source: snapshotv1api.VolumeSnapshotContentSource{
-					SnapshotHandle: &snapHandle,
-				},
-			},
-		}
-
-		// we create the VolumeSnapshotContent here instead of relying on the
-		// restore flow because we want to statically bind this VolumeSnapshot
-		// with a VolumeSnapshotContent that will be used as its source for pre-populating
-		// the volume that will be created as a result of the restore. To perform
-		// this static binding, a bi-directional link between the VolumeSnapshotContent
-		// and VolumeSnapshot objects have to be setup. Further, it is disallowed
-		// to convert a dynamically created VolumeSnapshotContent for static binding.
-		// See: https://github.com/kubernetes-csi/external-snapshotter/issues/274
-		if err := p.crClient.Create(context.TODO(), &vsc); err != nil {
-			return nil, errors.Wrapf(err,
-				"failed to create VolumeSnapshotContents %s",
-				vsc.GenerateName)
-		}
-		p.log.Infof("Created VolumesnapshotContents %s with static binding to volumesnapshot %s/%s",
-			vsc, newNamespace, vs.Name)
-
-		// Reset Spec to convert the VolumeSnapshot from using
-		// the dynamic VolumeSnapshotContent to the static one.
-		resetVolumeSnapshotSpecForRestore(&vs, &vsc.Name)
-
-		// Reset VolumeSnapshot annotation. By now, only change
-		// DeletionPolicy to Retain.
-		resetVolumeSnapshotAnnotation(&vs)
-	}
+	// Reset VolumeSnapshot annotation. By now, only change
+	// DeletionPolicy to Retain.
+	resetVolumeSnapshotAnnotation(&vs)
 
 	vsMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&vs)
 	if err != nil {
+		p.log.Errorf("Fail to convert VS %s to unstructured", vs.Namespace+"/"+vs.Name)
 		return nil, errors.WithStack(err)
 	}
 
@@ -195,6 +130,7 @@ func (p *volumeSnapshotRestoreItemAction) Cancel(
 	operationID string,
 	restore *velerov1api.Restore,
 ) error {
+	// CSI Specification doesn't support canceling a snapshot creation.
 	return nil
 }
 
@@ -206,7 +142,8 @@ func (p *volumeSnapshotRestoreItemAction) AreAdditionalItemsReady(
 }
 
 func NewVolumeSnapshotRestoreItemAction(
-	f client.Factory) plugincommon.HandlerInitializer {
+	f client.Factory,
+) plugincommon.HandlerInitializer {
 	return func(logger logrus.FieldLogger) (any, error) {
 		crClient, err := f.KubebuilderClient()
 		if err != nil {

pkg/restore/actions/csi/volumesnapshot_action_test.go

@@ -21,8 +21,6 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/google/go-cmp/cmp"
-	"github.com/google/go-cmp/cmp/cmpopts"
 	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
@@ -36,6 +34,7 @@ import (
 	factorymocks "github.com/vmware-tanzu/velero/pkg/client/mocks"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+	"github.com/vmware-tanzu/velero/pkg/util"
 )
 
 var (
@@ -89,7 +88,7 @@ func TestResetVolumeSnapshotSpecForRestore(t *testing.T) {
 			before := tc.vs.DeepCopy()
 			resetVolumeSnapshotSpecForRestore(&tc.vs, &tc.vscName)
 
-			assert.Equalf(t, tc.vs.Name, before.Name, "unexpected change to Object.Name, Want: %s; Got %s", tc.name, before.Name, tc.vs.Name)
+			assert.Equalf(t, tc.vs.Name, before.Name, "unexpected change to Object.Name, Want: %s; Got %s", before.Name, tc.vs.Name)
 			assert.Equal(t, tc.vs.Namespace, before.Namespace, "unexpected change to Object.Namespace, Want: %s; Got %s", tc.name, before.Namespace, tc.vs.Namespace)
 			assert.NotNil(t, tc.vs.Spec.Source)
 			assert.Nil(t, tc.vs.Spec.Source.PersistentVolumeClaimName)
@@ -103,15 +102,15 @@ func TestResetVolumeSnapshotSpecForRestore(t *testing.T) {
 }
 
 func TestVSExecute(t *testing.T) {
-	snapshotHandle := "vsc"
+	newVscName := util.GenerateSha256FromRestoreUIDAndVsName("restoreUID", "vsName")
 	tests := []struct {
-		name        string
-		item        runtime.Unstructured
-		vs          *snapshotv1api.VolumeSnapshot
-		restore     *velerov1api.Restore
-		expectErr   bool
-		createVS    bool
-		expectedVSC *snapshotv1api.VolumeSnapshotContent
+		name       string
+		item       runtime.Unstructured
+		vs         *snapshotv1api.VolumeSnapshot
+		restore    *velerov1api.Restore
+		expectErr  bool
+		createVS   bool
+		expectedVS *snapshotv1api.VolumeSnapshot
 	}{
 		{
 			name:      "Restore's RestorePVs is false",
@@ -119,45 +118,24 @@ func TestVSExecute(t *testing.T) {
 			expectErr: false,
 		},
 		{
-			name:      "Namespace remapping and VS already exists in cluster. Nothing change",
-			vs:        builder.ForVolumeSnapshot("ns", "name").Result(),
-			restore:   builder.ForRestore("velero", "restore").NamespaceMappings("ns", "newNS").Result(),
-			createVS:  true,
-			expectErr: false,
-		},
-		{
-			name:      "VS doesn't have VolumeSnapshotHandleAnnotation annotation",
-			vs:        builder.ForVolumeSnapshot("ns", "name").Result(),
-			restore:   builder.ForRestore("velero", "restore").NamespaceMappings("ns", "newNS").Result(),
-			expectErr: true,
-		},
-		{
-			name: "VS doesn't have DriverNameAnnotation annotation",
-			vs: builder.ForVolumeSnapshot("ns", "name").ObjectMeta(
-				builder.WithAnnotations(velerov1api.VolumeSnapshotHandleAnnotation, ""),
-			).Result(),
+			name:      "VS doesn't have VSC in status",
+			vs:        builder.ForVolumeSnapshot("ns", "name").ObjectMeta(builder.WithAnnotations("1", "1")).Status().Result(),
 			restore:   builder.ForRestore("velero", "restore").NamespaceMappings("ns", "newNS").Result(),
 			expectErr: true,
 		},
 		{
 			name: "Normal case, VSC should be created",
-			vs: builder.ForVolumeSnapshot("ns", "test").ObjectMeta(builder.WithAnnotationsMap(
-				map[string]string{
-					velerov1api.VolumeSnapshotHandleAnnotation: "vsc",
-					velerov1api.DriverNameAnnotation:           "pd.csi.storage.gke.io",
-				},
-			)).Result(),
-			restore:   builder.ForRestore("velero", "restore").Result(),
-			expectErr: false,
-			expectedVSC: builder.ForVolumeSnapshotContent("vsc").ObjectMeta(
-				builder.WithLabels(velerov1api.RestoreNameLabel, "restore"),
-			).VolumeSnapshotRef("ns", "test").
-				DeletionPolicy(snapshotv1api.VolumeSnapshotContentRetain).
-				Driver("pd.csi.storage.gke.io").
-				Source(snapshotv1api.VolumeSnapshotContentSource{
-					SnapshotHandle: &snapshotHandle,
-				}).
-				Result(),
+			vs: builder.ForVolumeSnapshot("ns", "vsName").ObjectMeta(
+				builder.WithAnnotationsMap(
+					map[string]string{
+						velerov1api.VolumeSnapshotHandleAnnotation: "vsc",
+						velerov1api.DriverNameAnnotation:           "pd.csi.storage.gke.io",
+					},
+				),
+			).SourceVolumeSnapshotContentName(newVscName).Status().BoundVolumeSnapshotContentName("vscName").Result(),
+			restore:    builder.ForRestore("velero", "restore").ObjectMeta(builder.WithUID("restoreUID")).Result(),
+			expectErr:  false,
+			expectedVS: builder.ForVolumeSnapshot("ns", "test").SourceVolumeSnapshotContentName(newVscName).Result(),
 		},
 	}
 
@@ -181,10 +159,11 @@ func TestVSExecute(t *testing.T) {
 				}
 			}
 
-			_, err := p.Execute(
+			result, err := p.Execute(
 				&velero.RestoreItemActionExecuteInput{
-					Item:    test.item,
-					Restore: test.restore,
+					Item:           test.item,
+					ItemFromBackup: test.item,
+					Restore:        test.restore,
 				},
 			)
 
@@ -192,18 +171,11 @@ func TestVSExecute(t *testing.T) {
 				require.NoError(t, err)
 			}
 
-			if test.expectedVSC != nil {
-				vscList := new(snapshotv1api.VolumeSnapshotContentList)
-				require.NoError(t, p.crClient.List(context.TODO(), vscList))
-				require.True(t, cmp.Equal(
-					*test.expectedVSC,
-					vscList.Items[0],
-					cmpopts.IgnoreFields(
-						snapshotv1api.VolumeSnapshotContent{},
-						"Kind", "APIVersion", "GenerateName", "Name",
-						"ResourceVersion",
-					),
-				))
+			if test.expectedVS != nil {
+				var vs snapshotv1api.VolumeSnapshot
+				require.NoError(t, runtime.DefaultUnstructuredConverter.FromUnstructured(
+					result.UpdatedItem.UnstructuredContent(), &vs))
+				require.Equal(t, test.expectedVS.Spec, vs.Spec)
 			}
 		})
 	}

pkg/restore/actions/csi/volumesnapshotcontent_action.go

@@ -20,11 +20,16 @@ import (
 	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	crclient "sigs.k8s.io/controller-runtime/pkg/client"
 
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/client"
+	plugincommon "github.com/vmware-tanzu/velero/pkg/plugin/framework/common"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	"github.com/vmware-tanzu/velero/pkg/util"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
 	"github.com/vmware-tanzu/velero/pkg/util/csi"
 )
@@ -32,7 +37,8 @@ import (
 // volumeSnapshotContentRestoreItemAction is a restore item action
 // plugin for Velero
 type volumeSnapshotContentRestoreItemAction struct {
-	log logrus.FieldLogger
+	log    logrus.FieldLogger
+	client crclient.Client
 }
 
 // AppliesTo returns information indicating VolumeSnapshotContentRestoreItemAction
@@ -51,34 +57,77 @@ func (p *volumeSnapshotContentRestoreItemAction) AppliesTo() (
 func (p *volumeSnapshotContentRestoreItemAction) Execute(
 	input *velero.RestoreItemActionExecuteInput,
 ) (*velero.RestoreItemActionExecuteOutput, error) {
-	p.log.Info("Starting VolumeSnapshotContentRestoreItemAction")
-	var snapCont snapshotv1api.VolumeSnapshotContent
 	if boolptr.IsSetToFalse(input.Restore.Spec.RestorePVs) {
 		p.log.Infof("Restore did not request for PVs to be restored %s/%s",
 			input.Restore.Namespace, input.Restore.Name)
 		return &velero.RestoreItemActionExecuteOutput{SkipRestore: true}, nil
 	}
+
+	p.log.Info("Starting VolumeSnapshotContentRestoreItemAction")
+
+	var vsc snapshotv1api.VolumeSnapshotContent
 	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(
-		input.Item.UnstructuredContent(), &snapCont); err != nil {
+		input.Item.UnstructuredContent(), &vsc); err != nil {
 		return &velero.RestoreItemActionExecuteOutput{},
 			errors.Wrapf(err, "failed to convert input.Item from unstructured")
 	}
 
+	var vscFromBackup snapshotv1api.VolumeSnapshotContent
+	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(
+		input.ItemFromBackup.UnstructuredContent(), &vscFromBackup); err != nil {
+		return &velero.RestoreItemActionExecuteOutput{},
+			errors.Errorf(err.Error(), "failed to convert input.ItemFromBackup from unstructured")
+	}
+
+	// If cross-namespace restore is configured, change the namespace
+	// for VolumeSnapshot object to be restored
+	newNamespace, ok := input.Restore.Spec.NamespaceMapping[vsc.Spec.VolumeSnapshotRef.Namespace]
+	if ok {
+		// Update the referenced VS namespace to the mapping one.
+		vsc.Spec.VolumeSnapshotRef.Namespace = newNamespace
+	}
+
+	// Reset VSC name to align with VS.
+	vsc.Name = util.GenerateSha256FromRestoreUIDAndVsName(
+		string(input.Restore.UID), vscFromBackup.Spec.VolumeSnapshotRef.Name)
+	// Also reset the referenced VS name.
+	vsc.Spec.VolumeSnapshotRef.Name = vsc.Name
+
+	// Reset the ResourceVersion and UID of referenced VolumeSnapshot.
+	vsc.Spec.VolumeSnapshotRef.ResourceVersion = ""
+	vsc.Spec.VolumeSnapshotRef.UID = ""
+
+	// Set the DeletionPolicy to Retain to avoid VS deletion will not trigger snapshot deletion
+	vsc.Spec.DeletionPolicy = snapshotv1api.VolumeSnapshotContentRetain
+
+	if vscFromBackup.Status != nil && vscFromBackup.Status.SnapshotHandle != nil {
+		vsc.Spec.Source.VolumeHandle = nil
+		vsc.Spec.Source.SnapshotHandle = vscFromBackup.Status.SnapshotHandle
+	} else {
+		p.log.Errorf("fail to get snapshot handle from VSC %s status", vsc.Name)
+		return nil, errors.Errorf("fail to get snapshot handle from VSC %s status", vsc.Name)
+	}
+
 	additionalItems := []velero.ResourceIdentifier{}
-	if csi.IsVolumeSnapshotContentHasDeleteSecret(&snapCont) {
+	if csi.IsVolumeSnapshotContentHasDeleteSecret(&vsc) {
 		additionalItems = append(additionalItems,
 			velero.ResourceIdentifier{
 				GroupResource: schema.GroupResource{Group: "", Resource: "secrets"},
-				Name:          snapCont.Annotations[velerov1api.PrefixedSecretNameAnnotation],
-				Namespace:     snapCont.Annotations[velerov1api.PrefixedSecretNamespaceAnnotation],
+				Name:          vsc.Annotations[velerov1api.PrefixedSecretNameAnnotation],
+				Namespace:     vsc.Annotations[velerov1api.PrefixedSecretNamespaceAnnotation],
 			},
 		)
 	}
 
+	vscMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&vsc)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
 	p.log.Infof("Returning from VolumeSnapshotContentRestoreItemAction with %d additionalItems",
 		len(additionalItems))
 	return &velero.RestoreItemActionExecuteOutput{
-		UpdatedItem:     input.Item,
+		UpdatedItem:     &unstructured.Unstructured{Object: vscMap},
 		AdditionalItems: additionalItems,
 	}, nil
 }
@@ -108,6 +157,15 @@ func (p *volumeSnapshotContentRestoreItemAction) AreAdditionalItemsReady(
 	return true, nil
 }
 
-func NewVolumeSnapshotContentRestoreItemAction(logger logrus.FieldLogger) (any, error) {
-	return &volumeSnapshotContentRestoreItemAction{logger}, nil
+func NewVolumeSnapshotContentRestoreItemAction(
+	f client.Factory,
+) plugincommon.HandlerInitializer {
+	return func(logger logrus.FieldLogger) (any, error) {
+		crClient, err := f.KubebuilderClient()
+		if err != nil {
+			return nil, err
+		}
+
+		return &volumeSnapshotContentRestoreItemAction{logger, crClient}, nil
+	}
 }

pkg/restore/actions/csi/volumesnapshotcontent_action_test.go

@@ -17,6 +17,8 @@ limitations under the License.
 package csi
 
 import (
+	"context"
+	"fmt"
 	"testing"
 
 	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
@@ -27,18 +29,25 @@ import (
 
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/builder"
+	factorymocks "github.com/vmware-tanzu/velero/pkg/client/mocks"
 	"github.com/vmware-tanzu/velero/pkg/kuberesource"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+	"github.com/vmware-tanzu/velero/pkg/util"
 )
 
 func TestVSCExecute(t *testing.T) {
+	snapshotHandleName := "testHandle"
+	newVscName := util.GenerateSha256FromRestoreUIDAndVsName("restoreUID", "vsName")
 	tests := []struct {
 		name          string
 		item          runtime.Unstructured
 		vsc           *snapshotv1api.VolumeSnapshotContent
 		restore       *velerov1api.Restore
 		expectErr     bool
+		createVSC     bool
 		expectedItems []velero.ResourceIdentifier
+		expectedVSC   *snapshotv1api.VolumeSnapshotContent
 	}{
 		{
 			name:      "Restore's RestorePVs is false",
@@ -46,14 +55,16 @@ func TestVSCExecute(t *testing.T) {
 			expectErr: false,
 		},
 		{
-			name: "Normal case, additional items should return",
+			name: "Normal case, additional items should return    ",
 			vsc: builder.ForVolumeSnapshotContent("test").ObjectMeta(builder.WithAnnotationsMap(
 				map[string]string{
 					velerov1api.PrefixedSecretNameAnnotation:      "name",
 					velerov1api.PrefixedSecretNamespaceAnnotation: "namespace",
 				},
-			)).Result(),
-			restore:   builder.ForRestore("velero", "restore").Result(),
+			)).VolumeSnapshotRef("velero", "vsName", "vsUID").
+				Status(&snapshotv1api.VolumeSnapshotContentStatus{SnapshotHandle: &snapshotHandleName}).Result(),
+			restore: builder.ForRestore("velero", "restore").ObjectMeta(builder.WithUID("restoreUID")).
+				NamespaceMappings("velero", "restore").Result(),
 			expectErr: false,
 			expectedItems: []velero.ResourceIdentifier{
 				{
@@ -62,26 +73,63 @@ func TestVSCExecute(t *testing.T) {
 					Name:          "name",
 				},
 			},
+			expectedVSC: builder.ForVolumeSnapshotContent(newVscName).ObjectMeta(builder.WithAnnotationsMap(
+				map[string]string{
+					velerov1api.PrefixedSecretNameAnnotation:      "name",
+					velerov1api.PrefixedSecretNamespaceAnnotation: "namespace",
+				},
+			)).VolumeSnapshotRef("restore", newVscName, "").
+				Source(snapshotv1api.VolumeSnapshotContentSource{SnapshotHandle: &snapshotHandleName}).
+				DeletionPolicy(snapshotv1api.VolumeSnapshotContentRetain).
+				Status(&snapshotv1api.VolumeSnapshotContentStatus{SnapshotHandle: &snapshotHandleName}).Result(),
+		},
+		{
+			name: "VSC exists in cluster, same as the normal case",
+			vsc: builder.ForVolumeSnapshotContent("test").ObjectMeta(builder.WithAnnotationsMap(
+				map[string]string{
+					velerov1api.PrefixedSecretNameAnnotation:      "name",
+					velerov1api.PrefixedSecretNamespaceAnnotation: "namespace",
+				},
+			)).VolumeSnapshotRef("velero", "vsName", "vsUID").
+				Status(&snapshotv1api.VolumeSnapshotContentStatus{SnapshotHandle: &snapshotHandleName}).Result(),
+			restore: builder.ForRestore("velero", "restore").ObjectMeta(builder.WithUID("restoreUID")).
+				NamespaceMappings("velero", "restore").Result(),
+			createVSC: true,
+			expectErr: false,
+			expectedVSC: builder.ForVolumeSnapshotContent(newVscName).ObjectMeta(builder.WithAnnotationsMap(
+				map[string]string{
+					velerov1api.PrefixedSecretNameAnnotation:      "name",
+					velerov1api.PrefixedSecretNamespaceAnnotation: "namespace",
+				},
+			)).VolumeSnapshotRef("restore", newVscName, "").
+				Source(snapshotv1api.VolumeSnapshotContentSource{SnapshotHandle: &snapshotHandleName}).
+				DeletionPolicy(snapshotv1api.VolumeSnapshotContentRetain).
+				Status(&snapshotv1api.VolumeSnapshotContentStatus{SnapshotHandle: &snapshotHandleName}).Result(),
 		},
 	}
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			p, err := NewVolumeSnapshotContentRestoreItemAction(logrus.StandardLogger())
-			require.NoError(t, err)
-
-			action := p.(*volumeSnapshotContentRestoreItemAction)
+			action := volumeSnapshotContentRestoreItemAction{
+				log:    logrus.StandardLogger(),
+				client: velerotest.NewFakeControllerRuntimeClient(t),
+			}
 
 			if test.vsc != nil {
 				vsMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(test.vsc)
 				require.NoError(t, err)
 				test.item = &unstructured.Unstructured{Object: vsMap}
+
+				if test.createVSC {
+					require.NoError(t, action.client.Create(context.TODO(), test.vsc))
+				}
 			}
 
 			output, err := action.Execute(
 				&velero.RestoreItemActionExecuteInput{
-					Item:    test.item,
-					Restore: test.restore,
+					Item:           test.item,
+					ItemFromBackup: test.item,
+					Restore:        test.restore,
 				},
 			)
 
@@ -89,6 +137,18 @@ func TestVSCExecute(t *testing.T) {
 				require.NoError(t, err)
 			}
 
+			if test.expectedVSC != nil {
+				vsc := new(snapshotv1api.VolumeSnapshotContent)
+				require.NoError(t,
+					runtime.DefaultUnstructuredConverter.FromUnstructured(
+						output.UpdatedItem.UnstructuredContent(),
+						vsc,
+					),
+				)
+
+				require.Equal(t, test.expectedVSC, vsc)
+			}
+
 			if len(test.expectedItems) > 0 {
 				require.Equal(t, test.expectedItems, output.AdditionalItems)
 			}
@@ -112,3 +172,20 @@ func TestVSCAppliesTo(t *testing.T) {
 		selector,
 	)
 }
+
+func TestNewVolumeSnapshotContentRestoreItemAction(t *testing.T) {
+	logger := logrus.StandardLogger()
+	crClient := velerotest.NewFakeControllerRuntimeClient(t)
+
+	f := &factorymocks.Factory{}
+	f.On("KubebuilderClient").Return(nil, fmt.Errorf(""))
+	plugin := NewVolumeSnapshotContentRestoreItemAction(f)
+	_, err := plugin(logger)
+	require.Error(t, err)
+
+	f1 := &factorymocks.Factory{}
+	f1.On("KubebuilderClient").Return(crClient, nil)
+	plugin1 := NewVolumeSnapshotContentRestoreItemAction(f1)
+	_, err1 := plugin1(logger)
+	require.NoError(t, err1)
+}