Summary of Changes:

Move PVC Request Size Patch to Backup CSI Action
Shifted the logic that patches the PVC request size (to match the corresponding VolumeSnapshot size) from the CSI Restore action to the CSI Backup action.
✅ This enables restoring a PVC independently using label selectors, without requiring the VolumeSnapshot to be restored first.

Include VolumeSnapshot in CSI Additional Items for PVC
Added VolumeSnapshot as an additional item in the PVC CSI backup logic to ensure necessary metadata is available during restore.

Include VolumeSnapshotContent in CSI Restore Additional Items
Added VolumeSnapshotContent to the additional items in the CSI restore action to support a more complete restore workflow.
✅ This to make sure those resources are restored even if filters out by label from the resources list to restore

Author:    Amos Mastbaum <[email protected]>
Signed-off-by: Amos Mastbaum <[email protected]>

fixing-after-michal

wait-for-vsc.Status.RestoreSize

wait-for-vsc.Status.RestoreSize

Update pkg/util/csi/volume_snapshot.go

Update pkg/util/csi/volume_snapshot.go

Update pkg/util/csi/volume_snapshot.go

Author: sseago

pkg/backup/actions/csi/pvc_action.go

@@ -35,6 +35,8 @@ import (
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 	crclient "sigs.k8s.io/controller-runtime/pkg/client"
 
+	"k8s.io/apimachinery/pkg/api/resource"
+
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	velerov2alpha1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
 	"github.com/vmware-tanzu/velero/pkg/client"
@@ -252,6 +254,24 @@ func (p *pvcBackupItemAction) Execute(
 		return nil, nil, "", nil, err
 	}
 
+	// Wait until VS associated VSC snapshot handle created before
+	// continue.we later requier the vsc restore size
+	vsc, err := csi.WaitUntilVSCHandleIsReady(
+		vs,
+		p.crClient,
+		p.log,
+		true,
+		backup.Spec.CSISnapshotTimeout.Duration,
+	)
+	if err != nil {
+		p.log.Errorf(
+			"Fail to wait VolumeSnapshot turned to ReadyToUse: %s",
+			err.Error(),
+		)
+		csi.CleanupVolumeSnapshot(vs, p.crClient, p.log)
+		return nil, nil, "", nil, errors.WithStack(err)
+	}
+
 	labels := map[string]string{
 		velerov1api.VolumeSnapshotLabel: vs.Name,
 		velerov1api.BackupNameLabel:     backup.Name,
@@ -279,24 +299,6 @@ func (p *pvcBackupItemAction) Execute(
 			"Backup":         backup.Name,
 		})
 
-		// Wait until VS associated VSC snapshot handle created before
-		// returning with the Async operation for data mover.
-		_, err := csi.WaitUntilVSCHandleIsReady(
-			vs,
-			p.crClient,
-			p.log,
-			true,
-			backup.Spec.CSISnapshotTimeout.Duration,
-		)
-		if err != nil {
-			dataUploadLog.Errorf(
-				"Fail to wait VolumeSnapshot turned to ReadyToUse: %s",
-				err.Error(),
-			)
-			csi.CleanupVolumeSnapshot(vs, p.crClient, p.log)
-			return nil, nil, "", nil, errors.WithStack(err)
-		}
-
 		dataUploadLog.Info("Starting data upload of backup")
 
 		dataUpload, err := createDataUpload(
@@ -340,6 +342,12 @@ func (p *pvcBackupItemAction) Execute(
 			dataUploadLog.Info("DataUpload is submitted successfully.")
 		}
 	} else {
+		err = setPVCRequestSizeToVSRestoreSize(&pvc, p.crClient, vs.Name, vsc, p.log)
+		if err != nil {
+			p.log.Errorf("Failed to set PVC request size: %s", err.Error())
+			return nil, nil, "", nil, errors.WithStack(err)
+		}
+
 		additionalItems = []velero.ResourceIdentifier{
 			{
 				GroupResource: kuberesource.VolumeSnapshots,
@@ -564,3 +572,61 @@ func NewPvcBackupItemAction(f client.Factory) plugincommon.HandlerInitializer {
 		}, nil
 	}
 }
+
+func setPVCRequestSizeToVSRestoreSize(
+	pvc *corev1api.PersistentVolumeClaim,
+	crClient crclient.Client,
+	volumeSnapshotName string,
+	vsc *snapshotv1api.VolumeSnapshotContent,
+	logger logrus.FieldLogger,
+) error {
+	vs := new(snapshotv1api.VolumeSnapshot)
+	if err := crClient.Get(context.TODO(),
+		crclient.ObjectKey{
+			Namespace: pvc.Namespace,
+			Name:      volumeSnapshotName,
+		},
+		vs,
+	); err != nil {
+		return errors.Wrapf(err, "Failed to get Volumesnapshot %s/%s to restore PVC %s/%s",
+			pvc.Namespace, volumeSnapshotName, pvc.Namespace, pvc.Name)
+	}
+
+	if vsc.Status.RestoreSize != nil {
+		logger.Debugf("Patching PVC request size to fit the volumesnapshot restore size %d", vsc.Status.RestoreSize)
+		restoreSize := *resource.NewQuantity(*vsc.Status.RestoreSize, resource.BinarySI)
+
+		// It is possible that the volume provider allocated a larger
+		// capacity volume than what was requested in the backed up PVC.
+		// In this scenario the volumesnapshot of the PVC will end being
+		// larger than its requested storage size.  Such a PVC, on restore
+		// as-is, will be stuck attempting to use a VolumeSnapshot as a
+		// data source for a PVC that is not large enough.
+		// To counter that, here we set the storage request on the PVC
+		// to the larger of the PVC's storage request and the size of the
+		// VolumeSnapshot
+		setPVCStorageResourceRequest(pvc, restoreSize, logger)
+	}
+
+	return nil
+}
+
+func setPVCStorageResourceRequest(
+	pvc *corev1api.PersistentVolumeClaim,
+	restoreSize resource.Quantity,
+	log logrus.FieldLogger,
+) {
+	{
+		if pvc.Spec.Resources.Requests == nil {
+			pvc.Spec.Resources.Requests = corev1api.ResourceList{}
+		}
+
+		storageReq, exists := pvc.Spec.Resources.Requests[corev1api.ResourceStorage]
+		if !exists || storageReq.Cmp(restoreSize) < 0 {
+			pvc.Spec.Resources.Requests[corev1api.ResourceStorage] = restoreSize
+			rs := pvc.Spec.Resources.Requests[corev1api.ResourceStorage]
+			log.Infof("Resetting storage requests for PVC %s/%s to %s",
+				pvc.Namespace, pvc.Name, rs.String())
+		}
+	}
+}

pkg/backup/actions/csi/pvc_action_test.go

@@ -37,14 +37,15 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	crclient "sigs.k8s.io/controller-runtime/pkg/client"
 
+	"k8s.io/apimachinery/pkg/api/resource"
+
 	"github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
 	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"
 	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/boolptr"
 )
 
 func TestExecute(t *testing.T) {
@@ -130,7 +131,7 @@ func TestExecute(t *testing.T) {
 		},
 		{
 			name:           "Test ResourcePolicy",
-			backup:         builder.ForBackup("velero", "test").ResourcePolicies("resourcePolicy").SnapshotVolumes(false).Result(),
+			backup:         builder.ForBackup("velero", "test").ResourcePolicies("resourcePolicy").SnapshotVolumes(false).CSISnapshotTimeout(time.Duration(3600) * time.Second).Result(),
 			resourcePolicy: builder.ForConfigMap("velero", "resourcePolicy").Data("policy", "{\"version\":\"v1\", \"volumePolicies\":[{\"conditions\":{\"csi\": {}},\"action\":{\"type\":\"snapshot\"}}]}").Result(),
 			pvc:            builder.ForPersistentVolumeClaim("velero", "testPVC").VolumeName("testPV").StorageClass("testSC").Phase(corev1.ClaimBound).Result(),
 			pv:             builder.ForPersistentVolume("testPV").CSI("hostpath", "testVolume").Result(),
@@ -170,7 +171,7 @@ func TestExecute(t *testing.T) {
 			pvcMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&tc.pvc)
 			require.NoError(t, err)
 
-			if boolptr.IsSetToTrue(tc.backup.Spec.SnapshotMoveData) == true {
+			if tc.pvc != nil {
 				go func() {
 					var vsList v1.VolumeSnapshotList
 					err := wait.PollUntilContextTimeout(context.Background(), 1*time.Second, 10*time.Second, true, func(ctx context.Context) (bool, error) {
@@ -412,3 +413,72 @@ func TestNewPVCBackupItemAction(t *testing.T) {
 	_, err1 := plugin1(logger)
 	require.NoError(t, err1)
 }
+
+func TestPVCRequestSize(t *testing.T) {
+	logger := logrus.New()
+
+	tests := []struct {
+		name         string
+		pvcInitial   string // initial storage request on the PVC (e.g. "1Gi" or "3Gi")
+		restoreSize  string // restore size set in VSC.Status.RestoreSize (e.g. "2Gi")
+		expectedSize string // expected storage request on the PVC after update
+	}{
+		{
+			name:         "UpdateRequired: PVC request is lower than restore size",
+			pvcInitial:   "1Gi",
+			restoreSize:  "2Gi",
+			expectedSize: "2Gi",
+		},
+		{
+			name:         "NoUpdateRequired: PVC request is larger than restore size",
+			pvcInitial:   "3Gi",
+			restoreSize:  "2Gi",
+			expectedSize: "3Gi",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			crClient := velerotest.NewFakeControllerRuntimeClient(t)
+
+			// Create a PVC with the initial storage request.
+			pvc := builder.ForPersistentVolumeClaim("velero", "testPVC").
+				VolumeName("testPV").
+				StorageClass("testSC").
+				Result()
+			pvc.Spec.Resources.Requests = corev1.ResourceList{
+				corev1.ResourceStorage: resource.MustParse(tc.pvcInitial),
+			}
+
+			// Create a VolumeSnapshot required for the lookup
+			vs := &snapshotv1api.VolumeSnapshot{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "testVS",
+					Namespace: "velero",
+				},
+			}
+			require.NoError(t, crClient.Create(context.Background(), vs))
+
+			// Create a VolumeSnapshotContent with restore size
+			rsQty := resource.MustParse(tc.restoreSize)
+			rsValue := rsQty.Value()
+			vsc := &snapshotv1api.VolumeSnapshotContent{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "testVSC",
+				},
+				Status: &snapshotv1api.VolumeSnapshotContentStatus{
+					RestoreSize: &rsValue,
+				},
+			}
+
+			// Call the function under test
+			err := setPVCRequestSizeToVSRestoreSize(pvc, crClient, "testVS", vsc, logger)
+			require.NoError(t, err)
+
+			// Verify that the PVC storage request is updated as expected.
+			updatedSize := pvc.Spec.Resources.Requests[corev1.ResourceStorage]
+			expected := resource.MustParse(tc.expectedSize)
+			require.Equal(t, 0, updatedSize.Cmp(expected), "PVC storage request should be %s", tc.expectedSize)
+		})
+	}
+}