optional VolumeAttachments check

Watching all VolumeAttachments in the cluster even when the CSI driver
itself doesn't support attach/detach causes unnecessary overhead. It
is now automatically enabled if and only if the driver reports the
PUBLISH_UNPUBLISH_VOLUME controller capability.

Checking the capability is better than checking the "skip attach" flag
in the CSIDriver object because the capability is readily available
and constant. It also has the advantage that the watch is disabled for
drivers like csi-host-path-driver which do not support attach/detach
but don't use "skip attach" (for whatever reason).

cmd/csi-provisioner/csi-provisioner.go

@@ -202,7 +202,13 @@ func main() {
 	claimLister := factory.Core().V1().PersistentVolumeClaims().Lister()
 
 	var csiNodeLister storagelistersv1.CSINodeLister
-	vaLister := factory.Storage().V1().VolumeAttachments().Lister()
+	var vaLister storagelistersv1.VolumeAttachmentLister
+	if controllerCapabilities[csi.ControllerServiceCapability_RPC_PUBLISH_UNPUBLISH_VOLUME] {
+		klog.Info("CSI driver supports PUBLISH_UNPUBLISH_VOLUME, watching VolumeAttachments")
+		vaLister = factory.Storage().V1().VolumeAttachments().Lister()
+	} else {
+		klog.Info("CSI driver does not support PUBLISH_UNPUBLISH_VOLUME, not watching VolumeAttachments")
+	}
 	var nodeLister v1.NodeLister
 	if ctrl.SupportsTopology(pluginCapabilities) {
 		csiNodeLister = factory.Storage().V1().CSINodes().Lister()

deploy/kubernetes/rbac.yaml

@@ -50,6 +50,10 @@ rules:
   - apiGroups: [""]
     resources: ["nodes"]
     verbs: ["get", "list", "watch"]
+  # Access to volumeattachments is only needed when the CSI driver
+  # has the PUBLISH_UNPUBLISH_VOLUME controller capability.
+  # In that case, external-provisioner will watch volumeattachments
+  # to determine when it is safe to delete a volume.
   - apiGroups: ["storage.k8s.io"]
     resources: ["volumeattachments"]
     verbs: ["get", "list", "watch"]

pkg/controller/controller.go

@@ -271,7 +271,10 @@ func GetDriverCapabilities(conn *grpc.ClientConn, timeout time.Duration) (rpc.Pl
 	return pluginCapabilities, controllerCapabilities, nil
 }
 
-// NewCSIProvisioner creates new CSI provisioner
+// NewCSIProvisioner creates new CSI provisioner.
+//
+// vaLister is optional and only needed when VolumeAttachments are
+// meant to be checked before deleting a volume.
 func NewCSIProvisioner(client kubernetes.Interface,
 	connectionTimeout time.Duration,
 	identity string,
@@ -1046,6 +1049,21 @@ func (p *csiProvisioner) Delete(ctx context.Context, volume *v1.PersistentVolume
 	deleteCtx, cancel := context.WithTimeout(ctx, p.timeout)
 	defer cancel()
 
+	if err := p.canDeleteVolume(volume); err != nil {
+		return err
+	}
+
+	_, err = p.csiClient.DeleteVolume(deleteCtx, &req)
+
+	return err
+}
+
+func (p *csiProvisioner) canDeleteVolume(volume *v1.PersistentVolume) error {
+	if p.vaLister == nil {
+		// Nothing to check.
+		return nil
+	}
+
 	// Verify if volume is attached to a node before proceeding with deletion
 	vaList, err := p.vaLister.List(labels.Everything())
 	if err != nil {
@@ -1058,9 +1076,7 @@ func (p *csiProvisioner) Delete(ctx context.Context, volume *v1.PersistentVolume
 		}
 	}
 
-	_, err = p.csiClient.DeleteVolume(deleteCtx, &req)
-
-	return err
+	return nil
 }
 
 func (p *csiProvisioner) SupportsBlock(ctx context.Context) bool {