Add in-place restore pre-flight check: PVC must be large enough for the source volume

Compare the existing PVC's capacity against the source volume size
recorded in the backup volume info (#10506) before any side effect and
skip the volume when it is too small, so the restore fails early instead
of running out of space midway. For the block data mover the source size
is the device size; for the file system data movers it is the logical
size of the backed-up files, a lower bound since file system metadata is
not accounted for.

The file system path reads the size from the volume info already carried
in RestoreData. The PVC CSI RIA has no access to the volume info, so the
restore engine carries the size on the PVC item through a Velero-internal
annotation, the same mechanism as the selected-node carrier; both carrier
annotations are stripped before the item is created in the cluster.

The check is skipped when the source size is unknown (backups taken
before it was recorded) or the PVC's capacity is not reported.

Signed-off-by: chlins <chlins.zhang@gmail.com>

changelogs/unreleased/10512-chlins

@@ -0,0 +1 @@
+Add in-place restore pre-flight check: PVC must be large enough for the backed-up data

internal/volume/volumes_information.go

@@ -368,6 +368,17 @@ func newPodVolumeInfoFromPVR(pvr *velerov1api.PodVolumeRestore) *PodVolumeRestor
 	}
 }
 
+// SourceSize returns the size of the source volume recorded at backup time, or 0 if unknown.
+func (v BackupVolumeInfo) SourceSize() int64 {
+	switch {
+	case v.SnapshotDataMovementInfo != nil:
+		return v.SnapshotDataMovementInfo.SourceSize
+	case v.PVBInfo != nil:
+		return v.PVBInfo.SourceSize
+	}
+	return 0
+}
+
 // PVInfo is used to store some PV information modified after creation.
 // Those information are lost after PV recreation.
 type PVInfo struct {

pkg/apis/velero/v1/labels_annotations.go

@@ -185,6 +185,13 @@ const (
 	// on the cluster. Using a carrier annotation avoids any dependency on the execution order
 	// of RestoreItemActions.
 	InplaceRestoreSelectedNodeAnnotation = "restore.velero.io/inplace-restore-selected-node"
+
+	// InplaceRestoreSourceSizeAnnotation is a Velero-internal carrier annotation set by the
+	// restore engine on a PVC item before RestoreItemActions run. It carries the size of the
+	// source volume recorded in the backup volume info, so the PVC CSI RestoreItemAction can
+	// run the in-place restore capacity pre-flight check without access to the volume info.
+	// The annotation is always stripped by the restore engine; it never lands on the cluster.
+	InplaceRestoreSourceSizeAnnotation = "restore.velero.io/inplace-restore-source-size"
 	// SkippedNoCSIPVAnnotation - Velero checks this annotation on processed PVC to
 	// find out if the snapshot was skipped b/c the PV is not provisioned via CSI
 	SkippedNoCSIPVAnnotation = "backup.velero.io/skipped-no-csi-pv"

pkg/podvolume/restorer.go

@@ -189,7 +189,12 @@ func (r *restorer) RestorePodVolumes(data RestoreData, tracker *volume.RestoreVo
 		// to write into, and they cannot write to it themselves until this
 		// restore's PodVolumeRestores complete.
 		if data.Restore.IsVolumeDataInplaceRestore() && pvc != nil {
-			if err := inplace.CheckPVCBoundToBackedUpPV(pvc, backedUpPVName(data.BackupVolumeInfos, data.SourceNamespace, pvc.Name), data.SourceNamespace); err != nil {
+			pvName := backedUpPVName(data.BackupVolumeInfos, data.SourceNamespace, pvc.Name)
+			if err := inplace.CheckPVCBoundToBackedUpPV(pvc, pvName, data.SourceNamespace); err != nil {
+				errs = append(errs, err)
+				continue
+			}
+			if err := inplace.CheckPVCCapacity(pvc, data.BackupVolumeInfos[pvName].SourceSize()); err != nil {
 				errs = append(errs, err)
 				continue
 			}

pkg/podvolume/restorer_test.go

@@ -27,6 +27,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	appsv1api "k8s.io/api/apps/v1"
 	corev1api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/kubernetes"
@@ -439,6 +440,37 @@ func TestRestorePodVolumes(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "in-place restore blocked when the PVC is too small for the source volume",
+			pvbs: []*velerov1api.PodVolumeBackup{
+				createPVBObj(true, true, 1, "kopia"),
+			},
+			inplace: true,
+			kubeClientObj: []runtime.Object{
+				createNodeAgentDaemonset(),
+				func() *corev1api.PersistentVolumeClaim {
+					pvc := createPVCObj(1)
+					pvc.Status.Capacity = corev1api.ResourceList{corev1api.ResourceStorage: resource.MustParse("100Mi")}
+					return pvc
+				}(),
+			},
+			ctlClientObj: []runtime.Object{
+				createBackupRepoObj(),
+			},
+			restoredPod:     createPodObj(true, true, true, 1),
+			sourceNamespace: "fake-ns",
+			bsl:             "fake-bsl",
+			volumeInfos: map[string]volume.BackupVolumeInfo{
+				"fake-pv-1": {PVCNamespace: "fake-ns", PVCName: "fake-pvc-1", PVBInfo: &volume.PodVolumeBackupInfo{SourceSize: 200 << 20}},
+			},
+			runtimeScheme: scheme,
+			errs: []expectError{
+				{
+					err:        "in-place restore pre-flight check failed, skipping volume data restore: PVC fake-ns/fake-pvc-1 capacity 100Mi is smaller than the backed-up volume size 209715200 bytes",
+					prefixOnly: true,
+				},
+			},
+		},
 		{
 			name: "in-place restore proceeds when the PVC is only used by the gated restored pod",
 			pvbs: []*velerov1api.PodVolumeBackup{

pkg/restore/actions/csi/pvc_action.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"strconv"
 	"time"
 
 	"github.com/cockroachdb/errors"
@@ -238,6 +239,9 @@ func (p *pvcRestoreItemAction) executeWithDataMove(logger *logrus.Entry, input *
 		if err := inplace.CheckPVCBoundToBackedUpPV(existingPVC, pvcFromBackup.Spec.VolumeName, pvcFromBackup.Namespace); err != nil {
 			return nil, errors.WithStack(err)
 		}
+		if err := inplace.CheckPVCCapacity(existingPVC, sourceSizeFromCarrier(pvc)); err != nil {
+			return nil, errors.WithStack(err)
+		}
 		if err := inplace.CheckPVCNotInUse(ctx, p.crClient, existingPVC, input.Restore.UID); err != nil {
 			return nil, errors.WithStack(err)
 		}
@@ -729,6 +733,13 @@ func (p *pvcRestoreItemAction) createVolumeSnapshot(ctx context.Context, logger
 	return vs, nil
 }
 
+// sourceSizeFromCarrier reads the source volume size the restore engine carries on the PVC
+// item from the backup volume info, or 0 if absent or malformed.
+func sourceSizeFromCarrier(pvc *corev1api.PersistentVolumeClaim) int64 {
+	size, _ := strconv.ParseInt(pvc.Annotations[velerov1api.InplaceRestoreSourceSizeAnnotation], 10, 64)
+	return size
+}
+
 func NewPvcRestoreItemAction(f client.Factory) plugincommon.HandlerInitializer {
 	return func(logger logrus.FieldLogger) (any, error) {
 		crClient, err := f.KubebuilderClient()

pkg/restore/actions/csi/pvc_action_test.go

@@ -757,6 +757,8 @@ func TestExecuteInplaceRestorePreflight(t *testing.T) {
 		name           string
 		pod            *corev1api.Pod
 		backedUpPVName string
+		sourceSize     string // carried on the PVC item by the restore engine
+		pvcCapacity    string
 		expectBlock    string
 	}{
 		{
@@ -774,6 +776,17 @@ func TestExecuteInplaceRestorePreflight(t *testing.T) {
 			backedUpPVName: "backupPV",
 			expectBlock:    "was bound to PV backupPV at backup time",
 		},
+		{
+			// Backed-up PV unknown so the same-volume skip does not apply.
+			name:        "PVC smaller than the source volume blocks the restore",
+			sourceSize:  "209715200",
+			pvcCapacity: "100Mi",
+			expectBlock: "capacity 100Mi is smaller than the backed-up volume size 209715200 bytes",
+		},
+		{
+			name:        "source size not carried skips the capacity check",
+			pvcCapacity: "100Mi",
+		},
 	}
 
 	for _, tc := range tests {
@@ -781,6 +794,9 @@ func TestExecuteInplaceRestorePreflight(t *testing.T) {
 			existingPVC := builder.ForPersistentVolumeClaim("velero", "testPVC").
 				VolumeName("testPV").
 				Phase(corev1api.ClaimBound).Result()
+			if tc.pvcCapacity != "" {
+				existingPVC.Status.Capacity = corev1api.ResourceList{corev1api.ResourceStorage: resource.MustParse(tc.pvcCapacity)}
+			}
 			existingPV := builder.ForPersistentVolume("testPV").Result()
 			backup := builder.ForBackup("velero", "testBackup").SnapshotMoveData(true).Result()
 			restore := builder.ForRestore("velero", "testRestore").Backup("testBackup").
@@ -811,7 +827,11 @@ func TestExecuteInplaceRestorePreflight(t *testing.T) {
 				kubeClient: fake.NewSimpleClientset(kubeObjects...),
 			}
 
-			pvcMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(pvcFromBackup.DeepCopy())
+			item := pvcFromBackup.DeepCopy()
+			if tc.sourceSize != "" {
+				item.Annotations[velerov1api.InplaceRestoreSourceSizeAnnotation] = tc.sourceSize
+			}
+			pvcMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(item)
 			require.NoError(t, err)
 			pvcFromBackupMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(pvcFromBackup)
 			require.NoError(t, err)

pkg/restore/inplace/preflight.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/cockroachdb/errors"
 	corev1api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/apimachinery/pkg/types"
 	crclient "sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -149,3 +150,26 @@ func CheckPVCBoundToBackedUpPV(existingPVC *corev1api.PersistentVolumeClaim, bac
 	return errors.Errorf("in-place restore pre-flight check failed, skipping volume data restore: PVC %s/%s is bound to PV %s, but was bound to PV %s at backup time",
 		existingPVC.Namespace, existingPVC.Name, existingPVC.Spec.VolumeName, backedUpPVName)
 }
+
+// CheckPVCCapacity verifies the existing PVC is large enough to hold the
+// backed-up volume, failing early instead of letting the restore run out of
+// space midway. sourceSize is the size of the source volume recorded at
+// backup time: the device size for the block data mover, the logical size of
+// the backed-up files for the file system data movers (a lower bound, since
+// file system metadata is not accounted for). The check is skipped when the
+// size is unknown (backups taken before it was recorded) or when the PVC's
+// capacity is not reported.
+func CheckPVCCapacity(existingPVC *corev1api.PersistentVolumeClaim, sourceSize int64) error {
+	if sourceSize <= 0 {
+		return nil
+	}
+	capacity, ok := existingPVC.Status.Capacity[corev1api.ResourceStorage]
+	if !ok || capacity.IsZero() {
+		return nil
+	}
+	if capacity.Cmp(*resource.NewQuantity(sourceSize, resource.BinarySI)) < 0 {
+		return errors.Errorf("in-place restore pre-flight check failed, skipping volume data restore: PVC %s/%s capacity %s is smaller than the backed-up volume size %d bytes",
+			existingPVC.Namespace, existingPVC.Name, capacity.String(), sourceSize)
+	}
+	return nil
+}

pkg/restore/inplace/preflight_test.go

@@ -22,6 +22,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	corev1api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
@@ -263,3 +264,67 @@ func TestCheckPVCBoundToBackedUpPV(t *testing.T) {
 		})
 	}
 }
+
+func TestCheckPVCCapacity(t *testing.T) {
+	pvc := func(capacity string) *corev1api.PersistentVolumeClaim {
+		p := &corev1api.PersistentVolumeClaim{
+			ObjectMeta: metav1.ObjectMeta{Name: "pvc-1", Namespace: "default"},
+		}
+		if capacity != "" {
+			p.Status.Capacity = corev1api.ResourceList{corev1api.ResourceStorage: resource.MustParse(capacity)}
+		}
+		return p
+	}
+	const mi = int64(1 << 20)
+
+	tests := []struct {
+		name        string
+		existingPVC *corev1api.PersistentVolumeClaim
+		sourceSize  int64
+		expectError string
+	}{
+		{
+			name:        "capacity larger than source volume, check passes",
+			existingPVC: pvc("100Mi"),
+			sourceSize:  50 * mi,
+		},
+		{
+			name:        "capacity equal to source volume, check passes",
+			existingPVC: pvc("100Mi"),
+			sourceSize:  100 * mi,
+		},
+		{
+			name:        "capacity smaller than source volume, check fails",
+			existingPVC: pvc("50Mi"),
+			sourceSize:  100 * mi,
+			expectError: "capacity 50Mi is smaller than the backed-up volume size 104857600 bytes",
+		},
+		{
+			name:        "unknown source size is skipped",
+			existingPVC: pvc("50Mi"),
+			sourceSize:  0,
+		},
+		{
+			name:        "missing capacity is skipped",
+			existingPVC: pvc(""),
+			sourceSize:  100 * mi,
+		},
+		{
+			name:        "capacity in decimal units compares by value",
+			existingPVC: pvc("104857600"),
+			sourceSize:  100 * mi,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			err := CheckPVCCapacity(tc.existingPVC, tc.sourceSize)
+			if tc.expectError == "" {
+				require.NoError(t, err)
+				return
+			}
+			require.Error(t, err)
+			assert.Contains(t, err.Error(), tc.expectError)
+		})
+	}
+}

pkg/restore/restore.go

@@ -27,6 +27,7 @@ import (
 	"reflect"
 	"slices"
 	"sort"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -1636,15 +1637,23 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso
 		return warnings, errs, itemExists
 	}
 
-	// Strip any pre-existing Velero-internal in-place restore carrier annotation coming from
-	// the backup metadata before RestoreItemActions run. The carrier is only trusted when it
-	// is set by a RestoreItemAction (the PVC CSI RIA) during this restore; a stale carrier
-	// baked into the backup must not be translated into the Kubernetes "selected-node"
-	// annotation, which could pin a newly provisioned PVC to a stale node.
-	if annotations := obj.GetAnnotations(); annotations != nil {
-		if _, present := annotations[velerov1api.InplaceRestoreSelectedNodeAnnotation]; present {
-			restoreLogger.Infof("Removing pre-existing %q annotation from backup metadata", velerov1api.InplaceRestoreSelectedNodeAnnotation)
-			delete(annotations, velerov1api.InplaceRestoreSelectedNodeAnnotation)
+	// Strip any pre-existing Velero-internal in-place restore carrier annotations coming from
+	// the backup metadata before RestoreItemActions run. A carrier is only trusted when it is
+	// set during this restore (by the engine below or by the PVC CSI RIA); a stale carrier
+	// baked into the backup must not be acted on, e.g. a stale "selected-node" could pin a
+	// newly provisioned PVC to a stale node.
+	stripInplaceRestoreCarrierAnnotations(obj)
+
+	// Carry the source volume size from the backup volume info to the PVC CSI RIA, which has no
+	// access to the volume info, so it can run the in-place restore capacity pre-flight check.
+	if groupResource == kuberesource.PersistentVolumeClaims {
+		pvName, _, _ := unstructured.NestedString(obj.Object, "spec", "volumeName")
+		if sourceSize := ctx.backupVolumeInfoMap[pvName].SourceSize(); sourceSize > 0 {
+			annotations := obj.GetAnnotations()
+			if annotations == nil {
+				annotations = map[string]string{}
+			}
+			annotations[velerov1api.InplaceRestoreSourceSizeAnnotation] = strconv.FormatInt(sourceSize, 10)
 			obj.SetAnnotations(annotations)
 		}
 	}
@@ -1788,15 +1797,13 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso
 	// while the carrier annotation passes through untouched. The carrier itself is always
 	// stripped so it never lands on the cluster.
 	if annotations := obj.GetAnnotations(); annotations != nil {
-		if selectedNode, present := annotations[velerov1api.InplaceRestoreSelectedNodeAnnotation]; present {
-			if selectedNode != "" {
-				restoreLogger.Infof("Restoring %q annotation with value %q from in-place restore carrier annotation", kube.KubeAnnSelectedNode, selectedNode)
-				annotations[kube.KubeAnnSelectedNode] = selectedNode
-			}
-			delete(annotations, velerov1api.InplaceRestoreSelectedNodeAnnotation)
+		if selectedNode := annotations[velerov1api.InplaceRestoreSelectedNodeAnnotation]; selectedNode != "" {
+			restoreLogger.Infof("Restoring %q annotation with value %q from in-place restore carrier annotation", kube.KubeAnnSelectedNode, selectedNode)
+			annotations[kube.KubeAnnSelectedNode] = selectedNode
 			obj.SetAnnotations(annotations)
 		}
 	}
+	stripInplaceRestoreCarrierAnnotations(obj)
 
 	// This comes after running item actions because we have built-in actions that restore
 	// a PVC's associated PV (if applicable). As part of the PV being restored, the 'pvsToProvision'
@@ -2513,6 +2520,22 @@ func resetMetadataAndStatus(obj *unstructured.Unstructured) (*unstructured.Unstr
 	return obj, nil
 }
 
+// inplaceRestoreCarrierAnnotations are the Velero-internal annotations used to pass data
+// between the restore engine and the in-place restore RestoreItemActions. They never land on
+// the cluster.
+var inplaceRestoreCarrierAnnotations = []string{
+	velerov1api.InplaceRestoreSelectedNodeAnnotation,
+	velerov1api.InplaceRestoreSourceSizeAnnotation,
+}
+
+func stripInplaceRestoreCarrierAnnotations(obj metav1.Object) {
+	annotations := obj.GetAnnotations()
+	for _, k := range inplaceRestoreCarrierAnnotations {
+		delete(annotations, k)
+	}
+	obj.SetAnnotations(annotations)
+}
+
 // addRestoreLabels labels the provided object with the restore name and the
 // restored backup's name.
 func addRestoreLabels(obj metav1.Object, restoreName, backupName string) {

pkg/restore/restore_test.go

@@ -5160,3 +5160,84 @@ func TestHasPodVolumeBackup(t *testing.T) {
 		})
 	}
 }
+
+func TestRestoreInplaceSourceSizeCarrierAnnotation(t *testing.T) {
+	newRequest := func(t *testing.T, h *harness, volumeInfos map[string]volume.BackupVolumeInfo) *Request {
+		t.Helper()
+		return &Request{
+			Log:     h.log,
+			Restore: defaultRestore().Result(),
+			Backup:  defaultBackup().Result(),
+			BackupReader: test.NewTarWriter(t).
+				AddItems("persistentvolumeclaims", builder.ForPersistentVolumeClaim("ns-1", "pvc-1").VolumeName("pv-1").Result()).
+				Done(),
+			BackupVolumeInfoMap: volumeInfos,
+		}
+	}
+
+	// captureCarrier records the source-size carrier the RIA sees on the item.
+	captureCarrier := func(seen *string) riav2.RestoreItemAction {
+		return &pluggableAction{
+			executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+				item := input.Item.(*unstructured.Unstructured)
+				*seen = item.GetAnnotations()[velerov1api.InplaceRestoreSourceSizeAnnotation]
+				return &velero.RestoreItemActionExecuteOutput{UpdatedItem: item}, nil
+			},
+		}
+	}
+
+	t.Run("source size from volume info is carried to RIAs and stripped from the cluster object", func(t *testing.T) {
+		h := newHarness(t)
+		h.AddItems(t, test.PVCs())
+		var seen string
+
+		warnings, errs := h.restorer.Restore(
+			newRequest(t, h, map[string]volume.BackupVolumeInfo{
+				"pv-1": {PVCNamespace: "ns-1", PVCName: "pvc-1", PVBInfo: &volume.PodVolumeBackupInfo{SourceSize: 31457288}},
+			}),
+			[]riav2.RestoreItemAction{captureCarrier(&seen)},
+			nil,
+		)
+		assertEmptyResults(t, warnings, errs)
+		assert.Equal(t, "31457288", seen)
+
+		got, err := h.DynamicClient.Resource(test.PVCs().GVR()).Namespace("ns-1").Get(t.Context(), "pvc-1", metav1.GetOptions{})
+		require.NoError(t, err)
+		assert.NotContains(t, got.GetAnnotations(), velerov1api.InplaceRestoreSourceSizeAnnotation)
+	})
+
+	t.Run("no carrier when the volume info has no source size", func(t *testing.T) {
+		h := newHarness(t)
+		h.AddItems(t, test.PVCs())
+		var seen string
+
+		warnings, errs := h.restorer.Restore(
+			newRequest(t, h, map[string]volume.BackupVolumeInfo{
+				"pv-1": {PVCNamespace: "ns-1", PVCName: "pvc-1", PVBInfo: &volume.PodVolumeBackupInfo{}},
+			}),
+			[]riav2.RestoreItemAction{captureCarrier(&seen)},
+			nil,
+		)
+		assertEmptyResults(t, warnings, errs)
+		assert.Empty(t, seen)
+	})
+
+	t.Run("stale carrier from the backup metadata is not trusted", func(t *testing.T) {
+		h := newHarness(t)
+		h.AddItems(t, test.PVCs())
+		var seen string
+
+		req := newRequest(t, h, nil)
+		req.BackupReader = test.NewTarWriter(t).
+			AddItems("persistentvolumeclaims", builder.ForPersistentVolumeClaim("ns-1", "pvc-1").
+				ObjectMeta(builder.WithAnnotations(velerov1api.InplaceRestoreSourceSizeAnnotation, "999")).Result()).
+			Done()
+		warnings, errs := h.restorer.Restore(req, []riav2.RestoreItemAction{captureCarrier(&seen)}, nil)
+		assertEmptyResults(t, warnings, errs)
+		assert.Empty(t, seen)
+
+		got, err := h.DynamicClient.Resource(test.PVCs().GVR()).Namespace("ns-1").Get(t.Context(), "pvc-1", metav1.GetOptions{})
+		require.NoError(t, err)
+		assert.NotContains(t, got.GetAnnotations(), velerov1api.InplaceRestoreSourceSizeAnnotation)
+	})
+}