From aee8f568534d50a290f5c7fb74845c1407a4934f Mon Sep 17 00:00:00 2001 From: Santosh Pillai Date: Thu, 22 Apr 2021 15:54:52 +0530 Subject: [PATCH] ceph: update podDisruptionBudget controller for OSD This PR has following changes around OSD pdbs - Detect node drains more reliably. When OSD is down and OSD pod is not scheduled to any node or if the scheduled node is not ready, then assume that node is draining. - Don't set no-out flag on the failure domain if the OSD is down due to reasons other than node drain (say disk failure) - update unit tests - update design doc to reflect above changes Signed-off-by: Santosh Pillai --- design/ceph/ceph-managed-disruptionbudgets.md | 4 +- .../ceph/disruption/clusterdisruption/osd.go | 243 +++++++++++---- .../disruption/clusterdisruption/osd_test.go | 286 +++++++++++++----- .../disruption/clusterdisruption/reconcile.go | 17 +- 4 files changed, 394 insertions(+), 156 deletions(-) diff --git a/design/ceph/ceph-managed-disruptionbudgets.md b/design/ceph/ceph-managed-disruptionbudgets.md index 079491572..236009e74 100644 --- a/design/ceph/ceph-managed-disruptionbudgets.md +++ b/design/ceph/ceph-managed-disruptionbudgets.md @@ -18,7 +18,7 @@ Even if an upgrade agent were only to drain one node at a time, Ceph would have The failure domain will be determined by the smallest failure domain of all the Ceph Pools in that cluster. We begin with creating a single PodDisruptionBudget for all the OSD with maxUnavailable=1. This will allow one OSD to go down anytime. Once the user drains a node and an OSD goes down, we determine the failure domain for the draining OSD (using the OSD deployment labels). Then we create blocking PodDisruptionBudgets (maxUnavailable=0) for all other failure domains and delete the main PodDisruptionBudget. This blocks OSDs from going down in multiple failure domains simultaneously. -Once the drained OSDs are back and all the pgs are active+clean, that is, the cluster is healed, the main PodDisruptionBudget is added back and the blocking ones are deleted. User can also add a timeout for the pgs to become healthy. If the timeout exceeds, the operator will ignore the pg health, add the main PodDisruptionBudget and delete the blocking ones. +Once the drained OSDs are back and all the pgs are active+clean, that is, the cluster is healed, the default PodDisruptionBudget (with maxUnavailable=1) is added back and the blocking ones are deleted. User can also add a timeout for the pgs to become healthy. If the timeout exceeds, the operator will ignore the pg health, add the main PodDisruptionBudget and delete the blocking ones. Detecting drains is not easy as they are a client side operation. The client cordons the node and continuously attempts to evict all pods from the node until it succeeds. Whenever an OSD goes into pending state, that is, `ReadyReplicas` count is 0, we assume that some drain operation is happening. @@ -52,6 +52,8 @@ a common way of deploying kubernetes. This will also work to mitigate manual dra When an node is drained, we will also delay it's DOWN/OUT process by placing a noout on that node. We will remove that noout after a timeout. +An OSD can be down due to reasons other than node drain, say, disk failure. In such a situation, if the pgs are unhealthy then rook will create a blocking PodDisruptionBudget on other failure domains to prevent further node drains on them. `noout` flag won't be set on node this is case. If the OSD is down but all the pgs are `active+clean`, the cluster will be treated as fully healthy. The default PodDisruptionBudget (with maxUnavailable=1) will be added back and the blocking ones will be deleted. + ### Mon, Mgr, MDS, RGW, RBDMirror Since there is no strict failure domain requirement for each of these, and they are not logically grouped, a stactic PDB will suffice. diff --git a/pkg/operator/ceph/disruption/clusterdisruption/osd.go b/pkg/operator/ceph/disruption/clusterdisruption/osd.go index 23fc3dba9..c9014a7b6 100644 --- a/pkg/operator/ceph/disruption/clusterdisruption/osd.go +++ b/pkg/operator/ceph/disruption/clusterdisruption/osd.go @@ -46,6 +46,7 @@ const ( osdPDBAppName = "rook-ceph-osd" drainingFailureDomainKey = "draining-failure-domain" drainingFailureDomainDurationKey = "draining-failure-domain-duration" + setNoOut = "set-no-out" pgHealthCheckDurationKey = "pg-health-check-duration" // DefaultMaintenanceTimeout is the period for which a drained failure domain will remain in noout DefaultMaintenanceTimeout = 30 * time.Minute @@ -68,9 +69,9 @@ func (r *ReconcileClusterDisruption) deletePDB(pdb *policyv1beta1.PodDisruptionB return nil } -// createOverallPDBforOSD creates a single PDB for all OSDs with maxUnavailable=1 +// createDefaultPDBforOSD creates a single PDB for all OSDs with maxUnavailable=1 // This allows all OSDs in a single failure domain to go down. -func (r *ReconcileClusterDisruption) createOverallPDBforOSD(namespace string) error { +func (r *ReconcileClusterDisruption) createDefaultPDBforOSD(namespace string) error { cephCluster, ok := r.clusterMap.GetCluster(namespace) if !ok { return errors.Errorf("failed to find the namespace %q in the clustermap", namespace) @@ -97,8 +98,8 @@ func (r *ReconcileClusterDisruption) createOverallPDBforOSD(namespace string) er err = r.client.Get(context.TODO(), pdbRequest, &policyv1beta1.PodDisruptionBudget{}) if err != nil { if apierrors.IsNotFound(err) { - logger.Info("all osds are up. pg health is active+clean") - logger.Infof("creating the main pdb %q with maxUnavailable=1 for all osd", osdPDBAppName) + logger.Info("all PGs are active+clean. Restoring default OSD pdb settings") + logger.Infof("creating the default pdb %q with maxUnavailable=1 for all osd", osdPDBAppName) return r.createPDB(pdb) } return errors.Wrapf(err, "failed to get pdb %q", pdb.Name) @@ -106,7 +107,7 @@ func (r *ReconcileClusterDisruption) createOverallPDBforOSD(namespace string) er return nil } -func (r *ReconcileClusterDisruption) deleteOverallPDBforOSD(namespace string) error { +func (r *ReconcileClusterDisruption) deleteDefaultPDBforOSD(namespace string) error { pdbRequest := types.NamespacedName{Name: osdPDBAppName, Namespace: namespace} pdb := &policyv1beta1.PodDisruptionBudget{ ObjectMeta: metav1.ObjectMeta{ @@ -121,7 +122,7 @@ func (r *ReconcileClusterDisruption) deleteOverallPDBforOSD(namespace string) er } return errors.Wrapf(err, "failed to get pdb %q", pdb.Name) } - logger.Infof("deleting the main pdb %q with maxUnavailable=1 for all osd", osdPDBAppName) + logger.Infof("deleting the default pdb %q with maxUnavailable=1 for all osd", osdPDBAppName) return r.deletePDB(pdb) } @@ -198,7 +199,7 @@ func (r *ReconcileClusterDisruption) initializePDBState(request reconcile.Reques if apierrors.IsNotFound(err) { // create configmap to track the draining failure domain - pdbStateMap.Data = map[string]string{drainingFailureDomainKey: ""} + pdbStateMap.Data = map[string]string{drainingFailureDomainKey: "", setNoOut: ""} err := r.client.Create(context.TODO(), pdbStateMap) if err != nil { return pdbStateMap, errors.Wrapf(err, "failed to create the PDB state map %q", pdbStateMapRequest) @@ -215,74 +216,100 @@ func (r *ReconcileClusterDisruption) reconcilePDBsForOSDs( pdbStateMap *corev1.ConfigMap, failureDomainType string, allFailureDomains, - drainingFailureDomains []string, -) error { + osdDownFailureDomains []string, + activeNodeDrains bool, +) (reconcile.Result, error) { + var osdDown bool + var drainingFailureDomain string + if len(osdDownFailureDomains) > 0 { + osdDown = true + drainingFailureDomain = osdDownFailureDomains[0] + } - pgHealthMsg, clean, err := cephclient.IsClusterClean(r.context.ClusterdContext, clusterInfo) + pgHealthMsg, pgClean, err := cephclient.IsClusterClean(r.context.ClusterdContext, clusterInfo) if err != nil { // If the error contains that message, this means the cluster is not up and running // No monitors are present and thus no ceph configuration has been created if strings.Contains(err.Error(), opcontroller.UninitializedCephConfigError) { - logger.Debugf("Ceph %q cluster not ready, cannot check Ceph status yet.", request.Namespace) - return nil + logger.Infof("Ceph %q cluster not ready, cannot check Ceph status yet.", request.Namespace) + return opcontroller.WaitForRequeueIfOperatorNotInitialized, nil } - return errors.Wrapf(err, "failed to check cluster health") + return reconcile.Result{}, errors.Wrapf(err, "failed to check cluster health") } - if pdbStateMap.Data == nil { - pdbStateMap.Data = make(map[string]string) - } - _, ok := pdbStateMap.Data[drainingFailureDomainKey] - if !ok { - pdbStateMap.Data[drainingFailureDomainKey] = "" - } - - var drainingFailureDomain string - activeDrains := len(drainingFailureDomains) > 0 - - // create blocking PDBs for all the OSDs that are not being drained right now - if activeDrains { - drainingFailureDomain = drainingFailureDomains[0] - logger.Infof("all %q failure domains: %v. currently draining: %q. pg health: %q", failureDomainType, - allFailureDomains, drainingFailureDomain, pgHealthMsg) - err := r.handleActiveDrains(allFailureDomains, drainingFailureDomain, failureDomainType, clusterInfo.Namespace, clean) + switch { + // osd is down but pgs are active+clean + case osdDown && pgClean: + lastDrainTimeStamp, err := getLastDrainTimeStamp(pdbStateMap, drainingFailureDomainDurationKey) if err != nil { - return errors.Wrap(err, "failed to handle active drains") + return reconcile.Result{}, errors.Wrapf(err, "failed to get last drain timestamp from the configmap %q", pdbStateMap.Name) } + timeSinceOSDDown := time.Since(lastDrainTimeStamp) + if timeSinceOSDDown > 30*time.Second { + logger.Infof("osd is down in failure domain %q is down for the last %.2f minutes, but pgs are active+clean", drainingFailureDomain, timeSinceOSDDown.Minutes()) + resetPDBConfig(pdbStateMap) + } else { + logger.Infof("osd is down in the failure domain %q, but pgs are active+clean. Requeuing in case pg status is not updated yet...", drainingFailureDomain) + return reconcile.Result{Requeue: true, RequeueAfter: 15 * time.Second}, nil + } + + // osd is down and pgs are not healthy + case osdDown && !pgClean: + logger.Infof("osd is down in failure domain %q and pgs are not active+clean. pg health: %q", drainingFailureDomain, pgHealthMsg) currentlyDrainingFD, ok := pdbStateMap.Data[drainingFailureDomainKey] if !ok || drainingFailureDomain != currentlyDrainingFD { pdbStateMap.Data[drainingFailureDomainKey] = drainingFailureDomain pdbStateMap.Data[drainingFailureDomainDurationKey] = time.Now().Format(time.RFC3339) } - } else if clean || r.hasPGHealthCheckTimedout(pdbStateMap) { + if activeNodeDrains { + pdbStateMap.Data[setNoOut] = "true" + } + + // osd is back up and either pgs have become healthy or pg healthy check timeout has elapsed + case !osdDown && (pgClean || r.hasPGHealthCheckTimedout(pdbStateMap)): // reset the configMap if cluster is clean or if the timeout for PGs to become active+clean has exceeded - logger.Debugf("no drains detected in %q failure domains: %v. pg health: %q", failureDomainType, allFailureDomains, pgHealthMsg) - pdbStateMap.Data[drainingFailureDomainKey] = "" - delete(pdbStateMap.Data, drainingFailureDomainDurationKey) - delete(pdbStateMap.Data, pgHealthCheckDurationKey) - } else { - logger.Infof("all osds are up. last drained failure domain: %q. pg health: %q. waiting for PGs to be active+clean or PG health check to timeout", - pdbStateMap.Data[drainingFailureDomainKey], pgHealthMsg) + logger.Debugf("no OSD is down in the %q failure domains: %v. pg health: %q", failureDomainType, allFailureDomains, pgHealthMsg) + resetPDBConfig(pdbStateMap) + + default: + logger.Infof("all %q failure domains: %v. osd is down in failure domain: %q. active node drains: %t. pg health: %q", failureDomainType, + allFailureDomains, drainingFailureDomain, activeNodeDrains, pgHealthMsg) } - err = r.updateNoout(clusterInfo, pdbStateMap, allFailureDomains) - if err != nil { - logger.Errorf("failed to update maintenance noout in cluster %q. %v", request, err) + if pdbStateMap.Data[setNoOut] == "true" { + err = r.updateNoout(clusterInfo, pdbStateMap, allFailureDomains) + if err != nil { + logger.Errorf("failed to update maintenance noout in cluster %q. %v", request, err) + } + } + + if pdbStateMap.Data[drainingFailureDomainKey] != "" && !pgClean { + // delete default OSD pdb and create blocking OSD pdbs + err := r.handleActiveDrains(allFailureDomains, pdbStateMap.Data[drainingFailureDomainKey], failureDomainType, clusterInfo.Namespace, pgClean) + if err != nil { + return reconcile.Result{}, errors.Wrap(err, "failed to handle active drains") + } + } else if pdbStateMap.Data[drainingFailureDomainKey] == "" { + // delete all blocking OSD pdb and restore the default OSD pdb + err := r.handleInactiveDrains(allFailureDomains, failureDomainType, clusterInfo.Namespace) + if err != nil { + return reconcile.Result{}, errors.Wrap(err, "failed to handle inactive drains") + } + // reset `set-no-out` flag on the configMap + pdbStateMap.Data[setNoOut] = "" } err = r.client.Update(context.TODO(), pdbStateMap) if err != nil { - return errors.Wrapf(err, "failed to update configMap %q in cluster %q", pdbStateMapName, request) + return reconcile.Result{}, errors.Wrapf(err, "failed to update configMap %q in cluster %q", pdbStateMapName, request) } - // handle inactive drains by deleting all blocking pdb and restoring the main pdb - if pdbStateMap.Data[drainingFailureDomainKey] == "" { - err := r.handleInactiveDrains(allFailureDomains, clusterInfo.Namespace, failureDomainType) - if err != nil { - return errors.Wrap(err, "failed to handle inactive drains") - } + // requeue if drain is still in progress + if len(pdbStateMap.Data[drainingFailureDomainKey]) > 0 { + return reconcile.Result{Requeue: true, RequeueAfter: 30 * time.Second}, nil } - return nil + + return reconcile.Result{}, nil } func (r *ReconcileClusterDisruption) handleActiveDrains(allFailureDomains []string, drainingFailureDomain, @@ -305,20 +332,20 @@ func (r *ReconcileClusterDisruption) handleActiveDrains(allFailureDomains []stri } } - // delete the main PDB for OSD + // delete the default PDB for OSD // This will allow all OSDs in the currently drained failure domain to be removed. - logger.Debug("deleting main pdb with maxUnavailable=1 for all osd") - err := r.deleteOverallPDBforOSD(namespace) + logger.Debug("deleting default pdb with maxUnavailable=1 for all osd") + err := r.deleteDefaultPDBforOSD(namespace) if err != nil { - return errors.Wrap(err, "failed to delete the main osd pdb") + return errors.Wrap(err, "failed to delete the default osd pdb") } return nil } -func (r *ReconcileClusterDisruption) handleInactiveDrains(allFailureDomains []string, namespace, failureDomainType string) error { - err := r.createOverallPDBforOSD(namespace) +func (r *ReconcileClusterDisruption) handleInactiveDrains(allFailureDomains []string, failureDomainType, namespace string) error { + err := r.createDefaultPDBforOSD(namespace) if err != nil { - return errors.Wrap(err, "failed to create main pdb") + return errors.Wrap(err, "failed to create default pdb") } for _, failureDomainName := range allFailureDomains { err := r.deleteBlockingPDBForOSD(namespace, failureDomainType, failureDomainName) @@ -375,37 +402,51 @@ func (r *ReconcileClusterDisruption) updateNoout(clusterInfo *cephclient.Cluster return nil } -func (r *ReconcileClusterDisruption) getOSDFailureDomains(clusterInfo *cephclient.ClusterInfo, request reconcile.Request, poolFailureDomain string) ([]string, []string, error) { +func (r *ReconcileClusterDisruption) getOSDFailureDomains(clusterInfo *cephclient.ClusterInfo, request reconcile.Request, poolFailureDomain string) ([]string, []string, []string, error) { osdDeploymentList := &appsv1.DeploymentList{} namespaceListOpts := client.InNamespace(request.Namespace) topologyLocationLabel := fmt.Sprintf(osd.TopologyLocationLabel, poolFailureDomain) err := r.client.List(context.TODO(), osdDeploymentList, client.MatchingLabels{k8sutil.AppAttr: osd.AppName}, namespaceListOpts) if err != nil { - return []string{}, []string{}, errors.Wrap(err, "failed to list osd deployments") + return nil, nil, nil, errors.Wrap(err, "failed to list osd deployments") } allFailureDomains := sets.NewString() - drainingFailureDomains := sets.NewString() + nodeDrainFailureDomains := sets.NewString() + osdDownFailureDomains := sets.NewString() for _, deployment := range osdDeploymentList.Items { labels := deployment.Spec.Template.ObjectMeta.GetLabels() failureDomainName := labels[topologyLocationLabel] if failureDomainName == "" { - return []string{}, []string{}, errors.Errorf("failed to get the topology location label %q in OSD deployment %q", + return nil, nil, nil, errors.Errorf("failed to get the topology location label %q in OSD deployment %q", topologyLocationLabel, deployment.Name) } + // Assume node drain if osd deployment ReadyReplicas count is 0 and OSD pod is not scheduled on a node if deployment.Status.ReadyReplicas < 1 { - if !drainingFailureDomains.Has(failureDomainName) { - drainingFailureDomains.Insert(failureDomainName) + if !osdDownFailureDomains.Has(failureDomainName) { + osdDownFailureDomains.Insert(failureDomainName) } - } else { - if !allFailureDomains.Has(failureDomainName) { - allFailureDomains.Insert(failureDomainName) + isDrained, err := hasOSDNodeDrained(r.client, request.Namespace, labels[osd.OsdIdLabelKey]) + if err != nil { + return nil, nil, nil, errors.Wrapf(err, "failed to check if osd %q node is drained", deployment.Name) + } + if isDrained { + logger.Infof("osd %q is down and a possible node drain is detected", deployment.Name) + if !nodeDrainFailureDomains.Has(failureDomainName) { + nodeDrainFailureDomains.Insert(failureDomainName) + } + } else { + logger.Infof("osd %q is down but no node drain is detected", deployment.Name) } } + + if !allFailureDomains.Has(failureDomainName) { + allFailureDomains.Insert(failureDomainName) + } } - return allFailureDomains.List(), drainingFailureDomains.List(), nil + return allFailureDomains.List(), nodeDrainFailureDomains.List(), osdDownFailureDomains.List(), nil } func (r *ReconcileClusterDisruption) hasPGHealthCheckTimedout(pdbStateMap *corev1.ConfigMap) bool { @@ -435,6 +476,74 @@ func (r *ReconcileClusterDisruption) hasPGHealthCheckTimedout(pdbStateMap *corev return false } +// hasNodeDrained returns true if OSD pod is not assigned to any node or if the OSD node is not schedulable +func hasOSDNodeDrained(c client.Client, namespace, osdID string) (bool, error) { + osdNodeName, err := getOSDNodeName(c, namespace, osdID) + if err != nil { + return false, errors.Wrapf(err, "failed to get node name assigned to OSD %q POD", osdID) + } + + if osdNodeName == "" { + logger.Debugf("osd %q POD is not assigned to any node. assuming node drain", osdID) + return true, nil + } + + node, err := getNode(c, osdNodeName) + if err != nil { + return false, errors.Wrapf(err, "failed to get node assigned to OSD %q POD", osdID) + } + return node.Spec.Unschedulable, nil +} + +func getOSDNodeName(c client.Client, namespace, osdID string) (string, error) { + pods := &corev1.PodList{} + listOpts := []client.ListOption{ + client.InNamespace(namespace), + client.MatchingLabels{osd.OsdIdLabelKey: osdID}, + } + + err := c.List(context.TODO(), pods, listOpts...) + if err != nil { + return "", errors.Wrapf(err, "failed to list pods for osd %q", osdID) + } + + if len(pods.Items) > 0 { + return pods.Items[0].Spec.NodeName, nil + } + return "", nil +} + +func getNode(c client.Client, nodeName string) (*corev1.Node, error) { + node := &corev1.Node{} + err := c.Get(context.TODO(), types.NamespacedName{Name: nodeName}, node) + if err != nil { + return nil, errors.Wrapf(err, "failed to get node %q", nodeName) + } + return node, nil +} + func getPDBName(failureDomainType, failureDomainName string) string { return k8sutil.TruncateNodeName(fmt.Sprintf("%s-%s-%s", osdPDBAppName, failureDomainType, "%s"), failureDomainName) } + +func getLastDrainTimeStamp(pdbStateMap *corev1.ConfigMap, key string) (time.Time, error) { + var err error + var lastDrainTimeStamp time.Time + lastDrainTimeStampString, ok := pdbStateMap.Data[key] + if !ok || len(lastDrainTimeStampString) == 0 { + return time.Now(), nil + } else { + lastDrainTimeStamp, err = time.Parse(time.RFC3339, pdbStateMap.Data[key]) + if err != nil { + return time.Time{}, errors.Wrapf(err, "failed to parse timestamp %q", pdbStateMap.Data[key]) + } + } + + return lastDrainTimeStamp, nil +} + +func resetPDBConfig(pdbStateMap *corev1.ConfigMap) { + pdbStateMap.Data[drainingFailureDomainKey] = "" + delete(pdbStateMap.Data, drainingFailureDomainDurationKey) + delete(pdbStateMap.Data, pgHealthCheckDurationKey) +} diff --git a/pkg/operator/ceph/disruption/clusterdisruption/osd_test.go b/pkg/operator/ceph/disruption/clusterdisruption/osd_test.go index 53703bfed..056bea1e8 100644 --- a/pkg/operator/ceph/disruption/clusterdisruption/osd_test.go +++ b/pkg/operator/ceph/disruption/clusterdisruption/osd_test.go @@ -45,13 +45,26 @@ const ( unHealthyCephStatus = `{"fsid":"613975f3-3025-4802-9de1-a2280b950e75","health":{"checks":{"OSD_DOWN":{"severity":"HEALTH_WARN","summary":{"message":"1 osds down"}},"OSD_HOST_DOWN":{"severity":"HEALTH_WARN","summary":{"message":"1 host (1 osds) down"}},"PG_AVAILABILITY":{"severity":"HEALTH_WARN","summary":{"message":"Reduced data availability: 101 pgs stale"}},"POOL_APP_NOT_ENABLED":{"severity":"HEALTH_WARN","summary":{"message":"application not enabled on 1 pool(s)"}}},"status":"HEALTH_WARN","overall_status":"HEALTH_WARN"},"election_epoch":12,"quorum":[0,1,2],"quorum_names":["rook-ceph-mon0","rook-ceph-mon2","rook-ceph-mon1"],"monmap":{"epoch":3,"fsid":"613975f3-3025-4802-9de1-a2280b950e75","modified":"2017-08-11 20:13:02.075679","created":"2017-08-11 20:12:35.314510","features":{"persistent":["kraken","luminous"],"optional":[]},"mons":[{"rank":0,"name":"rook-ceph-mon0","addr":"10.3.0.45:6789/0","public_addr":"10.3.0.45:6789/0"},{"rank":1,"name":"rook-ceph-mon2","addr":"10.3.0.249:6789/0","public_addr":"10.3.0.249:6789/0"},{"rank":2,"name":"rook-ceph-mon1","addr":"10.3.0.252:6789/0","public_addr":"10.3.0.252:6789/0"}]},"osdmap":{"osdmap":{"epoch":17,"num_osds":2,"num_up_osds":1,"num_in_osds":2,"full":false,"nearfull":true,"num_remapped_pgs":0}},"pgmap":{"pgs_by_state":[{"state_name":"stale+active+clean","count":101},{"state_name":"active+clean","count":99}],"num_pgs":200,"num_pools":2,"num_objects":243,"data_bytes":976793635,"bytes_used":13611479040,"bytes_avail":19825307648,"bytes_total":33436786688},"fsmap":{"epoch":1,"by_rank":[]},"mgrmap":{"epoch":3,"active_gid":14111,"active_name":"rook-ceph-mgr0","active_addr":"10.2.73.6:6800/9","available":true,"standbys":[],"modules":["restful","status"],"available_modules":["dashboard","prometheus","restful","status","zabbix"]},"servicemap":{"epoch":1,"modified":"0.000000","services":{}}}` ) +var nodeName = "node01" var namespace = "rook-ceph" var cephCluster = &cephv1.CephCluster{ - ObjectMeta: metav1.ObjectMeta{Name: "ceph-cluster"}} + ObjectMeta: metav1.ObjectMeta{Name: "ceph-cluster"}, +} -var pdbConfigMap = &corev1.ConfigMap{ - ObjectMeta: metav1.ObjectMeta{Name: pdbStateMapName, Namespace: namespace}} +var nodeObj = &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: nodeName}, + Spec: corev1.NodeSpec{ + Unschedulable: false, + }, +} + +var unschedulableNodeObj = &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: nodeName}, + Spec: corev1.NodeSpec{ + Unschedulable: true, + }, +} func fakeOSDDeployment(id, readyReplicas int) appsv1.Deployment { osd := appsv1.Deployment{ @@ -67,6 +80,7 @@ func fakeOSDDeployment(id, readyReplicas int) appsv1.Deployment { ObjectMeta: metav1.ObjectMeta{ Labels: map[string]string{ "topology-location-zone": fmt.Sprintf("zone-%d", id), + "ceph-osd-id": fmt.Sprintf("%d", id), }, }, }, @@ -78,6 +92,30 @@ func fakeOSDDeployment(id, readyReplicas int) appsv1.Deployment { return osd } +func fakeOSDPod(id int, nodeName string) corev1.Pod { + osdPod := corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: fmt.Sprintf("osd-%d", id), + Namespace: namespace, + Labels: map[string]string{ + "app": "rook-ceph-osd", + "ceph-osd-id": fmt.Sprintf("%d", id), + }, + }, + Spec: corev1.PodSpec{ + NodeName: nodeName, + }, + } + return osdPod +} + +func fakePDBConfigMap(drainingFailureDomain string) *corev1.ConfigMap { + return &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: pdbStateMapName, Namespace: namespace}, + Data: map[string]string{drainingFailureDomainKey: drainingFailureDomain, setNoOut: ""}, + } +} + func getFakeReconciler(t *testing.T, obj ...runtime.Object) *ReconcileClusterDisruption { scheme := scheme.Scheme err := policyv1beta1.AddToScheme(scheme) @@ -110,128 +148,193 @@ func getFakeClusterInfo() *client.ClusterInfo { func TestGetOSDFailureDomains(t *testing.T) { testcases := []struct { - label string + name string osds []appsv1.Deployment + osdPods []corev1.Pod + node *corev1.Node expectedAllFailureDomains []string expectedDrainingFailureDomains []string + expectedOsdDownFailureDomains []string }{ { - label: "case 1: all osds are running", + name: "case 1: all osds are running", osds: []appsv1.Deployment{fakeOSDDeployment(1, 1), fakeOSDDeployment(2, 1), fakeOSDDeployment(3, 1)}, + osdPods: []corev1.Pod{fakeOSDPod(1, nodeName), fakeOSDPod(2, nodeName), + fakeOSDPod(3, nodeName)}, + node: nodeObj, expectedAllFailureDomains: []string{"zone-1", "zone-2", "zone-3"}, + expectedOsdDownFailureDomains: []string{}, expectedDrainingFailureDomains: []string{}, }, { - label: "case 1: osd in zone-1 is pending", + name: "case 2: osd in zone-1 is pending and node is unscheduable", osds: []appsv1.Deployment{fakeOSDDeployment(1, 0), fakeOSDDeployment(2, 1), fakeOSDDeployment(3, 1)}, - expectedAllFailureDomains: []string{"zone-2", "zone-3"}, + osdPods: []corev1.Pod{fakeOSDPod(1, ""), fakeOSDPod(2, nodeName), + fakeOSDPod(3, nodeName)}, + node: nodeObj, + expectedAllFailureDomains: []string{"zone-1", "zone-2", "zone-3"}, + expectedOsdDownFailureDomains: []string{"zone-1"}, expectedDrainingFailureDomains: []string{"zone-1"}, }, { - label: "case 1: osds in zone-1 and zone-2 are pending", + name: "case 3: osd in zone-1 and zone-2 are pending and node is unscheduable", osds: []appsv1.Deployment{fakeOSDDeployment(1, 0), fakeOSDDeployment(2, 0), fakeOSDDeployment(3, 1)}, - expectedAllFailureDomains: []string{"zone-3"}, + osdPods: []corev1.Pod{fakeOSDPod(1, ""), fakeOSDPod(2, ""), + fakeOSDPod(3, nodeName)}, + node: nodeObj, + expectedAllFailureDomains: []string{"zone-1", "zone-2", "zone-3"}, + expectedOsdDownFailureDomains: []string{"zone-1", "zone-2"}, expectedDrainingFailureDomains: []string{"zone-1", "zone-2"}, }, + { + name: "case 4: osd in zone-1 is pending but osd node is scheduable", + osds: []appsv1.Deployment{fakeOSDDeployment(1, 0), fakeOSDDeployment(2, 1), + fakeOSDDeployment(3, 1)}, + osdPods: []corev1.Pod{fakeOSDPod(1, nodeName), fakeOSDPod(2, nodeName), + fakeOSDPod(3, nodeName)}, + node: nodeObj, + expectedAllFailureDomains: []string{"zone-1", "zone-2", "zone-3"}, + expectedOsdDownFailureDomains: []string{"zone-1"}, + expectedDrainingFailureDomains: []string{}, + }, + { + name: "case 5: osd in zone-1 is pending but osd node is not scheduable", + osds: []appsv1.Deployment{fakeOSDDeployment(1, 0), fakeOSDDeployment(2, 1), + fakeOSDDeployment(3, 1)}, + osdPods: []corev1.Pod{fakeOSDPod(1, nodeName), fakeOSDPod(2, nodeName), + fakeOSDPod(3, nodeName)}, + node: unschedulableNodeObj, + expectedAllFailureDomains: []string{"zone-1", "zone-2", "zone-3"}, + expectedOsdDownFailureDomains: []string{"zone-1"}, + expectedDrainingFailureDomains: []string{"zone-1"}, + }, } for _, tc := range testcases { - r := getFakeReconciler(t, cephCluster, &corev1.ConfigMap{}, tc.osds[0].DeepCopy(), - tc.osds[1].DeepCopy(), tc.osds[2].DeepCopy()) - clusterInfo := getFakeClusterInfo() - request := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}} - allfailureDomains, drainingFailureDomains, err := r.getOSDFailureDomains(clusterInfo, request, "zone") - assert.NoError(t, err) - assert.Equal(t, tc.expectedAllFailureDomains, allfailureDomains) - assert.Equal(t, tc.expectedDrainingFailureDomains, drainingFailureDomains) + t.Run(tc.name, func(t *testing.T) { + objs := []runtime.Object{ + cephCluster, + &corev1.ConfigMap{}, + tc.node, + } + for _, osdDeployment := range tc.osds { + objs = append(objs, osdDeployment.DeepCopy()) + } + for _, osdPod := range tc.osdPods { + objs = append(objs, osdPod.DeepCopy()) + } + r := getFakeReconciler(t, objs...) + clusterInfo := getFakeClusterInfo() + request := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}} + allfailureDomains, nodeDrainFailureDomains, osdDownFailureDomains, err := r.getOSDFailureDomains(clusterInfo, request, "zone") + assert.NoError(t, err) + assert.Equal(t, tc.expectedAllFailureDomains, allfailureDomains) + assert.Equal(t, tc.expectedDrainingFailureDomains, nodeDrainFailureDomains) + assert.Equal(t, tc.expectedOsdDownFailureDomains, osdDownFailureDomains) + }) } } func TestGetOSDFailureDomainsError(t *testing.T) { testcases := []struct { - label string + name string osds []appsv1.Deployment expectedAllFailureDomains []string expectedDrainingFailureDomains []string + expectedOsdDownFailureDomains []string }{ { - label: "case 1: one or more OSD deployment is missing crush location label", + name: "case 1: one or more OSD deployment is missing crush location label", osds: []appsv1.Deployment{fakeOSDDeployment(1, 1), fakeOSDDeployment(2, 1), fakeOSDDeployment(3, 1)}, - expectedAllFailureDomains: []string{}, - expectedDrainingFailureDomains: []string{}, + expectedAllFailureDomains: nil, + expectedDrainingFailureDomains: nil, + expectedOsdDownFailureDomains: nil, }, } for _, tc := range testcases { - osd := tc.osds[0].DeepCopy() - osd.Spec.Template.ObjectMeta.Labels["topology-location-zone"] = "" - r := getFakeReconciler(t, cephCluster, &corev1.ConfigMap{}, - tc.osds[1].DeepCopy(), tc.osds[2].DeepCopy(), osd) - clusterInfo := getFakeClusterInfo() - request := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}} - allfailureDomains, drainingFailureDomains, err := r.getOSDFailureDomains(clusterInfo, request, "zone") - assert.Error(t, err) - assert.Equal(t, tc.expectedAllFailureDomains, allfailureDomains) - assert.Equal(t, tc.expectedDrainingFailureDomains, drainingFailureDomains) + t.Run(tc.name, func(t *testing.T) { + osd := tc.osds[0].DeepCopy() + osd.Spec.Template.ObjectMeta.Labels["topology-location-zone"] = "" + r := getFakeReconciler(t, cephCluster, &corev1.ConfigMap{}, + tc.osds[1].DeepCopy(), tc.osds[2].DeepCopy(), osd) + clusterInfo := getFakeClusterInfo() + request := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}} + allfailureDomains, nodeDrainFailureDomains, osdDownFailureDomains, err := r.getOSDFailureDomains(clusterInfo, request, "zone") + assert.Error(t, err) + assert.Equal(t, tc.expectedAllFailureDomains, allfailureDomains) + assert.Equal(t, tc.expectedDrainingFailureDomains, nodeDrainFailureDomains) + assert.Equal(t, tc.expectedOsdDownFailureDomains, osdDownFailureDomains) + }) } } func TestReconcilePDBForOSD(t *testing.T) { - pdbConfig := pdbConfigMap.DeepCopy() - r := getFakeReconciler(t, cephCluster, pdbConfig) - clusterInfo := getFakeClusterInfo() - request := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}} - testcases := []struct { - label string + name string fakeCephStatus string fakeOSDDump string + configMap *corev1.ConfigMap allFailureDomains []string - drainingFailureDomains []string + osdDownFailureDomains []string + activeNodeDrains bool + expectedSetNoOutValue string expectedOSDPDBCount int expectedMaxUnavailableCount int expectedDrainingFailureDomainName string }{ { - label: "case 1: no draining failure domain and all pgs are healthy", + name: "case 1: no draining failure domain and all pgs are healthy", fakeCephStatus: healthyCephStatus, fakeOSDDump: `{"OSDs": [{"OSD": 3, "Up": 3, "In": 3}]}`, allFailureDomains: []string{"zone-1", "zone-2", "zone-3"}, - drainingFailureDomains: []string{}, + osdDownFailureDomains: []string{}, + configMap: fakePDBConfigMap(""), + activeNodeDrains: false, + expectedSetNoOutValue: "", expectedOSDPDBCount: 1, expectedMaxUnavailableCount: 1, expectedDrainingFailureDomainName: "", }, { - label: "case 2: zone-1 failure domain is draining and pgs are unhealthy", + name: "case 2: zone-1 failure domain is draining and pgs are unhealthy", fakeCephStatus: unHealthyCephStatus, fakeOSDDump: `{"OSDs": [{"OSD": 3, "Up": 3, "In": 2}]}`, allFailureDomains: []string{"zone-1", "zone-2", "zone-3"}, - drainingFailureDomains: []string{"zone-1"}, + osdDownFailureDomains: []string{"zone-1"}, + configMap: fakePDBConfigMap(""), + activeNodeDrains: true, + expectedSetNoOutValue: "true", expectedOSDPDBCount: 2, expectedMaxUnavailableCount: 0, expectedDrainingFailureDomainName: "zone-1", }, { - label: "case 3: zone-1 is back online. But pgs are still unhealthy from zone-1 drain", + name: "case 3: zone-1 is back online. But pgs are still unhealthy from zone-1 drain", fakeCephStatus: unHealthyCephStatus, fakeOSDDump: `{"OSDs": [{"OSD": 3, "Up": 3, "In": 3}]}`, allFailureDomains: []string{"zone-1", "zone-2", "zone-3"}, - drainingFailureDomains: []string{}, + osdDownFailureDomains: []string{}, + configMap: fakePDBConfigMap("zone-1"), + activeNodeDrains: true, + expectedSetNoOutValue: "", expectedOSDPDBCount: 2, expectedMaxUnavailableCount: 0, expectedDrainingFailureDomainName: "zone-1", }, { - label: "case 4: zone-1 is back online and pgs are also healthy", + name: "case 4: zone-1 is back online and pgs are also healthy", fakeCephStatus: healthyCephStatus, fakeOSDDump: `{"OSDs": [{"OSD": 3, "Up": 3, "In": 3}]}`, allFailureDomains: []string{"zone-1", "zone-2", "zone-3"}, - drainingFailureDomains: []string{}, + osdDownFailureDomains: []string{}, + configMap: fakePDBConfigMap(""), + activeNodeDrains: true, + expectedSetNoOutValue: "", expectedOSDPDBCount: 1, expectedMaxUnavailableCount: 1, expectedDrainingFailureDomainName: "", @@ -239,40 +342,47 @@ func TestReconcilePDBForOSD(t *testing.T) { } for _, tc := range testcases { - executor := &exectest.MockExecutor{} - executor.MockExecuteCommandWithOutputFile = func(command, outputFile string, args ...string) (string, error) { - logger.Infof("Command: %s %v", command, args) - if args[0] == "status" { - return tc.fakeCephStatus, nil + t.Run(tc.name, func(t *testing.T) { + r := getFakeReconciler(t, cephCluster, tc.configMap) + clusterInfo := getFakeClusterInfo() + request := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}} + executor := &exectest.MockExecutor{} + executor.MockExecuteCommandWithOutputFile = func(command, outputFile string, args ...string) (string, error) { + logger.Infof("Command: %s %v", command, args) + if args[0] == "status" { + return tc.fakeCephStatus, nil + } + if args[0] == "osd" && args[1] == "dump" { + return tc.fakeOSDDump, nil + } + return "", errors.Errorf("unexpected ceph command '%v'", args) } - if args[0] == "osd" && args[1] == "dump" { - return tc.fakeOSDDump, nil + r.context = &controllerconfig.Context{ClusterdContext: &clusterd.Context{Executor: executor}} + _, err := r.reconcilePDBsForOSDs(clusterInfo, request, tc.configMap, "zone", tc.allFailureDomains, tc.osdDownFailureDomains, tc.activeNodeDrains) + assert.NoError(t, err) + + // assert that pdb for osd are created correctly + existingPDBs := &policyv1beta1.PodDisruptionBudgetList{} + err = r.client.List(context.TODO(), existingPDBs) + assert.NoError(t, err) + assert.Equal(t, tc.expectedOSDPDBCount, len(existingPDBs.Items)) + for _, pdb := range existingPDBs.Items { + assert.Equal(t, tc.expectedMaxUnavailableCount, pdb.Spec.MaxUnavailable.IntValue()) } - return "", errors.Errorf("unexpected ceph command '%v'", args) - } - r.context = &controllerconfig.Context{ClusterdContext: &clusterd.Context{Executor: executor}} - err := r.reconcilePDBsForOSDs(clusterInfo, request, pdbConfig, "zone", tc.allFailureDomains, tc.drainingFailureDomains) - assert.NoError(t, err) - // assert that pdb for osd are created correctly - existingPDBs := &policyv1beta1.PodDisruptionBudgetList{} - err = r.client.List(context.TODO(), existingPDBs) - assert.NoError(t, err) - assert.Equal(t, tc.expectedOSDPDBCount, len(existingPDBs.Items)) - for _, pdb := range existingPDBs.Items { - assert.Equal(t, tc.expectedMaxUnavailableCount, pdb.Spec.MaxUnavailable.IntValue()) - } + // assert that config map is updated with correct failure domain + existingConfigMaps := &corev1.ConfigMapList{} + err = r.client.List(context.TODO(), existingConfigMaps) + assert.NoError(t, err) + assert.Equal(t, tc.expectedDrainingFailureDomainName, existingConfigMaps.Items[0].Data[drainingFailureDomainKey]) + assert.Equal(t, tc.expectedSetNoOutValue, existingConfigMaps.Items[0].Data[setNoOut]) + }) - // assert that config map is updated with correct failure domain - existingConfigMaps := &corev1.ConfigMapList{} - err = r.client.List(context.TODO(), existingConfigMaps) - assert.NoError(t, err) - assert.Equal(t, tc.expectedDrainingFailureDomainName, existingConfigMaps.Items[0].Data[drainingFailureDomainKey]) } } func TestPGHealthcheckTimeout(t *testing.T) { - pdbConfig := pdbConfigMap.DeepCopy() + pdbConfig := fakePDBConfigMap("") r := getFakeReconciler(t, cephCluster, pdbConfig) clusterInfo := getFakeClusterInfo() request := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}} @@ -292,23 +402,49 @@ func TestPGHealthcheckTimeout(t *testing.T) { r.pgHealthCheckTimeout = time.Duration(time.Minute * 10) // reconcile OSD PDB with active drains (on zone-1) and unhealthy PGs - err := r.reconcilePDBsForOSDs(clusterInfo, request, pdbConfig, "zone", []string{"zone-1", "zone-2"}, []string{"zone-1"}) + _, err := r.reconcilePDBsForOSDs(clusterInfo, request, pdbConfig, "zone", []string{"zone-1", "zone-2"}, []string{"zone-1"}, true) assert.NoError(t, err) assert.Equal(t, "zone-1", pdbConfig.Data[drainingFailureDomainKey]) + assert.Equal(t, "true", pdbConfig.Data[setNoOut]) // update the pgHealthCheckDuration time by -9 minutes pdbConfig.Data[pgHealthCheckDurationKey] = time.Now().Add(time.Duration(-7) * time.Minute).Format(time.RFC3339) // reconcile OSD PDB with no active drains and unhealthy PGs - err = r.reconcilePDBsForOSDs(clusterInfo, request, pdbConfig, "zone", []string{"zone-1", "zone-2"}, []string{}) + _, err = r.reconcilePDBsForOSDs(clusterInfo, request, pdbConfig, "zone", []string{"zone-1", "zone-2"}, []string{}, true) assert.NoError(t, err) // assert that pdb config map was not reset as the PG health check was not timed out assert.Equal(t, "zone-1", pdbConfig.Data[drainingFailureDomainKey]) + assert.Equal(t, "true", pdbConfig.Data[setNoOut]) // update the drainingFailureDomain time by -9 minutes pdbConfig.Data[pgHealthCheckDurationKey] = time.Now().Add(time.Duration(-11) * time.Minute).Format(time.RFC3339) // reconcile OSD PDB with no active drains and unhealthy PGs - err = r.reconcilePDBsForOSDs(clusterInfo, request, pdbConfig, "zone", []string{"zone-1", "zone-2"}, []string{}) + _, err = r.reconcilePDBsForOSDs(clusterInfo, request, pdbConfig, "zone", []string{"zone-1", "zone-2"}, []string{}, false) assert.NoError(t, err) // assert that pdb config map was reset as the PG health check was timed out assert.Equal(t, "", pdbConfig.Data[drainingFailureDomainKey]) + assert.Equal(t, "", pdbConfig.Data[setNoOut]) +} + +func TestHasNodeDrained(t *testing.T) { + osdPOD := fakeOSDPod(0, nodeName) + // Not expecting node drain because OSD pod is assigned to a schedulable node + r := getFakeReconciler(t, nodeObj, osdPOD.DeepCopy(), &corev1.ConfigMap{}) + expected, err := hasOSDNodeDrained(r.client, namespace, "0") + assert.NoError(t, err) + assert.False(t, expected) + + // Expecting node drain because OSD pod is assigned to an unschedulable node + r = getFakeReconciler(t, unschedulableNodeObj, osdPOD.DeepCopy(), &corev1.ConfigMap{}) + expected, err = hasOSDNodeDrained(r.client, namespace, "0") + assert.NoError(t, err) + assert.True(t, expected) + + // Expecting node drain because OSD pod is not assigned to any node + osdPodObj := osdPOD.DeepCopy() + osdPodObj.Spec.NodeName = "" + r = getFakeReconciler(t, nodeObj, osdPodObj, &corev1.ConfigMap{}) + expected, err = hasOSDNodeDrained(r.client, namespace, "0") + assert.NoError(t, err) + assert.True(t, expected) } diff --git a/pkg/operator/ceph/disruption/clusterdisruption/reconcile.go b/pkg/operator/ceph/disruption/clusterdisruption/reconcile.go index 0811040d8..e03278b6a 100644 --- a/pkg/operator/ceph/disruption/clusterdisruption/reconcile.go +++ b/pkg/operator/ceph/disruption/clusterdisruption/reconcile.go @@ -169,8 +169,8 @@ func (r *ReconcileClusterDisruption) reconcile(request reconcile.Request) (recon return reconcile.Result{}, nil } - // get all failure domains and the draining failure domains list - allFailureDomains, drainingFailureDomains, err := r.getOSDFailureDomains(clusterInfo, request, poolFailureDomain) + // get a list of all the failure domains, failure domains with failed OSDs and failure domains with drained nodes + allFailureDomains, nodeDrainFailureDomains, osdDownFailureDomains, err := r.getOSDFailureDomains(clusterInfo, request, poolFailureDomain) if err != nil { return reconcile.Result{}, err } @@ -181,17 +181,8 @@ func (r *ReconcileClusterDisruption) reconcile(request reconcile.Request) (recon return reconcile.Result{}, err } - err = r.reconcilePDBsForOSDs(clusterInfo, request, pdbStateMap, poolFailureDomain, allFailureDomains, drainingFailureDomains) - if err != nil { - return reconcile.Result{}, err - } - - disabledPDB, ok := pdbStateMap.Data[drainingFailureDomainKey] - if ok && len(disabledPDB) > 0 { - return reconcile.Result{Requeue: true, RequeueAfter: 30 * time.Second}, nil - } - - return reconcile.Result{}, nil + activeNodeDrains := len(nodeDrainFailureDomains) > 0 + return r.reconcilePDBsForOSDs(clusterInfo, request, pdbStateMap, poolFailureDomain, allFailureDomains, osdDownFailureDomains, activeNodeDrains) } // ClusterMap maintains the association between namespace and clusername