From fb599dd02be492fd9c097c94fa84c5cc32350480 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Kristoffer=20Gr=C3=B6nlund?= <kgronlund@suse.com>
Date: Thu, 19 Sep 2019 15:53:15 +0200
Subject: [PATCH] Ceph: Remove finalizer even if flex is disabled
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

When the flex driver is disabled, the check for
volume attachments will fail and the finalizer
is never removed. To avoid this, just log the
failure to list volumes and remove the
finalizer anyway.

Resolves #3912

Signed-off-by: Kristoffer Grönlund <kgronlund@suse.com>
---
 pkg/operator/ceph/cluster/controller.go      | 10 ++++-
 pkg/operator/ceph/cluster/controller_test.go | 45 +++++++++++++++++++-
 2 files changed, 53 insertions(+), 2 deletions(-)

diff --git a/pkg/operator/ceph/cluster/controller.go b/pkg/operator/ceph/cluster/controller.go
index 2a10f93b1..ea6ec34a7 100644
--- a/pkg/operator/ceph/cluster/controller.go
+++ b/pkg/operator/ceph/cluster/controller.go
@@ -58,6 +58,7 @@ import (
 const (
 	crushConfigMapName       = "rook-crush-config"
 	crushmapCreatedKey       = "initialCrushMapCreated"
+	enableFlexDriver         = "ROOK_ENABLE_FLEX_DRIVER"
 	clusterCreateInterval    = 6 * time.Second
 	clusterCreateTimeout     = 60 * time.Minute
 	updateClusterInterval    = 30 * time.Second
@@ -759,8 +760,15 @@ func (c *ClusterController) onDelete(obj interface{}) {
 }
 
 func (c *ClusterController) handleDelete(cluster *cephv1.CephCluster, retryInterval time.Duration) error {
-
+	flexDriverEnabled := os.Getenv(enableFlexDriver) != "false"
 	operatorNamespace := os.Getenv(k8sutil.PodNamespaceEnvVar)
+
+	if !flexDriverEnabled {
+		logger.Debugf("Flex driver disabled: no volume attachments for cluster %s (operator namespace: %s)",
+			cluster.Namespace, operatorNamespace)
+		return nil
+	}
+
 	retryCount := 0
 	for {
 		// TODO: filter this List operation by cluster namespace on the server side
diff --git a/pkg/operator/ceph/cluster/controller_test.go b/pkg/operator/ceph/cluster/controller_test.go
index 2fc9f26bd..37e83e4e8 100644
--- a/pkg/operator/ceph/cluster/controller_test.go
+++ b/pkg/operator/ceph/cluster/controller_test.go
@@ -34,13 +34,15 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-func TestClusterDelete(t *testing.T) {
+func TestClusterDeleteFlexEnabled(t *testing.T) {
 	nodeName := "node841"
 	clusterName := "cluster684"
 	pvName := "pvc-540"
 	rookSystemNamespace := "rook-system-6413"
 
+	os.Setenv("ROOK_ENABLE_FLEX_DRIVER", "true")
 	os.Setenv(k8sutil.PodNamespaceEnvVar, rookSystemNamespace)
+	defer os.Unsetenv("ROOK_ENABLE_FLEX_DRIVER")
 	defer os.Unsetenv(k8sutil.PodNamespaceEnvVar)
 
 	clientset := testop.New(3)
@@ -95,6 +97,47 @@ func TestClusterDelete(t *testing.T) {
 	assert.Equal(t, 2, listCount)
 }
 
+func TestClusterDeleteFlexDisabled(t *testing.T) {
+	clusterName := "cluster684"
+	rookSystemNamespace := "rook-system-6413"
+
+	os.Setenv("ROOK_ENABLE_FLEX_DRIVER", "false")
+	os.Setenv(k8sutil.PodNamespaceEnvVar, rookSystemNamespace)
+	defer os.Unsetenv("ROOK_ENABLE_FLEX_DRIVER")
+	defer os.Unsetenv(k8sutil.PodNamespaceEnvVar)
+
+	clientset := testop.New(3)
+	context := &clusterd.Context{
+		Clientset: clientset,
+	}
+
+	listCount := 0
+	volumeAttachmentController := &attachment.MockAttachment{
+		MockList: func(namespace string) (*rookalpha.VolumeList, error) {
+			listCount++
+			return &rookalpha.VolumeList{Items: []rookalpha.Volume{}}, nil
+
+		},
+	}
+	callbacks := []func(clusterSpec *cephv1.ClusterSpec) error{
+		func(clusterSpec *cephv1.ClusterSpec) error {
+			logger.Infof("test success callback")
+			os.Setenv("ROOK_ENABLE_FLEX_DRIVER", "true")
+			os.Setenv(k8sutil.PodNamespaceEnvVar, rookSystemNamespace)
+			defer os.Unsetenv("ROOK_ENABLE_FLEX_DRIVER")
+			return nil
+		},
+	}
+
+	// create the cluster controller and tell it that the cluster has been deleted
+	controller := NewClusterController(context, "", volumeAttachmentController, callbacks)
+	clusterToDelete := &cephv1.CephCluster{ObjectMeta: metav1.ObjectMeta{Namespace: clusterName}}
+	controller.handleDelete(clusterToDelete, time.Microsecond)
+
+	// Ensure that the listing of volume attachments was never called.
+	assert.Equal(t, 0, listCount)
+}
+
 func TestClusterChanged(t *testing.T) {
 	// a new node added, should be a change
 	old := cephv1.ClusterSpec{