diff --git a/changelogs/unreleased/10247-opbot-xd b/changelogs/unreleased/10247-opbot-xd new file mode 100644 index 000000000..d1caa9255 --- /dev/null +++ b/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. diff --git a/pkg/backup/actions/csi/pvc_action.go b/pkg/backup/actions/csi/pvc_action.go index b9debe031..ae0153b7c 100644 --- a/pkg/backup/actions/csi/pvc_action.go +++ b/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( diff --git a/pkg/backup/actions/csi/volumesnapshot_action.go b/pkg/backup/actions/csi/volumesnapshot_action.go index 49e690e93..b1f6050ef 100644 --- a/pkg/backup/actions/csi/volumesnapshot_action.go +++ b/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 { diff --git a/pkg/controller/backup_deletion_controller.go b/pkg/controller/backup_deletion_controller.go index cd74a3a27..0d5500972 100644 --- a/pkg/controller/backup_deletion_controller.go +++ b/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) } } diff --git a/pkg/exposer/csi_snapshot.go b/pkg/exposer/csi_snapshot.go index 2e8e08889..271ec914c 100644 --- a/pkg/exposer/csi_snapshot.go +++ b/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") } diff --git a/pkg/util/csi/volume_snapshot.go b/pkg/util/csi/volume_snapshot.go index e8fe9bead..bda9d2796 100644 --- a/pkg/util/csi/volume_snapshot.go +++ b/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", diff --git a/pkg/util/csi/volume_snapshot_test.go b/pkg/util/csi/volume_snapshot_test.go index 335cff6ee..61e76302b 100644 --- a/pkg/util/csi/volume_snapshot_test.go +++ b/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(