Merge pull request #6497 from sp98/osd-pdb-reconciler

ceph: OSD PDB reconciler changes

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

@@ -55,6 +55,8 @@ spec:
                       type: boolean
                     osdMaintenanceTimeout:
                       type: integer
+                    pgHealthCheckTimeout:
+                      type: integer
                     manageMachineDisruptionBudgets:
                       type: boolean
                 skipUpgradeChecks:
@@ -137,6 +139,8 @@ spec:
                           type: boolean
                         osdMaintenanceTimeout:
                           type: integer
+                        pgHealthCheckTimeout:
+                          type: integer
                         manageMachineDisruptionBudgets:
                           type: boolean
                     useAllNodes:
@@ -1372,6 +1376,8 @@ spec:
                   type: boolean
                 osdMaintenanceTimeout:
                   type: integer
+                pgHealthCheckTimeout:
+                  type: integer
                 manageMachineDisruptionBudgets:
                   type: boolean
             skipUpgradeChecks:
@@ -1413,6 +1419,8 @@ spec:
                       type: boolean
                     osdMaintenanceTimeout:
                       type: integer
+                    pgHealthCheckTimeout:
+                      type: integer
                     manageMachineDisruptionBudgets:
                       type: boolean
                 useAllNodes:

cluster/examples/kubernetes/ceph/cluster-on-pvc.yaml

@@ -168,6 +168,7 @@ spec:
   disruptionManagement:
     managePodBudgets: false
     osdMaintenanceTimeout: 30
+    pgHealthCheckTimeout: 0
     manageMachineDisruptionBudgets: false
     machineDisruptionBudgetNamespace: openshift-machine-api
   # security oriented settings

cluster/examples/kubernetes/ceph/cluster.yaml

@@ -211,6 +211,10 @@ 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
     # If true, the operator will create and manage MachineDisruptionBudgets to ensure OSDs are only fenced when the cluster is healthy.
     # Only available on OpenShift.
     manageMachineDisruptionBudgets: false

cluster/examples/kubernetes/ceph/crds.yaml

@@ -57,6 +57,8 @@ spec:
                       type: boolean
                     osdMaintenanceTimeout:
                       type: integer
+                    pgHealthCheckTimeout:
+                      type: integer
                     manageMachineDisruptionBudgets:
                       type: boolean
                 skipUpgradeChecks:
@@ -139,6 +141,8 @@ spec:
                           type: boolean
                         osdMaintenanceTimeout:
                           type: integer
+                        pgHealthCheckTimeout:
+                          type: integer
                         manageMachineDisruptionBudgets:
                           type: boolean
                     useAllNodes:

cluster/examples/kubernetes/ceph/pre-k8s-1.16/crds.yaml

@@ -65,6 +65,8 @@ spec:
                   type: boolean
                 osdMaintenanceTimeout:
                   type: integer
+                pgHealthCheckTimeout:
+                  type: integer
                 manageMachineDisruptionBudgets:
                   type: boolean
             skipUpgradeChecks:
@@ -106,6 +108,8 @@ spec:
                       type: boolean
                     osdMaintenanceTimeout:
                       type: integer
+                    pgHealthCheckTimeout:
+                      type: integer
                     manageMachineDisruptionBudgets:
                       type: boolean
                 useAllNodes:

design/ceph/ceph-managed-disruptionbudgets.md

@@ -14,26 +14,37 @@
 
 OSDs do not fit under the single PodDisruptionBudget pattern. Ceph's ability to tolerate pod disruptions in one failure domain is dependent on the overall health of the cluster.
 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 the next.
-Therefore, we will create a PodDisruptionBudget per failure domain that does not allow any evictions by default.
-When attempts to drain are detected, we will delete PodDisruption budget on one node at a time, progressing to the next only after ceph is healthy enough to avoid data loss/unavailability.
+
 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.
 
-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.
-We will use a heuristic to detect drains. We will create a canary deployment for each node with a nodeSelector for that node. Since it is likely that pod will only be
-removed from the node in the event of a drain, we will rely on the assumption that if that pod is not running, that node is being drained. This will not be a dangerous assumption
-as false positives for drains are not dangerous in this use case.
+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.
 
-Example flow:
-- A Ceph pool CRD is created.
-- The Rook operator creates a PDB with maxUnvailable of 0 for each failure domain.
-- A cluster upgrade agent wants to perform a kernel upgrade on the nodes.
-- It attempts to drain 1 or more nodes.
-- The drain attempt successfully evicts the canary pod.
-- The Rook operator interprets this as a drain request that it can grant by deleting the PDB.
-- The Rook operator deletes one PDB, and the blocked drain on that failure domain completes.
-- The OSDs on that node comeback up and all the necessary backfilling occurs, and all the osds are active+clean.
-- The Rook operator recreates the PDB on that failure domain.
-- The process is repeated with the subsequent nodes/failure domains.
+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.
+
+Example scenario:
+
+- Zone x
+  - Node a
+    - osd.0
+    - osd.1
+- Zone y
+  - Node b
+    - osd.2
+    - osd.3
+- Zone z
+  - Node c
+    - 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
+   - 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
 
 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

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

