fix: do not restore PV verbatim when volume data was skipped via VolumePolicy

When a VolumePolicy action=skip is used during backup, the volume data
is intentionally not backed up. During restore, Velero was restoring
the original PV identity (same VolumeHandle) which is dangerous:
- With Delete reclaim policy: underlying storage may no longer exist
- In cross-cluster restore: two clusters would share the same storage

This adds defense-in-depth in pv_restorer.go: when no snapshot is found
and the PV has a Delete reclaim policy, return errPVNeedsReprovisioning
to trigger dynamic re-provisioning instead of restoring the PV as-is.
The callers in restore.go (both new and legacy paths) handle this error
by inserting the PV into pvsToProvision.

For Retain reclaim policy PVs without snapshots, a warning is now logged
about the risks of cross-cluster restore scenarios.

Fixes #9318

Signed-off-by: Mateen Ali Anjum <mateenali66@gmail.com>

Author: mateenali66

pkg/restore/pv_restorer.go

@@ -18,6 +18,7 @@ package restore
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
@@ -31,6 +32,11 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
 )
 
+// errPVNeedsReprovisioning is returned when a PV has no snapshot and a Delete
+// reclaim policy, indicating the underlying storage may no longer exist and the
+// PV should be dynamically re-provisioned instead of restored as-is.
+var errPVNeedsReprovisioning = fmt.Errorf("persistent volume has no snapshot and Delete reclaim policy; it should be dynamically re-provisioned")
+
 type PVRestorer interface {
 	executePVAction(obj *unstructured.Unstructured) (*unstructured.Unstructured, error)
 }
@@ -65,6 +71,10 @@ func (r *pvRestorer) executePVAction(obj *unstructured.Unstructured) (*unstructu
 	}
 	if snapshotInfo == nil {
 		log.Infof("No snapshot found for persistent volume")
+		if hasDeleteReclaimPolicy(obj.Object) {
+			log.Warnf("Persistent volume has Delete reclaim policy but no snapshot; underlying storage may no longer exist, triggering dynamic re-provisioning")
+			return nil, errPVNeedsReprovisioning
+		}
 		return obj, nil
 	}
 

pkg/restore/pv_restorer_test.go

@@ -49,6 +49,7 @@ func TestExecutePVAction_NoSnapshotRestores(t *testing.T) {
 		volumeSnapshots []*volume.Snapshot
 		locations       []*api.VolumeSnapshotLocation
 		expectedErr     bool
+		expectedErrIs   error
 		expectedRes     *unstructured.Unstructured
 	}{
 		{
@@ -112,6 +113,23 @@ func TestExecutePVAction_NoSnapshotRestores(t *testing.T) {
 			},
 			expectedRes: newTestUnstructured().WithName("pv-1").WithSpec().Unstructured,
 		},
+		{
+			name:            "no snapshot and Delete reclaim policy: return errPVNeedsReprovisioning",
+			obj:             newTestUnstructured().WithName("pv-1").WithSpec().WithSpecField("persistentVolumeReclaimPolicy", "Delete").Unstructured,
+			restore:         builder.ForRestore(api.DefaultNamespace, "").RestorePVs(true).Result(),
+			backup:          defaultBackup().Result(),
+			volumeSnapshots: []*volume.Snapshot{},
+			expectedErr:     true,
+			expectedErrIs:   errPVNeedsReprovisioning,
+		},
+		{
+			name:            "no snapshot and Retain reclaim policy: return PV as-is",
+			obj:             newTestUnstructured().WithName("pv-1").WithSpec().WithSpecField("persistentVolumeReclaimPolicy", "Retain").Unstructured,
+			restore:         builder.ForRestore(api.DefaultNamespace, "").RestorePVs(true).Result(),
+			backup:          defaultBackup().Result(),
+			volumeSnapshots: []*volume.Snapshot{},
+			expectedRes:     newTestUnstructured().WithName("pv-1").WithSpec().WithSpecField("persistentVolumeReclaimPolicy", "Retain").Unstructured,
+		},
 	}
 
 	for _, tc := range tests {
@@ -135,6 +153,9 @@ func TestExecutePVAction_NoSnapshotRestores(t *testing.T) {
 			case true:
 				assert.Nil(t, res)
 				require.Error(t, err)
+				if tc.expectedErrIs != nil {
+					assert.Equal(t, tc.expectedErrIs, err)
+				}
 			case false:
 				assert.Equal(t, tc.expectedRes, res)
 				assert.NoError(t, err)

pkg/restore/restore.go

@@ -1223,6 +1223,11 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso
 			case volume.NativeSnapshot:
 				obj, err = ctx.handlePVHasNativeSnapshot(obj, resourceClient)
 				if err != nil {
+					if err == errPVNeedsReprovisioning {
+						restoreLogger.Infof("Dynamically re-provisioning persistent volume because it has Delete reclaim policy and no snapshot data.")
+						ctx.pvsToProvision.Insert(backupResourceName)
+						return warnings, errs, itemExists
+					}
 					errs.Add(namespace, err)
 					return warnings, errs, itemExists
 				}
@@ -1270,6 +1275,11 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso
 			case hasSnapshot(backupResourceName, ctx.volumeSnapshots):
 				obj, err = ctx.handlePVHasNativeSnapshot(obj, resourceClient)
 				if err != nil {
+					if err == errPVNeedsReprovisioning {
+						restoreLogger.Infof("Dynamically re-provisioning persistent volume because it has Delete reclaim policy and no snapshot data.")
+						ctx.pvsToProvision.Insert(backupResourceName)
+						return warnings, errs, itemExists
+					}
 					errs.Add(namespace, err)
 					return warnings, errs, itemExists
 				}
@@ -2557,6 +2567,9 @@ func (ctx *restoreContext) handlePVHasNativeSnapshot(obj *unstructured.Unstructu
 		ctx.log.Infof("Restoring persistent volume from snapshot.")
 		retObj, err = ctx.pvRestorer.executePVAction(retObj)
 		if err != nil {
+			if err == errPVNeedsReprovisioning {
+				return nil, err
+			}
 			return nil, fmt.Errorf("error executing PVAction for %s: %v", getResourceID(kuberesource.PersistentVolumes, "", oldName), err)
 		}
 
@@ -2600,6 +2613,11 @@ func (ctx *restoreContext) handleSkippedPVHasRetainPolicy(
 	logger logrus.FieldLogger,
 ) (*unstructured.Unstructured, error) {
 	logger.Infof("Restoring persistent volume as-is because it doesn't have a snapshot and its reclaim policy is not Delete.")
+	logger.Warnf("Warning: restoring PV %s with its original volume identity (VolumeHandle). "+
+		"If this volume's data was intentionally skipped during backup (e.g. via VolumePolicy action=skip), "+
+		"be aware that in cross-cluster restore scenarios this may cause two clusters to share the same "+
+		"underlying storage. Consider using Delete reclaim policy for volumes that are skipped during backup.",
+		obj.GetName())
 
 	// Check to see if the claimRef.namespace field needs to be remapped, and do so if necessary.
 	if _, err := remapClaimRefNS(ctx, obj); err != nil {