Merge pull request #8719 from rook/mergify/bp/release-1.6/pr-8698

ceph: reconcile osd pdb if allowed disruption is 0 (backport #8698)

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

@@ -407,6 +407,21 @@ func (r *ReconcileClusterDisruption) reconcilePDBsForOSDs(
 		return reconcile.Result{Requeue: true, RequeueAfter: 30 * time.Second}, nil
 	}
 
+	// requeue if allowed disruptions in the default PDB is 0
+	allowedDisruptions, err := r.getAllowedDisruptions(osdPDBAppName, request.Namespace)
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			logger.Debugf("default osd pdb %q not found. Skipping reconcile", osdPDBAppName)
+			return reconcile.Result{}, nil
+		}
+		return reconcile.Result{}, errors.Wrapf(err, "failed to get allowed disruptions count from default osd pdb %q.", osdPDBAppName)
+	}
+
+	if allowedDisruptions == 0 {
+		logger.Info("reconciling osd pdb reconciler as the allowed disruptions in default pdb is 0")
+		return reconcile.Result{Requeue: true, RequeueAfter: 30 * time.Second}, nil
+	}
+
 	return reconcile.Result{}, nil
 }
 
@@ -640,6 +655,30 @@ func getLastDrainTimeStamp(pdbStateMap *corev1.ConfigMap, key string) (time.Time
 	return lastDrainTimeStamp, nil
 }
 
+func (r *ReconcileClusterDisruption) getAllowedDisruptions(pdbName, namespace string) (int32, error) {
+	usePDBV1Beta1, err := k8sutil.UsePDBV1Beta1Version(r.context.ClusterdContext.Clientset)
+	if err != nil {
+		return -1, errors.Wrap(err, "failed to fetch pdb version")
+	}
+	if usePDBV1Beta1 {
+		pdb := &policyv1beta1.PodDisruptionBudget{}
+		err = r.client.Get(context.TODO(), types.NamespacedName{Name: pdbName, Namespace: namespace}, pdb)
+		if err != nil {
+			return -1, err
+		}
+
+		return pdb.Status.DisruptionsAllowed, nil
+	}
+
+	pdb := &policyv1.PodDisruptionBudget{}
+	err = r.client.Get(context.TODO(), types.NamespacedName{Name: pdbName, Namespace: namespace}, pdb)
+	if err != nil {
+		return -1, err
+	}
+
+	return pdb.Status.DisruptionsAllowed, nil
+}
+
 func resetPDBConfig(pdbStateMap *corev1.ConfigMap) {
 	pdbStateMap.Data[drainingFailureDomainKey] = ""
 	delete(pdbStateMap.Data, drainingFailureDomainDurationKey)

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

@@ -464,3 +464,31 @@ func TestHasNodeDrained(t *testing.T) {
 	assert.NoError(t, err)
 	assert.True(t, expected)
 }
+
+func TestGetAllowedDisruptions(t *testing.T) {
+	r := getFakeReconciler(t)
+	clientset := test.New(t, 3)
+	test.SetFakeKubernetesVersion(clientset, "v1.21.0")
+	r.context = &controllerconfig.Context{ClusterdContext: &clusterd.Context{Clientset: clientset}}
+
+	// Default PDB is not available
+	allowedDisruptions, err := r.getAllowedDisruptions(osdPDBAppName, namespace)
+	assert.Error(t, err)
+	assert.Equal(t, int32(-1), allowedDisruptions)
+
+	// Default PDB is available
+	pdb := &policyv1.PodDisruptionBudget{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      osdPDBAppName,
+			Namespace: namespace,
+		},
+		Status: policyv1.PodDisruptionBudgetStatus{
+			DisruptionsAllowed: int32(0),
+		},
+	}
+	err = r.client.Create(context.TODO(), pdb)
+	assert.NoError(t, err)
+	allowedDisruptions, err = r.getAllowedDisruptions(osdPDBAppName, namespace)
+	assert.NoError(t, err)
+	assert.Equal(t, int32(0), allowedDisruptions)
+}