Use watch instead of poll to check PVC/VolumeSnapshot deletion

Author: xing-yang

pkg/csi/service/common/common_controller_helper.go

@@ -27,7 +27,9 @@ import (
 	snapshotterClientSet "github.com/kubernetes-csi/external-snapshotter/client/v8/clientset/versioned"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/apimachinery/pkg/watch"
 	clientset "k8s.io/client-go/kubernetes"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
@@ -448,3 +450,103 @@ func WaitForVolumeSnapshotDeleted(ctx context.Context, client snapshotterClientS
 
 	return nil
 }
+
+// WaitForVolumeSnapshotDeletedWithWatch waits for a VolumeSnapshot to get deleted using Watch API instead of polling.
+func WaitForVolumeSnapshotDeletedWithWatch(ctx context.Context, client snapshotterClientSet.Interface,
+	snapshotName string, namespace string, timeout time.Duration) error {
+	log := logger.GetLogger(ctx)
+	timeoutSeconds := int64(timeout.Seconds())
+
+	log.Infof("Waiting up to %d seconds for VolumeSnapshot %s in namespace %s to be deleted using Watch API",
+		timeoutSeconds, snapshotName, namespace)
+
+	// First check if VolumeSnapshot already exists
+	_, err := client.SnapshotV1().VolumeSnapshots(namespace).Get(ctx, snapshotName, metav1.GetOptions{})
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			log.Infof("VolumeSnapshot %s/%s is already deleted", namespace, snapshotName)
+			return nil
+		}
+		return fmt.Errorf("unable to fetch VolumeSnapshot %s/%s with err: %+v", namespace, snapshotName, err)
+	}
+
+	// Set up watch for the specific VolumeSnapshot
+	watchSnapshot, err := client.SnapshotV1().VolumeSnapshots(namespace).Watch(
+		ctx,
+		metav1.ListOptions{
+			FieldSelector:  fields.OneTermEqualSelector("metadata.name", snapshotName).String(),
+			TimeoutSeconds: &timeoutSeconds,
+			Watch:          true,
+		})
+	if err != nil {
+		return fmt.Errorf("failed to watch VolumeSnapshot %s in namespace %s with Error: %+v",
+			snapshotName, namespace, err)
+	}
+	defer watchSnapshot.Stop()
+
+	// Watch for deletion events
+	for event := range watchSnapshot.ResultChan() {
+		switch event.Type {
+		case watch.Deleted:
+			log.Infof("VolumeSnapshot %s/%s was deleted", namespace, snapshotName)
+			return nil
+		case watch.Error:
+			return fmt.Errorf("error watching VolumeSnapshot %s/%s: %+v", namespace, snapshotName, event.Object)
+		}
+		// Continue watching for other event types (Added, Modified)
+	}
+
+	// If we reach here, the watch timed out
+	return fmt.Errorf("volumeSnapshot %s/%s was not deleted within %d seconds",
+		namespace, snapshotName, timeoutSeconds)
+}
+
+// WaitForPVCDeletedWithWatch waits for a PVC to get deleted using Watch API instead of polling.
+func WaitForPVCDeletedWithWatch(ctx context.Context, client clientset.Interface, pvcName string,
+	namespace string, timeout time.Duration) error {
+	log := logger.GetLogger(ctx)
+	timeoutSeconds := int64(timeout.Seconds())
+
+	log.Infof("Waiting up to %d seconds for PersistentVolumeClaim %s in namespace %s to be deleted using Watch API",
+		timeoutSeconds, pvcName, namespace)
+
+	// First check if PVC already exists
+	_, err := client.CoreV1().PersistentVolumeClaims(namespace).Get(ctx, pvcName, metav1.GetOptions{})
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			log.Infof("PersistentVolumeClaim %s/%s is already deleted", namespace, pvcName)
+			return nil
+		}
+		return fmt.Errorf("unable to fetch PersistentVolumeClaim %s/%s with err: %+v", namespace, pvcName, err)
+	}
+
+	// Set up watch for the specific PVC
+	watchPVC, err := client.CoreV1().PersistentVolumeClaims(namespace).Watch(
+		ctx,
+		metav1.ListOptions{
+			FieldSelector:  fields.OneTermEqualSelector("metadata.name", pvcName).String(),
+			TimeoutSeconds: &timeoutSeconds,
+			Watch:          true,
+		})
+	if err != nil {
+		return fmt.Errorf("failed to watch PersistentVolumeClaim %s in namespace %s with Error: %+v",
+			pvcName, namespace, err)
+	}
+	defer watchPVC.Stop()
+
+	// Watch for deletion events
+	for event := range watchPVC.ResultChan() {
+		switch event.Type {
+		case watch.Deleted:
+			log.Infof("PersistentVolumeClaim %s/%s was deleted", namespace, pvcName)
+			return nil
+		case watch.Error:
+			return fmt.Errorf("error watching PersistentVolumeClaim %s/%s: %+v", namespace, pvcName, event.Object)
+		}
+		// Continue watching for other event types (Added, Modified)
+	}
+
+	// If we reach here, the watch timed out
+	return fmt.Errorf("persistentVolumeClaim %s/%s was not deleted within %d seconds",
+		namespace, pvcName, timeoutSeconds)
+}

pkg/csi/service/wcpguest/controller.go

@@ -639,7 +639,7 @@ func (c *controller) DeleteVolume(ctx context.Context, req *csi.DeleteVolumeRequ
 		}
 
 		// Wait for PVC to be deleted from supervisor cluster
-		err = common.WaitForPVCDeleted(ctx, c.supervisorClient,
+		err = common.WaitForPVCDeletedWithWatch(ctx, c.supervisorClient,
 			req.VolumeId, c.supervisorNamespace,
 			time.Duration(getProvisionTimeoutInMin(ctx))*time.Minute)
 		if err != nil {
@@ -1796,7 +1796,7 @@ func (c *controller) DeleteSnapshot(ctx context.Context, req *csi.DeleteSnapshot
 		}
 
 		// Wait for VolumeSnapshot to be deleted from supervisor cluster
-		err = common.WaitForVolumeSnapshotDeleted(ctx, c.supervisorSnapshotterClient,
+		err = common.WaitForVolumeSnapshotDeletedWithWatch(ctx, c.supervisorSnapshotterClient,
 			supervisorVolumeSnapshotName, c.supervisorNamespace,
 			time.Duration(getSnapshotTimeoutInMin(ctx))*time.Minute)
 		if err != nil {