Add disabledCascadingDeletion flag

cmd/controller-manager/app/controller_manager.go

@@ -338,6 +338,7 @@ func StartControllers(s *options.ControllerManagerServer,
 		s.ClusterIDConfigMapName,
 		s.ClusterIDConfigMapNamespace,
 		s.OSBAPITimeOut,
+		s.DisableCascadingDeletion,
 	)
 	if err != nil {
 		return err

cmd/controller-manager/app/options/options.go

@@ -126,4 +126,5 @@ func (s *ControllerManagerServer) AddFlags(fs *pflag.FlagSet) {
 	utilfeature.DefaultMutableFeatureGate.AddFlag(fs)
 	fs.StringVar(&s.ClusterIDConfigMapName, "cluster-id-configmap-name", controller.DefaultClusterIDConfigMapName, "k8s name for clusterid configmap")
 	fs.StringVar(&s.ClusterIDConfigMapNamespace, "cluster-id-configmap-namespace", controller.DefaultClusterIDConfigMapNamespace, "k8s namespace for clusterid configmap")
+	fs.BoolVar(&s.DisableCascadingDeletion, "disable-cascading-deletion", s.DisableCascadingDeletion, "Disables cascading binding deletion.")
 }

pkg/apis/componentconfig/types.go

@@ -112,4 +112,7 @@ type ControllerManagerConfiguration struct {
 	ClusterIDConfigMapName string
 	// ClusterIDConfigMapNamespace is the k8s namespace that the clusterid configmap will be stored in.
 	ClusterIDConfigMapNamespace string
+
+	// DisableCascadingDeletion disables cascading binding deletion if the instance is deleted.
+	DisableCascadingDeletion bool
 }

pkg/controller/case_test.go

@@ -94,8 +94,12 @@ type controllerTest struct {
 	serviceBindingInformerStopCh chan struct{}
 }
 
-// newControllerTest creates a controllerTest instance with a ready to test running Controller
 func newControllerTest(t *testing.T) *controllerTest {
+	return newControllerTestWithConfig(t, false)
+}
+
+// newControllerTest creates a controllerTest instance with a ready to test running Controller
+func newControllerTestWithConfig(t *testing.T, disableCascadingDeletion bool) *controllerTest {
 	k8sClient := fakek8s.NewSimpleClientset()
 	k8sClient.CoreV1().Namespaces().Create(&corev1.Namespace{
 		ObjectMeta: metav1.ObjectMeta{
@@ -154,6 +158,7 @@ func newControllerTest(t *testing.T) *controllerTest {
 		"DefaultClusterIDConfigMapName",
 		"DefaultClusterIDConfigMapNamespace",
 		60*time.Second,
+		disableCascadingDeletion,
 	)
 	if err != nil {
 		t.Fatal(err)

pkg/controller/controller.go

@@ -96,6 +96,7 @@ func NewController(
 	clusterIDConfigMapName string,
 	clusterIDConfigMapNamespace string,
 	osbAPITimeOut time.Duration,
+	disableCascadingDeletion bool,
 ) (Controller, error) {
 	controller := &controller{
 		kubeClient:                  kubeClient,
@@ -119,6 +120,7 @@ func NewController(
 		clusterIDConfigMapName:      clusterIDConfigMapName,
 		clusterIDConfigMapNamespace: clusterIDConfigMapNamespace,
 		brokerClientCreateFunc:      brokerClientCreateFunc,
+		disableCascadingDeletion:    disableCascadingDeletion,
 	}
 	controller.brokerClientManager = NewBrokerClientManager(brokerClientCreateFunc)
 
@@ -240,6 +242,9 @@ type controller struct {
 	brokerClientManager *BrokerClientManager
 
 	brokerClientCreateFunc osb.CreateFunc
+
+	// disableCascadingDeletion disables cascading deletion feature.
+	disableCascadingDeletion bool
 }
 
 // Run runs the controller until the given stop channel can be read from.

pkg/controller/controller_flow_instance_test.go

@@ -160,6 +160,35 @@ func TestServiceInstanceDeleteWithExistingServiceBindings(t *testing.T) {
 	assert.NotZero(t, ct.NumberOfOSBDeprovisionCalls())
 }
 
+// TestServiceInstanceDeprovisionFailedWithExistingServiceBindings tests the cascading service binding deletion is blocked by disableCascadingDeletion flag.
+func TestServiceInstanceDeprovisionFailedWithExistingServiceBindings(t *testing.T) {
+	// GIVEN
+	utilfeature.DefaultMutableFeatureGate.Set(fmt.Sprintf("%s=true", features.CascadingDeletion))
+	defer utilfeature.DefaultMutableFeatureGate.Set(fmt.Sprintf("%s=false", features.CascadingDeletion))
+	ct := newControllerTestWithConfig(t, true)
+	defer ct.TearDown()
+
+	require.NoError(t, ct.CreateSimpleClusterServiceBroker())
+	require.NoError(t, ct.WaitForReadyBroker())
+	ct.AssertClusterServiceClassAndPlan(t)
+	assert.NoError(t, ct.CreateServiceInstance())
+	assert.NoError(t, ct.WaitForReadyInstance())
+	assert.NoError(t, ct.CreateBinding())
+	assert.NoError(t, ct.WaitForReadyBinding())
+
+	// WHEN
+	require.NoError(t, ct.Deprovision())
+
+	//THEN
+	assert.NoError(t, ct.WaitForReadyBinding())
+	assert.NoError(t, ct.WaitForInstanceCondition(v1beta1.ServiceInstanceCondition{
+		Type:   v1beta1.ServiceInstanceConditionReady,
+		Status: v1beta1.ConditionFalse,
+		Reason: "DeprovisionBlockedByExistingCredentials",
+	}))
+	assert.Zero(t, ct.NumberOfOSBDeprovisionCalls())
+}
+
 // TestServiceInstanceDeleteWithAsyncUpdateInProgress tests that you can delete
 // an instance during an async update.  That is, if you request a delete during
 // an instance update, the instance will be deleted when the update completes

pkg/controller/controller_instance.go

@@ -907,7 +907,7 @@ func (c *controller) reconcileServiceInstanceDelete(instance *v1beta1.ServiceIns
 	// We don't want to delete the instance if there are any bindings associated.
 	if err := c.checkServiceInstanceHasExistingBindings(instance); err != nil {
 		// if the CascadingDeletion feature flag is set, delete existing bindings instead of update the status with an error
-		if utilfeature.DefaultFeatureGate.Enabled(scfeatures.CascadingDeletion) {
+		if utilfeature.DefaultFeatureGate.Enabled(scfeatures.CascadingDeletion) && !c.disableCascadingDeletion {
 			err := c.deleteExistingBindings(instance)
 			if err != nil {
 				klog.V(4).Info(pcb.Messagef("unable to delete existing bindings: %s", err.Error()))

pkg/controller/controller_test.go

@@ -2455,6 +2455,7 @@ func newTestController(t *testing.T, config fakeosb.FakeClientConfiguration) (
 		DefaultClusterIDConfigMapName,
 		DefaultClusterIDConfigMapNamespace,
 		60*time.Second,
+		false,
 	)
 
 	if err != nil {

test/integration/controller_test.go

@@ -960,6 +960,7 @@ func newTestController(t *testing.T) (
 		controller.DefaultClusterIDConfigMapName,
 		controller.DefaultClusterIDConfigMapNamespace,
 		60*time.Second,
+		false,
 	)
 	t.Log("controller start")
 	if err != nil {