Fail the in-place restore pre-flight check when the backed-up pod already exists on the file system restore path (#10550)

PodVolumeRestores are only created for pods that Velero creates, so when
the pod already exists in the cluster the PVC-not-in-use pre-flight check
never runs and the volume data restore is skipped silently, while the
existing pod keeps consuming the PVC. Report an explicit pre-flight error
for such pods, aligned with the PVC CSI RIA behavior.

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

changelogs/unreleased/10550-chlins

@@ -0,0 +1 @@
+Fail the in-place restore pre-flight check when the backed-up pod exists

design/volume-data-inplace-restore/volume-data-inplace-restore.md

@@ -242,6 +242,8 @@ The "in use" semantics align with the Kubernetes `pvc-protection` controller: Po
 
 The check runs on both restore paths before any side effect on the existing PVC/PV: in the PVC CSI RIA before deleting the existing PVC, and before creating the `PodVolumeRestore` on the file system path. On the file system path, Pods gated by this restore's `restore-wait` init container (identified by the restore UID in its args, and not yet terminated) are exempted: they must mount the PVC for the node-agent to restore the data, and they cannot write to the volume until this restore's `PodVolumeRestore`s complete. Leftover Pods, controller-recreated Pods, and Pods gated by a different restore still block.
 
+On the file system path the check also covers the case where the backed-up Pod itself still exists in the cluster: `PodVolumeRestore`s are only created for a Pod that Velero creates, so an existing Pod would otherwise cause the volume data restore to be skipped silently while the Pod keeps consuming the PVC. In this case Velero reports a pre-flight error for the Pod instead of the plain "already exists" warning.
+
 This check is a fail-fast validation, not an atomic guarantee; the `pvc-protection` finalizer remains the actual safety gate for PVC deletion. A residual `VolumeAttachment` check (e.g. a `Failed` Pod imposed by the control plane after a non-graceful node shutdown, where the node never unmounted the volume) may be added as a future enhancement.
 
 #### 2. PVC is Bound to the Original PV

pkg/restore/restore.go

@@ -1945,6 +1945,31 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso
 		itemStatus := ctx.restoredItems[itemKey]
 		itemStatus.itemExists = itemExists
 		ctx.restoredItems[itemKey] = itemStatus
+
+		// PodVolumeRestores are only created for pods Velero creates, so an
+		// existing pod silently skips the volume data restore. For an in-place
+		// restore this fails the pre-flight check: the pod is still consuming
+		// the PVCs that were supposed to be restored in place. Otherwise it is
+		// only worth a warning.
+		if newGR == kuberesource.Pods {
+			pod := new(corev1api.Pod)
+			if err := runtime.DefaultUnstructuredConverter.FromUnstructured(obj.UnstructuredContent(), pod); err != nil {
+				errs.Add(namespace, err)
+				return warnings, errs, itemExists
+			}
+			if len(podvolume.GetVolumeBackupsForPod(ctx.podVolumeBackups, pod, originalNamespace)) > 0 {
+				if ctx.restore.IsVolumeDataInplaceRestore() {
+					err := errors.Errorf("in-place restore pre-flight check failed, skipping volume data restore: pod %s already exists and is still using the backed-up volumes: delete the pod and its owning workload and retry", kube.NamespaceAndName(obj))
+					restoreLogger.Error(err.Error())
+					errs.Add(namespace, err)
+				} else {
+					err := errors.Errorf("skipping volume data restore: pod %s already exists, its PodVolumeBackups will not be restored", kube.NamespaceAndName(obj))
+					restoreLogger.Warn(err.Error())
+					warnings.Add(namespace, err)
+				}
+			}
+		}
+
 		// Remove insubstantial metadata.
 		fromCluster, err = resetMetadataAndStatus(fromCluster)
 		if err != nil {

pkg/restore/restore_test.go

@@ -4333,6 +4333,99 @@ func TestRestoreWithPodVolume(t *testing.T) {
 	}
 }
 
+// TestRestoreInplaceExistingPodWithPodVolumeBackups verifies that an in-place
+// restore reports an error when the backed-up pod has PodVolumeBackups to
+// restore but already exists in the cluster: the PodVolumeRestores are never
+// created for an existing pod, so the volume data restore must not be skipped
+// silently. A regular restore keeps the plain "already exists" warning.
+func TestRestoreInplaceExistingPodWithPodVolumeBackups(t *testing.T) {
+	pvbs := []*velerov1api.PodVolumeBackup{
+		builder.ForPodVolumeBackup("velero", "pvb-1").PodName("pod-1").PodNamespace("ns-1").Volume("data").SnapshotID("foo").Result(),
+	}
+	backedUpPod := builder.ForPod("ns-1", "pod-1").
+		Volumes(builder.ForVolume("data").PersistentVolumeClaimSource("pvc-1").Result()).
+		Result()
+	existingPod := backedUpPod.DeepCopy()
+	existingPod.Spec.NodeName = "node-1"
+
+	tests := []struct {
+		name         string
+		restore      *velerov1api.Restore
+		pvbs         []*velerov1api.PodVolumeBackup
+		wantErrs     bool
+		wantWarnings bool
+		wantPVBSkip  bool
+	}{
+		{
+			name:     "in-place restore with an existing pod consuming the backed-up volumes fails the pre-flight check",
+			restore:  defaultRestore().ExistingVolumeDataPolicy(string(velerov1api.VolumeDataPolicyTypeFull)).Result(),
+			pvbs:     pvbs,
+			wantErrs: true,
+		},
+		{
+			name:     "in-place restore with existingResourcePolicy=update and an existing pod still fails the pre-flight check",
+			restore:  defaultRestore().ExistingVolumeDataPolicy(string(velerov1api.VolumeDataPolicyTypeFull)).ExistingResourcePolicy(string(velerov1api.ResourcePolicyTypeUpdate)).Result(),
+			pvbs:     pvbs,
+			wantErrs: true,
+		},
+		{
+			name:         "in-place restore with an existing pod without PodVolumeBackups only warns",
+			restore:      defaultRestore().ExistingVolumeDataPolicy(string(velerov1api.VolumeDataPolicyTypeIncremental)).Result(),
+			pvbs:         nil,
+			wantWarnings: true,
+		},
+		{
+			name:         "regular restore with an existing pod warns that the PodVolumeBackups are not restored",
+			restore:      defaultRestore().Result(),
+			pvbs:         pvbs,
+			wantWarnings: true,
+			wantPVBSkip:  true,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			h := newHarness(t)
+			restorer := new(uploadermocks.Restorer)
+			defer restorer.AssertExpectations(t)
+			h.restorer.podVolumeRestorerFactory = &fakePodVolumeRestorerFactory{restorer: restorer}
+
+			h.AddItems(t, test.Pods(existingPod))
+
+			tarball := test.NewTarWriter(t)
+			tarball.AddItems("pods", backedUpPod)
+
+			warnings, errs := h.restorer.Restore(
+				&Request{
+					Log:              h.log,
+					Restore:          tc.restore,
+					Backup:           defaultBackup().Result(),
+					PodVolumeBackups: tc.pvbs,
+					BackupReader:     tarball.Done(),
+				},
+				nil,
+				nil,
+			)
+
+			if tc.wantErrs {
+				require.Len(t, errs.Namespaces["ns-1"], 1)
+				assert.Contains(t, errs.Namespaces["ns-1"][0], "in-place restore pre-flight check failed")
+				assert.Contains(t, errs.Namespaces["ns-1"][0], "pod ns-1/pod-1 already exists")
+			} else {
+				assert.Empty(t, errs.Namespaces)
+			}
+			if tc.wantPVBSkip {
+				require.Len(t, warnings.Namespaces["ns-1"], 2)
+				assert.Contains(t, warnings.Namespaces["ns-1"][0], "PodVolumeBackups will not be restored")
+				assert.Contains(t, warnings.Namespaces["ns-1"][1], "already exists")
+			} else if tc.wantWarnings {
+				require.Len(t, warnings.Namespaces["ns-1"], 1)
+				assert.Contains(t, warnings.Namespaces["ns-1"][0], "already exists")
+			}
+		})
+	}
+}
+
 func TestResetMetadata(t *testing.T) {
 	tests := []struct {
 		name        string