@@ -823,10 +823,16 @@ type DisruptionManagementSpec struct {
 	ManagePodBudgets bool `json:"managePodBudgets,omitempty"`
 
 	// OSDMaintenanceTimeout sets how many additional minutes the DOWN/OUT interval is for drained failure domains
-	// it only works if managePodBudgetss is true.
+	// it only works if managePodBudgets is true.
 	// the default is 30 minutes
 	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.
+	PGHealthCheckTimeout time.Duration `json:"pgHealthCheckTimeout,omitempty"`
+
 	// This enables management of machinedisruptionbudgets
 	ManageMachineDisruptionBudgets bool `json:"manageMachineDisruptionBudgets,omitempty"`
 

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

@@ -19,6 +19,7 @@ package osd
 import (
 	"fmt"
 	"strconv"
+	"strings"
 
 	"github.com/rook/rook/pkg/operator/ceph/controller"
 )
@@ -32,6 +33,8 @@ const (
 	CephDeviceSetPVCIDLabelKey = "ceph.rook.io/DeviceSetPVCId"
 	// OSDOverPVCLabelKey is the Rook PVC label key
 	OSDOverPVCLabelKey = "ceph.rook.io/pvc"
+	// TopologyLocationLabel is the crush location label added to OSD deployments
+	TopologyLocationLabel = "topology-location-%s"
 )
 
 func makeStorageClassDeviceSetPVCLabel(storageClassDeviceSetName, pvcStorageClassDeviceSetPVCId string, setIndex int) map[string]string {
@@ -42,11 +45,26 @@ func makeStorageClassDeviceSetPVCLabel(storageClassDeviceSetName, pvcStorageClas
 	}
 }
 
-func (c *Cluster) getOSDLabels(osdID int, failureDomainValue string, portable bool) map[string]string {
-	stringID := fmt.Sprintf("%d", osdID)
+func (c *Cluster) getOSDLabels(osd OSDInfo, failureDomainValue string, portable bool) map[string]string {
+	stringID := fmt.Sprintf("%d", osd.ID)
 	labels := controller.CephDaemonAppLabels(AppName, c.clusterInfo.Namespace, "osd", stringID, true)
 	labels[OsdIdLabelKey] = stringID
 	labels[FailureDomainKey] = failureDomainValue
 	labels[portableKey] = strconv.FormatBool(portable)
+	for k, v := range getOSDTopologyLocationLabels(osd.Location) {
+		labels[k] = v
+	}
+	return labels
+}
+
+func getOSDTopologyLocationLabels(topologyLocation string) map[string]string {
+	labels := map[string]string{}
+	locations := strings.Split(topologyLocation, " ")
+	for _, location := range locations {
+		loc := strings.Split(location, "=")
+		if len(loc) == 2 {
+			labels[fmt.Sprintf(TopologyLocationLabel, loc[0])] = loc[1]
+		}
+	}
 	return labels
 }

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

@@ -0,0 +1,31 @@
+/*
+Copyright 2020 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package osd
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestOSDTopologyLabels(t *testing.T) {
+	fakeLocation := "root=default host=ocs-deviceset-gp2-1-data-0-wh5wl region=us-east-1 zone=us-east-1c"
+	result := getOSDTopologyLocationLabels(fakeLocation)
+	assert.Equal(t, "us-east-1", result["topology-location-region"])
+	assert.Equal(t, "ocs-deviceset-gp2-1-data-0-wh5wl", result["topology-location-host"])
+	assert.Equal(t, "us-east-1c", result["topology-location-zone"])
+}

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

@@ -318,6 +318,7 @@ func TestGetPVCHostName(t *testing.T) {
 	clientset := fake.NewSimpleClientset()
 	clusterInfo := &client.ClusterInfo{Namespace: "ns"}
 	c := &Cluster{context: &clusterd.Context{Clientset: clientset}, clusterInfo: clusterInfo}
+	osdInfo := OSDInfo{ID: 23}
 	pvcName := "test-pvc"
 
 	// fail to get the host name when there is no pod or deployment
@@ -330,7 +331,7 @@ func TestGetPVCHostName(t *testing.T) {
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "osd-23",
 			Namespace: c.clusterInfo.Namespace,
-			Labels:    c.getOSDLabels(23, "", true),
+			Labels:    c.getOSDLabels(osdInfo, "", true),
 		},
 	}
 	k8sutil.AddLabelToDeployment(OSDOverPVCLabelKey, pvcName, osdDeployment)
@@ -351,7 +352,7 @@ func TestGetPVCHostName(t *testing.T) {
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "osd-23",
 			Namespace: c.clusterInfo.Namespace,
-			Labels:    c.getOSDLabels(23, "", true),
+			Labels:    c.getOSDLabels(osdInfo, "", true),
 		},
 	}
 	osdPod.Labels = map[string]string{OSDOverPVCLabelKey: pvcName}

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

@@ -518,7 +518,7 @@ func (c *Cluster) makeDeployment(osdProps osdProperties, osd OSDInfo, provisionC
 	podTemplateSpec := v1.PodTemplateSpec{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:   AppName,
-			Labels: c.getOSDLabels(osd.ID, failureDomainValue, osdProps.portable),
+			Labels: c.getOSDLabels(osd, failureDomainValue, osdProps.portable),
 		},
 		Spec: v1.PodSpec{
 			RestartPolicy:      v1.RestartPolicyAlways,
@@ -563,7 +563,7 @@ func (c *Cluster) makeDeployment(osdProps osdProperties, osd OSDInfo, provisionC
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      deploymentName,
 			Namespace: c.clusterInfo.Namespace,
-			Labels:    c.getOSDLabels(osd.ID, failureDomainValue, osdProps.portable),
+			Labels:    c.getOSDLabels(osd, failureDomainValue, osdProps.portable),
 		},
 		Spec: apps.DeploymentSpec{
 			Selector: &metav1.LabelSelector{

pkg/operator/ceph/cluster/register_controllers.go

@@ -25,7 +25,6 @@ import (
 	"github.com/rook/rook/pkg/operator/ceph/disruption/controllerconfig"
 	"github.com/rook/rook/pkg/operator/ceph/disruption/machinedisruption"
 	"github.com/rook/rook/pkg/operator/ceph/disruption/machinelabel"
-	"github.com/rook/rook/pkg/operator/ceph/disruption/nodedrain"
 	"github.com/rook/rook/pkg/operator/ceph/file"
 	"github.com/rook/rook/pkg/operator/ceph/nfs"
 	"github.com/rook/rook/pkg/operator/ceph/object"
@@ -45,7 +44,6 @@ var (
 
 // AddToManagerFuncsMaintenance is a list of functions to add all Controllers to the Manager (entrypoint for controller)
 var AddToManagerFuncsMaintenance = []func(manager.Manager, *controllerconfig.Context) error{
-	nodedrain.Add,
 	clusterdisruption.Add,
 }
 

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

@@ -18,8 +18,6 @@ package clusterdisruption
 
 import (
 	"github.com/rook/rook/pkg/operator/ceph/disruption/controllerconfig"
-	"github.com/rook/rook/pkg/operator/ceph/disruption/nodedrain"
-	"github.com/rook/rook/pkg/operator/k8sutil"
 
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
@@ -29,8 +27,6 @@ import (
 
 	"github.com/pkg/errors"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
-	appsv1 "k8s.io/api/apps/v1"
-	policyv1beta1 "k8s.io/api/policy/v1beta1"
 	"k8s.io/apimachinery/pkg/types"
 )
 
@@ -49,11 +45,10 @@ func Add(mgr manager.Manager, context *controllerconfig.Context) error {
 	sharedClusterMap := &ClusterMap{}
 
 	reconcileClusterDisruption := &ReconcileClusterDisruption{
-		client:              mgr.GetClient(),
-		scheme:              mgrScheme,
-		context:             context,
-		clusterMap:          sharedClusterMap,
-		osdCrushLocationMap: &OSDCrushLocationMap{Context: context.ClusterdContext},
+		client:     mgr.GetClient(),
+		scheme:     mgrScheme,
+		context:    context,
+		clusterMap: sharedClusterMap,
 	}
 	reconciler := reconcile.Reconciler(reconcileClusterDisruption)
 	// Create a new controller
@@ -83,74 +78,6 @@ func Add(mgr manager.Manager, context *controllerconfig.Context) error {
 		return err
 	}
 
-	// Watch for PodDisruptionBudgets and enqueue the CephCluster in the namespace
-	err = c.Watch(
-		&source.Kind{Type: &policyv1beta1.PodDisruptionBudget{}},
-		&handler.EnqueueRequestsFromMapFunc{
-			ToRequests: handler.ToRequestsFunc(func(obj handler.MapObject) []reconcile.Request {
-				_, ok := obj.Object.(*policyv1beta1.PodDisruptionBudget)
-				if !ok {
-					// not a pdb, returning empty
-					logger.Errorf("PDB handler received non-PDB")
-					return []reconcile.Request{}
-				}
-				labels := obj.Meta.GetLabels()
-
-				// only enqueue osdDisruptionAppLabels
-				_, ok = labels[osdDisruptionAppName]
-				if !ok {
-					return []reconcile.Request{}
-				}
-				// // The name will be populated in the reconcile
-				namespace := obj.Meta.GetNamespace()
-				req := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}}
-
-				return []reconcile.Request{req}
-			}),
-		},
-	)
-	if err != nil {
-		return err
-	}
-
-	// Watch for canary Deployments created by the nodedrain controller and enqueue all Cephclusters
-	err = c.Watch(
-		&source.Kind{Type: &appsv1.Deployment{}},
-		&handler.EnqueueRequestsFromMapFunc{
-			ToRequests: handler.ToRequestsFunc(func(obj handler.MapObject) []reconcile.Request {
-				_, ok := obj.Object.(*appsv1.Deployment)
-				if !ok {
-					// not a Deployment, returning empty
-					logger.Errorf("deployment handler received non-Deployment")
-					return []reconcile.Request{}
-				}
-
-				// don't enqueue if it isn't a canary Deployment
-				labels := obj.Meta.GetLabels()
-				appLabel, ok := labels[k8sutil.AppAttr]
-				if !ok || appLabel != nodedrain.CanaryAppName {
-					return []reconcile.Request{}
-				}
-
-				// Enqueue all CephClusters
-				clusterNamespaces := sharedClusterMap.GetClusterNamespaces()
-				if len(clusterNamespaces) == 0 {
-					return []reconcile.Request{}
-				}
-				reqs := make([]reconcile.Request, 0)
-				for _, namespace := range clusterNamespaces {
-					// The name will be populated in the reconcile
-					reqs = append(reqs, reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}})
-				}
-
-				return reqs
-			}),
-		},
-	)
-	if err != nil {
-		return err
-	}
-
 	// Watch for CephBlockPools and enqueue the CephCluster in the namespace
 	err = c.Watch(&source.Kind{Type: &cephv1.CephBlockPool{}}, enqueueByNamespace)
 	if err != nil {

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

@@ -1,227 +0,0 @@
-/*
-Copyright 2019 The Rook Authors. All rights reserved.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-	http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package clusterdisruption
-
-import (
-	"context"
-	"sort"
-	"strconv"
-	"strings"
-	"sync"
-	"time"
-
-	"github.com/pkg/errors"
-	"github.com/rook/rook/pkg/clusterd"
-	cephClient "github.com/rook/rook/pkg/daemon/ceph/client"
-	"github.com/rook/rook/pkg/operator/ceph/cluster/osd"
-	"github.com/rook/rook/pkg/operator/k8sutil"
-	appsv1 "k8s.io/api/apps/v1"
-	corev1 "k8s.io/api/core/v1"
-	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/reconcile"
-)
-
-func (r *ReconcileClusterDisruption) getOsdDataList(clusterInfo *cephClient.ClusterInfo, request reconcile.Request, poolFailureDomain string) ([]OsdData, error) {
-	osdDeploymentList := &appsv1.DeploymentList{}
-	namespaceListOpts := client.InNamespace(request.Namespace)
-	err := r.client.List(context.TODO(), osdDeploymentList, client.MatchingLabels{k8sutil.AppAttr: osd.AppName}, namespaceListOpts)
-	if err != nil {
-		return nil, errors.Wrap(err, "could not list osd deployments")
-	}
-
-	osds := make([]OsdData, 0)
-
-	for _, deployment := range osdDeploymentList.Items {
-		osdData := OsdData{Deployment: deployment}
-		labels := deployment.Spec.Template.ObjectMeta.GetLabels()
-		osdID, ok := labels[osd.OsdIdLabelKey]
-		if !ok {
-			return nil, errors.Errorf("osd %q was not labeled with %q", deployment.GetName(), osd.OsdIdLabelKey)
-		}
-		osdIDInt, err := strconv.Atoi(osdID)
-		if err != nil {
-			return nil, errors.Wrapf(err, "failed to convert osd id %q in an int", osdID)
-		}
-		crushMeta, err := r.osdCrushLocationMap.Get(clusterInfo, osdIDInt)
-		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
-			//
-			// Or this means the ceph config hasn't been written on the operator yet
-			// The controller starts before we run WriteConnectionConfig()
-			if strings.Contains(err.Error(), "error calling conf_read_file") {
-				logger.Debugf("Ceph %q cluster is not ready, cannot check osd location yet.", request.Namespace)
-				return nil, nil
-			}
-			return nil, errors.Wrapf(err, "could not fetch info from ceph for osd %q", osdID)
-		}
-		// bypass the cache if the topology location is not populated in the cache
-		_, failureDomainKnown := crushMeta.Location[poolFailureDomain]
-		if !failureDomainKnown {
-			crushMeta, err = r.osdCrushLocationMap.get(clusterInfo, osdIDInt)
-			if err != nil {
-				return nil, errors.Wrapf(err, "could not fetch info from ceph for osd %q", osdID)
-			}
-		}
-
-		osdData.CrushMeta = crushMeta
-		osds = append(osds, osdData)
-
-	}
-	return osds, nil
-}
-
-// OsdData stores the deployment and the Crush Data of the osd together
-type OsdData struct {
-	Deployment appsv1.Deployment
-	CrushMeta  *cephClient.CrushFindResult
-}
-
-// OSDCrushLocationMap is used to maintain a cache of map of osd id to cephClientCrushhFindResults
-// the crush position of osds wrt to the failureDomain is not expected to change often, so a default Resync period
-// of half an hour is used, but if a use case arises where this is needed, ResyncPeriod should be made smaller.
-type OSDCrushLocationMap struct {
-	ResyncPeriod       time.Duration
-	Context            *clusterd.Context
-	clusterLocationMap map[string]map[int]cachedOSDLocation
-	mux                sync.Mutex
-}
-
-type cachedOSDLocation struct {
-	result     *cephClient.CrushFindResult
-	lastSynced time.Time
-}
-
-// Get takes an osd id and returns a CrushFindResult from cache
-func (o *OSDCrushLocationMap) Get(clusterInfo *cephClient.ClusterInfo, id int) (*cephClient.CrushFindResult, error) {
-	o.mux.Lock()
-	defer o.mux.Unlock()
-	if o.ResyncPeriod == 0 {
-		o.ResyncPeriod = 30 * time.Minute
-	}
-
-	// initialize clusterLocationMap
-	if len(o.clusterLocationMap) == 0 {
-		o.clusterLocationMap = make(map[string]map[int]cachedOSDLocation)
-	}
-	locationMap, ok := o.clusterLocationMap[clusterInfo.Namespace]
-	// initialize namespace map
-	if !ok {
-		o.clusterLocationMap[clusterInfo.Namespace] = make(map[int]cachedOSDLocation)
-	}
-
-	// sync of osd id not found in clusterNamespace
-	osdLocation, ok := locationMap[id]
-	if !ok {
-		osdResult, err := o.get(clusterInfo, id)
-		if err != nil {
-			return nil, errors.Wrapf(err, "failed to run `find` on osd %d in cluster %q", id, clusterInfo.Namespace)
-		}
-		o.clusterLocationMap[clusterInfo.Namespace][id] = cachedOSDLocation{result: osdResult, lastSynced: time.Now()}
-		return osdResult, nil
-	}
-
-	// sync if not synced for longer than ResyncPeriod
-	if time.Since(osdLocation.lastSynced) > o.ResyncPeriod {
-		osdResult, err := o.get(clusterInfo, id)
-		if err != nil {
-			return nil, errors.Wrapf(err, "failed to run `find` on osd %d in cluster %q", id, clusterInfo.Namespace)
-		}
-		o.clusterLocationMap[clusterInfo.Namespace][id] = cachedOSDLocation{result: osdResult, lastSynced: time.Now()}
-		return osdResult, nil
-	}
-
-	return osdLocation.result, nil
-
-}
-
-// uncached version
-func (o *OSDCrushLocationMap) get(clusterInfo *cephClient.ClusterInfo, id int) (*cephClient.CrushFindResult, error) {
-	osdResult, err := cephClient.FindOSDInCrushMap(o.Context, clusterInfo, id)
-	if err != nil {
-		return nil, errors.Wrapf(err, "failed running find on osd %d", id)
-	}
-	o.clusterLocationMap[clusterInfo.Namespace][id] = cachedOSDLocation{
-		result:     osdResult,
-		lastSynced: time.Now(),
-	}
-	return osdResult, nil
-}
-
-func getOSDsForNodes(osdDataList []OsdData, nodeList []*corev1.Node, failureDomainType string) ([]OsdData, error) {
-	nodeOsdDataList := make([]OsdData, 0)
-	for _, node := range nodeList {
-		if node == nil {
-			logger.Warningf("node in nodelist was nil")
-			continue
-		}
-		nodeTopologyMap := osd.ExtractOSDTopologyFromLabels(node.GetLabels())
-
-		for _, osdData := range osdDataList {
-			// get the crush location of the osd
-			crushFailureDomain, ok := osdData.CrushMeta.Location[failureDomainType]
-			if !ok {
-				return nil, errors.Errorf("could not find the CrushFindResult.Location[%q] for %q", failureDomainType, osdData.Deployment.GetName())
-			}
-			// get the crush location of the node
-			nodeFailureDomain, ok := nodeTopologyMap[failureDomainType]
-			if !ok {
-				return nil, errors.Errorf("could not find the %q failure domain on node %q", failureDomainType, node.GetName())
-			}
-
-			// check if the node and osd have the same crush location value for this particular crush location type
-			if cephClient.IsNormalizedCrushNameEqual(nodeFailureDomain, crushFailureDomain) || (failureDomainType == "host" && cephClient.IsNormalizedCrushNameEqual(node.GetName(), crushFailureDomain)) {
-				nodeOsdDataList = append(nodeOsdDataList, osdData)
-			}
-
-		}
-	}
-	return nodeOsdDataList, nil
-}
-
-func getFailureDomainMapForOsds(osdDataList []OsdData, failureDomainType string) (map[string][]OsdData, error) {
-	failureDomainMap := make(map[string][]OsdData)
-	unfoundOSDs := make([]string, 0)
-	var err error
-	for _, osdData := range osdDataList {
-		failureDomainValue, ok := osdData.CrushMeta.Location[failureDomainType]
-		if !ok {
-			logger.Errorf("failureDomain type %q not associated with %q", failureDomainType, osdData.Deployment.GetName())
-			unfoundOSDs = append(unfoundOSDs, osdData.Deployment.ObjectMeta.Name)
-		} else {
-			if len(failureDomainMap[failureDomainValue]) == 0 {
-				failureDomainMap[failureDomainValue] = make([]OsdData, 0)
-			}
-			failureDomainMap[failureDomainValue] = append(failureDomainMap[failureDomainValue], osdData)
-		}
-	}
-	if len(unfoundOSDs) > 0 {
-		err = errors.Errorf("failure domain type %q not associated with osds: [%q]", failureDomainType, strings.Join(unfoundOSDs, ","))
-	}
-	return failureDomainMap, err
-}
-
-func getSortedOSDMapKeys(m map[string][]OsdData) []string {
-	list := make([]string, len(m))
-	count := 0
-	for key := range m {
-		list[count] = key
-		count++
-	}
-	sort.Strings(list)
-	return list
-}

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

@@ -1,168 +0,0 @@
-/*
-Copyright 2019 The Rook Authors. All rights reserved.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-	http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package clusterdisruption
-
-import (
-	"fmt"
-	"testing"
-
-	cephClient "github.com/rook/rook/pkg/daemon/ceph/client"
-	"github.com/stretchr/testify/assert"
-	appsv1 "k8s.io/api/apps/v1"
-	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-)
-
-func TestGetOSDNodeAssociation(t *testing.T) {
-
-	hostNum := 5
-	hostNames := make([]string, hostNum)
-	for i := 0; i < hostNum; i++ {
-		hostNames[i] = fmt.Sprintf("worker-%d", i)
-	}
-
-	zoneNum := hostNum
-	zoneNames := make([]string, zoneNum)
-	for i := 0; i < zoneNum; i++ {
-		zoneNames[i] = fmt.Sprintf("zone-%d", i)
-	}
-
-	osdNum := hostNum * 3
-	osdNames := make([]string, osdNum)
-	for i := 0; i < osdNum; i++ {
-		osdNames[i] = fmt.Sprintf("osd-%d", i)
-	}
-
-	osdDataList := []OsdData{
-		{
-			Deployment: appsv1.Deployment{
-				ObjectMeta: metav1.ObjectMeta{
-					Name: osdNames[0],
-				},
-			},
-			CrushMeta: &cephClient.CrushFindResult{
-				Location: map[string]string{
-					"host": hostNames[0],
-					"zone": zoneNames[0],
-				},
-			},
-		},
-		{
-			Deployment: appsv1.Deployment{
-				ObjectMeta: metav1.ObjectMeta{
-					Name: osdNames[1],
-				},
-			},
-			CrushMeta: &cephClient.CrushFindResult{
-				Location: map[string]string{
-					"host": hostNames[1],
-					"zone": zoneNames[1],
-				},
-			},
-		},
-		{
-			Deployment: appsv1.Deployment{
-				ObjectMeta: metav1.ObjectMeta{
-					Name: osdNames[2],
-				},
-			},
-			CrushMeta: &cephClient.CrushFindResult{
-				Location: map[string]string{
-					"host": hostNames[1],
-					"zone": zoneNames[2],
-				},
-			},
-		},
-		{
-			Deployment: appsv1.Deployment{
-				ObjectMeta: metav1.ObjectMeta{
-					Name: osdNames[3],
-				},
-			},
-			CrushMeta: &cephClient.CrushFindResult{
-				Location: map[string]string{
-					"host": hostNames[1],
-					"zone": zoneNames[2],
-				},
-			},
-		},
-		{
-			Deployment: appsv1.Deployment{
-				ObjectMeta: metav1.ObjectMeta{
-					Name: osdNames[4],
-				},
-			},
-			CrushMeta: &cephClient.CrushFindResult{
-				Location: map[string]string{
-					"host": hostNames[2],
-					"zone": zoneNames[0],
-				},
-			},
-		},
-	}
-	nodeList := []*corev1.Node{
-		{
-			ObjectMeta: metav1.ObjectMeta{
-				Name: hostNames[0],
-				Labels: map[string]string{
-					corev1.LabelHostname:          hostNames[0],
-					corev1.LabelZoneFailureDomain: zoneNames[1],
-				},
-			},
-		},
-		{
-			ObjectMeta: metav1.ObjectMeta{
-				Name: hostNames[2],
-				Labels: map[string]string{
-					corev1.LabelHostname:          hostNames[2],
-					corev1.LabelZoneFailureDomain: zoneNames[2],
-				},
-			},
-		},
-		nil,
-	}
-
-	// osds from hostName[0] - (0)
-	// osds from hostNames[1] - (2,3,1)
-	// osds from hostNames[2] - (4)
-
-	// hosts from nodeList (0,2)
-	osdDataSubset, err := getOSDsForNodes(osdDataList, nodeList, "host")
-	assert.Nil(t, err)
-	assert.Len(t, osdDataSubset, 2)
-
-	failureDomainsMap, err := getFailureDomainMapForOsds(osdDataSubset, "host")
-	assert.Nil(t, err)
-	failureDomainSubset := getSortedOSDMapKeys(failureDomainsMap)
-	assert.Equal(t, []string{hostNames[0], hostNames[2]}, failureDomainSubset)
-
-	// osds from zone[0] - (0,4)
-	// osds from zone[1] - (1)
-	// osds from zone[2] - (2,3)
-
-	// zones from nodeList (0,2)
-	osdDataSubset, err = getOSDsForNodes(osdDataList, nodeList, "zone")
-	assert.Nil(t, err)
-	assert.Len(t, osdDataSubset, 3)
-
-	failureDomainsMap, err = getFailureDomainMapForOsds(osdDataSubset, "zone")
-	assert.Nil(t, err)
-
-	failureDomainSubset = getSortedOSDMapKeys(failureDomainsMap)
-	assert.Equal(t, []string{zoneNames[1], zoneNames[2]}, failureDomainSubset)
-
-}

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

@@ -30,7 +30,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
 
-func createFakeReconcileClusterDisruptionr(t *testing.T, obj ...runtime.Object) *ReconcileClusterDisruption {
+func createFakeReconcileClusterDisruption(t *testing.T, obj ...runtime.Object) *ReconcileClusterDisruption {
 	scheme := scheme.Scheme
 	err := policyv1beta1.AddToScheme(scheme)
 	if err != nil {
@@ -45,7 +45,7 @@ func createFakeReconcileClusterDisruptionr(t *testing.T, obj ...runtime.Object)
 }
 
 func TestReconcileMonPDB(t *testing.T) {
-	r := createFakeReconcileClusterDisruptionr(t, &cephv1.CephCluster{})
+	r := createFakeReconcileClusterDisruption(t, &cephv1.CephCluster{})
 	testCases := []struct {
 		label                string
 		cephCluster          *cephv1.CephCluster

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

@@ -23,10 +23,12 @@ import (
 	"time"
 
 	"github.com/pkg/errors"
+	cephClient "github.com/rook/rook/pkg/daemon/ceph/client"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/osd"
-	"github.com/rook/rook/pkg/operator/ceph/disruption/nodedrain"
+	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/k8sutil"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -37,85 +39,149 @@ import (
 	policyv1beta1 "k8s.io/api/policy/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/apimachinery/pkg/util/sets"
 	// cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 )
 
 const (
-	// PDBAppName is that app label value for pdbs targeting osds
-	PDBAppName         = "rook-ceph-osd-pdb"
-	disabledPDBKey     = "disabled-pdb"
-	disabledPDBTimeKey = "pdb-disabled-at"
+	// osdPDBAppName is that app label value for pdbs targeting osds
+	osdPDBAppName                    = "rook-ceph-osd"
+	drainingFailureDomainKey         = "draining-failure-domain"
+	drainingFailureDomainDurationKey = "draining-failure-domain-duration"
+	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"
 )
 
-func (r *ReconcileClusterDisruption) createPDBForOSD(deployment appsv1.Deployment) error {
-	deploymentLabels := deployment.ObjectMeta.GetLabels()
-	deploymentName := deployment.ObjectMeta.Name
-	namespace := deployment.ObjectMeta.Namespace
-	osdIDLabel, ok := deploymentLabels[osd.OsdIdLabelKey]
-	if !ok {
-		return errors.Errorf("could not find id label on osd %s/%s", namespace, deploymentName)
+func (r *ReconcileClusterDisruption) createPDB(pdb *policyv1beta1.PodDisruptionBudget) error {
+	err := r.client.Create(context.TODO(), pdb)
+	if err != nil && !kerrors.IsAlreadyExists(err) {
+		return errors.Wrapf(err, "failed to create pdb %q", pdb.Name)
 	}
+	return nil
+}
+
+func (r *ReconcileClusterDisruption) deletePDB(pdb *policyv1beta1.PodDisruptionBudget) error {
+	err := r.client.Delete(context.TODO(), pdb)
+	if err != nil && !kerrors.IsNotFound(err) {
+		return errors.Wrapf(err, "failed to delete pdb %q", pdb.Name)
+	}
+	return nil
+}
+
+// createOverallPDBforOSD 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 {
 	cephCluster, ok := r.clusterMap.GetCluster(namespace)
 	if !ok {
-		return errors.Errorf("the namespace %s was not found in the clustermap", namespace)
+		return errors.Errorf("failed to find the namespace %q in the clustermap", namespace)
 	}
-
+	pdbRequest := types.NamespacedName{Name: osdPDBAppName, Namespace: namespace}
 	pdb := &policyv1beta1.PodDisruptionBudget{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      deploymentName,
+			Name:      osdPDBAppName,
 			Namespace: namespace,
-			Labels: map[string]string{
-				k8sutil.AppAttr:   PDBAppName,
-				osd.OsdIdLabelKey: osdIDLabel,
-			},
 			OwnerReferences: []metav1.OwnerReference{
-				{
-					APIVersion: cephCluster.APIVersion,
-					Kind:       cephCluster.Kind,
-					Name:       cephCluster.ObjectMeta.GetName(),
-					UID:        cephCluster.GetUID(),
-				},
+				opcontroller.ClusterOwnerRef(cephCluster.GetName(), string(cephCluster.GetUID())),
+			},
+		},
+		Spec: policyv1beta1.PodDisruptionBudgetSpec{
+			MaxUnavailable: &intstr.IntOrString{IntVal: 1},
+			Selector: &metav1.LabelSelector{
+				MatchLabels: map[string]string{k8sutil.AppAttr: osd.AppName},
+			},
+		},
+	}
+
+	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)
+			return r.createPDB(pdb)
+		}
+		return errors.Wrapf(err, "failed to get pdb %q", pdb.Name)
+	}
+	return nil
+}
+
+func (r *ReconcileClusterDisruption) deleteOverallPDBforOSD(namespace string) error {
+	pdbRequest := types.NamespacedName{Name: osdPDBAppName, Namespace: namespace}
+	pdb := &policyv1beta1.PodDisruptionBudget{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      osdPDBAppName,
+			Namespace: namespace,
+		},
+	}
+	err := r.client.Get(context.TODO(), pdbRequest, &policyv1beta1.PodDisruptionBudget{})
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			return nil
+		}
+		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)
+	return r.deletePDB(pdb)
+}
+
+// createBlockingPDBForOSD creates individual blocking PDBs (maxUnavailable=0) for all the OSDs in
+// failure domains that are not draining
+func (r *ReconcileClusterDisruption) createBlockingPDBForOSD(namespace, failureDomainType, failureDomainName string) error {
+	cephCluster, ok := r.clusterMap.GetCluster(namespace)
+	if !ok {
+		return errors.Errorf("failed to find the namespace %q in the clustermap", namespace)
+	}
+
+	pdbName := getPDBName(failureDomainType, failureDomainName)
+	pdbRequest := types.NamespacedName{Name: pdbName, Namespace: namespace}
+	pdb := &policyv1beta1.PodDisruptionBudget{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      pdbName,
+			Namespace: namespace,
+			OwnerReferences: []metav1.OwnerReference{
+				opcontroller.ClusterOwnerRef(cephCluster.GetName(), string(cephCluster.GetUID())),
 			},
 		},
 		Spec: policyv1beta1.PodDisruptionBudgetSpec{
 			MaxUnavailable: &intstr.IntOrString{IntVal: 0},
-			Selector:       deployment.Spec.Selector,
+			Selector: &metav1.LabelSelector{
+				MatchLabels: map[string]string{fmt.Sprintf(osd.TopologyLocationLabel, failureDomainType): failureDomainName},
+			},
 		},
 	}
-
-	err := r.client.Create(context.TODO(), pdb)
-	if err != nil && !kerrors.IsAlreadyExists(err) {
-		return errors.Wrapf(err, "could not create pdb for osd: %s in namespace %s", osdIDLabel, namespace)
+	err := r.client.Get(context.TODO(), pdbRequest, &policyv1beta1.PodDisruptionBudget{})
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			logger.Infof("creating temporary blocking pdb %q with maxUnavailable=0 for %q failure domain %q", pdbName, failureDomainType, failureDomainName)
+			return r.createPDB(pdb)
+		}
+		return errors.Wrapf(err, "failed to get pdb %q", pdb.Name)
 	}
 	return nil
 }
 
-func (r *ReconcileClusterDisruption) deletePDB(deployment appsv1.Deployment) error {
-	deploymentLabels := deployment.ObjectMeta.GetLabels()
-	deploymentName := deployment.ObjectMeta.Name
-	namespace := deployment.ObjectMeta.Namespace
-
-	osdIDLabel, ok := deploymentLabels[osd.OsdIdLabelKey]
-	if !ok {
-		return errors.Errorf("could not find id label on osd %s/%s", namespace, deploymentName)
-	}
+func (r *ReconcileClusterDisruption) deleteBlockingPDBForOSD(namespace, failureDomainType, failureDomainName string) error {
+	pdbName := getPDBName(failureDomainType, failureDomainName)
+	pdbRequest := types.NamespacedName{Name: pdbName, Namespace: namespace}
 	pdb := &policyv1beta1.PodDisruptionBudget{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      deploymentName,
+			Name:      pdbName,
 			Namespace: namespace,
 		},
 	}
-	err := r.client.Delete(context.TODO(), pdb)
-	if err != nil && !kerrors.IsNotFound(err) {
-		return errors.Wrapf(err, "could not delete pdb for osd: %s in namespace %s", osdIDLabel, namespace)
+	err := r.client.Get(context.TODO(), pdbRequest, &policyv1beta1.PodDisruptionBudget{})
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			return nil
+		}
+		return errors.Wrapf(err, "failed to get pdb %q", pdb.Name)
 	}
-	return nil
+	logger.Infof("deleting temporary blocking pdb with %q with maxUnavailable=0 for %q failure domain %q", pdbName, failureDomainType, failureDomainName)
+	return r.deletePDB(pdb)
 }
 
-func (r *ReconcileClusterDisruption) initializePDBState(request reconcile.Request, osdDataList []OsdData) (*corev1.ConfigMap, error) {
+func (r *ReconcileClusterDisruption) initializePDBState(request reconcile.Request) (*corev1.ConfigMap, error) {
 	pdbStateMap := &corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      pdbStateMapName,
@@ -129,23 +195,14 @@ func (r *ReconcileClusterDisruption) initializePDBState(request reconcile.Reques
 	err := r.client.Get(context.TODO(), pdbStateMapRequest, pdbStateMap)
 
 	if kerrors.IsNotFound(err) {
-		// create configmap and PDBs for all nodes labeled by failuredomain
-		logger.Infof("inititalizing pod disruption budgets for osds")
-		// one pdb is created per OSD, but after initialization they are created/deleted in failuredomain groups
-		for _, osdData := range osdDataList {
-			err := r.createPDBForOSD(osdData.Deployment)
-			if err != nil {
-				return pdbStateMap, errors.Wrapf(err, "failed to create pdb for osd deployment %q. %v", osdData.Deployment.ObjectMeta.GetName(), err)
-			}
-		}
-		pdbStateMap.Data = map[string]string{disabledPDBKey: ""}
-		// create configmap
+		// create configmap to track the draining failure domain
+		pdbStateMap.Data = map[string]string{drainingFailureDomainKey: ""}
 		err := r.client.Create(context.TODO(), pdbStateMap)
 		if err != nil {
-			return pdbStateMap, errors.Wrapf(err, "could not create the PDB state map %s", pdbStateMapRequest)
+			return pdbStateMap, errors.Wrapf(err, "failed to create the PDB state map %q", pdbStateMapRequest)
 		}
 	} else if err != nil {
-		return pdbStateMap, errors.Wrapf(err, "could not get the pdbStateMap %s", pdbStateMapRequest)
+		return pdbStateMap, errors.Wrapf(err, "failed to get the pdbStateMap %s", pdbStateMapRequest)
 	}
 	return pdbStateMap, nil
 }
@@ -154,11 +211,10 @@ func (r *ReconcileClusterDisruption) reconcilePDBsForOSDs(
 	clusterInfo *cephclient.ClusterInfo,
 	request reconcile.Request,
 	pdbStateMap *corev1.ConfigMap,
-	poolFailureDomain string,
-	allFailureDomainsMap,
-	drainingFailureDomainsMap map[string][]OsdData,
+	failureDomainType string,
+	allFailureDomains,
+	drainingFailureDomains []string,
 ) error {
-	drainingFailureDomains := getSortedOSDMapKeys(drainingFailureDomainsMap)
 
 	pgHealthMsg, clean, err := cephclient.IsClusterClean(r.context.ClusterdContext, clusterInfo)
 	if err != nil {
@@ -168,130 +224,215 @@ func (r *ReconcileClusterDisruption) reconcilePDBsForOSDs(
 			logger.Debugf("Ceph %q cluster not ready, cannot check Ceph status yet.", request.Namespace)
 			return nil
 		}
-		return errors.Wrapf(err, "could not check cluster health")
+		return errors.Wrapf(err, "failed to check cluster health")
 	}
+
 	if pdbStateMap.Data == nil {
 		pdbStateMap.Data = make(map[string]string)
 	}
-	_, ok := pdbStateMap.Data[disabledPDBKey]
+	_, ok := pdbStateMap.Data[drainingFailureDomainKey]
 	if !ok {
-		pdbStateMap.Data[disabledPDBKey] = ""
-	}
-	var disabledPDBTime time.Time
-	timeString, disabledPDBTimeSet := pdbStateMap.Data[disabledPDBTimeKey]
-	if disabledPDBTimeSet {
-		disabledPDBTime, err = time.Parse(time.RFC3339, timeString)
-		if err != nil {
-			logger.Errorf("could not parse timestamp %v. %v", disabledPDBTime, err)
-			disabledPDBTime = time.Now()
-			pdbStateMap.Data[disabledPDBTimeKey] = disabledPDBTime.Format(time.RFC3339)
-		}
-	}
-	recentlyChanged := disabledPDBTimeSet && time.Since(disabledPDBTime) < time.Minute
-	shouldChange := clean && !recentlyChanged
-	activeDrains := len(drainingFailureDomains) != 0
-	if activeDrains {
-		logger.Infof("pg health: %q. detected drains on %q: %v", pgHealthMsg, poolFailureDomain, drainingFailureDomains)
-	}
-	if shouldChange {
-		if activeDrains {
-			pdbStateMap.Data[disabledPDBKey] = drainingFailureDomains[0]
-			pdbStateMap.Data[disabledPDBTimeKey] = time.Now().Format(time.RFC3339)
-		} else {
-			pdbStateMap.Data[disabledPDBKey] = ""
-			delete(pdbStateMap.Data, disabledPDBTimeKey)
-		}
+		pdbStateMap.Data[drainingFailureDomainKey] = ""
 	}
 
-	err = r.updateNoout(clusterInfo, pdbStateMap, allFailureDomainsMap)
+	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)
+		if err != nil {
+			return errors.Wrap(err, "failed to handle active drains")
+		}
+		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) {
+		// 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)
+	}
+
+	err = r.updateNoout(clusterInfo, pdbStateMap, allFailureDomains)
 	if err != nil {
-		logger.Errorf("could not update maintenance noout in cluster %q with ceph image. %v", request, err)
+		logger.Errorf("failed to update maintenance noout in cluster %q. %v", request, err)
 	}
 
 	err = r.client.Update(context.TODO(), pdbStateMap)
 	if err != nil {
-		return errors.Wrapf(err, "could not update %q in cluster %q", pdbStateMapName, request)
+		return errors.Wrapf(err, "failed to update configMap %q in cluster %q", pdbStateMapName, request)
 	}
-	drainingFailureDomain, ok := pdbStateMap.Data[disabledPDBKey]
-	if ok && clean && len(drainingFailureDomain) > 0 {
 
-		canaryLabels := client.MatchingLabels{k8sutil.AppAttr: nodedrain.CanaryAppName, poolFailureDomain: drainingFailureDomain}
-
-		// list and delete only if it's old
-		drainingCanaryList := &appsv1.DeploymentList{}
-		err := r.client.List(context.TODO(), drainingCanaryList, canaryLabels, client.InNamespace(r.context.OperatorNamespace))
+	// 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.Wrapf(err, "could not list canary pods by labels %q", canaryLabels)
+			return errors.Wrap(err, "failed to handle inactive drains")
 		}
-		// refresh old canaries in draining failure domain
-		for _, drainingCanary := range drainingCanaryList.Items {
-			if time.Since(drainingCanary.GetCreationTimestamp().Time) > time.Minute && drainingCanary.Status.ReadyReplicas < 1 {
-				localdrainingCanary := drainingCanary
-				err := r.client.Delete(context.TODO(), &localdrainingCanary)
+	}
+	return nil
+}
+
+func (r *ReconcileClusterDisruption) handleActiveDrains(allFailureDomains []string, drainingFailureDomain,
+	failureDomainType, namespace string, isClean bool) error {
+
+	for _, failureDomainName := range allFailureDomains {
+		// create blocking PDB for failure domains not currently draining
+		if failureDomainName != drainingFailureDomain {
+			err := r.createBlockingPDBForOSD(namespace, failureDomainType, failureDomainName)
+			if err != nil {
+				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 {
-					logger.Warningf("could not delete canary deployment %q in namespace %q. %v", drainingCanary.GetName(), drainingCanary.GetNamespace(), err)
+					return errors.Wrapf(err, "failed to delete pdb for %q failure domain %q. %v", failureDomainType, failureDomainName, err)
 				}
 			}
 		}
 	}
-	for failureDomain, osdDataList := range allFailureDomainsMap {
-		for _, osdData := range osdDataList {
-			var err error
-			if failureDomain == pdbStateMap.Data[disabledPDBKey] {
-				err = r.deletePDB(osdData.Deployment)
-			} else {
-				err = r.createPDBForOSD(osdData.Deployment)
-			}
-			if err != nil {
-				return errors.Wrapf(err, "failed to reconcile pdb for osd deployment %q. %v", osdData.Deployment.ObjectMeta.GetName(), err)
-			}
-		}
-	}
 
+	// delete the main 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)
+	if err != nil {
+		return errors.Wrap(err, "failed to delete the main osd pdb")
+	}
 	return nil
 }
 
-func (r *ReconcileClusterDisruption) updateNoout(clusterInfo *cephclient.ClusterInfo, pdbStateMap *corev1.ConfigMap, allFailureDomainsMap map[string][]OsdData) error {
-	disabledFailureDomain := pdbStateMap.Data[disabledPDBKey]
+func (r *ReconcileClusterDisruption) handleInactiveDrains(allFailureDomains []string, namespace, failureDomainType string) error {
+	err := r.createOverallPDBforOSD(namespace)
+	if err != nil {
+		return errors.Wrap(err, "failed to create main pdb")
+	}
+	for _, failureDomainName := range allFailureDomains {
+		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)
+		}
+		logger.Debugf("deleted temporary blocking pdb for %q failure domain %q.", failureDomainType, failureDomainName)
+	}
+	return nil
+}
+
+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, "could not get osddump for reconciling maintenance noout in namespace %s", clusterInfo.Namespace)
+		return errors.Wrapf(err, "failed to get osddump for reconciling maintenance noout in namespace %s", clusterInfo.Namespace)
 	}
-	for failureDomain := range allFailureDomainsMap {
-		disabledFailureDomainTimeStampKey := fmt.Sprintf("%s-noout-last-set-at", failureDomain)
-		if disabledFailureDomain == failureDomain {
+	for _, failureDomainName := range allFailureDomains {
+		drainingFailureDomainTimeStampKey := fmt.Sprintf("%s-noout-last-set-at", failureDomainName)
+		if drainingFailureDomain == failureDomainName {
 
 			// get the time stamp
-			nooutSetTimeString, ok := pdbStateMap.Data[disabledFailureDomainTimeStampKey]
+			nooutSetTimeString, ok := pdbStateMap.Data[drainingFailureDomainTimeStampKey]
 			if !ok || len(nooutSetTimeString) == 0 {
 				// initialize it if it's not set
-				pdbStateMap.Data[disabledFailureDomainTimeStampKey] = time.Now().Format(time.RFC3339)
+				pdbStateMap.Data[drainingFailureDomainTimeStampKey] = time.Now().Format(time.RFC3339)
 			}
 			// parse the timestamp
-			nooutSetTime, err := time.Parse(time.RFC3339, pdbStateMap.Data[disabledFailureDomainTimeStampKey])
+			nooutSetTime, err := time.Parse(time.RFC3339, pdbStateMap.Data[drainingFailureDomainTimeStampKey])
 			if err != nil {
-				return errors.Wrapf(err, "could not parse timestamp %s for failureDomain %s", pdbStateMap.Data[disabledFailureDomainTimeStampKey], nooutSetTime)
+				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, failureDomain, nooutFlag); err != nil {
+				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, failureDomain, nooutFlag); err != nil {
+				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 {
 			// ensure noout unset
-			if _, err := osdDump.UpdateFlagOnCrushUnit(r.context.ClusterdContext, clusterInfo, false, failureDomain, nooutFlag); err != nil {
+			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, disabledFailureDomainTimeStampKey)
+			delete(pdbStateMap.Data, drainingFailureDomainTimeStampKey)
 		}
 	}
 	return nil
 }
+
+func (r *ReconcileClusterDisruption) getOSDFailureDomains(clusterInfo *cephClient.ClusterInfo, request reconcile.Request, poolFailureDomain 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")
+	}
+
+	allFailureDomains := sets.NewString()
+	drainingFailureDomains := 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",
+				topologyLocationLabel, deployment.Name)
+		}
+
+		if deployment.Status.ReadyReplicas < 1 {
+			if !drainingFailureDomains.Has(failureDomainName) {
+				drainingFailureDomains.Insert(failureDomainName)
+			}
+		} else {
+			if !allFailureDomains.Has(failureDomainName) {
+				allFailureDomains.Insert(failureDomainName)
+			}
+		}
+	}
+	return allFailureDomains.List(), drainingFailureDomains.List(), 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
+}
+
+func getPDBName(failureDomainType, failureDomainName string) string {
+	return k8sutil.TruncateNodeName(fmt.Sprintf("%s-%s-%s", osdPDBAppName, failureDomainType, "%s"), failureDomainName)
+}

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

@@ -0,0 +1,315 @@
+/*
+Copyright 2020 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package clusterdisruption
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/pkg/errors"
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	"github.com/rook/rook/pkg/client/clientset/versioned/scheme"
+	"github.com/rook/rook/pkg/clusterd"
+	"github.com/rook/rook/pkg/daemon/ceph/client"
+	"github.com/rook/rook/pkg/operator/ceph/disruption/controllerconfig"
+	exectest "github.com/rook/rook/pkg/util/exec/test"
+	"github.com/stretchr/testify/assert"
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
+	policyv1beta1 "k8s.io/api/policy/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+)
+
+const (
+	healthyCephStatus   = `{"fsid":"877a47e0-7f6c-435e-891a-76983ab8c509","health":{"checks":{},"status":"HEALTH_OK"},"election_epoch":12,"quorum":[0,1,2],"quorum_names":["a","b","c"],"monmap":{"epoch":3,"fsid":"877a47e0-7f6c-435e-891a-76983ab8c509","modified":"2020-11-02 09:58:23.015313","created":"2020-11-02 09:57:37.719235","min_mon_release":14,"min_mon_release_name":"nautilus","features":{"persistent":["kraken","luminous","mimic","osdmap-prune","nautilus"],"optional":[]},"mons":[{"rank":0,"name":"a","public_addrs":{"addrvec":[{"type":"v2","addr":"172.30.74.42:3300","nonce":0},{"type":"v1","addr":"172.30.74.42:6789","nonce":0}]},"addr":"172.30.74.42:6789/0","public_addr":"172.30.74.42:6789/0"},{"rank":1,"name":"b","public_addrs":{"addrvec":[{"type":"v2","addr":"172.30.101.61:3300","nonce":0},{"type":"v1","addr":"172.30.101.61:6789","nonce":0}]},"addr":"172.30.101.61:6789/0","public_addr":"172.30.101.61:6789/0"},{"rank":2,"name":"c","public_addrs":{"addrvec":[{"type":"v2","addr":"172.30.250.55:3300","nonce":0},{"type":"v1","addr":"172.30.250.55:6789","nonce":0}]},"addr":"172.30.250.55:6789/0","public_addr":"172.30.250.55:6789/0"}]},"osdmap":{"osdmap":{"epoch":19,"num_osds":3,"num_up_osds":3,"num_in_osds":3,"num_remapped_pgs":0}},"pgmap":{"pgs_by_state":[{"state_name":"active+clean","count":96}],"num_pgs":96,"num_pools":3,"num_objects":79,"data_bytes":81553681,"bytes_used":3255447552,"bytes_avail":1646011994112,"bytes_total":1649267441664,"read_bytes_sec":853,"write_bytes_sec":5118,"read_op_per_sec":1,"write_op_per_sec":0},"fsmap":{"epoch":9,"id":1,"up":1,"in":1,"max":1,"by_rank":[{"filesystem_id":1,"rank":0,"name":"ocs-storagecluster-cephfilesystem-b","status":"up:active","gid":14161},{"filesystem_id":1,"rank":0,"name":"ocs-storagecluster-cephfilesystem-a","status":"up:standby-replay","gid":24146}],"up:standby":0},"mgrmap":{"epoch":10,"active_gid":14122,"active_name":"a","active_addrs":{"addrvec":[{"type":"v2","addr":"10.131.0.28:6800","nonce":1},{"type":"v1","addr":"10.131.0.28:6801","nonce":1}]}}}`
+	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 namespace = "rook-ceph"
+
+var cephCluster = &cephv1.CephCluster{
+	ObjectMeta: metav1.ObjectMeta{Name: "ceph-cluster"}}
+
+var pdbConfigMap = &corev1.ConfigMap{
+	ObjectMeta: metav1.ObjectMeta{Name: pdbStateMapName, Namespace: namespace}}
+
+func fakeOSDDeployment(id, readyReplicas int) appsv1.Deployment {
+	osd := appsv1.Deployment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      fmt.Sprintf("osd-%d", id),
+			Namespace: namespace,
+			Labels: map[string]string{
+				"app": "rook-ceph-osd",
+			},
+		},
+		Spec: appsv1.DeploymentSpec{
+			Template: v1.PodTemplateSpec{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						"topology-location-zone": fmt.Sprintf("zone-%d", id),
+					},
+				},
+			},
+		},
+		Status: appsv1.DeploymentStatus{
+			ReadyReplicas: int32(readyReplicas),
+		},
+	}
+	return osd
+}
+
+func getFakeReconciler(t *testing.T, obj ...runtime.Object) *ReconcileClusterDisruption {
+	scheme := scheme.Scheme
+	err := policyv1beta1.AddToScheme(scheme)
+	if err != nil {
+		assert.Fail(t, "failed to build scheme")
+	}
+	err = appsv1.AddToScheme(scheme)
+	if err != nil {
+		assert.Fail(t, "failed to build scheme")
+	}
+	err = corev1.AddToScheme(scheme)
+	if err != nil {
+		assert.Fail(t, "failed to build scheme")
+	}
+	client := fake.NewFakeClientWithScheme(scheme, obj...)
+
+	return &ReconcileClusterDisruption{
+		client:     client,
+		scheme:     scheme,
+		clusterMap: &ClusterMap{clusterMap: map[string]*cephv1.CephCluster{namespace: cephCluster}},
+	}
+}
+
+func getFakeClusterInfo() *client.ClusterInfo {
+	sharedClusterMap := &ClusterMap{}
+	sharedClusterMap.UpdateClusterMap(namespace, cephCluster)
+	return sharedClusterMap.GetClusterInfo(namespace)
+
+}
+
+func TestGetOSDFailureDomains(t *testing.T) {
+	testcases := []struct {
+		label                          string
+		osds                           []appsv1.Deployment
+		expectedAllFailureDomains      []string
+		expectedDrainingFailureDomains []string
+	}{
+		{
+			label: "case 1: all osds are running",
+			osds: []appsv1.Deployment{fakeOSDDeployment(1, 1), fakeOSDDeployment(2, 1),
+				fakeOSDDeployment(3, 1)},
+			expectedAllFailureDomains:      []string{"zone-1", "zone-2", "zone-3"},
+			expectedDrainingFailureDomains: []string{},
+		},
+		{
+			label: "case 1: osd in zone-1 is pending",
+			osds: []appsv1.Deployment{fakeOSDDeployment(1, 0), fakeOSDDeployment(2, 1),
+				fakeOSDDeployment(3, 1)},
+			expectedAllFailureDomains:      []string{"zone-2", "zone-3"},
+			expectedDrainingFailureDomains: []string{"zone-1"},
+		},
+		{
+			label: "case 1: osds in zone-1 and zone-2 are pending",
+			osds: []appsv1.Deployment{fakeOSDDeployment(1, 0), fakeOSDDeployment(2, 0),
+				fakeOSDDeployment(3, 1)},
+			expectedAllFailureDomains:      []string{"zone-3"},
+			expectedDrainingFailureDomains: []string{"zone-1", "zone-2"},
+		},
+	}
+
+	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)
+	}
+}
+
+func TestGetOSDFailureDomainsError(t *testing.T) {
+	testcases := []struct {
+		label                          string
+		osds                           []appsv1.Deployment
+		expectedAllFailureDomains      []string
+		expectedDrainingFailureDomains []string
+	}{
+		{
+			label: "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{},
+		},
+	}
+
+	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)
+	}
+}
+
+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
+		fakeCephStatus                    string
+		fakeOSDDump                       string
+		allFailureDomains                 []string
+		drainingFailureDomains            []string
+		expectedOSDPDBCount               int
+		expectedMaxUnavailableCount       int
+		expectedDrainingFailureDomainName string
+	}{
+		{
+			label:                             "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{},
+			expectedOSDPDBCount:               1,
+			expectedMaxUnavailableCount:       1,
+			expectedDrainingFailureDomainName: "",
+		},
+		{
+			label:                             "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"},
+			expectedOSDPDBCount:               2,
+			expectedMaxUnavailableCount:       0,
+			expectedDrainingFailureDomainName: "zone-1",
+		},
+		{
+			label:                             "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{},
+			expectedOSDPDBCount:               2,
+			expectedMaxUnavailableCount:       0,
+			expectedDrainingFailureDomainName: "zone-1",
+		},
+		{
+			label:                             "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{},
+			expectedOSDPDBCount:               1,
+			expectedMaxUnavailableCount:       1,
+			expectedDrainingFailureDomainName: "",
+		},
+	}
+
+	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
+			}
+			if args[0] == "osd" && args[1] == "dump" {
+				return tc.fakeOSDDump, nil
+			}
+			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])
+	}
+}
+
+func TestPGHealthcheckTimeout(t *testing.T) {
+	pdbConfig := pdbConfigMap.DeepCopy()
+	r := getFakeReconciler(t, cephCluster, pdbConfig)
+	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 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)
+	}
+	r.context = &controllerconfig.Context{ClusterdContext: &clusterd.Context{Executor: executor}}
+	// 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"})
+	assert.NoError(t, err)
+	assert.Equal(t, "zone-1", pdbConfig.Data[drainingFailureDomainKey])
+
+	// 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{})
+	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])
+
+	// 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{})
+	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])
+}

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

@@ -25,12 +25,8 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/osd"
-	"github.com/rook/rook/pkg/operator/ceph/disruption/nodedrain"
-	"github.com/rook/rook/pkg/operator/k8sutil"
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
-	appsv1 "k8s.io/api/apps/v1"
-	corev1 "k8s.io/api/core/v1"
 	policyv1beta1 "k8s.io/api/policy/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -109,43 +105,6 @@ func getMinimumFailureDomain(poolList []cephv1.PoolSpec) string {
 	return osd.CRUSHMapLevelsOrdered[minfailureDomainIndex]
 }
 
-func (r *ReconcileClusterDisruption) getOngoingDrains(request reconcile.Request) ([]*corev1.Node, error) {
-	// Get the canary deployments
-	canaryDeploymentList := &appsv1.DeploymentList{}
-	operatorNameSpaceListOpts := client.InNamespace(request.Namespace)
-	err := r.client.List(context.TODO(), canaryDeploymentList, client.MatchingLabels{k8sutil.AppAttr: nodedrain.CanaryAppName}, operatorNameSpaceListOpts)
-	if err != nil {
-		return nil, errors.Wrap(err, "could not list canary deployments")
-	}
-
-	ongoingDrains := make([]*corev1.Node, 0)
-	for _, deployment := range canaryDeploymentList.Items {
-		if deployment.Status.ReadyReplicas < 1 {
-			nodeHostname, ok := deployment.Spec.Template.Spec.NodeSelector[corev1.LabelHostname]
-			if !ok {
-				logger.Errorf("could not find a the nodeSelector key %q for canary deployment %q", corev1.LabelHostname, deployment.ObjectMeta.Name)
-				continue
-			}
-
-			nodeList := &corev1.NodeList{}
-			err = r.client.List(context.TODO(), nodeList, client.MatchingLabels{corev1.LabelHostname: nodeHostname})
-			if err != nil {
-				return nil, errors.Errorf("could not get node: %s ", nodeHostname)
-			} else if len(nodeList.Items) < 1 {
-				logger.Infof("node %q does not exist, deleting corresponding drain-canary", nodeHostname)
-				err = r.client.DeleteAllOf(context.TODO(), &appsv1.Deployment{}, client.MatchingLabels{corev1.LabelHostname: nodeHostname, k8sutil.AppAttr: nodedrain.CanaryAppName})
-				if err != nil {
-					return nil, errors.Wrapf(err, "could not delete drain-canary for host %q", nodeHostname)
-				}
-			} else if len(nodeList.Items) > 1 {
-				logger.Warningf("found more than one node with %s=%s", corev1.LabelHostname, nodeHostname)
-			}
-			ongoingDrains = append(ongoingDrains, &nodeList.Items[0])
-		}
-	}
-	return ongoingDrains, nil
-}
-
 // Setting naive minAvailable for RGW at: n - 1
 func (r *ReconcileClusterDisruption) reconcileCephObjectStore(cephObjectStoreList *cephv1.CephObjectStoreList) error {
 	for _, objectStore := range cephObjectStoreList.Items {

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

@@ -25,18 +25,21 @@ import (
 
 	"github.com/coreos/pkg/capnslog"
 	cephClient "github.com/rook/rook/pkg/daemon/ceph/client"
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	"github.com/rook/rook/pkg/operator/ceph/disruption/controllerconfig"
+	"github.com/rook/rook/pkg/operator/k8sutil"
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	appsv1 "k8s.io/api/apps/v1"
+	policyv1beta1 "k8s.io/api/policy/v1beta1"
 )
 
 const (
-	osdDisruptionAppName = "rook-ceph-osd-disruption"
-	controllerName       = "clusterdisruption-controller"
+	controllerName = "clusterdisruption-controller"
 	// pdbStateMapName for the clusterdisruption pdb state map
 	pdbStateMapName = "rook-ceph-pdbstatemap"
 )
@@ -46,17 +49,20 @@ var (
 
 	// Implement reconcile.Reconciler so the controller can reconcile objects
 	_ reconcile.Reconciler = &ReconcileClusterDisruption{}
+
+	// delete legacy drain canary pods and blocking OSD podDisruptionBudgets
+	deleteLegacyResources = true
 )
 
 // 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
-	osdCrushLocationMap *OSDCrushLocationMap
-	maintenanceTimeout  time.Duration
+	scheme               *runtime.Scheme
+	client               client.Client
+	context              *controllerconfig.Context
+	clusterMap           *ClusterMap
+	maintenanceTimeout   time.Duration
+	pgHealthCheckTimeout time.Duration
 }
 
 // Reconcile reconciles a node and ensures that it has a drain-detection deployment
@@ -111,14 +117,33 @@ func (r *ReconcileClusterDisruption) reconcile(request reconcile.Request) (recon
 		// feature disabled for this cluster. not requeueing
 		return reconcile.Result{Requeue: false}, nil
 	}
-	//signal to the nodedrain controller to start
-	r.context.ReconcileCanaries.Update(true)
-	r.maintenanceTimeout = cephCluster.Spec.DisruptionManagement.OSDMaintenanceTimeout
+
+	if deleteLegacyResources {
+		// delete any legacy blocking pdbs for osd
+		err := r.deleteLegacyPDBForOSD(clusterInfo.Namespace)
+		if err != nil {
+			return reconcile.Result{}, err
+		}
+		logger.Info("deleted all legacy blocking pdbs for osds")
+
+		// delete any legacy node drain canary pods
+		err = r.deleteDrainCanaryPods(clusterInfo.Namespace)
+		if err != nil {
+			return reconcile.Result{}, err
+		}
+		logger.Info("deleted all legacy node drain canary pods")
+
+		deleteLegacyResources = false
+	}
+
+	r.maintenanceTimeout = cephCluster.Spec.DisruptionManagement.OSDMaintenanceTimeout * time.Minute
 	if r.maintenanceTimeout == 0 {
 		r.maintenanceTimeout = DefaultMaintenanceTimeout
 		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 {
@@ -148,46 +173,28 @@ func (r *ReconcileClusterDisruption) reconcile(request reconcile.Request) (recon
 		return reconcile.Result{}, nil
 	}
 
-	// get the osds with crush data populated
-	osdDataList, err := r.getOsdDataList(clusterInfo, request, poolFailureDomain)
+	// get all failure domains and the draining failure domains list
+	allFailureDomains, drainingFailureDomains, err := r.getOSDFailureDomains(clusterInfo, request, poolFailureDomain)
 	if err != nil {
 		return reconcile.Result{}, err
 	}
 
-	// get the list of nodes with ongoing drains
-	drainingNodes, err := r.getOngoingDrains(request)
+	// get the map that stores currently draining failure domain
+	pdbStateMap, err := r.initializePDBState(request)
 	if err != nil {
 		return reconcile.Result{}, err
 	}
 
-	drainingOSDs, err := getOSDsForNodes(osdDataList, drainingNodes, poolFailureDomain)
+	err = r.reconcilePDBsForOSDs(clusterInfo, request, pdbStateMap, poolFailureDomain, allFailureDomains, drainingFailureDomains)
 	if err != nil {
 		return reconcile.Result{}, err
 	}
 
-	allFailureDomainsMap, err := getFailureDomainMapForOsds(osdDataList, poolFailureDomain)
-	if err != nil {
-		logger.Error(err)
-	}
-	drainingFailureDomainsMap, err := getFailureDomainMapForOsds(drainingOSDs, poolFailureDomain)
-	if err != nil {
-		logger.Error(err)
-	}
-
-	// get the map that stores which PDBs are intentionally down
-	pdbStateMap, err := r.initializePDBState(request, osdDataList)
-	if err != nil {
-		return reconcile.Result{}, err
-	}
-
-	err = r.reconcilePDBsForOSDs(clusterInfo, request, pdbStateMap, poolFailureDomain, allFailureDomainsMap, drainingFailureDomainsMap)
-	if err != nil {
-		return reconcile.Result{}, err
-	}
-	disabledPDB, ok := pdbStateMap.Data[disabledPDBKey]
+	disabledPDB, ok := pdbStateMap.Data[drainingFailureDomainKey]
 	if ok && len(disabledPDB) > 0 {
 		return reconcile.Result{Requeue: true, RequeueAfter: 30 * time.Second}, nil
 	}
+
 	return reconcile.Result{}, nil
 }
 
@@ -247,7 +254,7 @@ func (c *ClusterMap) GetCluster(namespace string) (*cephv1.CephCluster, bool) {
 	return cluster, true
 }
 
-// GetClusterMap returns the internal clustermap for iteration purporses
+// GetClusterNamespaces returns the internal clustermap for iteration purporses
 func (c *ClusterMap) GetClusterNamespaces() []string {
 	defer c.mux.Unlock()
 	c.mux.Lock()
@@ -257,3 +264,21 @@ func (c *ClusterMap) GetClusterNamespaces() []string {
 	}
 	return namespaces
 }
+
+func (r *ReconcileClusterDisruption) deleteDrainCanaryPods(namespace string) error {
+	err := r.client.DeleteAllOf(context.TODO(), &appsv1.Deployment{}, client.InNamespace(namespace),
+		client.MatchingLabels{k8sutil.AppAttr: "rook-ceph-drain-canary"})
+	if err != nil && !kerrors.IsNotFound(err) {
+		return errors.Wrap(err, "failed to delete all the legacy drain-canary pods")
+	}
+	return nil
+}
+
+func (r *ReconcileClusterDisruption) deleteLegacyPDBForOSD(namespace string) error {
+	err := r.client.DeleteAllOf(context.TODO(), &policyv1beta1.PodDisruptionBudget{}, client.InNamespace(namespace),
+		client.MatchingLabels{k8sutil.AppAttr: "rook-ceph-osd-pdb"})
+	if err != nil && !kerrors.IsNotFound(err) {
+		return errors.Wrap(err, "failed to delete all the legacy blocking podDisruptionBugets")
+	}
+	return nil
+}

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

@@ -1,148 +0,0 @@
-/*
-Copyright 2019 The Rook Authors. All rights reserved.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-	http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package nodedrain
-
-import (
-	"reflect"
-
-	"github.com/rook/rook/pkg/operator/ceph/cluster/osd"
-	"github.com/rook/rook/pkg/operator/ceph/disruption/controllerconfig"
-	"github.com/rook/rook/pkg/operator/k8sutil"
-
-	appsv1 "k8s.io/api/apps/v1"
-	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/types"
-	"sigs.k8s.io/controller-runtime/pkg/controller"
-	"sigs.k8s.io/controller-runtime/pkg/event"
-	"sigs.k8s.io/controller-runtime/pkg/handler"
-	"sigs.k8s.io/controller-runtime/pkg/manager"
-	"sigs.k8s.io/controller-runtime/pkg/predicate"
-	"sigs.k8s.io/controller-runtime/pkg/reconcile"
-	"sigs.k8s.io/controller-runtime/pkg/source"
-)
-
-const (
-	controllerName = "nodedrain-controller"
-
-	// CanaryAppName is applied to nodedrain canary components (pods, deployments) with the key app
-	CanaryAppName = "rook-ceph-drain-canary"
-	NodeNameLabel = "node_name"
-)
-
-// Add adds a new Controller based on nodedrain.ReconcileNode and registers the relevant watches and handlers
-func Add(mgr manager.Manager, context *controllerconfig.Context) error {
-	reconcileNode := &ReconcileNode{
-		client:  mgr.GetClient(),
-		scheme:  mgr.GetScheme(),
-		context: context,
-	}
-	reconciler := reconcile.Reconciler(reconcileNode)
-	// Create a new controller
-	c, err := controller.New(controllerName, mgr, controller.Options{Reconciler: reconciler})
-	if err != nil {
-		return err
-	}
-
-	// Watch for changes to the nodes
-	pred := predicate.Funcs{
-		UpdateFunc: func(e event.UpdateEvent) bool {
-			nodeOld, ok := e.ObjectOld.DeepCopyObject().(*corev1.Node)
-			if !ok {
-				return false
-			}
-			nodeNew, ok := e.ObjectNew.DeepCopyObject().(*corev1.Node)
-			if !ok {
-				return false
-			}
-			return !reflect.DeepEqual(nodeOld.Spec, nodeNew.Spec)
-		},
-	}
-	err = c.Watch(&source.Kind{Type: &corev1.Node{}}, &handler.EnqueueRequestForObject{}, pred)
-	if err != nil {
-		return err
-	}
-	// Watch for changes to the node canary deployments
-	err = c.Watch(
-		&source.Kind{Type: &appsv1.Deployment{}},
-		&handler.EnqueueRequestsFromMapFunc{
-			ToRequests: handler.ToRequestsFunc(func(obj handler.MapObject) []reconcile.Request {
-				deployment, ok := obj.Object.(*appsv1.Deployment)
-				if !ok {
-					return []reconcile.Request{}
-				}
-				labels := deployment.GetLabels()
-				if appName, ok := labels[k8sutil.AppAttr]; !ok || appName != CanaryAppName {
-					return []reconcile.Request{}
-				}
-				nodeName, ok := deployment.Spec.Template.ObjectMeta.Labels[NodeNameLabel]
-				if !ok {
-					return []reconcile.Request{}
-				}
-				req := reconcile.Request{NamespacedName: types.NamespacedName{Name: nodeName}}
-				return []reconcile.Request{req}
-			}),
-		},
-	)
-	if err != nil {
-		return err
-	}
-
-	// Watch for changes to the osd pod nodename and enqueue their nodes
-	err = c.Watch(
-		&source.Kind{Type: &corev1.Pod{}},
-		&handler.EnqueueRequestsFromMapFunc{
-			ToRequests: handler.ToRequestsFunc(func(obj handler.MapObject) []reconcile.Request {
-				pod, ok := obj.Object.(*corev1.Pod)
-				if !ok {
-					return []reconcile.Request{}
-				}
-				labels := pod.GetLabels()
-				if appName, ok := labels[k8sutil.AppAttr]; !ok || appName != osd.AppName {
-					return []reconcile.Request{}
-				}
-				nodeName := pod.Spec.NodeName
-				if len(nodeName) < 1 {
-					return []reconcile.Request{}
-				}
-				req := reconcile.Request{NamespacedName: types.NamespacedName{Name: nodeName}}
-				return []reconcile.Request{req}
-			}),
-		},
-		predicate.Funcs{
-			UpdateFunc: func(event event.UpdateEvent) bool {
-				oldPod, ok := event.ObjectOld.(*corev1.Pod)
-				if !ok {
-					return false
-				}
-				newPod, ok := event.ObjectNew.(*corev1.Pod)
-				if !ok {
-					return false
-				}
-				// only enqueue if the nodename has changed
-				if oldPod.Spec.NodeName == newPod.Spec.NodeName {
-					return false
-				}
-				return true
-			},
-		},
-	)
-	if err != nil {
-		return err
-	}
-
-	return nil
-}

pkg/operator/ceph/disruption/nodedrain/doc.go

@@ -1,22 +0,0 @@
-/*
-Copyright 2019 The Rook Authors. All rights reserved.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-	http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-/*
-Package nodedrain implements the controller for ensuring that drain detection deployments exist.
-The design and purpose for drain detection is found at:
-https://github.com/rook/rook/blob/master/design/ceph/ceph-managed-disruptionbudgets.md
-*/
-package nodedrain

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

@@ -1,291 +0,0 @@
-/*
-Copyright 2019 The Rook Authors. All rights reserved.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-	http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package nodedrain
-
-import (
-	"context"
-	"fmt"
-	"os"
-	"time"
-
-	kerrors "k8s.io/apimachinery/pkg/api/errors"
-
-	"github.com/coreos/pkg/capnslog"
-	"github.com/pkg/errors"
-
-	"github.com/rook/rook/pkg/operator/ceph/cluster/osd"
-	"github.com/rook/rook/pkg/operator/ceph/disruption/controllerconfig"
-	"github.com/rook/rook/pkg/operator/k8sutil"
-
-	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/types"
-	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
-	"sigs.k8s.io/controller-runtime/pkg/reconcile"
-
-	appsv1 "k8s.io/api/apps/v1"
-	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-)
-
-var (
-	logger = capnslog.NewPackageLogger("github.com/rook/rook", "nodedrain-controller")
-	// Implement reconcile.Reconciler so the controller can reconcile objects
-	_ reconcile.Reconciler = &ReconcileNode{}
-)
-
-// ReconcileNode reconciles ReplicaSets
-type ReconcileNode struct {
-	// client can be used to retrieve objects from the APIServer.
-	scheme  *runtime.Scheme
-	client  client.Client
-	context *controllerconfig.Context
-}
-
-// Reconcile reconciles a node and ensures that it has a drain-detection deployment
-// attached to it.
-// The Controller will requeue the Request to be processed again if an error is non-nil or
-// Result.Requeue is true, otherwise upon completion it will remove the work from the queue.
-func (r *ReconcileNode) Reconcile(request reconcile.Request) (reconcile.Result, error) {
-	// workaround because the rook logging mechanism is not compatible with the controller-runtime logging interface
-	result, err := r.reconcile(request)
-	if err != nil {
-		logger.Error(err)
-	}
-	return result, err
-}
-
-func (r *ReconcileNode) reconcile(request reconcile.Request) (reconcile.Result, error) {
-
-	if !r.context.ReconcileCanaries.Get() {
-		return reconcile.Result{}, nil
-	}
-
-	logger.Debugf("reconciling node: %q", request.Name)
-
-	// Create or Update the deployment default/foo
-	deploy := &appsv1.Deployment{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      k8sutil.TruncateNodeName(fmt.Sprintf("%s-%%s", CanaryAppName), request.Name),
-			Namespace: r.context.OperatorNamespace,
-		},
-	}
-
-	node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: request.Name}}
-
-	err := r.client.Get(context.TODO(), request.NamespacedName, node)
-	if kerrors.IsNotFound(err) {
-		// delete any canary deployments if the node doesn't exist
-		logger.Infof("deleting drain-canary deployment for deleted node %q", request.Name)
-		err = r.client.Delete(context.TODO(), deploy)
-		if kerrors.IsNotFound(err) {
-			return reconcile.Result{}, nil
-		} else if err != nil {
-			return reconcile.Result{}, errors.Wrapf(err, "could not delete drain-canary deployment for deleted node %q", request.Name)
-		}
-		return reconcile.Result{}, nil
-	} else if err != nil {
-		return reconcile.Result{}, errors.Errorf("could not get node %q", request.NamespacedName)
-	}
-
-	nodeHostnameLabel, ok := node.ObjectMeta.Labels[corev1.LabelHostname]
-	if !ok {
-		return reconcile.Result{}, errors.Errorf("Label key %q does not exist on node %q", corev1.LabelHostname, request.NamespacedName)
-	}
-
-	osdPodList := &corev1.PodList{}
-	err = r.client.List(context.TODO(), osdPodList, client.MatchingLabels{k8sutil.AppAttr: osd.AppName})
-	if err != nil {
-		return reconcile.Result{}, errors.Wrap(err, "could not list the osd pods")
-	}
-
-	// map with tolerations as keys and empty struct as values for uniqueness
-	uniqueTolerations := controllerconfig.TolerationSet{}
-
-	occupiedByOSD := false
-	for _, osdPod := range osdPodList.Items {
-		if osdPod.Spec.NodeName == request.Name {
-			labels := osdPod.GetLabels()
-			deploymentList := &appsv1.DeploymentList{}
-			labelSelector := map[string]string{
-				osd.OsdIdLabelKey: labels[osd.OsdIdLabelKey],
-				k8sutil.AppAttr:   osd.AppName,
-			}
-			var deployment appsv1.Deployment
-			err := r.client.List(context.TODO(), deploymentList, client.MatchingLabels(labelSelector), client.InNamespace(osdPod.GetNamespace()))
-			if err != nil || len(deploymentList.Items) < 1 {
-				logger.Errorf("cannot find deployment for osd id %q in namespace %q", labels[osd.OsdIdLabelKey], osdPod.GetNamespace())
-			} else {
-				deployment = deploymentList.Items[0]
-			}
-			if len(deploymentList.Items) > 1 {
-				logger.Errorf("found multiple deployments for osd id %q in namespace %q: %+v", labels[osd.OsdIdLabelKey], osdPod.GetNamespace(), deploymentList)
-			}
-
-			occupiedByOSD = true
-			// get the osd tolerations
-			for _, osdToleration := range deployment.Spec.Template.Spec.Tolerations {
-				if osdToleration.Key == "node.kubernetes.io/unschedulable" {
-					logger.Errorf(
-						"osd %q in namespace %q tolerates the drain taint, but the drain canary will not.",
-						labels[osd.OsdIdLabelKey],
-						labels[k8sutil.ClusterAttr],
-					)
-				} else {
-					uniqueTolerations.Add(osdToleration)
-				}
-			}
-		}
-	}
-
-	// CreateOrUpdate the deployment
-	mutateFunc := func() error {
-
-		// labels for the pod, the deployment, and the deploymentSelector
-		selectorLabels := map[string]string{
-			corev1.LabelHostname: nodeHostnameLabel,
-			k8sutil.AppAttr:      CanaryAppName,
-			NodeNameLabel:        node.GetName(),
-		}
-
-		nodeSelector := map[string]string{corev1.LabelHostname: nodeHostnameLabel}
-
-		// Deployment selector is immutable so we set this value only if
-		// a new object is going to be created
-		if deploy.ObjectMeta.CreationTimestamp.IsZero() {
-			deploy.Spec.Selector = &metav1.LabelSelector{
-				MatchLabels: selectorLabels,
-			}
-		}
-
-		ownerReferences := []metav1.OwnerReference{}
-		// get the operator Deployment to use as an owner reference
-		operatorPodKey := types.NamespacedName{Name: os.Getenv(k8sutil.PodNameEnvVar), Namespace: r.context.OperatorNamespace}
-		operatorDeployment, err := getDeploymentForPod(r.client, operatorPodKey)
-		if err != nil {
-			logger.Errorf("could not find rook operator deployment for pod %+v. %v", operatorPodKey, err)
-		} else {
-			operatorDeployment := operatorDeployment
-			controllerBool := true
-			operatorOwnerRef := metav1.OwnerReference{
-				APIVersion: operatorDeployment.APIVersion,
-				Kind:       operatorDeployment.Kind,
-				UID:        operatorDeployment.GetUID(),
-				Name:       operatorDeployment.GetName(),
-				Controller: &controllerBool,
-			}
-			ownerReferences = append(ownerReferences, operatorOwnerRef)
-		}
-		deploy.ObjectMeta.OwnerReferences = ownerReferences
-
-		// update the deployment labels
-		topology := osd.ExtractOSDTopologyFromLabels(node.GetLabels())
-		for key, value := range topology {
-			selectorLabels[key] = value
-		}
-		deploy.ObjectMeta.Labels = selectorLabels
-
-		// update the Deployment pod template
-		deploy.Spec.Template = corev1.PodTemplateSpec{
-			ObjectMeta: metav1.ObjectMeta{Labels: selectorLabels},
-			Spec: corev1.PodSpec{
-				NodeSelector: nodeSelector,
-				Containers:   newDoNothingContainers(r.context.RookImage),
-				Tolerations:  uniqueTolerations.ToList(),
-			},
-		}
-
-		return nil
-	}
-	if occupiedByOSD {
-		op, err := controllerutil.CreateOrUpdate(context.TODO(), r.client, deploy, mutateFunc)
-		if err != nil {
-			return reconcile.Result{}, errors.Wrapf(err, "node reconcile failed on op %q", op)
-		}
-		msg := fmt.Sprintf("deployment successfully reconciled for node %q. operation: %q", request.Name, op)
-		if op == controllerutil.OperationResultCreated {
-			logger.Info(msg)
-		} else {
-			logger.Debug(msg)
-		}
-	} else {
-		logger.Debugf("not watching for drains on node %q as there are no osds running there.", request.Name)
-
-		// get the canary deployment
-		canarayDeployment := &appsv1.Deployment{}
-		key := types.NamespacedName{Name: deploy.GetName(), Namespace: deploy.GetNamespace()}
-		err := r.client.Get(context.TODO(), key, canarayDeployment)
-		if err != nil && !kerrors.IsNotFound(err) {
-			return reconcile.Result{}, errors.Wrapf(err, "could not fetch deployment %q", key)
-		} else if kerrors.IsNotFound(err) {
-			return reconcile.Result{}, nil
-		}
-
-		// delete the canary deployments that aren't triggered by drains, but are on nodes that aren't occupied by OSDs
-		// if it's on a draining node, we don't want to kill the canary.
-		if canarayDeployment.Status.ReadyReplicas > 0 {
-			logger.Infof("deleting drain-canary deployment on node %q as OSDs are no longer scheduled here", request.Name)
-			err := r.client.Delete(context.TODO(), canarayDeployment)
-			if err != nil {
-				return reconcile.Result{}, errors.Wrapf(err, "could not delete deployment %q", key)
-			}
-		}
-	}
-	return reconcile.Result{Requeue: true, RequeueAfter: time.Minute * 15}, nil
-}
-
-// returns a container that does nothing
-func newDoNothingContainers(rookImage string) []corev1.Container {
-	return []corev1.Container{{
-		Image:   rookImage,
-		Name:    "sleep",
-		Command: []string{"/tini"},
-		Args:    []string{"--", "sleep", "infinity"},
-	}}
-}
-
-func getDeploymentForPod(client client.Client, podKey types.NamespacedName) (*appsv1.Deployment, error) {
-	// get pod
-	operatorPod := &corev1.Pod{}
-	err := client.Get(context.TODO(), podKey, operatorPod)
-	if err != nil {
-		return nil, errors.Wrapf(err, "could not get pod %+v", podKey)
-	} else {
-		for _, rsRef := range operatorPod.ObjectMeta.OwnerReferences {
-			if *rsRef.Controller {
-				// get rs
-				replicaSet := &appsv1.ReplicaSet{}
-				replicaSetKey := types.NamespacedName{Name: rsRef.Name, Namespace: podKey.Namespace}
-				err := client.Get(context.TODO(), replicaSetKey, replicaSet)
-				if err != nil {
-					return nil, errors.Wrapf(err, "could not get replicaset %+v", replicaSetKey)
-				} else {
-					for _, depRef := range replicaSet.ObjectMeta.OwnerReferences {
-						if *depRef.Controller {
-							// get deployment
-							operatorDeployment := &appsv1.Deployment{}
-							operatorDeploymentKey := types.NamespacedName{Name: depRef.Name, Namespace: podKey.Namespace}
-							err := client.Get(context.TODO(), operatorDeploymentKey, operatorDeployment)
-							return operatorDeployment, err
-						}
-					}
-				}
-			}
-		}
-	}
-	return nil, errors.Errorf("could not get deployment for pod %+v", podKey)
-}

tests/framework/installer/ceph_manifests.go

@@ -150,6 +150,8 @@ spec:
                       type: boolean
                     osdMaintenanceTimeout:
                       type: integer
+                    pgHealthCheckTimeout:
+                      type: integer
                     manageMachineDisruptionBudgets:
                       type: boolean
                 skipUpgradeChecks:
@@ -232,6 +234,8 @@ spec:
                           type: boolean
                         osdMaintenanceTimeout:
                           type: integer
+                        pgHealthCheckTimeout:
+                          type: integer
                         manageMachineDisruptionBudgets:
                           type: boolean
                     useAllNodes:
@@ -1453,6 +1457,8 @@ spec:
                   type: boolean
                 osdMaintenanceTimeout:
                   type: integer
+                pgHealthCheckTimeout:
+                  type: integer
                 manageMachineDisruptionBudgets:
                   type: boolean
             skipUpgradeChecks:
@@ -1492,6 +1498,8 @@ spec:
                       type: boolean
                     osdMaintenanceTimeout:
                       type: integer
+                    pgHealthCheckTimeout:
+                      type: integer
                     manageMachineDisruptionBudgets:
                       type: boolean
                 useAllNodes:
@@ -3426,6 +3434,7 @@ spec:
   disruptionManagement:
     managePodBudgets: false
     osdMaintenanceTimeout: 30
+    pgHealthCheckTimeout: 0
     manageMachineDisruptionBudgets: false
     machineDisruptionBudgetNamespace: openshift-machine-api`
 	}