osd: allow one healthy OSD to be drained

One ore more OSDs can be down and PGs can still be active+clean (data
was rebalanced to other available OSDs). This PR sets
maxunavailable=1+downOSDs to allow one healthy to be drained.

Signed-off-by: Santosh Pillai <sapillai@redhat.com>
(cherry picked from commit a816b4a971)

Documentation/CRDs/Cluster/ceph-cluster-crd.md

@@ -102,9 +102,6 @@ For more details on the mons and when to choose a number other than `3`, see the
 * `disruptionManagement`: The section for configuring management of daemon disruptions
     * `managePodBudgets`: if `true`, the operator will create and manage PodDisruptionBudgets for OSD, Mon, RGW, and MDS daemons. OSD PDBs are managed dynamically via the strategy outlined in the [design](https://github.com/rook/rook/blob/master/design/ceph/ceph-managed-disruptionbudgets.md). The operator will block eviction of OSDs by default and unblock them safely when drains are detected.
     * `osdMaintenanceTimeout`: is a duration in minutes that determines how long an entire failureDomain like `region/zone/host` will be held in `noout` (in addition to the default DOWN/OUT interval) when it is draining. The default value is `30` minutes.
-    * `pgHealthCheckTimeout`: A duration in minutes that the operator will wait for the placement groups to become healthy (see `pgHealthyRegex`) after a drain was completed and OSDs came back up.
-    Operator will continue with the next drain if the timeout exceeds.
-    No values or `0` means that the operator will wait until the placement groups are healthy before unblocking the next drain.
     * `pgHealthyRegex`: The regular expression that is used to determine which PG states should be considered healthy.
     The default is `^(active\+clean|active\+clean\+scrubbing|active\+clean\+scrubbing\+deep)$`.
 * `removeOSDsIfOutAndSafeToRemove`: If `true` the operator will remove the OSDs that are down and whose data has been restored to other OSDs. In Ceph terms, the OSDs are `out` and `safe-to-destroy` when they are removed.

Documentation/CRDs/specification.md

@@ -5766,10 +5766,7 @@ time.Duration
 </td>
 <td>
 <em>(Optional)</em>
-<p>PGHealthCheckTimeout is the time (in minutes) that the operator will wait for the placement groups to become
-healthy (active+clean) after a drain was completed and OSDs came back up. Rook will continue with the next drain
-if the timeout exceeds. It only works if managePodBudgets is true.
-No values or 0 means that the operator will wait until the placement groups are healthy before unblocking the next drain.</p>
+<p>DEPRECATED: PGHealthCheckTimeout is no longer implemented</p>
 </td>
 </tr>
 <tr>

deploy/charts/rook-ceph-cluster/values.yaml

@@ -390,10 +390,6 @@ cephClusterSpec:
     # A duration in minutes that determines how long an entire failureDomain like `region/zone/host` will be held in `noout` (in addition to the
     # default DOWN/OUT interval) when it is draining. This is only relevant when  `managePodBudgets` is `true`. The default value is `30` minutes.
     osdMaintenanceTimeout: 30
-    # A duration in minutes that the operator will wait for the placement groups to become healthy (active+clean) after a drain was completed and OSDs came back up.
-    # Operator will continue with the next drain if the timeout exceeds. It only works if `managePodBudgets` is `true`.
-    # No values or 0 means that the operator will wait until the placement groups are healthy before unblocking the next drain.
-    pgHealthCheckTimeout: 0
 
   # Configure the healthcheck and liveness probes for ceph pods.
   # Valid values for daemons are 'mon', 'osd', 'status'

deploy/charts/rook-ceph/templates/resources.yaml

@@ -1342,11 +1342,7 @@ spec:
                       format: int64
                       type: integer
                     pgHealthCheckTimeout:
-                      description: |-
-                        PGHealthCheckTimeout is the time (in minutes) that the operator will wait for the placement groups to become
-                        healthy (active+clean) after a drain was completed and OSDs came back up. Rook will continue with the next drain
-                        if the timeout exceeds. It only works if managePodBudgets is true.
-                        No values or 0 means that the operator will wait until the placement groups are healthy before unblocking the next drain.
+                      description: 'DEPRECATED: PGHealthCheckTimeout is no longer implemented'
                       format: int64
                       type: integer
                     pgHealthyRegex:

deploy/examples/cluster-on-local-pvc.yaml

@@ -257,4 +257,3 @@ spec:
   disruptionManagement:
     managePodBudgets: true
     osdMaintenanceTimeout: 30
-    pgHealthCheckTimeout: 0

deploy/examples/cluster-on-pvc-minikube.yaml

@@ -188,7 +188,6 @@ spec:
   disruptionManagement:
     managePodBudgets: true
     osdMaintenanceTimeout: 30
-    pgHealthCheckTimeout: 0
   cephConfig:
     global:
       mon_warn_on_pool_no_redundancy: "false"

deploy/examples/cluster-on-pvc.yaml

@@ -182,7 +182,6 @@ spec:
   disruptionManagement:
     managePodBudgets: true
     osdMaintenanceTimeout: 30
-    pgHealthCheckTimeout: 0
   # security oriented settings
   # security:
   # Settings to enable key rotation for KEK(Key Encryption Key).

deploy/examples/cluster.yaml

@@ -308,10 +308,6 @@ spec:
     # A duration in minutes that determines how long an entire failureDomain like `region/zone/host` will be held in `noout` (in addition to the
     # default DOWN/OUT interval) when it is draining. This is only relevant when  `managePodBudgets` is `true`. The default value is `30` minutes.
     osdMaintenanceTimeout: 30
-    # A duration in minutes that the operator will wait for the placement groups to become healthy (active+clean) after a drain was completed and OSDs came back up.
-    # Operator will continue with the next drain if the timeout exceeds. It only works if `managePodBudgets` is `true`.
-    # No values or 0 means that the operator will wait until the placement groups are healthy before unblocking the next drain.
-    pgHealthCheckTimeout: 0
 
   # csi defines CSI Driver settings applied per cluster.
   csi:

deploy/examples/crds.yaml

@@ -1340,11 +1340,7 @@ spec:
                       format: int64
                       type: integer
                     pgHealthCheckTimeout:
-                      description: |-
-                        PGHealthCheckTimeout is the time (in minutes) that the operator will wait for the placement groups to become
-                        healthy (active+clean) after a drain was completed and OSDs came back up. Rook will continue with the next drain
-                        if the timeout exceeds. It only works if managePodBudgets is true.
-                        No values or 0 means that the operator will wait until the placement groups are healthy before unblocking the next drain.
+                      description: 'DEPRECATED: PGHealthCheckTimeout is no longer implemented'
                       format: int64
                       type: integer
                     pgHealthyRegex:

design/ceph/ceph-managed-disruptionbudgets.md

@@ -16,13 +16,25 @@ OSDs do not fit under the single PodDisruptionBudget pattern. Ceph's ability to
 Even if an upgrade agent were only to drain one node at a time, Ceph would have to wait until there were no undersized PGs before moving on the next.
 
 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 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.
+#### Types of OSD PDBs:
+- `Default` PDB:
+    - Named as `rook-ceph-osd`
+    - It allows one healthy OSD to go down, by setting `maxUnavailable=1`, on any failure domain.
+    - Sometimes one or more OSDs can be down (disk failure, etc) without any node drain but PGs would still be `active+clean`. In that case, the `maxUnavailable` is set to `1+number of down OSDs`
+- `Blocking` PDBs
+    - Named as `rook-ceph-osd-<failureDomainType>-<FailureDomainName>`. For example: `rook-ceph-osd-zone-zone-a`
+    - These PDBs are created on the entire failure domain.
+    - `maxUnavailable` is set to 0 to prevent any OSD pod from draining.
 
-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.
+We begin with creating the default PodDisruptionBudget for all the OSDs. 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.
 
-Example scenario:
+Once the drained OSDs are back and all the pgs are active+clean, that is, the cluster is healed, the default PodDisruptionBudget is added back and the blocking ones are deleted.
+
+#### Detecting Node Drains:
+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. If a node on which the OSD is suppose to run, is `unscheduleable` then the operator considers that node to be draining.
+
+#### Example scenario:
 
 - Zone x
   - Node a
@@ -37,22 +49,31 @@ Example scenario:
     - osd.4
     - osd.5
 
-1. Rook Operator creates a single PDB that covers all OSDs with maxUnavailable=1.
-2. When Rook Operator sees an OSD go down (for example, osd.0 goes down):
-   - Create a PDB for each failure domain (zones y and z) with maxUnavailable=0 where the OSD did *not* go down.
-   - Delete the original PDB that covers all OSDs
+1. A default PDB `rook-ceph-osd`, with maxUnavailable=1, is created for all OSDs.
+2. User drains `Node a` for maintenance
+3. Operator notices that an OSD has gone down (for example, `osd.0` on `Node a` and `Zone x`):
+   - Creates a blocking PDBs `rook-ceph-osd-zone-zone-y` and `rook-ceph-osd-zone-zone-z`
+   - Deletes the default PDB that covers all OSDs
    - Now all remaining OSDs in zone x would be allowed to be drained
-3. When Rook sees the OSDs are back up and all PGs are clean
-   - Restore the PDB that covers all OSDs with maxUnavailable=1
-   - Delete the PDBs (in zone y and z) where maxUnavailable=0
+4. When `Node-a` is back, all of its OSDs are running and all PGs are `active+clean`:
+   - Restores the default PDB `rook-ceph-osd` (maxUnavailable=1)
+   - Deletes the blocking PDBs `rook-ceph-osd-zone-zone-y` and `rook-ceph-osd-zone-zone-z`
 
 An example of an operator that will attempt to do rolling upgrades of nodes is the Machine Config Operator in openshift. Based on what I have seen in
 [SIG cluster lifecycle](https://github.com/kubernetes/community/tree/master/sig-cluster-lifecycle), kubernetes deployments based on cluster-api approach will be
 a common way of deploying kubernetes. This will also work to mitigate manual drains from accidentally disrupting storage.
 
-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.
+#### Preventing unnecessary data migration:
+- If a node is down due to planned maintenance, we don't want the data of the drained OSDs to be migrated to other OSDs.
+- Operator adds `noout` flag to the failure domain on which the node was drained.
+- This `noout` is removed after `OSDMaintenanceTimeout` is elapsed. `OSDMaintenanceTimeout` defaults to 30 minutes but can be configured from the cephCluster CR.
+- `noout` is not added if the OSD is down but there is no node drain.
+
+#### OSDs down due to reasons other than node drain:
+- OSDs can be down due to various reasons other than a node drain event. For example, disk failure.
+- If the PGs are active+clean even after the OSDs are down, then the operator will update the `maxUnavailable` count to `1+number of down OSDs` in the main PDB.
+- This will allow other healthy OSDs to be drained.
 
 ### Mon, Mgr, MDS, RGW, RBDMirror
 

pkg/apis/ceph.rook.io/v1/types.go

@@ -2847,10 +2847,7 @@ type DisruptionManagementSpec struct {
 	// +optional
 	OSDMaintenanceTimeout time.Duration `json:"osdMaintenanceTimeout,omitempty"`
 
-	// PGHealthCheckTimeout is the time (in minutes) that the operator will wait for the placement groups to become
-	// healthy (active+clean) after a drain was completed and OSDs came back up. Rook will continue with the next drain
-	// if the timeout exceeds. It only works if managePodBudgets is true.
-	// No values or 0 means that the operator will wait until the placement groups are healthy before unblocking the next drain.
+	// DEPRECATED: PGHealthCheckTimeout is no longer implemented
 	// +optional
 	PGHealthCheckTimeout time.Duration `json:"pgHealthCheckTimeout,omitempty"`
 

pkg/daemon/ceph/client/osd.go

@@ -448,3 +448,22 @@ func SetPrimaryAffinity(context *clusterd.Context, clusterInfo *ClusterInfo, osd
 	logger.Infof("successfully applied osd.%d primary-affinity %q", osdID, affinity)
 	return nil
 }
+
+type OSDMetadata struct {
+	Id       int    `json:"id"`
+	HostName string `json:"hostname"`
+}
+
+// GetOSDMetadata returns the output of `ceph osd metadata`
+func GetOSDMetadata(context *clusterd.Context, clusterInfo *ClusterInfo) (*[]OSDMetadata, error) {
+	args := []string{"osd", "metadata"}
+	buf, err := NewCephCommand(context, clusterInfo, args).Run()
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get osd metadata")
+	}
+	var osdMetadata []OSDMetadata
+	if err := json.Unmarshal(buf, &osdMetadata); err != nil {
+		return nil, errors.Wrap(err, "failed to unmarshal osd metadata response")
+	}
+	return &osdMetadata, nil
+}

pkg/operator/ceph/cluster/osd/osd.go

@@ -611,7 +611,8 @@ func (c *Cluster) getPVCHostName(pvcName string) (string, error) {
 	return "", errors.Errorf("node selector not found on deployment for osd with pvc %q", pvcName)
 }
 
-func getOSDID(d *appsv1.Deployment) (int, error) {
+// GetOSDID returns OSD ID from the OSD deployment
+func GetOSDID(d *appsv1.Deployment) (int, error) {
 	osdID, err := strconv.Atoi(d.Labels[OsdIdLabelKey])
 	if err != nil {
 		// add a question to the user AFTER the error text to help them recover from user error
@@ -624,7 +625,7 @@ func (c *Cluster) getOSDInfo(d *appsv1.Deployment) (OSDInfo, error) {
 	container := d.Spec.Template.Spec.Containers[0]
 	var osd OSDInfo
 
-	osdID, err := getOSDID(d)
+	osdID, err := GetOSDID(d)
 	if err != nil {
 		return OSDInfo{}, err
 	}

pkg/operator/ceph/cluster/osd/update.go

@@ -236,7 +236,7 @@ func (c *Cluster) getOSDUpdateInfo(errs *provisionErrors) (*updateQueue, *existe
 	updateQueue := newUpdateQueueWithCapacity(len(deps.Items))
 	existenceList := newExistenceListWithCapacity(len(deps.Items))
 	for i := range deps.Items {
-		id, err := getOSDID(&deps.Items[i]) // avoid implicit memory aliasing by indexing
+		id, err := GetOSDID(&deps.Items[i]) // avoid implicit memory aliasing by indexing
 		if err != nil {
 			// add a question to the user AFTER the error text to help them recover from user error
 			errs.addError("%v. did a user create their own deployment with label %q?", selector, err)

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

@@ -80,8 +80,6 @@ func Add(mgr manager.Manager, context *controllerconfig.Context) error {
 		return err
 	}
 
-	// Only reconcile for PDB update event when allowed disruptions for the main OSD PDB is 0.
-	// This means that one of the OSD is down due to node drain or any other reason
 	pdbPredicate := predicate.Funcs{
 		CreateFunc: func(e event.CreateEvent) bool {
 			// Do not reconcile when PDB is created
@@ -92,7 +90,8 @@ func Add(mgr manager.Manager, context *controllerconfig.Context) error {
 			if !ok {
 				return false
 			}
-			return pdb.Name == osdPDBAppName && pdb.Status.DisruptionsAllowed == 0
+			// reconcile for the main PDB update event when first OSD goes down, that is,  when `DisruptionsAllowed` gets updated to 0.
+			return pdb.Name == osdPDBAppName && pdb.Spec.MaxUnavailable.IntVal == 1 && pdb.Status.DisruptionsAllowed == 0
 		},
 		DeleteFunc: func(e event.DeleteEvent) bool {
 			// Do not reconcile when PDB is deleted

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

@@ -46,7 +46,6 @@ const (
 	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
 	nooutFlag                 = "noout"
@@ -70,7 +69,7 @@ func (r *ReconcileClusterDisruption) deletePDB(pdb client.Object) error {
 
 // 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) createDefaultPDBforOSD(namespace string) error {
+func (r *ReconcileClusterDisruption) createDefaultPDBforOSD(namespace string, maxUnavailable int) error {
 	cephCluster, ok := r.clusterMap.GetCluster(namespace)
 	if !ok {
 		return errors.Errorf("failed to find the namespace %q in the clustermap", namespace)
@@ -83,10 +82,11 @@ func (r *ReconcileClusterDisruption) createDefaultPDBforOSD(namespace string) er
 	selector := &metav1.LabelSelector{
 		MatchLabels: map[string]string{k8sutil.AppAttr: osdPDBAppName},
 	}
+
 	pdb := &policyv1.PodDisruptionBudget{
 		ObjectMeta: objectMeta,
 		Spec: policyv1.PodDisruptionBudgetSpec{
-			MaxUnavailable: &intstr.IntOrString{IntVal: 1},
+			MaxUnavailable: &intstr.IntOrString{IntVal: int32(maxUnavailable)}, // nolint:gosec // G115 - no overflow is expected for maxUnavailable count
 			Selector:       selector,
 		},
 	}
@@ -96,15 +96,22 @@ func (r *ReconcileClusterDisruption) createDefaultPDBforOSD(namespace string) er
 		return errors.Wrapf(err, "failed to set owner reference to pdb %v", pdb)
 	}
 
-	err = r.client.Get(r.context.OpManagerContext, pdbRequest, &policyv1.PodDisruptionBudget{})
+	existingPDB := &policyv1.PodDisruptionBudget{}
+	err = r.client.Get(r.context.OpManagerContext, pdbRequest, existingPDB)
 	if err != nil {
 		if apierrors.IsNotFound(err) {
 			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)
+			logger.Infof("creating the default pdb %q with maxUnavailable=%d for all osd", osdPDBAppName, maxUnavailable)
 			return r.createPDB(pdb)
 		}
 		return errors.Wrapf(err, "failed to get pdb %q", pdb.Name)
 	}
+
+	existingPDB.Spec = pdb.Spec
+	err = r.client.Update(r.context.OpManagerContext, existingPDB)
+	if err != nil {
+		return errors.Wrapf(err, "failed to update existing pdb %q", existingPDB.Name)
+	}
 	return nil
 }
 
@@ -201,7 +208,6 @@ func (r *ReconcileClusterDisruption) initializePDBState(request reconcile.Reques
 		Namespace: request.Namespace,
 	}
 	err := r.client.Get(r.context.OpManagerContext, pdbStateMapRequest, pdbStateMap)
-
 	if apierrors.IsNotFound(err) {
 		// create configmap to track the draining failure domain
 		pdbStateMap.Data = map[string]string{drainingFailureDomainKey: "", setNoOut: ""}
@@ -221,17 +227,11 @@ func (r *ReconcileClusterDisruption) reconcilePDBsForOSDs(
 	pdbStateMap *corev1.ConfigMap,
 	failureDomainType string,
 	allFailureDomains,
-	osdDownFailureDomains []string,
-	activeNodeDrains bool,
+	osdDownFailureDomains,
+	nodeDrainFailureDomains []string,
+	downOSDs []int,
 	pgHealthyRegex string,
 ) (reconcile.Result, error) {
-	var osdDown bool
-	var drainingFailureDomain string
-	if len(osdDownFailureDomains) > 0 {
-		osdDown = true
-		drainingFailureDomain = osdDownFailureDomains[0]
-	}
-
 	pgHealthMsg, pgClean, err := cephclient.IsClusterClean(r.context.ClusterdContext, clusterInfo, pgHealthyRegex)
 	if err != nil {
 		// If the error contains that message, this means the cluster is not up and running
@@ -244,53 +244,71 @@ func (r *ReconcileClusterDisruption) reconcilePDBsForOSDs(
 		return opcontroller.WaitForRequeueIfCephClusterNotReady, nil
 	}
 
-	switch {
-	// osd is down
-	case osdDown:
-		logger.Infof("osd is down in failure domain %q. 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)
-		}
-		if activeNodeDrains {
-			pdbStateMap.Data[setNoOut] = "true"
-		}
+	osdDown := len(downOSDs) > 0
+	maxUnavailableOSDCount := 1
 
-	// 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 OSD is down in the %q failure domains: %v. pg health: %q", failureDomainType, allFailureDomains, pgHealthMsg)
+	// switch block to update the PDB state config map based on the PG status and running OSDs
+	switch {
+	case !osdDown && pgClean:
+		logger.Infof("OSDs are up and PGs are clean. PG status: %q", pgHealthMsg)
+		maxUnavailableOSDCount = 1
 		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)
-	}
-
-	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)
+	case osdDown && pgClean:
+		logger.Infof("OSD(s) %v are down but PGs are clean. PG Status: %q", downOSDs, pgHealthMsg)
+		// In case of a node drain event, the OSD pods can get drained rapidly and it would take some time for rook to fetch
+		// the correct PG status. So wait for 60 seconds when OSD is down and node drain event is detected
+		if len(nodeDrainFailureDomains) > 0 {
+			lastNodeDrainTimeStamp, err := getLastNodeDrainTimeStamp(pdbStateMap, drainingFailureDomainDurationKey)
+			if err != nil {
+				return reconcile.Result{}, errors.Wrapf(err, "failed to get last node drain timestamp from the configmap %q", pdbStateMap.Name)
+			}
+			if time.Since(lastNodeDrainTimeStamp) < 60*time.Second {
+				logger.Infof("node drain is detected. Requeue to ensure that correct PG status is read.")
+				maxUnavailableOSDCount = 1
+			} else {
+				maxUnavailableOSDCount = len(downOSDs) + 1
+				resetPDBConfig(pdbStateMap)
+			}
+		} else {
+			maxUnavailableOSDCount = len(downOSDs) + 1
+			resetPDBConfig(pdbStateMap)
 		}
+
+	case osdDown && !pgClean:
+		setPDBConfig(pdbStateMap, osdDownFailureDomains, nodeDrainFailureDomains)
+		logger.Infof("OSD(s) %v are down and PGs are not clean. PGs Status: %q", downOSDs, pgHealthMsg)
+
+	// no-op. Wait for the PGs to become healthy from the previous node drain event
+	case !osdDown && !pgClean && len(pdbStateMap.Data[drainingFailureDomainKey]) > 1:
+		logger.Infof("OSDs are up but PGs are not clean from previous drain event. PGs Status: %q", pgHealthMsg)
 	}
 
+	// handle drains based on the PDB config map
 	if pdbStateMap.Data[drainingFailureDomainKey] != "" {
+		logger.Infof("OSD failure Domains : %q", allFailureDomains)
+		logger.Infof("Draining Failure Domain: %q", pdbStateMap.Data[drainingFailureDomainKey])
+		logger.Infof("Set noout on draining Failure Domain: %q", pdbStateMap.Data[setNoOut])
 		// delete default OSD pdb and create blocking OSD pdbs
-		err := r.handleActiveDrains(allFailureDomains, pdbStateMap.Data[drainingFailureDomainKey], failureDomainType, clusterInfo.Namespace, pgClean)
+		err := r.handleActiveDrains(allFailureDomains, pdbStateMap.Data[drainingFailureDomainKey], failureDomainType, clusterInfo.Namespace)
 		if err != nil {
 			return reconcile.Result{}, errors.Wrap(err, "failed to handle active drains")
 		}
 	} else if pdbStateMap.Data[drainingFailureDomainKey] == "" {
+		logger.Infof("`maxUnavailable` for the main OSD PDB is set to %d", maxUnavailableOSDCount)
 		// delete all blocking OSD pdb and restore the default OSD pdb
-		err := r.handleInactiveDrains(allFailureDomains, failureDomainType, clusterInfo.Namespace)
+		err = r.handleInactiveDrains(allFailureDomains, failureDomainType, clusterInfo.Namespace, maxUnavailableOSDCount)
 		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.updateNoout(clusterInfo, pdbStateMap, allFailureDomains)
+	if err != nil {
+		logger.Errorf("failed to update maintenance noout in cluster %q. %v", request, err)
+	}
+
+	// update PDB configmap
 	err = r.client.Update(clusterInfo.Context, pdbStateMap)
 	if err != nil {
 		if errors.Is(err, context.Canceled) {
@@ -299,31 +317,11 @@ func (r *ReconcileClusterDisruption) reconcilePDBsForOSDs(
 		return reconcile.Result{}, errors.Wrapf(err, "failed to update configMap %q in cluster %q", pdbStateMapName, request)
 	}
 
-	// requeue if drain is still in progress
-	if len(pdbStateMap.Data[drainingFailureDomainKey]) > 0 {
-		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
+	return r.requeuePDBController(request)
 }
 
 func (r *ReconcileClusterDisruption) handleActiveDrains(allFailureDomains []string, drainingFailureDomain,
-	failureDomainType, namespace string, isClean bool) error {
+	failureDomainType, namespace string) error {
 
 	for _, failureDomainName := range allFailureDomains {
 		// create blocking PDB for failure domains not currently draining
@@ -333,11 +331,9 @@ func (r *ReconcileClusterDisruption) handleActiveDrains(allFailureDomains []stri
 				return errors.Wrapf(err, "failed to create blocking pdb for %q failure domain %q", failureDomainType, failureDomainName)
 			}
 		} else {
-			if isClean {
-				err := r.deleteBlockingPDBForOSD(namespace, failureDomainType, failureDomainName)
-				if err != nil {
-					return errors.Wrapf(err, "failed to delete pdb for %q failure domain %q. %v", failureDomainType, failureDomainName, err)
-				}
+			err := r.deleteBlockingPDBForOSD(namespace, failureDomainType, failureDomainName)
+			if err != nil {
+				return errors.Wrapf(err, "failed to delete blocking pdb for %q failure domain %q. %v", failureDomainType, failureDomainName, err)
 			}
 		}
 	}
@@ -352,8 +348,8 @@ func (r *ReconcileClusterDisruption) handleActiveDrains(allFailureDomains []stri
 	return nil
 }
 
-func (r *ReconcileClusterDisruption) handleInactiveDrains(allFailureDomains []string, failureDomainType, namespace string) error {
-	err := r.createDefaultPDBforOSD(namespace)
+func (r *ReconcileClusterDisruption) handleInactiveDrains(allFailureDomains []string, failureDomainType, namespace string, maxUnavailable int) error {
+	err := r.createDefaultPDBforOSD(namespace, maxUnavailable)
 	if err != nil {
 		return errors.Wrap(err, "failed to create default pdb")
 	}
@@ -368,38 +364,43 @@ func (r *ReconcileClusterDisruption) handleInactiveDrains(allFailureDomains []st
 }
 
 func (r *ReconcileClusterDisruption) updateNoout(clusterInfo *cephclient.ClusterInfo, pdbStateMap *corev1.ConfigMap, allFailureDomains []string) error {
-	drainingFailureDomain := pdbStateMap.Data[drainingFailureDomainKey]
 	osdDump, err := cephclient.GetOSDDump(r.context.ClusterdContext, clusterInfo)
 	if err != nil {
 		return errors.Wrapf(err, "failed to get osddump for reconciling maintenance noout in namespace %s", clusterInfo.Namespace)
 	}
 	for _, failureDomainName := range allFailureDomains {
 		drainingFailureDomainTimeStampKey := fmt.Sprintf("%s-noout-last-set-at", failureDomainName)
-		if drainingFailureDomain == failureDomainName {
-
-			// get the time stamp
-			nooutSetTimeString, ok := pdbStateMap.Data[drainingFailureDomainTimeStampKey]
-			if !ok || len(nooutSetTimeString) == 0 {
-				// initialize it if it's not set
-				pdbStateMap.Data[drainingFailureDomainTimeStampKey] = time.Now().Format(time.RFC3339)
-			}
-			// parse the timestamp
-			nooutSetTime, err := time.Parse(time.RFC3339, pdbStateMap.Data[drainingFailureDomainTimeStampKey])
-			if err != nil {
-				return errors.Wrapf(err, "failed to parse timestamp %s for failureDomain %s", pdbStateMap.Data[drainingFailureDomainTimeStampKey], nooutSetTime)
-			}
-			if time.Since(nooutSetTime) >= r.maintenanceTimeout {
-				// noout expired
-				if _, err := osdDump.UpdateFlagOnCrushUnit(r.context.ClusterdContext, clusterInfo, false, failureDomainName, nooutFlag); err != nil {
-					return errors.Wrapf(err, "failed to update flag on crush unit when noout expired.")
+		if pdbStateMap.Data[drainingFailureDomainKey] == failureDomainName {
+			if pdbStateMap.Data[setNoOut] == "true" {
+				// get the time stamp
+				nooutSetTimeString, ok := pdbStateMap.Data[drainingFailureDomainTimeStampKey]
+				if !ok || len(nooutSetTimeString) == 0 {
+					// initialize it if it's not set
+					pdbStateMap.Data[drainingFailureDomainTimeStampKey] = time.Now().Format(time.RFC3339)
+				}
+				// parse the timestamp
+				nooutSetTime, err := time.Parse(time.RFC3339, pdbStateMap.Data[drainingFailureDomainTimeStampKey])
+				if err != nil {
+					return errors.Wrapf(err, "failed to parse timestamp %s for failureDomain %s", pdbStateMap.Data[drainingFailureDomainTimeStampKey], nooutSetTime)
+				}
+				if time.Since(nooutSetTime) >= r.maintenanceTimeout {
+					// noout expired
+					if _, err := osdDump.UpdateFlagOnCrushUnit(r.context.ClusterdContext, clusterInfo, false, failureDomainName, nooutFlag); err != nil {
+						return errors.Wrapf(err, "failed to update flag on crush unit when noout expired.")
+					}
+				} else {
+					// set noout
+					if _, err := osdDump.UpdateFlagOnCrushUnit(r.context.ClusterdContext, clusterInfo, true, failureDomainName, nooutFlag); err != nil {
+						return errors.Wrapf(err, "failed to update flag on crush unit while setting noout.")
+					}
 				}
 			} else {
-				// set noout
-				if _, err := osdDump.UpdateFlagOnCrushUnit(r.context.ClusterdContext, clusterInfo, true, failureDomainName, nooutFlag); err != nil {
-					return errors.Wrapf(err, "failed to update flag on crush unit while setting noout.")
+				if _, err := osdDump.UpdateFlagOnCrushUnit(r.context.ClusterdContext, clusterInfo, false, failureDomainName, nooutFlag); err != nil {
+					return errors.Wrapf(err, "failed to update flag on crush unit when ensuring noout is unset.")
 				}
+				// delete the timestamp
+				delete(pdbStateMap.Data, drainingFailureDomainTimeStampKey)
 			}
-
 		} else {
 			// ensure noout unset
 			if _, err := osdDump.UpdateFlagOnCrushUnit(r.context.ClusterdContext, clusterInfo, false, failureDomainName, nooutFlag); err != nil {
@@ -412,24 +413,30 @@ func (r *ReconcileClusterDisruption) updateNoout(clusterInfo *cephclient.Cluster
 	return nil
 }
 
-func (r *ReconcileClusterDisruption) getOSDFailureDomains(clusterInfo *cephclient.ClusterInfo, request reconcile.Request, poolFailureDomain string) ([]string, []string, []string, error) {
+func (r *ReconcileClusterDisruption) getOSDFailureDomains(clusterInfo *cephclient.ClusterInfo, request reconcile.Request, poolFailureDomain string) ([]string, []string, []string, []int, error) {
 	osdDeploymentList := &appsv1.DeploymentList{}
 	namespaceListOpts := client.InNamespace(request.Namespace)
 	topologyLocationLabel := fmt.Sprintf(osd.TopologyLocationLabel, poolFailureDomain)
 	err := r.client.List(clusterInfo.Context, osdDeploymentList, client.MatchingLabels{k8sutil.AppAttr: osd.AppName}, namespaceListOpts)
 	if err != nil {
-		return nil, nil, nil, errors.Wrap(err, "failed to list osd deployments")
+		return nil, nil, nil, nil, errors.Wrap(err, "failed to list osd deployments")
 	}
 
 	allFailureDomains := sets.New[string]()
 	nodeDrainFailureDomains := sets.New[string]()
 	osdDownFailureDomains := sets.New[string]()
+	downOSDs := []int{}
+
+	osdMetadata, err := cephclient.GetOSDMetadata(r.context.ClusterdContext, clusterInfo)
+	if err != nil {
+		return nil, nil, nil, nil, errors.Wrapf(err, "failed to get OSD status")
+	}
 
 	for _, deployment := range osdDeploymentList.Items {
-		labels := deployment.Spec.Template.ObjectMeta.GetLabels()
+		labels := deployment.GetLabels()
 		failureDomainName := labels[topologyLocationLabel]
 		if failureDomainName == "" {
-			return nil, nil, nil, errors.Errorf("failed to get the topology location label %q in OSD deployment %q",
+			return nil, nil, nil, nil, errors.Errorf("failed to get the topology location label %q in OSD deployment %q",
 				topologyLocationLabel, deployment.Name)
 		}
 
@@ -438,93 +445,58 @@ func (r *ReconcileClusterDisruption) getOSDFailureDomains(clusterInfo *cephclien
 			if !osdDownFailureDomains.Has(failureDomainName) {
 				osdDownFailureDomains.Insert(failureDomainName)
 			}
-			isDrained, err := hasOSDNodeDrained(clusterInfo.Context, r.client, request.Namespace, labels[osd.OsdIdLabelKey])
+
+			osdID, err := osd.GetOSDID(&deployment)
 			if err != nil {
-				return nil, nil, nil, errors.Wrapf(err, "failed to check if osd %q node is drained", deployment.Name)
+				return nil, nil, nil, nil, errors.Wrapf(err, "failed to get ID for the OSD deployment %q", 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)
+			downOSDs = append(downOSDs, osdID)
+
+			// check if OSD is down on unscheduleable node
+			var osdNodeName string
+			for _, metadata := range *osdMetadata {
+				if metadata.Id == osdID {
+					osdNodeName = metadata.HostName
+				}
+			}
+			if osdNodeName != "" {
+				isDrained, err := hasOSDNodeDrained(clusterInfo.Context, r.client, osdNodeName)
+				if err != nil {
+					return nil, 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 on node %q and a possible node drain is detected", deployment.Name, osdNodeName)
+					if !nodeDrainFailureDomains.Has(failureDomainName) {
+						nodeDrainFailureDomains.Insert(failureDomainName)
+					}
+				} else {
+					if !strings.HasSuffix(deployment.Name, "-debug") {
+						logger.Infof("osd %q is down on node %q but no node drain is detected", deployment.Name, osdNodeName)
+					}
 				}
 			} else {
-				if !strings.HasSuffix(deployment.Name, "-debug") {
-					logger.Infof("osd %q is down but no node drain is detected", deployment.Name)
-				}
+				logger.Warningf("failed to get the node name for the OSD %d", osdID)
+				continue
 			}
+
 		}
 
 		if !allFailureDomains.Has(failureDomainName) {
 			allFailureDomains.Insert(failureDomainName)
 		}
 	}
-	return sets.List(allFailureDomains), sets.List(nodeDrainFailureDomains), sets.List(osdDownFailureDomains), nil
-}
-
-func (r *ReconcileClusterDisruption) hasPGHealthCheckTimedout(pdbStateMap *corev1.ConfigMap) bool {
-	if r.pgHealthCheckTimeout == 0 {
-		logger.Debug("pg health check timeout is not set in the cluster. waiting for PGs to get active+clean")
-		return false
-	}
-
-	timeString, ok := pdbStateMap.Data[pgHealthCheckDurationKey]
-	if !ok || len(timeString) == 0 {
-		pdbStateMap.Data[pgHealthCheckDurationKey] = time.Now().Format(time.RFC3339)
-	} else {
-		pgHealthCheckDuration, err := time.Parse(time.RFC3339, timeString)
-		if err != nil {
-			logger.Errorf("failed to parse timestamp %v. %v", pgHealthCheckDuration, err)
-			pdbStateMap.Data[pgHealthCheckDurationKey] = time.Now().Format(time.RFC3339)
-			return false
-		}
-		timeElapsed := time.Since(pgHealthCheckDuration)
-		if timeElapsed >= r.pgHealthCheckTimeout {
-			logger.Info("timed out waiting for the PGs to become active+clean")
-			return true
-		}
-		timeleft := r.pgHealthCheckTimeout - timeElapsed
-		logger.Infof("waiting for %d minute(s) for PGs to become active+clean", int(timeleft.Minutes()))
-	}
-	return false
+	return sets.List(allFailureDomains), sets.List(nodeDrainFailureDomains), sets.List(osdDownFailureDomains), downOSDs, nil
 }
 
 // hasOSDNodeDrained returns true if OSD pod is not assigned to any node or if the OSD node is not schedulable
-func hasOSDNodeDrained(ctx context.Context, c client.Client, namespace, osdID string) (bool, error) {
-	osdNodeName, err := getOSDNodeName(ctx, 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
-	}
-
+func hasOSDNodeDrained(ctx context.Context, c client.Client, osdNodeName string) (bool, error) {
 	node, err := getNode(ctx, c, osdNodeName)
 	if err != nil {
-		return false, errors.Wrapf(err, "failed to get node assigned to OSD %q POD", osdID)
+		return false, errors.Wrapf(err, "failed to get node %q", osdNodeName)
 	}
 	return node.Spec.Unschedulable, nil
 }
 
-func getOSDNodeName(ctx context.Context, 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(ctx, 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(ctx context.Context, c client.Client, nodeName string) (*corev1.Node, error) {
 	node := &corev1.Node{}
 	err := c.Get(ctx, types.NamespacedName{Name: nodeName}, node)
@@ -538,18 +510,68 @@ func getPDBName(failureDomainType, failureDomainName string) string {
 	return k8sutil.TruncateNodeName(fmt.Sprintf("%s-%s-%s", osdPDBAppName, failureDomainType, "%s"), failureDomainName)
 }
 
-func (r *ReconcileClusterDisruption) getAllowedDisruptions(pdbName, namespace string) (int32, error) {
-	pdb := &policyv1.PodDisruptionBudget{}
-	err := r.client.Get(r.context.OpManagerContext, 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)
-	delete(pdbStateMap.Data, pgHealthCheckDurationKey)
+	// reset `set-no-out` flag on the configMap
+	pdbStateMap.Data[setNoOut] = ""
+}
+
+// setPDBConfig updates the OSD PDB config map. If there are unschedulable nodes (that is, a node drain event)
+// then those failureDomains are given higher precedence than the failureDomains where OSDs might be down
+// due to some reason but node is schedulable. `Noout` is set only if nodes are unscheduleable.
+func setPDBConfig(pdbStateMap *corev1.ConfigMap, osdDownFailureDomains, nodeDrainFailureDomains []string) {
+	if len(pdbStateMap.Data[drainingFailureDomainKey]) == 0 {
+		if len(nodeDrainFailureDomains) > 0 {
+			pdbStateMap.Data[drainingFailureDomainKey] = nodeDrainFailureDomains[0]
+			pdbStateMap.Data[setNoOut] = "true"
+		} else if len(osdDownFailureDomains) > 0 {
+			pdbStateMap.Data[drainingFailureDomainKey] = osdDownFailureDomains[0]
+			pdbStateMap.Data[setNoOut] = ""
+		}
+		pdbStateMap.Data[drainingFailureDomainDurationKey] = time.Now().Format(time.RFC3339)
+	}
+}
+
+// requeuePDBController returns requeue request with timeout if:
+// - allowedDisruption in main PDB is 0, that is, One or more OSD went down.
+// - MaxUnavailable in the main PDB is > 1, that is, OSDs are down but PGs might be clean.
+// - default OSD PDB is not available.
+func (r *ReconcileClusterDisruption) requeuePDBController(request reconcile.Request) (reconcile.Result, error) {
+	defaultPDB := &policyv1.PodDisruptionBudget{}
+	err := r.client.Get(r.context.OpManagerContext, types.NamespacedName{Name: osdPDBAppName, Namespace: request.Namespace}, defaultPDB)
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			logger.Debugf("default osd pdb %q not found", osdPDBAppName)
+			logger.Info("reconciling osd pdb controller")
+			return reconcile.Result{Requeue: true, RequeueAfter: 15 * time.Second}, nil
+		} else {
+			return reconcile.Result{}, errors.Wrapf(err, "failed to get allowed disruptions count from default osd pdb %q.", osdPDBAppName)
+		}
+	}
+
+	if defaultPDB.Status.DisruptionsAllowed == 0 || defaultPDB.Spec.MaxUnavailable.IntVal > 1 {
+		logger.Info("reconciling osd pdb controller")
+		return reconcile.Result{Requeue: true, RequeueAfter: 30 * time.Second}, nil
+	}
+
+	logger.Info("successfully reconciled OSD PDB controller")
+	return reconcile.Result{}, nil
+}
+
+func getLastNodeDrainTimeStamp(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 {
+		currentTimeStamp := time.Now()
+		pdbStateMap.Data[key] = currentTimeStamp.Format(time.RFC3339)
+		return currentTimeStamp, 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
 }

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

@@ -20,7 +20,6 @@ import (
 	"context"
 	"fmt"
 	"testing"
-	"time"
 
 	"github.com/pkg/errors"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
@@ -47,25 +46,19 @@ const (
 	healthyCephStatusRemapped = `{"fsid":"e32d91a2-24ff-4953-bc4a-6864d31dd2a0","health":{"status":"HEALTH_OK","checks":{},"mutes":[]},"election_epoch":3,"quorum":[0],"quorum_names":["a"],"quorum_age":1177701,"monmap":{"epoch":1,"min_mon_release_name":"reef","num_mons":1},"osdmap":{"epoch":1800,"num_osds":5,"num_up_osds":5,"osd_up_since":1699834324,"num_in_osds":5,"osd_in_since":1699834304,"num_remapped_pgs":11},"pgmap":{"pgs_by_state":[{"state_name":"active+clean","count":174},{"state_name":"active+remapped+backfilling","count":10},{"state_name":"active+clean+remapped","count":1}],"num_pgs":185,"num_pools":9,"num_objects":2383,"data_bytes":2222656224,"bytes_used":8793104384,"bytes_avail":18050441216,"bytes_total":26843545600,"misplaced_objects":139,"misplaced_total":7149,"misplaced_ratio":0.019443278780248985,"recovering_objects_per_sec":10,"recovering_bytes_per_sec":9739877,"recovering_keys_per_sec":0,"num_objects_recovered":62,"num_bytes_recovered":58471087,"num_keys_recovered":0,"write_bytes_sec":2982994,"read_op_per_sec":0,"write_op_per_sec":26},"fsmap":{"epoch":1,"by_rank":[],"up:standby":0},"mgrmap":{"available":true,"num_standbys":0,"modules":["iostat","nfs","prometheus","restful"],"services":{"prometheus":"http://10.244.0.36:9283/"}},"servicemap":{"epoch":1,"modified":"0.000000","services":{}},"progress_events":{}}`
 )
 
-var nodeName = "node01"
 var namespace = "rook-ceph"
 
 var cephCluster = &cephv1.CephCluster{
 	ObjectMeta: metav1.ObjectMeta{Name: "ceph-cluster"},
 }
 
-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 getNodeObject(name string, unschedulable bool) *corev1.Node {
+	return &corev1.Node{
+		ObjectMeta: metav1.ObjectMeta{Name: name},
+		Spec: corev1.NodeSpec{
+			Unschedulable: unschedulable,
+		},
+	}
 }
 
 func fakeOSDDeployment(id, readyReplicas int) appsv1.Deployment {
@@ -74,17 +67,9 @@ func fakeOSDDeployment(id, readyReplicas int) appsv1.Deployment {
 			Name:      fmt.Sprintf("osd-%d", id),
 			Namespace: namespace,
 			Labels: map[string]string{
-				"app": "rook-ceph-osd",
-			},
-		},
-		Spec: appsv1.DeploymentSpec{
-			Template: corev1.PodTemplateSpec{
-				ObjectMeta: metav1.ObjectMeta{
-					Labels: map[string]string{
-						"topology-location-zone": fmt.Sprintf("zone-%d", id),
-						"ceph-osd-id":            fmt.Sprintf("%d", id),
-					},
-				},
+				"app":                    "rook-ceph-osd",
+				"topology-location-zone": fmt.Sprintf("zone-%d", id),
+				"ceph-osd-id":            fmt.Sprintf("%d", id),
 			},
 		},
 		Status: appsv1.DeploymentStatus{
@@ -94,23 +79,6 @@ 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},
@@ -147,66 +115,62 @@ func TestGetOSDFailureDomains(t *testing.T) {
 	testcases := []struct {
 		name                           string
 		osds                           []appsv1.Deployment
-		osdPods                        []corev1.Pod
-		node                           *corev1.Node
+		osdMetadata                    string
+		nodes                          []corev1.Node
 		expectedAllFailureDomains      []string
 		expectedDrainingFailureDomains []string
 		expectedOsdDownFailureDomains  []string
+		expectedDownOSDs               []int
 	}{
 		{
 			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,
+			nodes:                          []corev1.Node{*getNodeObject("node-1", false), *getNodeObject("node-2", false), *getNodeObject("node-3", false)},
 			expectedAllFailureDomains:      []string{"zone-1", "zone-2", "zone-3"},
 			expectedOsdDownFailureDomains:  []string{},
 			expectedDrainingFailureDomains: []string{},
+			expectedDownOSDs:               []int{},
 		},
 		{
 			name: "case 2: osd in zone-1 is pending and node is unschedulable",
 			osds: []appsv1.Deployment{fakeOSDDeployment(1, 0), fakeOSDDeployment(2, 1),
 				fakeOSDDeployment(3, 1)},
-			osdPods: []corev1.Pod{fakeOSDPod(1, ""), fakeOSDPod(2, nodeName),
-				fakeOSDPod(3, nodeName)},
-			node:                           nodeObj,
+			nodes:                          []corev1.Node{*getNodeObject("node-1", true), *getNodeObject("node-2", false), *getNodeObject("node-3", false)},
 			expectedAllFailureDomains:      []string{"zone-1", "zone-2", "zone-3"},
 			expectedOsdDownFailureDomains:  []string{"zone-1"},
 			expectedDrainingFailureDomains: []string{"zone-1"},
+			expectedDownOSDs:               []int{1},
 		},
 		{
 			name: "case 3: osd in zone-1 and zone-2 are pending and node is unschedulable",
 			osds: []appsv1.Deployment{fakeOSDDeployment(1, 0), fakeOSDDeployment(2, 0),
 				fakeOSDDeployment(3, 1)},
-			osdPods: []corev1.Pod{fakeOSDPod(1, ""), fakeOSDPod(2, ""),
-				fakeOSDPod(3, nodeName)},
-			node:                           nodeObj,
+			nodes:                          []corev1.Node{*getNodeObject("node-1", true), *getNodeObject("node-2", true), *getNodeObject("node-3", false)},
 			expectedAllFailureDomains:      []string{"zone-1", "zone-2", "zone-3"},
 			expectedOsdDownFailureDomains:  []string{"zone-1", "zone-2"},
 			expectedDrainingFailureDomains: []string{"zone-1", "zone-2"},
+			expectedDownOSDs:               []int{1, 2},
 		},
 		{
 			name: "case 4: osd in zone-1 is pending but osd node is schedulable",
 			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,
+			nodes:                          []corev1.Node{*getNodeObject("node-1", false), *getNodeObject("node-2", false), *getNodeObject("node-3", false)},
 			expectedAllFailureDomains:      []string{"zone-1", "zone-2", "zone-3"},
 			expectedOsdDownFailureDomains:  []string{"zone-1"},
 			expectedDrainingFailureDomains: []string{},
+			expectedDownOSDs:               []int{1},
 		},
 		{
-			name: "case 5: osd in zone-1 is pending but osd node is not schedulable",
-			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,
+			name: "case 5: osd in zone-3 is pending and the osd node is not schedulable",
+			osds: []appsv1.Deployment{fakeOSDDeployment(1, 1), fakeOSDDeployment(2, 1),
+				fakeOSDDeployment(3, 0)},
+			nodes:                          []corev1.Node{*getNodeObject("node-1", false), *getNodeObject("node-2", false), *getNodeObject("node-3", true)},
 			expectedAllFailureDomains:      []string{"zone-1", "zone-2", "zone-3"},
-			expectedOsdDownFailureDomains:  []string{"zone-1"},
-			expectedDrainingFailureDomains: []string{"zone-1"},
+			expectedOsdDownFailureDomains:  []string{"zone-3"},
+			expectedDrainingFailureDomains: []string{"zone-3"},
+			expectedDownOSDs:               []int{3},
 		},
 	}
 
@@ -215,23 +179,34 @@ func TestGetOSDFailureDomains(t *testing.T) {
 			objs := []runtime.Object{
 				cephCluster,
 				&corev1.ConfigMap{},
-				tc.node,
+			}
+			for _, node := range tc.nodes {
+				objs = append(objs, node.DeepCopy())
 			}
 			for _, osdDeployment := range tc.osds {
 				objs = append(objs, osdDeployment.DeepCopy())
 			}
-			for _, osdPod := range tc.osdPods {
-				objs = append(objs, osdPod.DeepCopy())
+
+			executor := &exectest.MockExecutor{}
+			executor.MockExecuteCommandWithOutput = func(command string, args ...string) (string, error) {
+				logger.Infof("Command: %s %v", command, args)
+				if args[0] == "osd" && args[1] == "metadata" {
+					return `[{"id": 1, "hostname": "node-1"}, {"id": 2, "hostname": "node-2"}, {"id": 3, "hostname": "node-3"}]`, nil
+				}
+				return "", errors.Errorf("unexpected ceph command '%v'", args)
 			}
+
 			r := getFakeReconciler(t, objs...)
 			clusterInfo := getFakeClusterInfo()
 			clusterInfo.Context = context.TODO()
+			r.context = &controllerconfig.Context{ClusterdContext: &clusterd.Context{Executor: executor}}
 			request := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}}
-			allfailureDomains, nodeDrainFailureDomains, osdDownFailureDomains, err := r.getOSDFailureDomains(clusterInfo, request, "zone")
+			allfailureDomains, nodeDrainFailureDomains, osdDownFailureDomains, downOSDs, 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)
+			assert.Equal(t, tc.expectedDownOSDs, downOSDs)
 		})
 	}
 }
@@ -243,6 +218,7 @@ func TestGetOSDFailureDomainsError(t *testing.T) {
 		expectedAllFailureDomains      []string
 		expectedDrainingFailureDomains []string
 		expectedOsdDownFailureDomains  []string
+		expectedDownOSDs               []int
 	}{
 		{
 			name: "case 1: one or more OSD deployment is missing crush location label",
@@ -251,23 +227,34 @@ func TestGetOSDFailureDomainsError(t *testing.T) {
 			expectedAllFailureDomains:      nil,
 			expectedDrainingFailureDomains: nil,
 			expectedOsdDownFailureDomains:  nil,
+			expectedDownOSDs:               nil,
 		},
 	}
 
 	for _, tc := range testcases {
 		t.Run(tc.name, func(t *testing.T) {
+			executor := &exectest.MockExecutor{}
+			executor.MockExecuteCommandWithOutput = func(command string, args ...string) (string, error) {
+				logger.Infof("Command: %s %v", command, args)
+				if args[0] == "osd" && args[1] == "metadata" {
+					return `[{"id": 1, "hostname": "node-1"}, {"id": 2, "hostname": "node-2"}, {"id": 3, "hostname": "node-3"}]`, nil
+				}
+				return "", errors.Errorf("unexpected ceph command '%v'", args)
+			}
 			osd := tc.osds[0].DeepCopy()
-			osd.Spec.Template.ObjectMeta.Labels["topology-location-zone"] = ""
+			osd.Labels["topology-location-zone"] = ""
 			r := getFakeReconciler(t, cephCluster, &corev1.ConfigMap{},
 				tc.osds[1].DeepCopy(), tc.osds[2].DeepCopy(), osd)
+			r.context = &controllerconfig.Context{ClusterdContext: &clusterd.Context{Executor: executor}}
 			clusterInfo := getFakeClusterInfo()
 			clusterInfo.Context = context.TODO()
 			request := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}}
-			allfailureDomains, nodeDrainFailureDomains, osdDownFailureDomains, err := r.getOSDFailureDomains(clusterInfo, request, "zone")
+			allfailureDomains, nodeDrainFailureDomains, osdDownFailureDomains, downOSDs, 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)
+			assert.Equal(t, tc.expectedDownOSDs, downOSDs)
 		})
 	}
 }
@@ -276,11 +263,11 @@ func TestReconcilePDBForOSD(t *testing.T) {
 	testcases := []struct {
 		name                              string
 		fakeCephStatus                    string
-		fakeOSDDump                       string
 		configMap                         *corev1.ConfigMap
 		allFailureDomains                 []string
 		osdDownFailureDomains             []string
-		activeNodeDrains                  bool
+		activeNodeDrains                  []string
+		downOSDs                          []int
 		pgHealthyRegex                    string
 		expectedSetNoOutValue             string
 		expectedOSDPDBCount               int
@@ -288,13 +275,13 @@ func TestReconcilePDBForOSD(t *testing.T) {
 		expectedDrainingFailureDomainName string
 	}{
 		{
-			name:                              "case 1: no draining failure domain and all pgs are healthy",
+			name:                              "case 1: No draining failure domains, all OSDs are up/in and all pgs are healthy",
 			fakeCephStatus:                    healthyCephStatus,
-			fakeOSDDump:                       `{"OSDs": [{"OSD": 3, "Up": 3, "In": 3}]}`,
 			allFailureDomains:                 []string{"zone-1", "zone-2", "zone-3"},
 			osdDownFailureDomains:             []string{},
+			downOSDs:                          []int{},
+			activeNodeDrains:                  []string{},
 			configMap:                         fakePDBConfigMap(""),
-			activeNodeDrains:                  false,
 			pgHealthyRegex:                    "",
 			expectedSetNoOutValue:             "",
 			expectedOSDPDBCount:               1,
@@ -304,11 +291,11 @@ func TestReconcilePDBForOSD(t *testing.T) {
 		{
 			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"},
 			osdDownFailureDomains:             []string{"zone-1"},
+			downOSDs:                          []int{1},
 			configMap:                         fakePDBConfigMap(""),
-			activeNodeDrains:                  true,
+			activeNodeDrains:                  []string{"zone-1"},
 			pgHealthyRegex:                    "",
 			expectedSetNoOutValue:             "true",
 			expectedOSDPDBCount:               2,
@@ -318,11 +305,11 @@ func TestReconcilePDBForOSD(t *testing.T) {
 		{
 			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"},
 			osdDownFailureDomains:             []string{},
+			downOSDs:                          []int{},
 			configMap:                         fakePDBConfigMap("zone-1"),
-			activeNodeDrains:                  true,
+			activeNodeDrains:                  []string{},
 			pgHealthyRegex:                    "",
 			expectedSetNoOutValue:             "",
 			expectedOSDPDBCount:               2,
@@ -332,11 +319,11 @@ func TestReconcilePDBForOSD(t *testing.T) {
 		{
 			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"},
 			osdDownFailureDomains:             []string{},
+			downOSDs:                          []int{},
 			configMap:                         fakePDBConfigMap("zone-1"),
-			activeNodeDrains:                  true,
+			activeNodeDrains:                  []string{},
 			pgHealthyRegex:                    "",
 			expectedSetNoOutValue:             "",
 			expectedOSDPDBCount:               1,
@@ -346,17 +333,32 @@ func TestReconcilePDBForOSD(t *testing.T) {
 		{
 			name:                              "case 5: remapped pgs are regarded as clean if pgHealthyRegex allows it",
 			fakeCephStatus:                    healthyCephStatusRemapped,
-			fakeOSDDump:                       `{"OSDs": [{"OSD": 3, "Up": 3, "In": 3}]}`,
 			allFailureDomains:                 []string{"zone-1", "zone-2", "zone-3"},
 			osdDownFailureDomains:             []string{},
+			downOSDs:                          []int{},
 			configMap:                         fakePDBConfigMap("zone-1"),
-			activeNodeDrains:                  true,
+			activeNodeDrains:                  []string{},
 			pgHealthyRegex:                    `^(active\+clean|active\+clean\+scrubbing|active\+clean\+scrubbing\+deep|active\+clean\+remapped|active\+remapped\+backfilling)$`,
 			expectedSetNoOutValue:             "",
 			expectedOSDPDBCount:               1,
 			expectedMaxUnavailableCount:       1,
 			expectedDrainingFailureDomainName: "",
 		},
+		{
+			name:                  "case 6: Cluster is healthy but OSDs are down. MaxUnavailable should be set to 1 + number of down OSDs",
+			fakeCephStatus:        healthyCephStatus,
+			allFailureDomains:     []string{"zone-1", "zone-2", "zone-3"},
+			osdDownFailureDomains: []string{},
+			// OSD 0 and 1 are down but ceph health is good. So maxUnavailable should be set to 3.
+			downOSDs:                          []int{0, 1},
+			configMap:                         fakePDBConfigMap("zone-1"),
+			activeNodeDrains:                  []string{},
+			pgHealthyRegex:                    "",
+			expectedSetNoOutValue:             "",
+			expectedOSDPDBCount:               1,
+			expectedMaxUnavailableCount:       3,
+			expectedDrainingFailureDomainName: "",
+		},
 	}
 
 	for _, tc := range testcases {
@@ -372,7 +374,7 @@ func TestReconcilePDBForOSD(t *testing.T) {
 					return tc.fakeCephStatus, nil
 				}
 				if args[0] == "osd" && args[1] == "dump" {
-					return tc.fakeOSDDump, nil
+					return `{"OSDs": [{"OSD": 3, "Up": 3, "In": 3}]}`, nil
 				}
 				return "", errors.Errorf("unexpected ceph command '%v'", args)
 			}
@@ -381,7 +383,8 @@ func TestReconcilePDBForOSD(t *testing.T) {
 			// check for PDBV1 version
 			test.SetFakeKubernetesVersion(clientset, "v1.21.0")
 			r.context = &controllerconfig.Context{ClusterdContext: &clusterd.Context{Executor: executor, Clientset: clientset}}
-			_, err := r.reconcilePDBsForOSDs(clusterInfo, request, tc.configMap, "zone", tc.allFailureDomains, tc.osdDownFailureDomains, tc.activeNodeDrains, tc.pgHealthyRegex)
+
+			_, err := r.reconcilePDBsForOSDs(clusterInfo, request, tc.configMap, "zone", tc.allFailureDomains, tc.osdDownFailureDomains, tc.activeNodeDrains, tc.downOSDs, tc.pgHealthyRegex)
 			assert.NoError(t, err)
 
 			// assert that pdb for osd are created correctly
@@ -399,105 +402,75 @@ func TestReconcilePDBForOSD(t *testing.T) {
 			assert.Equal(t, tc.expectedDrainingFailureDomainName, existingConfigMaps.Items[0].Data[drainingFailureDomainKey])
 			assert.Equal(t, tc.expectedSetNoOutValue, existingConfigMaps.Items[0].Data[setNoOut])
 		})
-
 	}
 }
 
-func TestPGHealthcheckTimeout(t *testing.T) {
-	pdbConfig := fakePDBConfigMap("")
-	r := getFakeReconciler(t, cephCluster, pdbConfig)
-	clusterInfo := getFakeClusterInfo()
-	clusterInfo.Context = context.TODO()
-	request := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}}
-	executor := &exectest.MockExecutor{}
-	executor.MockExecuteCommandWithOutput = func(command string, args ...string) (string, error) {
-		logger.Infof("Command: %s %v", command, args)
-		if args[0] == "status" {
-			return unHealthyCephStatus, nil
-		}
-		if args[0] == "osd" && args[1] == "dump" {
-			return `{"OSDs": [{"OSD": 3, "Up": 3, "In": 3}]}`, nil
-		}
-		return "", errors.Errorf("unexpected ceph command '%v'", args)
-	}
-	clientset := test.New(t, 3)
-	r.context = &controllerconfig.Context{ClusterdContext: &clusterd.Context{Executor: executor, Clientset: clientset}}
-	// set PG health check timeout to 10 minutes
-	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"}, 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{}, 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{}, 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)
+	osdDeployment := fakeOSDDeployment(1, 1)
 	ctx := context.TODO()
 	// Not expecting node drain because OSD pod is assigned to a schedulable node
-	r := getFakeReconciler(t, nodeObj, osdPOD.DeepCopy(), &corev1.ConfigMap{})
-	expected, err := hasOSDNodeDrained(ctx, r.client, namespace, "0")
+	r := getFakeReconciler(t, getNodeObject("node-1", false), osdDeployment.DeepCopy(), &corev1.ConfigMap{})
+	expected, err := hasOSDNodeDrained(ctx, r.client, "node-1")
 	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(ctx, 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(ctx, r.client, namespace, "0")
+	osdDeployment = fakeOSDDeployment(2, 2)
+	r = getFakeReconciler(t, getNodeObject("node-2", true), osdDeployment.DeepCopy(), &corev1.ConfigMap{})
+	expected, err = hasOSDNodeDrained(ctx, r.client, "node-2")
 	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,
+func TestSetPDBConfig(t *testing.T) {
+	testcases := []struct {
+		name                          string
+		pdbConfig                     *corev1.ConfigMap
+		osdDownFailureDomains         []string
+		drainingFailureDomains        []string
+		expectedFailureDomainKeyValue string
+		expecteNoOutSetting           string
+	}{
+		{
+			name:                          "case 1: Empty PDB config and no draining failureDomain",
+			pdbConfig:                     fakePDBConfigMap(""),
+			osdDownFailureDomains:         []string{"zone-1", "zone-2"},
+			drainingFailureDomains:        []string{},
+			expectedFailureDomainKeyValue: "zone-1",
+			expecteNoOutSetting:           "",
 		},
-		Status: policyv1.PodDisruptionBudgetStatus{
-			DisruptionsAllowed: int32(0),
+		{
+			name:                          "case 2: Non-empty PDB config and no draining failureDomain",
+			pdbConfig:                     fakePDBConfigMap("zone-2"),
+			osdDownFailureDomains:         []string{"zone-1", "zone-2"},
+			drainingFailureDomains:        []string{},
+			expectedFailureDomainKeyValue: "zone-2",
+			expecteNoOutSetting:           "",
+		},
+		{
+			name:                          "case 3: Node drain event should set the no-out flag on the failure domain",
+			pdbConfig:                     fakePDBConfigMap(""),
+			osdDownFailureDomains:         []string{"zone-1", "zone-2"},
+			drainingFailureDomains:        []string{"zone-1"},
+			expectedFailureDomainKeyValue: "zone-1",
+			expecteNoOutSetting:           "true",
+		},
+		{
+			name:                          "case 3: failure domain with drained nodes should get higher precedence",
+			pdbConfig:                     fakePDBConfigMap(""),
+			osdDownFailureDomains:         []string{"zone-1", "zone-2", "zone-3"},
+			drainingFailureDomains:        []string{"zone-3"},
+			expectedFailureDomainKeyValue: "zone-3",
+			expecteNoOutSetting:           "true",
 		},
 	}
-	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)
+
+	for _, tc := range testcases {
+		t.Run(tc.name, func(t *testing.T) {
+			setPDBConfig(tc.pdbConfig, tc.osdDownFailureDomains, tc.drainingFailureDomains)
+			assert.Equal(t, tc.expectedFailureDomainKeyValue, tc.pdbConfig.Data[drainingFailureDomainKey])
+			assert.Equal(t, tc.expecteNoOutSetting, tc.pdbConfig.Data[setNoOut])
+		})
+	}
+
 }

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

@@ -56,12 +56,11 @@ var (
 // ReconcileClusterDisruption reconciles ReplicaSets
 type ReconcileClusterDisruption struct {
 	// client can be used to retrieve objects from the APIServer.
-	scheme               *runtime.Scheme
-	client               client.Client
-	context              *controllerconfig.Context
-	clusterMap           *ClusterMap
-	maintenanceTimeout   time.Duration
-	pgHealthCheckTimeout time.Duration
+	scheme             *runtime.Scheme
+	client             client.Client
+	context            *controllerconfig.Context
+	clusterMap         *ClusterMap
+	maintenanceTimeout time.Duration
 }
 
 // Reconcile reconciles a node and ensures that it has a drain-detection deployment
@@ -135,8 +134,6 @@ func (r *ReconcileClusterDisruption) reconcile(request reconcile.Request) (recon
 		logger.Debugf("Using default maintenance timeout: %v", r.maintenanceTimeout)
 	}
 
-	r.pgHealthCheckTimeout = cephCluster.Spec.DisruptionManagement.PGHealthCheckTimeout * time.Minute
-
 	//  reconcile the pools and get the failure domain
 	cephObjectStoreList, cephFilesystemList, poolFailureDomain, poolCount, err := r.processPools(request)
 	if err != nil {
@@ -161,7 +158,7 @@ func (r *ReconcileClusterDisruption) reconcile(request reconcile.Request) (recon
 	}
 
 	// 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)
+	allFailureDomains, nodeDrainFailureDomains, osdDownFailureDomains, downOSDs, err := r.getOSDFailureDomains(clusterInfo, request, poolFailureDomain)
 	if err != nil {
 		return reconcile.Result{}, err
 	}
@@ -172,9 +169,8 @@ func (r *ReconcileClusterDisruption) reconcile(request reconcile.Request) (recon
 		return reconcile.Result{}, err
 	}
 
-	activeNodeDrains := len(nodeDrainFailureDomains) > 0
 	pgHealthyRegex := cephCluster.Spec.DisruptionManagement.PGHealthyRegex
-	return r.reconcilePDBsForOSDs(clusterInfo, request, pdbStateMap, poolFailureDomain, allFailureDomains, osdDownFailureDomains, activeNodeDrains, pgHealthyRegex)
+	return r.reconcilePDBsForOSDs(clusterInfo, request, pdbStateMap, poolFailureDomain, allFailureDomains, osdDownFailureDomains, nodeDrainFailureDomains, downOSDs, pgHealthyRegex)
 }
 
 // ClusterMap maintains the association between namespace and clusername

tests/framework/installer/ceph_manifests.go

@@ -183,7 +183,6 @@ spec:
   disruptionManagement:
     managePodBudgets: true
     osdMaintenanceTimeout: 30
-    pgHealthCheckTimeout: 0
   healthCheck:
     daemonHealth:
       mon: