Merge pull request #3869 from rohantmp/canaryTolerations

Tolerations for drain detection canaries

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
 	}

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{{

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)
+	}
+}