Implement original VolumeSnapshotContent deletion for legacy backups

Signed-off-by: Priyansh Choudhary <im1706@gmail.com>

internal/delete/actions/csi/volumesnapshotcontent_action.go

@@ -81,6 +81,17 @@ func (p *volumeSnapshotContentDeleteItemAction) Execute(
 
 	p.log.Infof("Deleting VolumeSnapshotContent %s", snapCont.Name)
 
+	// Try to delete the original VSC from the cluster first.
+	// This handles legacy (pre-1.15) backups where the original VSC
+	// with DeletionPolicy=Retain still exists in the cluster.
+	originalVSCName := snapCont.Name
+	if cleaned := p.tryDeleteOriginalVSC(context.TODO(), originalVSCName); cleaned {
+		p.log.Infof("Successfully deleted original VolumeSnapshotContent %s from cluster, skipping temp VSC creation", originalVSCName)
+		return nil
+	}
+
+	// create a temp VSC to trigger cloud snapshot deletion
+	// (for backups where the original VSC no longer exists in cluster)
 	uuid, err := uuid.NewRandom()
 	if err != nil {
 		p.log.WithError(err).Errorf("Fail to generate the UUID to create VSC %s", snapCont.Name)
@@ -155,6 +166,47 @@ func (p *volumeSnapshotContentDeleteItemAction) Execute(
 	return nil
 }
 
+// tryDeleteOriginalVSC attempts to find and delete the original VSC from
+// the cluster (legacy pre-1.15 backups). It patches the DeletionPolicy to
+// Delete so the CSI driver also removes the cloud snapshot, then deletes
+// the VSC object itself.
+// Returns true if the original VSC was found and deletion was initiated.
+func (p *volumeSnapshotContentDeleteItemAction) tryDeleteOriginalVSC(
+	ctx context.Context,
+	vscName string,
+) bool {
+	existing := new(snapshotv1api.VolumeSnapshotContent)
+	if err := p.crClient.Get(ctx, crclient.ObjectKey{Name: vscName}, existing); err != nil {
+		if apierrors.IsNotFound(err) {
+			p.log.Debugf("Original VolumeSnapshotContent %s not found in cluster, will use temp VSC flow", vscName)
+		} else {
+			p.log.Debugf("Error looking up original VolumeSnapshotContent %s, will use temp VSC flow", vscName)
+		}
+		return false
+	}
+
+	p.log.Debugf("Found original VolumeSnapshotContent %s in cluster (legacy backup), cleaning up directly", vscName)
+
+	// Patch DeletionPolicy to Delete so the CSI driver removes the cloud snapshot
+	if existing.Spec.DeletionPolicy != snapshotv1api.VolumeSnapshotContentDelete {
+		original := existing.DeepCopy()
+		existing.Spec.DeletionPolicy = snapshotv1api.VolumeSnapshotContentDelete
+		if err := p.crClient.Patch(ctx, existing, crclient.MergeFrom(original)); err != nil {
+			p.log.WithError(err).Debugf("Failed to patch DeletionPolicy on original VSC %s, will use temp VSC flow", vscName)
+			return false
+		}
+		p.log.Debugf("Patched DeletionPolicy to Delete on original VolumeSnapshotContent %s", vscName)
+	}
+
+	// Delete the original VSC — the CSI driver will clean up the cloud snapshot
+	if err := p.crClient.Delete(ctx, existing); err != nil && !apierrors.IsNotFound(err) {
+		p.log.WithError(err).Debugf("Failed to delete original VolumeSnapshotContent %s, will use temp VSC flow", vscName)
+		return false
+	}
+
+	return true
+}
+
 var checkVSCReadiness = func(
 	ctx context.Context,
 	vsc *snapshotv1api.VolumeSnapshotContent,

internal/delete/actions/csi/volumesnapshotcontent_action_test.go

@@ -25,6 +25,8 @@ import (
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/require"
+	corev1api "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -239,6 +241,88 @@ func TestCheckVSCReadiness(t *testing.T) {
 	}
 }
 
+func TestTryDeleteOriginalVSC(t *testing.T) {
+	tests := []struct {
+		name      string
+		vscName   string
+		existing  *snapshotv1api.VolumeSnapshotContent
+		createIt  bool
+		expectRet bool
+	}{
+		{
+			name:      "VSC not found in cluster, returns false",
+			vscName:   "not-found",
+			expectRet: false,
+		},
+		{
+			name:    "VSC found with Retain policy, patches and deletes",
+			vscName: "legacy-vsc",
+			existing: &snapshotv1api.VolumeSnapshotContent{
+				ObjectMeta: metav1.ObjectMeta{Name: "legacy-vsc"},
+				Spec: snapshotv1api.VolumeSnapshotContentSpec{
+					DeletionPolicy: snapshotv1api.VolumeSnapshotContentRetain,
+					Driver:         "disk.csi.azure.com",
+					Source: snapshotv1api.VolumeSnapshotContentSource{
+						SnapshotHandle: stringPtr("snap-123"),
+					},
+					VolumeSnapshotRef: corev1api.ObjectReference{
+						Name:      "vs-1",
+						Namespace: "default",
+					},
+				},
+			},
+			createIt:  true,
+			expectRet: true,
+		},
+		{
+			name:    "VSC found with Delete policy already, just deletes",
+			vscName: "already-delete-vsc",
+			existing: &snapshotv1api.VolumeSnapshotContent{
+				ObjectMeta: metav1.ObjectMeta{Name: "already-delete-vsc"},
+				Spec: snapshotv1api.VolumeSnapshotContentSpec{
+					DeletionPolicy: snapshotv1api.VolumeSnapshotContentDelete,
+					Driver:         "disk.csi.azure.com",
+					Source: snapshotv1api.VolumeSnapshotContentSource{
+						SnapshotHandle: stringPtr("snap-456"),
+					},
+					VolumeSnapshotRef: corev1api.ObjectReference{
+						Name:      "vs-2",
+						Namespace: "default",
+					},
+				},
+			},
+			createIt:  true,
+			expectRet: true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			crClient := velerotest.NewFakeControllerRuntimeClient(t)
+			logger := logrus.StandardLogger()
+			p := &volumeSnapshotContentDeleteItemAction{
+				log:      logger,
+				crClient: crClient,
+			}
+
+			if test.createIt && test.existing != nil {
+				require.NoError(t, crClient.Create(t.Context(), test.existing))
+			}
+
+			result := p.tryDeleteOriginalVSC(t.Context(), test.vscName)
+			require.Equal(t, test.expectRet, result)
+
+			// If cleanup succeeded, verify the VSC is gone
+			if test.expectRet {
+				err := crClient.Get(t.Context(), crclient.ObjectKey{Name: test.vscName},
+					&snapshotv1api.VolumeSnapshotContent{})
+				require.True(t, apierrors.IsNotFound(err),
+					"VSC should have been deleted from cluster")
+			}
+		})
+	}
+}
+
 func boolPtr(b bool) *bool {
 	return &b
 }