Merge pull request #7726 from sp98/pdb-controller-changes

ceph: update podDisruptionBudget controller for OSD
2021-05-11 15:02:22 +02:00
parent 04a226ce52 aee8f56853
commit 87154dfe9c
4 changed files with 394 additions and 156 deletions
@@ -18,7 +18,7 @@ Even if an upgrade agent were only to drain one node at a time, Ceph would have
 The failure domain will be determined by the smallest failure domain of all the Ceph Pools in that cluster.
 We begin with creating a single PodDisruptionBudget for all the OSD with maxUnavailable=1. This will allow one OSD to go down anytime. Once the user drains a node and an OSD goes down, we determine the failure domain for the draining OSD (using the OSD deployment labels). Then we create blocking PodDisruptionBudgets (maxUnavailable=0) for all other failure domains and delete the main PodDisruptionBudget. This blocks OSDs from going down in multiple failure domains simultaneously.

-Once the drained OSDs are back and all the pgs are active+clean, that is, the cluster is healed, the main PodDisruptionBudget is added back and the blocking ones are deleted. User can also add a timeout for the pgs to become healthy. If the timeout exceeds, the operator will ignore the pg health, add the main PodDisruptionBudget and delete the blocking ones.
+Once the drained OSDs are back and all the pgs are active+clean, that is, the cluster is healed, the default PodDisruptionBudget (with maxUnavailable=1) is added back and the blocking ones are deleted. User can also add a timeout for the pgs to become healthy. If the timeout exceeds, the operator will ignore the pg health, add the main PodDisruptionBudget and delete the blocking ones.

 Detecting drains is not easy as they are a client side operation. The client cordons the node and continuously attempts to evict all pods from the node until it succeeds. Whenever an OSD goes into pending state, that is, `ReadyReplicas` count is 0, we assume that some drain operation is happening.

@@ -52,6 +52,8 @@ a common way of deploying kubernetes. This will also work to mitigate manual dra

 When an node is drained, we will also delay it's DOWN/OUT process by placing a noout on that node. We will remove that noout after a timeout.

+An OSD can be down due to reasons other than node drain, say, disk failure. In such a situation, if the pgs are unhealthy then rook will create a blocking PodDisruptionBudget on other failure domains to prevent further node drains on them. `noout` flag won't be set on node this is case. If the OSD is down but all the pgs are `active+clean`, the cluster will be treated as fully healthy. The default PodDisruptionBudget (with maxUnavailable=1) will be added back and the blocking ones will be deleted.
+
 ### Mon, Mgr, MDS, RGW, RBDMirror

 Since there is no strict failure domain requirement for each of these, and they are not logically grouped, a stactic PDB will suffice.
