From 195f20ad011bbdd7cfbe925487c1c3a3aaa8a022 Mon Sep 17 00:00:00 2001 From: Rohan CJ Date: Thu, 12 Sep 2019 12:16:56 +0530 Subject: [PATCH] Pick up tolerations for drain canary pod from the osd. This change is necessary drain canary pod needs to be schedulable on the same nodes as the OSDs. Since it affects multiple OSDs, we need to gather tolerations from multiple OSDs. Signed-off-by: Rohan CJ --- pkg/operator/ceph/disruption/nodedrain/add.go | 26 +++- .../ceph/disruption/nodedrain/reconcile.go | 27 +++- .../disruption/nodedrain/reconcile_test.go | 116 ++++++++++++++++++ 3 files changed, 164 insertions(+), 5 deletions(-) create mode 100644 pkg/operator/ceph/disruption/nodedrain/reconcile_test.go diff --git a/pkg/operator/ceph/disruption/nodedrain/add.go b/pkg/operator/ceph/disruption/nodedrain/add.go index c7ee3a121..29c294dcf 100644 --- a/pkg/operator/ceph/disruption/nodedrain/add.go +++ b/pkg/operator/ceph/disruption/nodedrain/add.go @@ -42,6 +42,7 @@ const ( // 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 @@ -71,10 +72,27 @@ func Add(mgr manager.Manager, context *controllerconfig.Context) error { return err } // Watch for changes to the node canary deployments - err = c.Watch(&source.Kind{Type: &appsv1.Deployment{}}, &handler.EnqueueRequestForOwner{ - OwnerType: &corev1.Node{}, - IsController: true, - }) + 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 } diff --git a/pkg/operator/ceph/disruption/nodedrain/reconcile.go b/pkg/operator/ceph/disruption/nodedrain/reconcile.go index b09379dab..ec3279911 100644 --- a/pkg/operator/ceph/disruption/nodedrain/reconcile.go +++ b/pkg/operator/ceph/disruption/nodedrain/reconcile.go @@ -88,11 +88,26 @@ func (r *ReconcileNode) reconcile(request reconcile.Request) (reconcile.Result, if err != nil { return reconcile.Result{}, fmt.Errorf("could not list the osd pods: %+v", err) } + + // map with tolerations as keys and empty struct as values for uniqueness + uniqueTolerations := make(map[corev1.Toleration]struct{}) occupiedByOSD := false for _, osdPod := range osdPodList.Items { if osdPod.Spec.NodeName == request.Name { occupiedByOSD = true - break + // get the osd tolerations + for _, osdToleration := range osdPod.Spec.Tolerations { + if osdToleration.Key == "node.kubernetes.io/unschedulable" { + labels := osdPod.GetLabels() + logger.Errorf( + "osd %s in namespace %s tolerates the drain taint, but the drain canary will not.", + labels[osd.OsdIdLabelKey], + labels[k8sutil.ClusterAttr], + ) + } else { + uniqueTolerations[osdToleration] = struct{}{} + } + } } } // Create or Update the deployment default/foo @@ -110,6 +125,7 @@ func (r *ReconcileNode) reconcile(request reconcile.Request) (reconcile.Result, deploymentLabels := map[string]string{ corev1.LabelHostname: nodeHostnameLabel, k8sutil.AppAttr: CanaryAppName, + NodeNameLabel: node.GetName(), } nodeSelector := map[string]string{corev1.LabelHostname: nodeHostnameLabel} @@ -131,6 +147,7 @@ func (r *ReconcileNode) reconcile(request reconcile.Request) (reconcile.Result, Spec: corev1.PodSpec{ NodeSelector: nodeSelector, Containers: newDoNothingContainers(), + Tolerations: tolerationMapToList(uniqueTolerations), }, } @@ -148,6 +165,14 @@ func (r *ReconcileNode) reconcile(request reconcile.Request) (reconcile.Result, return reconcile.Result{}, nil } +func tolerationMapToList(tolerationMap map[corev1.Toleration]struct{}) []corev1.Toleration { + tolerationList := make([]corev1.Toleration, 0) + for toleration := range tolerationMap { + tolerationList = append(tolerationList, toleration) + } + return tolerationList +} + // returns a container that does nothing func newDoNothingContainers() []corev1.Container { return []corev1.Container{{ diff --git a/pkg/operator/ceph/disruption/nodedrain/reconcile_test.go b/pkg/operator/ceph/disruption/nodedrain/reconcile_test.go new file mode 100644 index 000000000..84f19484c --- /dev/null +++ b/pkg/operator/ceph/disruption/nodedrain/reconcile_test.go @@ -0,0 +1,116 @@ +/* +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 ( + "testing" + + "github.com/stretchr/testify/assert" + + corev1 "k8s.io/api/core/v1" +) + +func TestCanaryTolerations(t *testing.T) { + uniqueTolerationsManual := []corev1.Toleration{ + // key1 + // exists + { + Key: "key1", + Operator: corev1.TolerationOpExists, + Effect: corev1.TaintEffectNoSchedule, + }, + { + Key: "key1", + Operator: corev1.TolerationOpExists, + Effect: corev1.TaintEffectPreferNoSchedule, + }, + // equals with different values + { + Key: "key1", + Operator: corev1.TolerationOpEqual, + Value: "value1", + Effect: corev1.TaintEffectNoSchedule, + }, + { + Key: "key1", + Operator: corev1.TolerationOpEqual, + Value: "value2", + Effect: corev1.TaintEffectNoSchedule, + }, + + // with different effects + { + Key: "key1", + Operator: corev1.TolerationOpEqual, + Value: "value2", + Effect: corev1.TaintEffectNoExecute, + }, + // key2 + { + Key: "key2", + Operator: corev1.TolerationOpExists, + Effect: corev1.TaintEffectNoSchedule, + }, + { + Key: "key2", + Operator: corev1.TolerationOpEqual, + Value: "value1", + Effect: corev1.TaintEffectNoSchedule, + }, + { + Key: "key2", + Operator: corev1.TolerationOpEqual, + Value: "value2", + Effect: corev1.TaintEffectNoSchedule, + }, + { + Key: "key2", + Operator: corev1.TolerationOpEqual, + Value: "value2", + Effect: corev1.TaintEffectNoExecute, + }, + } + + tolerationsWithDuplicates := make([]corev1.Toleration, 0) + for i := range uniqueTolerationsManual { + tolerationsWithDuplicates = append(tolerationsWithDuplicates, uniqueTolerationsManual[i]) + + //append the previous one again if it's within range, else append the last one + if i > 0 { + tolerationsWithDuplicates = append(tolerationsWithDuplicates, uniqueTolerationsManual[i-1]) + } else { + tolerationsWithDuplicates = append(tolerationsWithDuplicates, uniqueTolerationsManual[len(uniqueTolerationsManual)-1]) + } + } + uniqueTolerationsMap := make(map[corev1.Toleration]struct{}) + for _, toleration := range tolerationsWithDuplicates { + uniqueTolerationsMap[toleration] = struct{}{} + } + + uniqueTolerations := tolerationMapToList(uniqueTolerationsMap) + + assert.Equal(t, len(uniqueTolerationsManual), len(uniqueTolerations)) + for _, tolerationI := range uniqueTolerationsManual { + found := false + for _, tolerationJ := range uniqueTolerations { + if tolerationI == tolerationJ { + found = true + } + } + assert.True(t, found) + } +}