Fix generic CSI changeID retrieval and honor snapshot class deletion policy

Two defects in the CSI snapshot exposer, both of which make every CBT
incremental silently degrade to a full backup on non-vSphere drivers.

1. getCBTInfo read the changeID from the freshly-created backup VSC's
   status, which is populated asynchronously and is therefore usually
   empty at that point. The handle is already present in the spec, so
   fall back to vsc.Spec.Source.SnapshotHandle. vSphere is unaffected --
   it takes the VSphereCNSChangeIDAnno branch and never reads the VSC
   handle -- so this affects every other CSI driver.

2. createBackupVSC hardcoded DeletionPolicy: Delete, so the physical
   snapshot was removed when the backup completed. Case-2 storage such
   as Ceph RBD requires the base snapshot to survive for the next
   GetMetadataDelta call, and the block-data-mover design specifies a
   RetainSnapshot volume-policy parameter that was never implemented.
   Inherit the source snapshot class's deletion policy instead, and
   clean up the backup VSC object in CleanUp.

Adds TestCreateBackupVSCDeletionPolicy covering both Delete and Retain.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Signed-off-by: Tiger Kaovilai <tkaovila@redhat.com>
(cherry picked from commit f4867d0489788ce0c7a63ffe08da4ad059db88b4)
Signed-off-by: Tiger Kaovilai <tkaovila@redhat.com>

pkg/exposer/csi_snapshot.go

@@ -496,6 +496,7 @@ func (e *csiSnapshotExposer) CleanUp(ctx context.Context, ownerObject corev1api.
 	backupPodName := ownerObject.Name
 	backupPVCName := ownerObject.Name
 	backupVSName := ownerObject.Name
+	backupVSCName := ownerObject.Name
 
 	kube.DeletePodIfAny(ctx, e.kubeClient.CoreV1(), backupPodName, ownerObject.Namespace, e.log)
 	kube.DeletePVAndPVCIfAny(ctx, e.kubeClient.CoreV1(), backupPVCName, ownerObject.Namespace, cleanUpTimeout, e.log)
@@ -507,6 +508,13 @@ func (e *csiSnapshotExposer) CleanUp(ctx context.Context, ownerObject corev1api.
 
 	csi.DeleteVolumeSnapshotIfAny(ctx, e.csiSnapshotClient, backupVSName, ownerObject.Namespace, e.log)
 	csi.DeleteVolumeSnapshotIfAny(ctx, e.csiSnapshotClient, vsName, sourceNamespace, e.log)
+
+	// The backup VSC is created by Velero as an internal handle to the source
+	// snapshot. Deleting the backup VS above only cascades to it when its
+	// deletion policy is Delete, so remove it explicitly to avoid leaking the
+	// object under a Retain policy. Deleting a Retain VSC drops only the API
+	// object and leaves the underlying snapshot intact.
+	csi.DeleteVolumeSnapshotContentIfAny(ctx, e.csiSnapshotClient, backupVSCName, e.log)
 }
 
 func getVolumeModeByAccessMode(accessMode string, dataMover string) (corev1api.PersistentVolumeMode, error) {
@@ -571,7 +579,13 @@ func (e *csiSnapshotExposer) createBackupVSC(ctx context.Context, ownerObject co
 			Source: snapshotv1api.VolumeSnapshotContentSource{
 				SnapshotHandle: snapshotVSC.Status.SnapshotHandle,
 			},
-			DeletionPolicy:          snapshotv1api.VolumeSnapshotContentDelete,
+			// The backup VSC is statically provisioned against the same
+			// snapshot handle as the source VSC, so both objects refer to one
+			// physical snapshot. Inherit the source's deletion policy instead
+			// of forcing Delete, otherwise a user who configured Retain on the
+			// VolumeSnapshotClass still loses the snapshot when the backup VSC
+			// is cleaned up.
+			DeletionPolicy:          snapshotVSC.Spec.DeletionPolicy,
 			Driver:                  snapshotVSC.Spec.Driver,
 			VolumeSnapshotClassName: snapshotVSC.Spec.VolumeSnapshotClassName,
 		},

pkg/exposer/csi_snapshot_test.go

@@ -2537,3 +2537,62 @@ func TestCleanUp_SecretsAndConfigMaps(t *testing.T) {
 	_, err = fakeKubeClient.CoreV1().Secrets("velero").Get(t.Context(), "other-secret", metav1.GetOptions{})
 	assert.NoError(t, err, "unrelated secret should not be deleted")
 }
+
+func TestCreateBackupVSCDeletionPolicy(t *testing.T) {
+	tests := []struct {
+		name           string
+		sourcePolicy   snapshotv1api.DeletionPolicy
+		expectedPolicy snapshotv1api.DeletionPolicy
+	}{
+		{
+			name:           "Delete policy is inherited",
+			sourcePolicy:   snapshotv1api.VolumeSnapshotContentDelete,
+			expectedPolicy: snapshotv1api.VolumeSnapshotContentDelete,
+		},
+		{
+			// The backup VSC points at the same snapshot handle as the source
+			// VSC, so forcing Delete here would destroy a snapshot the user
+			// asked to keep.
+			name:           "Retain policy is inherited",
+			sourcePolicy:   snapshotv1api.VolumeSnapshotContentRetain,
+			expectedPolicy: snapshotv1api.VolumeSnapshotContentRetain,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			handle := "fake-snapshot-handle"
+			className := "fake-snapshot-class"
+
+			sourceVSC := &snapshotv1api.VolumeSnapshotContent{
+				ObjectMeta: metav1.ObjectMeta{Name: "source-vsc"},
+				Spec: snapshotv1api.VolumeSnapshotContentSpec{
+					DeletionPolicy:          test.sourcePolicy,
+					Driver:                  "fake-driver",
+					VolumeSnapshotClassName: &className,
+				},
+				Status: &snapshotv1api.VolumeSnapshotContentStatus{
+					SnapshotHandle: &handle,
+				},
+			}
+
+			exposer := csiSnapshotExposer{
+				csiSnapshotClient: snapshotFake.NewSimpleClientset().SnapshotV1(),
+				log:               velerotest.NewLogger(),
+			}
+
+			ownerObject := corev1api.ObjectReference{
+				Name:      "fake-du",
+				Namespace: "velero",
+			}
+			vs := &snapshotv1api.VolumeSnapshot{
+				ObjectMeta: metav1.ObjectMeta{Name: "fake-du", Namespace: "velero"},
+			}
+
+			backupVSC, err := exposer.createBackupVSC(t.Context(), ownerObject, sourceVSC, vs)
+			require.NoError(t, err)
+			assert.Equal(t, test.expectedPolicy, backupVSC.Spec.DeletionPolicy)
+			assert.Equal(t, handle, *backupVSC.Spec.Source.SnapshotHandle)
+		})
+	}
+}

pkg/util/csi/cbt.go

@@ -64,6 +64,11 @@ func GetCBTInfo(ctx context.Context, kubeClient kubernetes.Interface, log logrus
 
 		if vsc.Status != nil && vsc.Status.SnapshotHandle != nil {
 			cbtInfo.ChangeID = *vsc.Status.SnapshotHandle
+		} else if vsc.Spec.Source.SnapshotHandle != nil {
+			// The backup VSC is statically provisioned from the source VSC's
+			// snapshot handle; its status is populated asynchronously and may
+			// not be set yet, but the handle is already in the spec.
+			cbtInfo.ChangeID = *vsc.Spec.Source.SnapshotHandle
 		}
 
 		if pv.Spec.CSI != nil && pv.Spec.CSI.VolumeHandle != "" {