@@ -46,6 +46,7 @@ const (
 	osdPDBAppName                    = "rook-ceph-osd"
 	drainingFailureDomainKey         = "draining-failure-domain"
 	drainingFailureDomainDurationKey = "draining-failure-domain-duration"
+	setNoOut                         = "set-no-out"
 	pgHealthCheckDurationKey         = "pg-health-check-duration"
 	// DefaultMaintenanceTimeout is the period for which a drained failure domain will remain in noout
 	DefaultMaintenanceTimeout = 30 * time.Minute
@@ -68,9 +69,9 @@ func (r *ReconcileClusterDisruption) deletePDB(pdb *policyv1beta1.PodDisruptionB
 	return nil
 }

-// createOverallPDBforOSD creates a single PDB for all OSDs with maxUnavailable=1
+// createDefaultPDBforOSD creates a single PDB for all OSDs with maxUnavailable=1
 // This allows all OSDs in a single failure domain to go down.
-func (r *ReconcileClusterDisruption) createOverallPDBforOSD(namespace string) error {
+func (r *ReconcileClusterDisruption) createDefaultPDBforOSD(namespace string) error {
 	cephCluster, ok := r.clusterMap.GetCluster(namespace)
 	if !ok {
 		return errors.Errorf("failed to find the namespace %q in the clustermap", namespace)
@@ -97,8 +98,8 @@ func (r *ReconcileClusterDisruption) createOverallPDBforOSD(namespace string) er
 	err = r.client.Get(context.TODO(), pdbRequest, &policyv1beta1.PodDisruptionBudget{})
 	if err != nil {
 		if apierrors.IsNotFound(err) {
-			logger.Info("all osds are up. pg health is active+clean")
-			logger.Infof("creating the main pdb %q with maxUnavailable=1 for all osd", osdPDBAppName)
+			logger.Info("all PGs are active+clean. Restoring default OSD pdb settings")
+			logger.Infof("creating the default pdb %q with maxUnavailable=1 for all osd", osdPDBAppName)
 			return r.createPDB(pdb)
 		}
 		return errors.Wrapf(err, "failed to get pdb %q", pdb.Name)
@@ -106,7 +107,7 @@ func (r *ReconcileClusterDisruption) createOverallPDBforOSD(namespace string) er
 	return nil
 }

-func (r *ReconcileClusterDisruption) deleteOverallPDBforOSD(namespace string) error {
+func (r *ReconcileClusterDisruption) deleteDefaultPDBforOSD(namespace string) error {
 	pdbRequest := types.NamespacedName{Name: osdPDBAppName, Namespace: namespace}
 	pdb := &policyv1beta1.PodDisruptionBudget{
 		ObjectMeta: metav1.ObjectMeta{
@@ -121,7 +122,7 @@ func (r *ReconcileClusterDisruption) deleteOverallPDBforOSD(namespace string) er
 		}
 		return errors.Wrapf(err, "failed to get pdb %q", pdb.Name)
 	}
-	logger.Infof("deleting the main pdb %q with maxUnavailable=1 for all osd", osdPDBAppName)
+	logger.Infof("deleting the default pdb %q with maxUnavailable=1 for all osd", osdPDBAppName)
 	return r.deletePDB(pdb)
 }

@@ -198,7 +199,7 @@ func (r *ReconcileClusterDisruption) initializePDBState(request reconcile.Reques

 	if apierrors.IsNotFound(err) {
 		// create configmap to track the draining failure domain
-		pdbStateMap.Data = map[string]string{drainingFailureDomainKey: ""}
+		pdbStateMap.Data = map[string]string{drainingFailureDomainKey: "", setNoOut: ""}
 		err := r.client.Create(context.TODO(), pdbStateMap)
 		if err != nil {
 			return pdbStateMap, errors.Wrapf(err, "failed to create the PDB state map %q", pdbStateMapRequest)
@@ -215,74 +216,100 @@ func (r *ReconcileClusterDisruption) reconcilePDBsForOSDs(
 	pdbStateMap *corev1.ConfigMap,
 	failureDomainType string,
 	allFailureDomains,
-	drainingFailureDomains []string,
-) error {
+	osdDownFailureDomains []string,
+	activeNodeDrains bool,
+) (reconcile.Result, error) {
+	var osdDown bool
+	var drainingFailureDomain string
+	if len(osdDownFailureDomains) > 0 {
+		osdDown = true
+		drainingFailureDomain = osdDownFailureDomains[0]
+	}

-	pgHealthMsg, clean, err := cephclient.IsClusterClean(r.context.ClusterdContext, clusterInfo)
+	pgHealthMsg, pgClean, err := cephclient.IsClusterClean(r.context.ClusterdContext, clusterInfo)
 	if err != nil {
 		// If the error contains that message, this means the cluster is not up and running
 		// No monitors are present and thus no ceph configuration has been created
 		if strings.Contains(err.Error(), opcontroller.UninitializedCephConfigError) {
-			logger.Debugf("Ceph %q cluster not ready, cannot check Ceph status yet.", request.Namespace)
-			return nil
+			logger.Infof("Ceph %q cluster not ready, cannot check Ceph status yet.", request.Namespace)
+			return opcontroller.WaitForRequeueIfOperatorNotInitialized, nil
 		}
-		return errors.Wrapf(err, "failed to check cluster health")
+		return reconcile.Result{}, errors.Wrapf(err, "failed to check cluster health")
 	}

-	if pdbStateMap.Data == nil {
-		pdbStateMap.Data = make(map[string]string)
-	}
-	_, ok := pdbStateMap.Data[drainingFailureDomainKey]
-	if !ok {
-		pdbStateMap.Data[drainingFailureDomainKey] = ""
-	}
-
-	var drainingFailureDomain string
-	activeDrains := len(drainingFailureDomains) > 0
-
-	// create blocking PDBs for all the OSDs that are not being drained right now
-	if activeDrains {
-		drainingFailureDomain = drainingFailureDomains[0]
-		logger.Infof("all %q failure domains: %v. currently draining: %q. pg health: %q", failureDomainType,
-			allFailureDomains, drainingFailureDomain, pgHealthMsg)
-		err := r.handleActiveDrains(allFailureDomains, drainingFailureDomain, failureDomainType, clusterInfo.Namespace, clean)
+	switch {
+	// osd is down but pgs are active+clean
+	case osdDown && pgClean:
+		lastDrainTimeStamp, err := getLastDrainTimeStamp(pdbStateMap, drainingFailureDomainDurationKey)
 		if err != nil {
-			return errors.Wrap(err, "failed to handle active drains")
+			return reconcile.Result{}, errors.Wrapf(err, "failed to get last drain timestamp from the configmap %q", pdbStateMap.Name)
 		}
+		timeSinceOSDDown := time.Since(lastDrainTimeStamp)
+		if timeSinceOSDDown > 30*time.Second {
+			logger.Infof("osd is down in failure domain %q is down for the last %.2f minutes, but pgs are active+clean", drainingFailureDomain, timeSinceOSDDown.Minutes())
+			resetPDBConfig(pdbStateMap)
+		} else {
+			logger.Infof("osd is down in the failure domain %q, but pgs are active+clean. Requeuing in case pg status is not updated yet...", drainingFailureDomain)
+			return reconcile.Result{Requeue: true, RequeueAfter: 15 * time.Second}, nil
+		}
+
+	// osd is down and pgs are not healthy
+	case osdDown && !pgClean:
+		logger.Infof("osd is down in failure domain %q and pgs are not active+clean. pg health: %q", drainingFailureDomain, pgHealthMsg)
 		currentlyDrainingFD, ok := pdbStateMap.Data[drainingFailureDomainKey]
 		if !ok || drainingFailureDomain != currentlyDrainingFD {
 			pdbStateMap.Data[drainingFailureDomainKey] = drainingFailureDomain
 			pdbStateMap.Data[drainingFailureDomainDurationKey] = time.Now().Format(time.RFC3339)
 		}
-	} else if clean || r.hasPGHealthCheckTimedout(pdbStateMap) {
+		if activeNodeDrains {
+			pdbStateMap.Data[setNoOut] = "true"
+		}
+
+	// osd is back up and either pgs have become healthy or pg healthy check timeout has elapsed
+	case !osdDown && (pgClean || r.hasPGHealthCheckTimedout(pdbStateMap)):
 		// reset the configMap if cluster is clean or if the timeout for PGs to become active+clean has exceeded
-		logger.Debugf("no drains detected in %q failure domains: %v. pg health: %q", failureDomainType, allFailureDomains, pgHealthMsg)
-		pdbStateMap.Data[drainingFailureDomainKey] = ""
-		delete(pdbStateMap.Data, drainingFailureDomainDurationKey)
-		delete(pdbStateMap.Data, pgHealthCheckDurationKey)
-	} else {
-		logger.Infof("all osds are up. last drained failure domain: %q.  pg health: %q. waiting for PGs to be active+clean or PG health check to timeout",
-			pdbStateMap.Data[drainingFailureDomainKey], pgHealthMsg)
+		logger.Debugf("no OSD is down in the %q failure domains: %v. pg health: %q", failureDomainType, allFailureDomains, pgHealthMsg)
+		resetPDBConfig(pdbStateMap)
+
+	default:
+		logger.Infof("all %q failure domains: %v. osd is down in failure domain: %q. active node drains: %t. pg health: %q", failureDomainType,
+			allFailureDomains, drainingFailureDomain, activeNodeDrains, pgHealthMsg)
 	}

-	err = r.updateNoout(clusterInfo, pdbStateMap, allFailureDomains)
-	if err != nil {
-		logger.Errorf("failed to update maintenance noout in cluster %q. %v", request, err)
+	if pdbStateMap.Data[setNoOut] == "true" {
+		err = r.updateNoout(clusterInfo, pdbStateMap, allFailureDomains)
+		if err != nil {
+			logger.Errorf("failed to update maintenance noout in cluster %q. %v", request, err)
+		}
+	}
+
+	if pdbStateMap.Data[drainingFailureDomainKey] != "" && !pgClean {
+		// delete default OSD pdb and create blocking OSD pdbs
+		err := r.handleActiveDrains(allFailureDomains, pdbStateMap.Data[drainingFailureDomainKey], failureDomainType, clusterInfo.Namespace, pgClean)
+		if err != nil {
+			return reconcile.Result{}, errors.Wrap(err, "failed to handle active drains")
+		}
+	} else if pdbStateMap.Data[drainingFailureDomainKey] == "" {
+		// delete all blocking OSD pdb and restore the default OSD pdb
+		err := r.handleInactiveDrains(allFailureDomains, failureDomainType, clusterInfo.Namespace)
+		if err != nil {
+			return reconcile.Result{}, errors.Wrap(err, "failed to handle inactive drains")
+		}
+		// reset `set-no-out` flag on the configMap
+		pdbStateMap.Data[setNoOut] = ""
 	}

 	err = r.client.Update(context.TODO(), pdbStateMap)
 	if err != nil {
-		return errors.Wrapf(err, "failed to update configMap %q in cluster %q", pdbStateMapName, request)
+		return reconcile.Result{}, errors.Wrapf(err, "failed to update configMap %q in cluster %q", pdbStateMapName, request)
 	}

-	// handle inactive drains by deleting all blocking pdb and restoring the main pdb
-	if pdbStateMap.Data[drainingFailureDomainKey] == "" {
-		err := r.handleInactiveDrains(allFailureDomains, clusterInfo.Namespace, failureDomainType)
-		if err != nil {
-			return errors.Wrap(err, "failed to handle inactive drains")
-		}
+	// requeue if drain is still in progress
+	if len(pdbStateMap.Data[drainingFailureDomainKey]) > 0 {
+		return reconcile.Result{Requeue: true, RequeueAfter: 30 * time.Second}, nil
 	}
-	return nil
+
+	return reconcile.Result{}, nil
 }

 func (r *ReconcileClusterDisruption) handleActiveDrains(allFailureDomains []string, drainingFailureDomain,
@@ -305,20 +332,20 @@ func (r *ReconcileClusterDisruption) handleActiveDrains(allFailureDomains []stri
 		}
 	}

-	// delete the main PDB for OSD
+	// delete the default PDB for OSD
 	// This will allow all OSDs in the currently drained failure domain to be removed.
-	logger.Debug("deleting main pdb with maxUnavailable=1 for all osd")
-	err := r.deleteOverallPDBforOSD(namespace)
+	logger.Debug("deleting default pdb with maxUnavailable=1 for all osd")
+	err := r.deleteDefaultPDBforOSD(namespace)
 	if err != nil {
-		return errors.Wrap(err, "failed to delete the main osd pdb")
+		return errors.Wrap(err, "failed to delete the default osd pdb")
 	}
 	return nil
 }

-func (r *ReconcileClusterDisruption) handleInactiveDrains(allFailureDomains []string, namespace, failureDomainType string) error {
-	err := r.createOverallPDBforOSD(namespace)
+func (r *ReconcileClusterDisruption) handleInactiveDrains(allFailureDomains []string, failureDomainType, namespace string) error {
+	err := r.createDefaultPDBforOSD(namespace)
 	if err != nil {
-		return errors.Wrap(err, "failed to create main pdb")
+		return errors.Wrap(err, "failed to create default pdb")
 	}
 	for _, failureDomainName := range allFailureDomains {
 		err := r.deleteBlockingPDBForOSD(namespace, failureDomainType, failureDomainName)
@@ -375,37 +402,51 @@ func (r *ReconcileClusterDisruption) updateNoout(clusterInfo *cephclient.Cluster
 	return nil
 }

-func (r *ReconcileClusterDisruption) getOSDFailureDomains(clusterInfo *cephclient.ClusterInfo, request reconcile.Request, poolFailureDomain string) ([]string, []string, error) {
+func (r *ReconcileClusterDisruption) getOSDFailureDomains(clusterInfo *cephclient.ClusterInfo, request reconcile.Request, poolFailureDomain string) ([]string, []string, []string, error) {
 	osdDeploymentList := &appsv1.DeploymentList{}
 	namespaceListOpts := client.InNamespace(request.Namespace)
 	topologyLocationLabel := fmt.Sprintf(osd.TopologyLocationLabel, poolFailureDomain)
 	err := r.client.List(context.TODO(), osdDeploymentList, client.MatchingLabels{k8sutil.AppAttr: osd.AppName}, namespaceListOpts)
 	if err != nil {
-		return []string{}, []string{}, errors.Wrap(err, "failed to list osd deployments")
+		return nil, nil, nil, errors.Wrap(err, "failed to list osd deployments")
 	}

 	allFailureDomains := sets.NewString()
-	drainingFailureDomains := sets.NewString()
+	nodeDrainFailureDomains := sets.NewString()
+	osdDownFailureDomains := sets.NewString()

 	for _, deployment := range osdDeploymentList.Items {
 		labels := deployment.Spec.Template.ObjectMeta.GetLabels()
 		failureDomainName := labels[topologyLocationLabel]
 		if failureDomainName == "" {
-			return []string{}, []string{}, errors.Errorf("failed to get the topology location label %q in OSD deployment %q",
+			return nil, nil, nil, errors.Errorf("failed to get the topology location label %q in OSD deployment %q",
 				topologyLocationLabel, deployment.Name)
 		}

+		// Assume node drain if osd deployment ReadyReplicas count is 0 and OSD pod is not scheduled on a node
 		if deployment.Status.ReadyReplicas < 1 {
-			if !drainingFailureDomains.Has(failureDomainName) {
-				drainingFailureDomains.Insert(failureDomainName)
+			if !osdDownFailureDomains.Has(failureDomainName) {
+				osdDownFailureDomains.Insert(failureDomainName)
 			}
-		} else {
-			if !allFailureDomains.Has(failureDomainName) {
-				allFailureDomains.Insert(failureDomainName)
+			isDrained, err := hasOSDNodeDrained(r.client, request.Namespace, labels[osd.OsdIdLabelKey])
+			if err != nil {
+				return nil, nil, nil, errors.Wrapf(err, "failed to check if osd %q node is drained", deployment.Name)
+			}
+			if isDrained {
+				logger.Infof("osd %q is down and a possible node drain is detected", deployment.Name)
+				if !nodeDrainFailureDomains.Has(failureDomainName) {
+					nodeDrainFailureDomains.Insert(failureDomainName)
+				}
+			} else {
+				logger.Infof("osd %q is down but no node drain is detected", deployment.Name)
 			}
 		}
+
+		if !allFailureDomains.Has(failureDomainName) {
+			allFailureDomains.Insert(failureDomainName)
+		}
 	}
-	return allFailureDomains.List(), drainingFailureDomains.List(), nil
+	return allFailureDomains.List(), nodeDrainFailureDomains.List(), osdDownFailureDomains.List(), nil
 }

 func (r *ReconcileClusterDisruption) hasPGHealthCheckTimedout(pdbStateMap *corev1.ConfigMap) bool {
@@ -435,6 +476,74 @@ func (r *ReconcileClusterDisruption) hasPGHealthCheckTimedout(pdbStateMap *corev
 	return false
 }

+// hasNodeDrained returns true if OSD pod is not assigned to any node or if the OSD node is not schedulable
+func hasOSDNodeDrained(c client.Client, namespace, osdID string) (bool, error) {
+	osdNodeName, err := getOSDNodeName(c, namespace, osdID)
+	if err != nil {
+		return false, errors.Wrapf(err, "failed to get node name assigned to OSD %q POD", osdID)
+	}
+
+	if osdNodeName == "" {
+		logger.Debugf("osd %q POD is not assigned to any node. assuming node drain", osdID)
+		return true, nil
+	}
+
+	node, err := getNode(c, osdNodeName)
+	if err != nil {
+		return false, errors.Wrapf(err, "failed to get node assigned to OSD %q POD", osdID)
+	}
+	return node.Spec.Unschedulable, nil
+}
+
+func getOSDNodeName(c client.Client, namespace, osdID string) (string, error) {
+	pods := &corev1.PodList{}
+	listOpts := []client.ListOption{
+		client.InNamespace(namespace),
+		client.MatchingLabels{osd.OsdIdLabelKey: osdID},
+	}
+
+	err := c.List(context.TODO(), pods, listOpts...)
+	if err != nil {
+		return "", errors.Wrapf(err, "failed to list pods for osd %q", osdID)
+	}
+
+	if len(pods.Items) > 0 {
+		return pods.Items[0].Spec.NodeName, nil
+	}
+	return "", nil
+}
+
+func getNode(c client.Client, nodeName string) (*corev1.Node, error) {
+	node := &corev1.Node{}
+	err := c.Get(context.TODO(), types.NamespacedName{Name: nodeName}, node)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to get node %q", nodeName)
+	}
+	return node, nil
+}
+
 func getPDBName(failureDomainType, failureDomainName string) string {
 	return k8sutil.TruncateNodeName(fmt.Sprintf("%s-%s-%s", osdPDBAppName, failureDomainType, "%s"), failureDomainName)
 }
+
+func getLastDrainTimeStamp(pdbStateMap *corev1.ConfigMap, key string) (time.Time, error) {
+	var err error
+	var lastDrainTimeStamp time.Time
+	lastDrainTimeStampString, ok := pdbStateMap.Data[key]
+	if !ok || len(lastDrainTimeStampString) == 0 {
+		return time.Now(), nil
+	} else {
+		lastDrainTimeStamp, err = time.Parse(time.RFC3339, pdbStateMap.Data[key])
+		if err != nil {
+			return time.Time{}, errors.Wrapf(err, "failed to parse timestamp %q", pdbStateMap.Data[key])
+		}
+	}
+
+	return lastDrainTimeStamp, nil
+}
+
+func resetPDBConfig(pdbStateMap *corev1.ConfigMap) {
+	pdbStateMap.Data[drainingFailureDomainKey] = ""
+	delete(pdbStateMap.Data, drainingFailureDomainDurationKey)
+	delete(pdbStateMap.Data, pgHealthCheckDurationKey)
+}
@@ -45,13 +45,26 @@ const (
 	unHealthyCephStatus = `{"fsid":"613975f3-3025-4802-9de1-a2280b950e75","health":{"checks":{"OSD_DOWN":{"severity":"HEALTH_WARN","summary":{"message":"1 osds down"}},"OSD_HOST_DOWN":{"severity":"HEALTH_WARN","summary":{"message":"1 host (1 osds) down"}},"PG_AVAILABILITY":{"severity":"HEALTH_WARN","summary":{"message":"Reduced data availability: 101 pgs stale"}},"POOL_APP_NOT_ENABLED":{"severity":"HEALTH_WARN","summary":{"message":"application not enabled on 1 pool(s)"}}},"status":"HEALTH_WARN","overall_status":"HEALTH_WARN"},"election_epoch":12,"quorum":[0,1,2],"quorum_names":["rook-ceph-mon0","rook-ceph-mon2","rook-ceph-mon1"],"monmap":{"epoch":3,"fsid":"613975f3-3025-4802-9de1-a2280b950e75","modified":"2017-08-11 20:13:02.075679","created":"2017-08-11 20:12:35.314510","features":{"persistent":["kraken","luminous"],"optional":[]},"mons":[{"rank":0,"name":"rook-ceph-mon0","addr":"10.3.0.45:6789/0","public_addr":"10.3.0.45:6789/0"},{"rank":1,"name":"rook-ceph-mon2","addr":"10.3.0.249:6789/0","public_addr":"10.3.0.249:6789/0"},{"rank":2,"name":"rook-ceph-mon1","addr":"10.3.0.252:6789/0","public_addr":"10.3.0.252:6789/0"}]},"osdmap":{"osdmap":{"epoch":17,"num_osds":2,"num_up_osds":1,"num_in_osds":2,"full":false,"nearfull":true,"num_remapped_pgs":0}},"pgmap":{"pgs_by_state":[{"state_name":"stale+active+clean","count":101},{"state_name":"active+clean","count":99}],"num_pgs":200,"num_pools":2,"num_objects":243,"data_bytes":976793635,"bytes_used":13611479040,"bytes_avail":19825307648,"bytes_total":33436786688},"fsmap":{"epoch":1,"by_rank":[]},"mgrmap":{"epoch":3,"active_gid":14111,"active_name":"rook-ceph-mgr0","active_addr":"10.2.73.6:6800/9","available":true,"standbys":[],"modules":["restful","status"],"available_modules":["dashboard","prometheus","restful","status","zabbix"]},"servicemap":{"epoch":1,"modified":"0.000000","services":{}}}`
 )

+var nodeName = "node01"
 var namespace = "rook-ceph"

 var cephCluster = &cephv1.CephCluster{
-	ObjectMeta: metav1.ObjectMeta{Name: "ceph-cluster"}}
+	ObjectMeta: metav1.ObjectMeta{Name: "ceph-cluster"},
+}

-var pdbConfigMap = &corev1.ConfigMap{
-	ObjectMeta: metav1.ObjectMeta{Name: pdbStateMapName, Namespace: namespace}}
+var nodeObj = &corev1.Node{
+	ObjectMeta: metav1.ObjectMeta{Name: nodeName},
+	Spec: corev1.NodeSpec{
+		Unschedulable: false,
+	},
+}
+
+var unschedulableNodeObj = &corev1.Node{
+	ObjectMeta: metav1.ObjectMeta{Name: nodeName},
+	Spec: corev1.NodeSpec{
+		Unschedulable: true,
+	},
+}

 func fakeOSDDeployment(id, readyReplicas int) appsv1.Deployment {
 	osd := appsv1.Deployment{
@@ -67,6 +80,7 @@ func fakeOSDDeployment(id, readyReplicas int) appsv1.Deployment {
 				ObjectMeta: metav1.ObjectMeta{
 					Labels: map[string]string{
 						"topology-location-zone": fmt.Sprintf("zone-%d", id),
+						"ceph-osd-id":            fmt.Sprintf("%d", id),
 					},
 				},
 			},
@@ -78,6 +92,30 @@ func fakeOSDDeployment(id, readyReplicas int) appsv1.Deployment {
 	return osd
 }

+func fakeOSDPod(id int, nodeName string) corev1.Pod {
+	osdPod := corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      fmt.Sprintf("osd-%d", id),
+			Namespace: namespace,
+			Labels: map[string]string{
+				"app":         "rook-ceph-osd",
+				"ceph-osd-id": fmt.Sprintf("%d", id),
+			},
+		},
+		Spec: corev1.PodSpec{
+			NodeName: nodeName,
+		},
+	}
+	return osdPod
+}
+
+func fakePDBConfigMap(drainingFailureDomain string) *corev1.ConfigMap {
+	return &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{Name: pdbStateMapName, Namespace: namespace},
+		Data:       map[string]string{drainingFailureDomainKey: drainingFailureDomain, setNoOut: ""},
+	}
+}
+
 func getFakeReconciler(t *testing.T, obj ...runtime.Object) *ReconcileClusterDisruption {
 	scheme := scheme.Scheme
 	err := policyv1beta1.AddToScheme(scheme)
@@ -110,128 +148,193 @@ func getFakeClusterInfo() *client.ClusterInfo {

 func TestGetOSDFailureDomains(t *testing.T) {
 	testcases := []struct {
-		label                          string
+		name                           string
 		osds                           []appsv1.Deployment
+		osdPods                        []corev1.Pod
+		node                           *corev1.Node
 		expectedAllFailureDomains      []string
 		expectedDrainingFailureDomains []string
+		expectedOsdDownFailureDomains  []string
 	}{
 		{
-			label: "case 1: all osds are running",
+			name: "case 1: all osds are running",
 			osds: []appsv1.Deployment{fakeOSDDeployment(1, 1), fakeOSDDeployment(2, 1),
 				fakeOSDDeployment(3, 1)},
+			osdPods: []corev1.Pod{fakeOSDPod(1, nodeName), fakeOSDPod(2, nodeName),
+				fakeOSDPod(3, nodeName)},
+			node:                           nodeObj,
 			expectedAllFailureDomains:      []string{"zone-1", "zone-2", "zone-3"},
+			expectedOsdDownFailureDomains:  []string{},
 			expectedDrainingFailureDomains: []string{},
 		},
 		{
-			label: "case 1: osd in zone-1 is pending",
+			name: "case 2: osd in zone-1 is pending and node is unscheduable",
 			osds: []appsv1.Deployment{fakeOSDDeployment(1, 0), fakeOSDDeployment(2, 1),
 				fakeOSDDeployment(3, 1)},
-			expectedAllFailureDomains:      []string{"zone-2", "zone-3"},
+			osdPods: []corev1.Pod{fakeOSDPod(1, ""), fakeOSDPod(2, nodeName),
+				fakeOSDPod(3, nodeName)},
+			node:                           nodeObj,
+			expectedAllFailureDomains:      []string{"zone-1", "zone-2", "zone-3"},
+			expectedOsdDownFailureDomains:  []string{"zone-1"},
 			expectedDrainingFailureDomains: []string{"zone-1"},
 		},
 		{
-			label: "case 1: osds in zone-1 and zone-2 are pending",
+			name: "case 3: osd in zone-1 and zone-2 are pending and node is unscheduable",
 			osds: []appsv1.Deployment{fakeOSDDeployment(1, 0), fakeOSDDeployment(2, 0),
 				fakeOSDDeployment(3, 1)},
-			expectedAllFailureDomains:      []string{"zone-3"},
+			osdPods: []corev1.Pod{fakeOSDPod(1, ""), fakeOSDPod(2, ""),
+				fakeOSDPod(3, nodeName)},
+			node:                           nodeObj,
+			expectedAllFailureDomains:      []string{"zone-1", "zone-2", "zone-3"},
+			expectedOsdDownFailureDomains:  []string{"zone-1", "zone-2"},
 			expectedDrainingFailureDomains: []string{"zone-1", "zone-2"},
 		},
+		{
+			name: "case 4: osd in zone-1 is pending but osd node is scheduable",
+			osds: []appsv1.Deployment{fakeOSDDeployment(1, 0), fakeOSDDeployment(2, 1),
+				fakeOSDDeployment(3, 1)},
+			osdPods: []corev1.Pod{fakeOSDPod(1, nodeName), fakeOSDPod(2, nodeName),
+				fakeOSDPod(3, nodeName)},
+			node:                           nodeObj,
+			expectedAllFailureDomains:      []string{"zone-1", "zone-2", "zone-3"},
+			expectedOsdDownFailureDomains:  []string{"zone-1"},
+			expectedDrainingFailureDomains: []string{},
+		},
+		{
+			name: "case 5: osd in zone-1 is pending but osd node is not scheduable",
+			osds: []appsv1.Deployment{fakeOSDDeployment(1, 0), fakeOSDDeployment(2, 1),
+				fakeOSDDeployment(3, 1)},
+			osdPods: []corev1.Pod{fakeOSDPod(1, nodeName), fakeOSDPod(2, nodeName),
+				fakeOSDPod(3, nodeName)},
+			node:                           unschedulableNodeObj,
+			expectedAllFailureDomains:      []string{"zone-1", "zone-2", "zone-3"},
+			expectedOsdDownFailureDomains:  []string{"zone-1"},
+			expectedDrainingFailureDomains: []string{"zone-1"},
+		},
 	}

 	for _, tc := range testcases {
-		r := getFakeReconciler(t, cephCluster, &corev1.ConfigMap{}, tc.osds[0].DeepCopy(),
-			tc.osds[1].DeepCopy(), tc.osds[2].DeepCopy())
-		clusterInfo := getFakeClusterInfo()
-		request := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}}
-		allfailureDomains, drainingFailureDomains, err := r.getOSDFailureDomains(clusterInfo, request, "zone")
-		assert.NoError(t, err)
-		assert.Equal(t, tc.expectedAllFailureDomains, allfailureDomains)
-		assert.Equal(t, tc.expectedDrainingFailureDomains, drainingFailureDomains)
+		t.Run(tc.name, func(t *testing.T) {
+			objs := []runtime.Object{
+				cephCluster,
+				&corev1.ConfigMap{},
+				tc.node,
+			}
+			for _, osdDeployment := range tc.osds {
+				objs = append(objs, osdDeployment.DeepCopy())
+			}
+			for _, osdPod := range tc.osdPods {
+				objs = append(objs, osdPod.DeepCopy())
+			}
+			r := getFakeReconciler(t, objs...)
+			clusterInfo := getFakeClusterInfo()
+			request := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}}
+			allfailureDomains, nodeDrainFailureDomains, osdDownFailureDomains, err := r.getOSDFailureDomains(clusterInfo, request, "zone")
+			assert.NoError(t, err)
+			assert.Equal(t, tc.expectedAllFailureDomains, allfailureDomains)
+			assert.Equal(t, tc.expectedDrainingFailureDomains, nodeDrainFailureDomains)
+			assert.Equal(t, tc.expectedOsdDownFailureDomains, osdDownFailureDomains)
+		})
 	}
 }

 func TestGetOSDFailureDomainsError(t *testing.T) {
 	testcases := []struct {
-		label                          string
+		name                           string
 		osds                           []appsv1.Deployment
 		expectedAllFailureDomains      []string
 		expectedDrainingFailureDomains []string
+		expectedOsdDownFailureDomains  []string
 	}{
 		{
-			label: "case 1: one or more OSD deployment is missing crush location label",
+			name: "case 1: one or more OSD deployment is missing crush location label",
 			osds: []appsv1.Deployment{fakeOSDDeployment(1, 1), fakeOSDDeployment(2, 1),
 				fakeOSDDeployment(3, 1)},
-			expectedAllFailureDomains:      []string{},
-			expectedDrainingFailureDomains: []string{},
+			expectedAllFailureDomains:      nil,
+			expectedDrainingFailureDomains: nil,
+			expectedOsdDownFailureDomains:  nil,
 		},
 	}

 	for _, tc := range testcases {
-		osd := tc.osds[0].DeepCopy()
-		osd.Spec.Template.ObjectMeta.Labels["topology-location-zone"] = ""
-		r := getFakeReconciler(t, cephCluster, &corev1.ConfigMap{},
-			tc.osds[1].DeepCopy(), tc.osds[2].DeepCopy(), osd)
-		clusterInfo := getFakeClusterInfo()
-		request := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}}
-		allfailureDomains, drainingFailureDomains, err := r.getOSDFailureDomains(clusterInfo, request, "zone")
-		assert.Error(t, err)
-		assert.Equal(t, tc.expectedAllFailureDomains, allfailureDomains)
-		assert.Equal(t, tc.expectedDrainingFailureDomains, drainingFailureDomains)
+		t.Run(tc.name, func(t *testing.T) {
+			osd := tc.osds[0].DeepCopy()
+			osd.Spec.Template.ObjectMeta.Labels["topology-location-zone"] = ""
+			r := getFakeReconciler(t, cephCluster, &corev1.ConfigMap{},
+				tc.osds[1].DeepCopy(), tc.osds[2].DeepCopy(), osd)
+			clusterInfo := getFakeClusterInfo()
+			request := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}}
+			allfailureDomains, nodeDrainFailureDomains, osdDownFailureDomains, err := r.getOSDFailureDomains(clusterInfo, request, "zone")
+			assert.Error(t, err)
+			assert.Equal(t, tc.expectedAllFailureDomains, allfailureDomains)
+			assert.Equal(t, tc.expectedDrainingFailureDomains, nodeDrainFailureDomains)
+			assert.Equal(t, tc.expectedOsdDownFailureDomains, osdDownFailureDomains)
+		})
 	}
 }

 func TestReconcilePDBForOSD(t *testing.T) {
-	pdbConfig := pdbConfigMap.DeepCopy()
-	r := getFakeReconciler(t, cephCluster, pdbConfig)
-	clusterInfo := getFakeClusterInfo()
-	request := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}}
-
 	testcases := []struct {
-		label                             string
+		name                              string
 		fakeCephStatus                    string
 		fakeOSDDump                       string
+		configMap                         *corev1.ConfigMap
 		allFailureDomains                 []string
-		drainingFailureDomains            []string
+		osdDownFailureDomains             []string
+		activeNodeDrains                  bool
+		expectedSetNoOutValue             string
 		expectedOSDPDBCount               int
 		expectedMaxUnavailableCount       int
 		expectedDrainingFailureDomainName string
 	}{
 		{
-			label:                             "case 1: no draining failure domain and all pgs are healthy",
+			name:                              "case 1: no draining failure domain and all pgs are healthy",
 			fakeCephStatus:                    healthyCephStatus,
 			fakeOSDDump:                       `{"OSDs": [{"OSD": 3, "Up": 3, "In": 3}]}`,
 			allFailureDomains:                 []string{"zone-1", "zone-2", "zone-3"},
-			drainingFailureDomains:            []string{},
+			osdDownFailureDomains:             []string{},
+			configMap:                         fakePDBConfigMap(""),
+			activeNodeDrains:                  false,
+			expectedSetNoOutValue:             "",
 			expectedOSDPDBCount:               1,
 			expectedMaxUnavailableCount:       1,
 			expectedDrainingFailureDomainName: "",
 		},
 		{
-			label:                             "case 2: zone-1 failure domain is draining and pgs are unhealthy",
+			name:                              "case 2: zone-1 failure domain is draining and pgs are unhealthy",
 			fakeCephStatus:                    unHealthyCephStatus,
 			fakeOSDDump:                       `{"OSDs": [{"OSD": 3, "Up": 3, "In": 2}]}`,
 			allFailureDomains:                 []string{"zone-1", "zone-2", "zone-3"},
-			drainingFailureDomains:            []string{"zone-1"},
+			osdDownFailureDomains:             []string{"zone-1"},
+			configMap:                         fakePDBConfigMap(""),
+			activeNodeDrains:                  true,
+			expectedSetNoOutValue:             "true",
 			expectedOSDPDBCount:               2,
 			expectedMaxUnavailableCount:       0,
 			expectedDrainingFailureDomainName: "zone-1",
 		},
 		{
-			label:                             "case 3: zone-1 is back online. But pgs are still unhealthy from zone-1 drain",
+			name:                              "case 3: zone-1 is back online. But pgs are still unhealthy from zone-1 drain",
 			fakeCephStatus:                    unHealthyCephStatus,
 			fakeOSDDump:                       `{"OSDs": [{"OSD": 3, "Up": 3, "In": 3}]}`,
 			allFailureDomains:                 []string{"zone-1", "zone-2", "zone-3"},
-			drainingFailureDomains:            []string{},
+			osdDownFailureDomains:             []string{},
+			configMap:                         fakePDBConfigMap("zone-1"),
+			activeNodeDrains:                  true,
+			expectedSetNoOutValue:             "",
 			expectedOSDPDBCount:               2,
 			expectedMaxUnavailableCount:       0,
 			expectedDrainingFailureDomainName: "zone-1",
 		},
 		{
-			label:                             "case 4: zone-1 is back online and pgs are also healthy",
+			name:                              "case 4: zone-1 is back online and pgs are also healthy",
 			fakeCephStatus:                    healthyCephStatus,
 			fakeOSDDump:                       `{"OSDs": [{"OSD": 3, "Up": 3, "In": 3}]}`,
 			allFailureDomains:                 []string{"zone-1", "zone-2", "zone-3"},
-			drainingFailureDomains:            []string{},
+			osdDownFailureDomains:             []string{},
+			configMap:                         fakePDBConfigMap(""),
+			activeNodeDrains:                  true,
+			expectedSetNoOutValue:             "",
 			expectedOSDPDBCount:               1,
 			expectedMaxUnavailableCount:       1,
 			expectedDrainingFailureDomainName: "",
@@ -239,40 +342,47 @@ func TestReconcilePDBForOSD(t *testing.T) {
 	}

 	for _, tc := range testcases {
-		executor := &exectest.MockExecutor{}
-		executor.MockExecuteCommandWithOutputFile = func(command, outputFile string, args ...string) (string, error) {
-			logger.Infof("Command: %s %v", command, args)
-			if args[0] == "status" {
-				return tc.fakeCephStatus, nil
+		t.Run(tc.name, func(t *testing.T) {
+			r := getFakeReconciler(t, cephCluster, tc.configMap)
+			clusterInfo := getFakeClusterInfo()
+			request := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}}
+			executor := &exectest.MockExecutor{}
+			executor.MockExecuteCommandWithOutputFile = func(command, outputFile string, args ...string) (string, error) {
+				logger.Infof("Command: %s %v", command, args)
+				if args[0] == "status" {
+					return tc.fakeCephStatus, nil
+				}
+				if args[0] == "osd" && args[1] == "dump" {
+					return tc.fakeOSDDump, nil
+				}
+				return "", errors.Errorf("unexpected ceph command '%v'", args)
 			}
-			if args[0] == "osd" && args[1] == "dump" {
-				return tc.fakeOSDDump, nil
+			r.context = &controllerconfig.Context{ClusterdContext: &clusterd.Context{Executor: executor}}
+			_, err := r.reconcilePDBsForOSDs(clusterInfo, request, tc.configMap, "zone", tc.allFailureDomains, tc.osdDownFailureDomains, tc.activeNodeDrains)
+			assert.NoError(t, err)
+
+			// assert that pdb for osd are created correctly
+			existingPDBs := &policyv1beta1.PodDisruptionBudgetList{}
+			err = r.client.List(context.TODO(), existingPDBs)
+			assert.NoError(t, err)
+			assert.Equal(t, tc.expectedOSDPDBCount, len(existingPDBs.Items))
+			for _, pdb := range existingPDBs.Items {
+				assert.Equal(t, tc.expectedMaxUnavailableCount, pdb.Spec.MaxUnavailable.IntValue())
 			}
-			return "", errors.Errorf("unexpected ceph command '%v'", args)
-		}
-		r.context = &controllerconfig.Context{ClusterdContext: &clusterd.Context{Executor: executor}}
-		err := r.reconcilePDBsForOSDs(clusterInfo, request, pdbConfig, "zone", tc.allFailureDomains, tc.drainingFailureDomains)
-		assert.NoError(t, err)

-		// assert that pdb for osd are created correctly
-		existingPDBs := &policyv1beta1.PodDisruptionBudgetList{}
-		err = r.client.List(context.TODO(), existingPDBs)
-		assert.NoError(t, err)
-		assert.Equal(t, tc.expectedOSDPDBCount, len(existingPDBs.Items))
-		for _, pdb := range existingPDBs.Items {
-			assert.Equal(t, tc.expectedMaxUnavailableCount, pdb.Spec.MaxUnavailable.IntValue())
-		}
+			// assert that config map is updated with correct failure domain
+			existingConfigMaps := &corev1.ConfigMapList{}
+			err = r.client.List(context.TODO(), existingConfigMaps)
+			assert.NoError(t, err)
+			assert.Equal(t, tc.expectedDrainingFailureDomainName, existingConfigMaps.Items[0].Data[drainingFailureDomainKey])
+			assert.Equal(t, tc.expectedSetNoOutValue, existingConfigMaps.Items[0].Data[setNoOut])
+		})

-		// assert that config map is updated with correct failure domain
-		existingConfigMaps := &corev1.ConfigMapList{}
-		err = r.client.List(context.TODO(), existingConfigMaps)
-		assert.NoError(t, err)
-		assert.Equal(t, tc.expectedDrainingFailureDomainName, existingConfigMaps.Items[0].Data[drainingFailureDomainKey])
 	}
 }

 func TestPGHealthcheckTimeout(t *testing.T) {
-	pdbConfig := pdbConfigMap.DeepCopy()
+	pdbConfig := fakePDBConfigMap("")
 	r := getFakeReconciler(t, cephCluster, pdbConfig)
 	clusterInfo := getFakeClusterInfo()
 	request := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}}
@@ -292,23 +402,49 @@ func TestPGHealthcheckTimeout(t *testing.T) {
 	r.pgHealthCheckTimeout = time.Duration(time.Minute * 10)

 	// reconcile OSD PDB with active drains (on zone-1) and unhealthy PGs
-	err := r.reconcilePDBsForOSDs(clusterInfo, request, pdbConfig, "zone", []string{"zone-1", "zone-2"}, []string{"zone-1"})
+	_, err := r.reconcilePDBsForOSDs(clusterInfo, request, pdbConfig, "zone", []string{"zone-1", "zone-2"}, []string{"zone-1"}, true)
 	assert.NoError(t, err)
 	assert.Equal(t, "zone-1", pdbConfig.Data[drainingFailureDomainKey])
+	assert.Equal(t, "true", pdbConfig.Data[setNoOut])

 	// update the pgHealthCheckDuration time by -9 minutes
 	pdbConfig.Data[pgHealthCheckDurationKey] = time.Now().Add(time.Duration(-7) * time.Minute).Format(time.RFC3339)
 	// reconcile OSD PDB with no active drains and unhealthy PGs
-	err = r.reconcilePDBsForOSDs(clusterInfo, request, pdbConfig, "zone", []string{"zone-1", "zone-2"}, []string{})
+	_, err = r.reconcilePDBsForOSDs(clusterInfo, request, pdbConfig, "zone", []string{"zone-1", "zone-2"}, []string{}, true)
 	assert.NoError(t, err)
 	// assert that pdb config map was not reset as the PG health check was not timed out
 	assert.Equal(t, "zone-1", pdbConfig.Data[drainingFailureDomainKey])
+	assert.Equal(t, "true", pdbConfig.Data[setNoOut])

 	// update the drainingFailureDomain time by -9 minutes
 	pdbConfig.Data[pgHealthCheckDurationKey] = time.Now().Add(time.Duration(-11) * time.Minute).Format(time.RFC3339)
 	// reconcile OSD PDB with no active drains and unhealthy PGs
-	err = r.reconcilePDBsForOSDs(clusterInfo, request, pdbConfig, "zone", []string{"zone-1", "zone-2"}, []string{})
+	_, err = r.reconcilePDBsForOSDs(clusterInfo, request, pdbConfig, "zone", []string{"zone-1", "zone-2"}, []string{}, false)
 	assert.NoError(t, err)
 	// assert that pdb config map was reset as the PG health check was timed out
 	assert.Equal(t, "", pdbConfig.Data[drainingFailureDomainKey])
+	assert.Equal(t, "", pdbConfig.Data[setNoOut])
+}
+
+func TestHasNodeDrained(t *testing.T) {
+	osdPOD := fakeOSDPod(0, nodeName)
+	// Not expecting node drain because OSD pod is assigned to a schedulable node
+	r := getFakeReconciler(t, nodeObj, osdPOD.DeepCopy(), &corev1.ConfigMap{})
+	expected, err := hasOSDNodeDrained(r.client, namespace, "0")
+	assert.NoError(t, err)
+	assert.False(t, expected)
+
+	// Expecting node drain because OSD pod is assigned to an unschedulable node
+	r = getFakeReconciler(t, unschedulableNodeObj, osdPOD.DeepCopy(), &corev1.ConfigMap{})
+	expected, err = hasOSDNodeDrained(r.client, namespace, "0")
+	assert.NoError(t, err)
+	assert.True(t, expected)
+
+	// Expecting node drain because OSD pod is not assigned to any node
+	osdPodObj := osdPOD.DeepCopy()
+	osdPodObj.Spec.NodeName = ""
+	r = getFakeReconciler(t, nodeObj, osdPodObj, &corev1.ConfigMap{})
+	expected, err = hasOSDNodeDrained(r.client, namespace, "0")
+	assert.NoError(t, err)
+	assert.True(t, expected)
 }
@@ -169,8 +169,8 @@ func (r *ReconcileClusterDisruption) reconcile(request reconcile.Request) (recon
 		return reconcile.Result{}, nil
 	}

-	// get all failure domains and the draining failure domains list
-	allFailureDomains, drainingFailureDomains, err := r.getOSDFailureDomains(clusterInfo, request, poolFailureDomain)
+	// get a list of all the failure domains, failure domains with failed OSDs and failure domains with drained nodes
+	allFailureDomains, nodeDrainFailureDomains, osdDownFailureDomains, err := r.getOSDFailureDomains(clusterInfo, request, poolFailureDomain)
 	if err != nil {
 		return reconcile.Result{}, err
 	}
@@ -181,17 +181,8 @@ func (r *ReconcileClusterDisruption) reconcile(request reconcile.Request) (recon
 		return reconcile.Result{}, err
 	}

-	err = r.reconcilePDBsForOSDs(clusterInfo, request, pdbStateMap, poolFailureDomain, allFailureDomains, drainingFailureDomains)
-	if err != nil {
-		return reconcile.Result{}, err
-	}
-
-	disabledPDB, ok := pdbStateMap.Data[drainingFailureDomainKey]
-	if ok && len(disabledPDB) > 0 {
-		return reconcile.Result{Requeue: true, RequeueAfter: 30 * time.Second}, nil
-	}
-
-	return reconcile.Result{}, nil
+	activeNodeDrains := len(nodeDrainFailureDomains) > 0
+	return r.reconcilePDBsForOSDs(clusterInfo, request, pdbStateMap, poolFailureDomain, allFailureDomains, osdDownFailureDomains, activeNodeDrains)
 }

 // ClusterMap maintains the association between namespace and clusername