ceph: fix mon pdb reconcile

pkg/operator/ceph/disruption/clusterdisruption/mon.go

@@ -61,9 +61,10 @@ func (r *ReconcileClusterDisruption) reconcileMonPDB(cephCluster *cephv1.CephClu
 			MinAvailable: &intstr.IntOrString{IntVal: minAvailable},
 		},
 	}
+
 	err := r.reconcileStaticPDB(pdbRequest, pdb)
 	if err != nil {
-		return errors.Wrap(err, "could not reconcile mon pdb")
+		return errors.Wrap(err, "failed to reconcile mon pdb")
 	}
 	return nil
 }

pkg/operator/ceph/disruption/clusterdisruption/mon_test.go

@@ -0,0 +1,103 @@
+/*
+Copyright 2020 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package clusterdisruption
+
+import (
+	"context"
+	"testing"
+
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	"github.com/rook/rook/pkg/client/clientset/versioned/scheme"
+	"github.com/stretchr/testify/assert"
+	policyv1beta1 "k8s.io/api/policy/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+)
+
+func createFakeReconcileClusterDisruptionr(t *testing.T, obj ...runtime.Object) *ReconcileClusterDisruption {
+	scheme := scheme.Scheme
+	err := policyv1beta1.AddToScheme(scheme)
+	if err != nil {
+		assert.Fail(t, "failed to build scheme")
+	}
+	client := fake.NewFakeClientWithScheme(scheme, obj...)
+
+	return &ReconcileClusterDisruption{
+		client: client,
+		scheme: scheme,
+	}
+}
+
+func TestReconcileMonPDB(t *testing.T) {
+	r := createFakeReconcileClusterDisruptionr(t, &cephv1.CephCluster{})
+	testCases := []struct {
+		label                string
+		cephCluster          *cephv1.CephCluster
+		expectedMinAvailable int32
+		errorExpected        bool
+	}{
+		{
+			label: "case 1: 0 mons",
+			cephCluster: &cephv1.CephCluster{
+				ObjectMeta: metav1.ObjectMeta{Name: "rook", Namespace: "rook-ceph"},
+			},
+			expectedMinAvailable: 0,
+			errorExpected:        true,
+		},
+		{
+			label: "case 2: 3 mons",
+			cephCluster: &cephv1.CephCluster{
+				ObjectMeta: metav1.ObjectMeta{Name: "rook", Namespace: "rook-ceph"},
+				Spec: cephv1.ClusterSpec{
+					Mon: cephv1.MonSpec{
+						Count: 3,
+					},
+				},
+			},
+			expectedMinAvailable: 2,
+			errorExpected:        false,
+		},
+		{
+			label: "case 3: 5 mons",
+			cephCluster: &cephv1.CephCluster{
+				ObjectMeta: metav1.ObjectMeta{Name: "rook", Namespace: "rook-ceph"},
+				Spec: cephv1.ClusterSpec{
+					Mon: cephv1.MonSpec{
+						Count: 5,
+					},
+				},
+			},
+			expectedMinAvailable: 3,
+			errorExpected:        false,
+		},
+	}
+
+	for _, tc := range testCases {
+		err := r.reconcileMonPDB(tc.cephCluster)
+		assert.NoError(t, err)
+		existingPDB := &policyv1beta1.PodDisruptionBudget{}
+		err = r.client.Get(context.TODO(), types.NamespacedName{Name: pdbName, Namespace: tc.cephCluster.Namespace}, existingPDB)
+		if tc.errorExpected {
+			assert.Error(t, err)
+			continue
+		}
+		assert.NoError(t, err)
+		assert.Equalf(t, tc.expectedMinAvailable, int32(existingPDB.Spec.MinAvailable.IntValue()), "[%s]: incorrect minAvailable count in pdb", tc.label)
+	}
+}

pkg/operator/ceph/disruption/clusterdisruption/static_pdb.go

@@ -19,30 +19,37 @@ package clusterdisruption
 import (
 	"context"
 
+	"github.com/pkg/errors"
+
 	policyv1beta1 "k8s.io/api/policy/v1beta1"
-	"k8s.io/apimachinery/pkg/api/errors"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/types"
 )
 
-const (
-	// MonPDBAppName for monitor daemon poddisruptionbudgets
-	MonPDBAppName = "rook-ceph-mon-pdb"
-)
-
-func (r *ReconcileClusterDisruption) createStaticPDB(request types.NamespacedName, pdb *policyv1beta1.PodDisruptionBudget) error {
+func (r *ReconcileClusterDisruption) createStaticPDB(pdb *policyv1beta1.PodDisruptionBudget) error {
 	err := r.client.Create(context.TODO(), pdb)
 	if err != nil {
-		return err
+		return errors.Wrapf(err, "failed to create pdb %q", pdb.Name)
 	}
 	return nil
 }
 
 func (r *ReconcileClusterDisruption) reconcileStaticPDB(request types.NamespacedName, pdb *policyv1beta1.PodDisruptionBudget) error {
-	err := r.client.Get(context.TODO(), request, pdb)
-	if errors.IsNotFound(err) {
-		return r.createStaticPDB(request, pdb)
-	} else if err != nil {
-		return err
+	existingPDB := &policyv1beta1.PodDisruptionBudget{}
+	err := r.client.Get(context.TODO(), request, existingPDB)
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			return r.createStaticPDB(pdb)
+		}
+		return errors.Wrapf(err, "failed to get pdb %q", pdb.Name)
+	}
+
+	if *existingPDB.Spec.MinAvailable != *pdb.Spec.MinAvailable {
+		err := r.client.Delete(context.TODO(), existingPDB)
+		if err != nil {
+			return errors.Wrapf(err, "failed to delete pdb %q", pdb.Name)
+		}
+		return r.createStaticPDB(pdb)
 	}
 	return nil
 }