Merge pull request #10512 from chlins/feat/inplace-preflight-check3

Add in-place restore pre-flight check: PVC must be large enough for the backed-up data

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

@@ -365,6 +365,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)
 		}
@@ -728,6 +732,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)
+	})
+}