Refactor: Replace context.TODO() with properly plumbed contexts in CSI actions (#10247)

Signed-off-by: opbot_xd <awasthikrishna23052005@gmail.com>

changelogs/unreleased/10247-opbot-xd

@@ -0,0 +1 @@
+Refactor: Replace context.TODO() with properly plumbed contexts in CSI backup actions and utility functions to enable proper cancellation of in-flight API requests.

pkg/backup/actions/csi/pvc_action.go

@@ -212,6 +212,7 @@ func (p *pvcBackupItemAction) validatePVCAndPV(
 }
 
 func (p *pvcBackupItemAction) createVolumeSnapshot(
+	ctx context.Context,
 	pvc corev1api.PersistentVolumeClaim,
 	backup *velerov1api.Backup,
 	policySnapshotClass string,
@@ -222,7 +223,7 @@ func (p *pvcBackupItemAction) createVolumeSnapshot(
 	p.log.Debugf("Fetching storage class for PV %s", *pvc.Spec.StorageClassName)
 	storageClass := new(storagev1api.StorageClass)
 	if err := p.crClient.Get(
-		context.TODO(), crclient.ObjectKey{Name: *pvc.Spec.StorageClassName},
+		ctx, crclient.ObjectKey{Name: *pvc.Spec.StorageClassName},
 		storageClass,
 	); err != nil {
 		return nil, errors.Wrap(err, "error getting storage class")
@@ -230,6 +231,7 @@ func (p *pvcBackupItemAction) createVolumeSnapshot(
 
 	p.log.Debugf("Fetching VolumeSnapshotClass for %s", storageClass.Provisioner)
 	vsClass, err := csi.GetVolumeSnapshotClass(
+		ctx,
 		storageClass.Provisioner,
 		backup,
 		&pvc,
@@ -266,7 +268,7 @@ func (p *pvcBackupItemAction) createVolumeSnapshot(
 		},
 	}
 
-	if err := p.crClient.Create(context.TODO(), vs); err != nil {
+	if err := p.crClient.Create(ctx, vs); err != nil {
 		return nil, errors.Wrapf(
 			err, "error creating volume snapshot",
 		)
@@ -295,6 +297,8 @@ func (p *pvcBackupItemAction) Execute(
 ) {
 	p.log.Info("Starting PVCBackupItemAction")
 
+	ctx := context.Background()
+
 	if valid := p.validateBackup(*backup); !valid {
 		return item, nil, "", nil, nil
 	}
@@ -319,7 +323,7 @@ func (p *pvcBackupItemAction) Execute(
 	}
 
 	// Ensure PVC-to-Pod cache is built for this namespace (lazy per-namespace caching)
-	if err := p.ensurePVCPodCacheForNamespace(context.TODO(), pvc.Namespace); err != nil {
+	if err := p.ensurePVCPodCacheForNamespace(ctx, pvc.Namespace); err != nil {
 		return nil, nil, "", nil, err
 	}
 
@@ -347,7 +351,7 @@ func (p *pvcBackupItemAction) Execute(
 	// created but never processed (the DataUpload controller runs inside node-agent),
 	// causing the backup to hang until itemOperationTimeout expires.
 	if boolptr.IsSetToTrue(backup.Spec.SnapshotMoveData) && datamover.IsBuiltInDataMover(backup.Spec.DataMover) {
-		if err := nodeagent.IsReady(context.TODO(), backup.Namespace, p.crClient, p.log); err != nil {
+		if err := nodeagent.IsReady(ctx, backup.Namespace, p.crClient, p.log); err != nil {
 			p.log.WithError(err).Error("cannot perform snapshot data movement without running node-agent pods")
 			return nil, nil, "", nil, errors.Wrap(err, "CSI PVC BIA cannot proceed: node-agent is not ready for snapshot data movement")
 		}
@@ -360,7 +364,7 @@ func (p *pvcBackupItemAction) Execute(
 		p.log.Infof("Volume policy specifies snapshotClass=%s for PVC %s/%s", policySnapshotClass, pvc.Namespace, pvc.Name)
 	}
 
-	vs, err := p.getVolumeSnapshotReference(context.TODO(), pvc, backup, policySnapshotClass)
+	vs, err := p.getVolumeSnapshotReference(ctx, pvc, backup, policySnapshotClass)
 	if err != nil {
 		return nil, nil, "", nil, err
 	}
@@ -376,7 +380,7 @@ func (p *pvcBackupItemAction) Execute(
 	if err != nil {
 		p.log.Errorf("Failed to wait for VolumeSnapshot %s/%s to become ReadyToUse within timeout %v: %s",
 			vs.Namespace, vs.Name, backup.Spec.CSISnapshotTimeout.Duration, err.Error())
-		csi.CleanupVolumeSnapshot(vs, p.crClient, p.log)
+		csi.CleanupVolumeSnapshot(ctx, vs, p.crClient, p.log)
 		return nil, nil, "", nil, errors.WithStack(err)
 	}
 
@@ -427,7 +431,7 @@ func (p *pvcBackupItemAction) Execute(
 
 			// TODO: need to use DeleteVolumeSnapshotIfAny, after data mover
 			// adopting the controller-runtime client.
-			if deleteErr := p.crClient.Delete(context.TODO(), vs); deleteErr != nil {
+			if deleteErr := p.crClient.Delete(ctx, vs); deleteErr != nil {
 				if !apierrors.IsNotFound(deleteErr) {
 					dataUploadLog.WithError(deleteErr).Error("fail to delete VolumeSnapshot")
 				}
@@ -841,7 +845,7 @@ func (p *pvcBackupItemAction) getVolumeSnapshotReference(
 	}
 
 	// Legacy fallback: create individual VS
-	return p.createVolumeSnapshot(pvc, backup, policySnapshotClass)
+	return p.createVolumeSnapshot(ctx, pvc, backup, policySnapshotClass)
 }
 
 func (p *pvcBackupItemAction) findExistingVSForBackup(

pkg/backup/actions/csi/volumesnapshot_action.go

@@ -78,6 +78,8 @@ func (p *volumeSnapshotBackupItemAction) Execute(
 ) {
 	p.log.Infof("Executing VolumeSnapshotBackupItemAction")
 
+	ctx := context.Background()
+
 	vs := new(snapshotv1api.VolumeSnapshot)
 	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(
 		item.UnstructuredContent(), vs); err != nil {
@@ -90,7 +92,7 @@ func (p *volumeSnapshotBackupItemAction) Execute(
 			WithField("Backup", fmt.Sprintf("%s/%s", backup.Namespace, backup.Name)).
 			WithField("BackupPhase", backup.Status.Phase).Debugf("Cleaning VolumeSnapshots.")
 
-		csi.DeleteReadyVolumeSnapshot(*vs, p.crClient, p.log)
+		csi.DeleteReadyVolumeSnapshot(ctx, *vs, p.crClient, p.log)
 		return item, nil, "", nil, nil
 	}
 
@@ -115,11 +117,9 @@ func (p *volumeSnapshotBackupItemAction) Execute(
 	p.log.Infof("Getting VolumesnapshotContent for Volumesnapshot %s/%s",
 		vs.Namespace, vs.Name)
 
-	ctx := context.TODO()
-
 	vsc, err := csi.GetVSCForVS(ctx, vs, p.crClient)
 	if err != nil {
-		csi.CleanupVolumeSnapshot(vs, p.crClient, p.log)
+		csi.CleanupVolumeSnapshot(ctx, vs, p.crClient, p.log)
 		return nil, nil, "", nil, errors.WithStack(err)
 	}
 
@@ -187,7 +187,7 @@ func (p *volumeSnapshotBackupItemAction) Execute(
 		)
 
 		if vscPatchError := p.crClient.Patch(
-			context.TODO(),
+			ctx,
 			vsc,
 			crclient.MergeFrom(originVSC),
 		); vscPatchError != nil {
@@ -203,7 +203,7 @@ func (p *volumeSnapshotBackupItemAction) Execute(
 	originVS := vs.DeepCopy()
 	kubeutil.AddAnnotations(&vs.ObjectMeta, annotations)
 	if err := p.crClient.Patch(
-		context.TODO(),
+		ctx,
 		vs,
 		crclient.MergeFrom(originVS),
 	); err != nil {

pkg/controller/backup_deletion_controller.go

@@ -531,7 +531,7 @@ func (r *backupDeletionReconciler) deleteCSIVolumeSnapshotsIfAny(ctx context.Con
 	}
 	for _, item := range vsList.Items {
 		vs := item
-		csi.CleanupVolumeSnapshot(&vs, r.Client, log)
+		csi.CleanupVolumeSnapshot(ctx, &vs, r.Client, log)
 	}
 }
 

pkg/exposer/csi_snapshot.go

@@ -157,7 +157,7 @@ func (e *csiSnapshotExposer) Expose(ctx context.Context, ownerObject corev1api.O
 
 	curLog.Info("Volumesnapshot is ready")
 
-	vsc, err := csi.GetVolumeSnapshotContentForVolumeSnapshot(volumeSnapshot, e.csiSnapshotClient)
+	vsc, err := csi.GetVolumeSnapshotContentForVolumeSnapshot(ctx, volumeSnapshot, e.csiSnapshotClient)
 	if err != nil {
 		return errors.Wrap(err, "error to get volume snapshot content")
 	}

pkg/util/csi/volume_snapshot.go

@@ -112,6 +112,7 @@ func WaitVolumeSnapshotReady(
 // GetVolumeSnapshotContentForVolumeSnapshot returns the VolumeSnapshotContent
 // object associated with the VolumeSnapshot.
 func GetVolumeSnapshotContentForVolumeSnapshot(
+	ctx context.Context,
 	volSnap *snapshotv1api.VolumeSnapshot,
 	snapshotClient snapshotter.SnapshotV1Interface,
 ) (*snapshotv1api.VolumeSnapshotContent, error) {
@@ -120,7 +121,7 @@ func GetVolumeSnapshotContentForVolumeSnapshot(
 	}
 
 	vsc, err := snapshotClient.VolumeSnapshotContents().Get(
-		context.TODO(),
+		ctx,
 		*volSnap.Status.BoundVolumeSnapshotContentName,
 		metav1.GetOptions{},
 	)
@@ -309,6 +310,7 @@ func patchVSC(
 }
 
 func GetVolumeSnapshotClass(
+	ctx context.Context,
 	provisioner string,
 	backup *velerov1api.Backup,
 	pvc *corev1api.PersistentVolumeClaim,
@@ -317,7 +319,7 @@ func GetVolumeSnapshotClass(
 	policySnapshotClass string,
 ) (*snapshotv1api.VolumeSnapshotClass, error) {
 	snapshotClasses := new(snapshotv1api.VolumeSnapshotClassList)
-	err := crClient.List(context.TODO(), snapshotClasses)
+	err := crClient.List(ctx, snapshotClasses)
 	if err != nil {
 		return nil, errors.Wrap(err, "error listing VolumeSnapshotClass")
 	}
@@ -517,13 +519,14 @@ func IsVolumeSnapshotContentHasDeleteSecret(vsc *snapshotv1api.VolumeSnapshotCon
 
 // IsVolumeSnapshotExists returns whether a specific volumesnapshot object exists.
 func IsVolumeSnapshotExists(
+	ctx context.Context,
 	ns,
 	name string,
 	crClient crclient.Client,
 ) bool {
 	vs := new(snapshotv1api.VolumeSnapshot)
 	err := crClient.Get(
-		context.TODO(),
+		ctx,
 		crclient.ObjectKey{Namespace: ns, Name: name},
 		vs,
 	)
@@ -532,24 +535,26 @@ func IsVolumeSnapshotExists(
 }
 
 func SetVolumeSnapshotContentDeletionPolicy(
+	ctx context.Context,
 	vscName string,
 	crClient crclient.Client,
 	policy snapshotv1api.DeletionPolicy,
 ) (*snapshotv1api.VolumeSnapshotContent, error) {
 	vsc := new(snapshotv1api.VolumeSnapshotContent)
-	if err := crClient.Get(context.TODO(), crclient.ObjectKey{Name: vscName}, vsc); err != nil {
+	if err := crClient.Get(ctx, crclient.ObjectKey{Name: vscName}, vsc); err != nil {
 		return nil, err
 	}
 
 	originVSC := vsc.DeepCopy()
 	vsc.Spec.DeletionPolicy = policy
 
-	return vsc, crClient.Patch(context.TODO(), vsc, crclient.MergeFrom(originVSC))
+	return vsc, crClient.Patch(ctx, vsc, crclient.MergeFrom(originVSC))
 }
 
 // CleanupVolumeSnapshot deletes the VolumeSnapshot and the associated VolumeSnapshotContent.  It will make sure the
 // physical snapshot is also deleted.
 func CleanupVolumeSnapshot(
+	ctx context.Context,
 	volSnap *snapshotv1api.VolumeSnapshot,
 	crClient crclient.Client,
 	log logrus.FieldLogger,
@@ -557,7 +562,7 @@ func CleanupVolumeSnapshot(
 	log.Infof("Deleting Volumesnapshot %s/%s", volSnap.Namespace, volSnap.Name)
 	vs := new(snapshotv1api.VolumeSnapshot)
 	err := crClient.Get(
-		context.TODO(),
+		ctx,
 		crclient.ObjectKey{Name: volSnap.Name, Namespace: volSnap.Namespace},
 		vs,
 	)
@@ -570,6 +575,7 @@ func CleanupVolumeSnapshot(
 		// we patch the DeletionPolicy of the VolumeSnapshotContent to set it to Delete.
 		// This ensures that the volume snapshot in the storage provider is also deleted.
 		_, err := SetVolumeSnapshotContentDeletionPolicy(
+			ctx,
 			*vs.Status.BoundVolumeSnapshotContentName,
 			crClient,
 			snapshotv1api.VolumeSnapshotContentDelete,
@@ -579,7 +585,7 @@ func CleanupVolumeSnapshot(
 				vs.Namespace, vs.Name)
 		}
 	}
-	err = crClient.Delete(context.TODO(), vs)
+	err = crClient.Delete(ctx, vs)
 	if err != nil {
 		log.Debugf("Failed to delete volumesnapshot %s/%s: %v", vs.Namespace, vs.Name, err)
 	} else {
@@ -589,6 +595,7 @@ func CleanupVolumeSnapshot(
 }
 
 func DeleteReadyVolumeSnapshot(
+	ctx context.Context,
 	vs snapshotv1api.VolumeSnapshot,
 	client crclient.Client,
 	logger logrus.FieldLogger,
@@ -610,6 +617,7 @@ func DeleteReadyVolumeSnapshot(
 		// Patch the DeletionPolicy of the VolumeSnapshotContent to set it to Retain.
 		// This ensures that the volume snapshot in the storage provider is kept.
 		if vsc, err = SetVolumeSnapshotContentDeletionPolicy(
+			ctx,
 			*vs.Status.BoundVolumeSnapshotContentName,
 			client,
 			snapshotv1api.VolumeSnapshotContentRetain,
@@ -619,11 +627,11 @@ func DeleteReadyVolumeSnapshot(
 			return
 		}
 
-		if err := client.Delete(context.TODO(), vsc); err != nil {
+		if err := client.Delete(ctx, vsc); err != nil {
 			logger.WithError(err).Warnf("Failed to delete the VolumeSnapshotContent %s", vsc.Name)
 		}
 	}
-	if err := client.Delete(context.TODO(), &vs); err != nil {
+	if err := client.Delete(ctx, &vs); err != nil {
 		logger.WithError(err).Warnf("Failed to delete VolumeSnapshot %s", vs.Namespace+"/"+vs.Name)
 	} else {
 		logger.Infof("Deleted VolumeSnapshot %s and VolumeSnapshotContent %s",

pkg/util/csi/volume_snapshot_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package csi
 
 import (
+	"context"
 	"errors"
 	"testing"
 	"time"
@@ -286,7 +287,7 @@ func TestGetVolumeSnapshotContentForVolumeSnapshot(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			fakeClient := snapshotFake.NewSimpleClientset(test.clientObj...)
 
-			vs, err := GetVolumeSnapshotContentForVolumeSnapshot(test.snapshotObj, fakeClient.SnapshotV1())
+			vs, err := GetVolumeSnapshotContentForVolumeSnapshot(context.TODO(), test.snapshotObj, fakeClient.SnapshotV1())
 			if err != nil {
 				require.EqualError(t, err, test.err)
 			} else {
@@ -1032,6 +1033,7 @@ func TestGetVolumeSnapshotClass(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			actualSnapshotClass, actualError := GetVolumeSnapshotClass(
+				context.TODO(),
 				tc.driverName, tc.backup, tc.pvc, logrus.New(), fakeClient, "")
 			if tc.expectError {
 				require.Error(t, actualError)
@@ -1458,7 +1460,7 @@ func TestIsVolumeSnapshotExists(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			actual := IsVolumeSnapshotExists(tc.vs.Namespace, tc.vs.Name, fakeClient)
+			actual := IsVolumeSnapshotExists(context.TODO(), tc.vs.Namespace, tc.vs.Name, fakeClient)
 			assert.Equal(t, tc.expected, actual)
 		})
 	}
@@ -1529,7 +1531,7 @@ func TestSetVolumeSnapshotContentDeletionPolicy(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			fakeClient := velerotest.NewFakeControllerRuntimeClient(t, tc.objs...)
-			_, err := SetVolumeSnapshotContentDeletionPolicy(tc.inputVSCName, fakeClient, tc.policy)
+			_, err := SetVolumeSnapshotContentDeletionPolicy(context.TODO(), tc.inputVSCName, fakeClient, tc.policy)
 			if tc.expectError {
 				assert.Error(t, err)
 			} else {
@@ -1586,7 +1588,7 @@ func TestDeleteVolumeSnapshots(t *testing.T) {
 			)
 			logger := logging.DefaultLogger(logrus.DebugLevel, logging.FormatText)
 
-			DeleteReadyVolumeSnapshot(tc.vs, client, logger)
+			DeleteReadyVolumeSnapshot(context.TODO(), tc.vs, client, logger)
 
 			vsList := new(snapshotv1api.VolumeSnapshotList)
 			err := client.List(