Do not get PV for externally deleting volume

Currently, we get each released PV every 15s, and in parallel. If there are a lot of released PV and we cannot finish all the get in 15s, it will starve other request by making the queue waiting for client-side throttling very long.

Even in a normal cluster, these requests are taking majority of all get requests from KCM (58% or 470 qps) in our stress test.

pkg/controller/volume/persistentvolume/pv_controller.go

@@ -1324,6 +1324,22 @@ func (ctrl *PersistentVolumeController) deleteVolumeOperation(ctx context.Contex
 	logger := klog.FromContext(ctx)
 	logger.V(4).Info("DeleteVolumeOperation started", "volumeName", volume.Name)
 
+	plugin, err := ctrl.findDeletablePlugin(volume)
+	if err != nil {
+		if _, err := ctrl.updateVolumePhaseWithEvent(ctx, volume, v1.VolumeFailed, v1.EventTypeWarning, events.VolumeFailedDelete, err.Error()); err != nil {
+			logger.V(4).Info("DeleteVolumeOperation: failed to mark volume as failed", "volumeName", volume.Name, "err", err)
+			// Save failed, retry on the next deletion attempt
+			return "", err
+		}
+	}
+	if plugin == nil {
+		// External deleter is requested, do nothing
+		logger.V(3).Info("External deleter for volume requested, ignoring", "volumeName", volume.Name)
+		return "", nil
+	}
+	pluginName := plugin.GetPluginName()
+	logger.V(5).Info("Found a deleter plugin for volume", "pluginName", pluginName, "volumeName", volume.Name)
+
 	// This method may have been waiting for a volume lock for some time.
 	// Previous deleteVolumeOperation might just have saved an updated version, so
 	// read current volume state now.
@@ -1343,7 +1359,7 @@ func (ctrl *PersistentVolumeController) deleteVolumeOperation(ctx context.Contex
 		return "", nil
 	}
 
-	pluginName, deleted, err := ctrl.doDeleteVolume(ctx, volume)
+	err = ctrl.doDeleteVolume(ctx, plugin, volume)
 	if err != nil {
 		// Delete failed, update the volume and emit an event.
 		logger.V(3).Info("Deletion of volume failed", "volumeName", volume.Name, "err", err)
@@ -1365,10 +1381,6 @@ func (ctrl *PersistentVolumeController) deleteVolumeOperation(ctx context.Contex
 		// the volume in every syncVolume() call.
 		return pluginName, err
 	}
-	if !deleted {
-		// The volume waits for deletion by an external plugin. Do nothing.
-		return pluginName, nil
-	}
 
 	logger.V(4).Info("DeleteVolumeOperation: success", "volumeName", volume.Name)
 	// Delete the volume
@@ -1490,29 +1502,17 @@ func (ctrl *PersistentVolumeController) findNonScheduledPodsByPVC(pvc *v1.Persis
 // the volume plugin name. Also, it returns 'true', when the volume was deleted and
 // 'false' when the volume cannot be deleted because the deleter is external. No
 // error should be reported in this case.
-func (ctrl *PersistentVolumeController) doDeleteVolume(ctx context.Context, volume *v1.PersistentVolume) (string, bool, error) {
+func (ctrl *PersistentVolumeController) doDeleteVolume(ctx context.Context, plugin vol.DeletableVolumePlugin, volume *v1.PersistentVolume) error {
 	logger := klog.FromContext(ctx)
 	logger.V(4).Info("doDeleteVolume", "volumeName", volume.Name)
 	var err error
 
-	plugin, err := ctrl.findDeletablePlugin(volume)
-	if err != nil {
-		return "", false, err
-	}
-	if plugin == nil {
-		// External deleter is requested, do nothing
-		logger.V(3).Info("External deleter for volume requested, ignoring", "volumeName", volume.Name)
-		return "", false, nil
-	}
-
-	// Plugin found
 	pluginName := plugin.GetPluginName()
-	logger.V(5).Info("Found a deleter plugin for volume", "pluginName", pluginName, "volumeName", volume.Name)
 	spec := vol.NewSpecFromPersistentVolume(volume, false)
 	deleter, err := plugin.NewDeleter(logger, spec)
 	if err != nil {
 		// Cannot create deleter
-		return pluginName, false, fmt.Errorf("failed to create deleter for volume %q: %w", volume.Name, err)
+		return fmt.Errorf("failed to create deleter for volume %q: %w", volume.Name, err)
 	}
 
 	opComplete := util.OperationCompleteHook(pluginName, "volume_delete")
@@ -1520,13 +1520,13 @@ func (ctrl *PersistentVolumeController) doDeleteVolume(ctx context.Context, volu
 	opComplete(volumetypes.CompleteFuncParam{Err: &err})
 	if err != nil {
 		// Deleter failed
-		return pluginName, false, err
+		return err
 	}
 	logger.V(2).Info("Volume deleted", "volumeName", volume.Name)
 	if err := ctrl.removeDeletionProtectionFinalizer(ctx, volume); err != nil {
-		return pluginName, true, err
+		return err
 	}
-	return pluginName, true, nil
+	return nil
 }
 
 func (ctrl *PersistentVolumeController) removeDeletionProtectionFinalizer(ctx context.Context, volume *v1.PersistentVolume) error {
@@ -1769,25 +1769,27 @@ func (ctrl *PersistentVolumeController) provisionClaimOperation(
 		logger.V(3).Info(strerr)
 		ctrl.eventRecorder.Event(claim, v1.EventTypeWarning, events.ProvisioningFailed, strerr)
 
-		var deleteErr error
-		var deleted bool
-		for i := 0; i < ctrl.createProvisionedPVRetryCount; i++ {
-			_, deleted, deleteErr = ctrl.doDeleteVolume(ctx, volume)
-			if deleteErr == nil && deleted {
-				// Delete succeeded
-				logger.V(4).Info("provisionClaimOperation: cleaning volume succeeded", "PVC", klog.KObj(claim), "volumeName", volume.Name)
-				break
-			}
-			if !deleted {
+		delPlugin, deleteErr := ctrl.findDeletablePlugin(volume)
+		if deleteErr == nil {
+			if delPlugin == nil {
 				// This is unreachable code, the volume was provisioned by an
 				// internal plugin and therefore there MUST be an internal
 				// plugin that deletes it.
 				logger.Error(nil, "Error finding internal deleter for volume plugin", "plugin", plugin.GetPluginName())
-				break
+				return pluginName, nil
+			}
+
+			for i := 0; i < ctrl.createProvisionedPVRetryCount; i++ {
+				deleteErr = ctrl.doDeleteVolume(ctx, delPlugin, volume)
+				if deleteErr == nil {
+					// Delete succeeded
+					logger.V(4).Info("provisionClaimOperation: cleaning volume succeeded", "PVC", klog.KObj(claim), "volumeName", volume.Name)
+					break
+				}
+				// Delete failed, try again after a while.
+				logger.V(3).Info("Failed to delete volume", "volumeName", volume.Name, "err", deleteErr)
+				time.Sleep(ctrl.createProvisionedPVInterval)
 			}
-			// Delete failed, try again after a while.
-			logger.V(3).Info("Failed to delete volume", "volumeName", volume.Name, "err", deleteErr)
-			time.Sleep(ctrl.createProvisionedPVInterval)
 		}
 
 		if deleteErr != nil {