From 0a4d542e15ea8d4c2a9c7ee10e5d6c3142b82d4d Mon Sep 17 00:00:00 2001
From: Patrick Ohly <patrick.ohly@intel.com>
Date: Wed, 16 Sep 2020 09:15:07 +0200
Subject: [PATCH] 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).

(cherry picked from commit c53bc294019b45a31be29743d39a43b0df3bd110)
---
 cmd/csi-provisioner/csi-provisioner.go |  8 +++++++-
 deploy/kubernetes/rbac.yaml            |  4 ++++
 pkg/controller/controller.go           | 24 ++++++++++++++++++++----
 3 files changed, 31 insertions(+), 5 deletions(-)

diff --git a/cmd/csi-provisioner/csi-provisioner.go b/cmd/csi-provisioner/csi-provisioner.go
index 43b1d38b91..c0163367f2 100644
--- a/cmd/csi-provisioner/csi-provisioner.go
+++ b/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()
diff --git a/deploy/kubernetes/rbac.yaml b/deploy/kubernetes/rbac.yaml
index 48450f89ac..92fa4920b8 100644
--- a/deploy/kubernetes/rbac.yaml
+++ b/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"]
diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go
index 9c0cb33762..2385c4d628 100644
--- a/pkg/controller/controller.go
+++ b/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 {