Enabling the ability to drain daemon set pods, with priority

cmd/kops/rollingupdatecluster.go

@@ -93,6 +93,16 @@ var (
 		  --fail-on-validate-error="false" \
 		  --node-interval 8m \
 		  --instance-group nodes
+
+		# Roll the k8s-cluster.example.com kops cluster,
+		# only roll the node instancegroup,
+		# use the new drain and validate functionality,
+		# do not ignore daemon sets
+		kops rolling-update cluster k8s-cluster.example.com --yes \
+		  --fail-on-validate-error="false" \
+		  --node-interval 8m \
+		  --instance-group nodes \
+		  --ignore-daemonsets="false"
 		`))
 
 	rollingupdateShort = i18n.T(`Rolling update a cluster.`)
@@ -141,6 +151,9 @@ type RollingUpdateOptions struct {
 	// InstanceGroupRoles is the list of roles we should rolling-update
 	// if not specified, all instance groups will be updated
 	InstanceGroupRoles []string
+
+	// IgnoreDaemonsets when a node is drained, daemon set pods will also be drained if false.  Default is true.
+	IgnoreDaemonsets bool
 }
 
 func (o *RollingUpdateOptions) InitDefaults() {
@@ -157,6 +170,8 @@ func (o *RollingUpdateOptions) InitDefaults() {
 
 	o.PostDrainDelay = 5 * time.Second
 	o.ValidationTimeout = 15 * time.Minute
+
+	o.IgnoreDaemonsets = true
 }
 
 func NewCmdRollingUpdateCluster(f *util.Factory, out io.Writer) *cobra.Command {
@@ -186,6 +201,7 @@ func NewCmdRollingUpdateCluster(f *util.Factory, out io.Writer) *cobra.Command {
 
 	cmd.Flags().BoolVar(&options.FailOnDrainError, "fail-on-drain-error", true, "The rolling-update will fail if draining a node fails.")
 	cmd.Flags().BoolVar(&options.FailOnValidate, "fail-on-validate-error", true, "The rolling-update will fail if the cluster fails to validate.")
+	cmd.Flags().BoolVar(&options.IgnoreDaemonsets, "ignore-daemonsets", true, "Whether to ignore draining daemon set pods on the node.  Default is true.")
 
 	cmd.Run = func(cmd *cobra.Command, args []string) {
 		err := rootCommand.ProcessArgs(args)
@@ -409,6 +425,7 @@ func RunRollingUpdateCluster(f *util.Factory, out io.Writer, options *RollingUpd
 		// TODO should we expose this to the UI?
 		ValidateTickDuration:    30 * time.Second,
 		ValidateSuccessDuration: 10 * time.Second,
+		IgnoreDaemonsets:        options.IgnoreDaemonsets,
 	}
 	return d.RollingUpdate(groups, cluster, list)
 }

pkg/drain/drain.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"io"
 	"math"
+	"sort"
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
@@ -145,6 +146,7 @@ func (d *Helper) GetPodsForDeletion(nodeName string) (*podDeleteList, []error) {
 
 	for _, pod := range podList.Items {
 		var status podDeleteStatus
+		var priority podDeletePriorty = podDeletionPriorityHighest
 		for _, filter := range d.makeFilters() {
 			status = filter(pod)
 			if !status.delete {
@@ -153,13 +155,27 @@ func (d *Helper) GetPodsForDeletion(nodeName string) (*podDeleteList, []error) {
 				// through any additional filters
 				break
 			}
+			if status.priority < priority {
+				//a previous filter had determined that this pod should be deleted at a lower priority
+				// use that priority
+				priority = status.priority
+			}
 		}
+		status.priority = priority
 		pods = append(pods, podDelete{
 			pod:    pod,
 			status: status,
 		})
 	}
 
+	//sort the list to come up with a sensible order for deleting pods
+	//for backwards compatibility, the default priority is highest.  If a daemon-set is found
+	//and ignore-daemonsets is set to false, the priority will be the lowest.
+	//This sort will return the order of pods to delete, with the highest priority first.
+	sort.SliceStable(pods, func(i, j int) bool {
+		return pods[i].status.priority > pods[j].status.priority
+	})
+
 	list := &podDeleteList{items: pods}
 
 	if errs := list.errors(); len(errs) > 0 {

pkg/drain/drain_test.go

@@ -0,0 +1,186 @@
+/*
+Copyright 2020 The Kubernetes Authors.
+
+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 drain
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	appsv1 "k8s.io/api/apps/v1"
+	v1 "k8s.io/api/core/v1"
+	v1meta "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes/fake"
+)
+
+func getTestSetup(objects []runtime.Object) *fake.Clientset {
+	k8sClient := fake.NewSimpleClientset(objects...)
+	return k8sClient
+}
+
+func dummyDaemonSet(name string) appsv1.DaemonSet {
+	return appsv1.DaemonSet{
+		ObjectMeta: v1meta.ObjectMeta{
+			Name:      name,
+			Namespace: "kube-system",
+		},
+	}
+}
+
+func dummyPod(podMap map[string]string) v1.Pod {
+	pod := v1.Pod{
+		ObjectMeta: v1meta.ObjectMeta{
+			Name:      podMap["name"],
+			Namespace: "kube-system",
+		},
+		Spec: v1.PodSpec{
+			NodeName: "node1",
+		},
+		Status: v1.PodStatus{
+			Phase: v1.PodPhase(podMap["phase"]),
+			ContainerStatuses: []v1.ContainerStatus{
+				{
+					Name:  "container1",
+					Ready: podMap["ready"] == "true",
+				},
+			},
+		},
+	}
+	if podMap["controller"] == "DaemonSet" {
+		pod.SetOwnerReferences([]v1meta.OwnerReference{
+			{
+				Name:       podMap["controllerName"],
+				Kind:       appsv1.SchemeGroupVersion.WithKind("DaemonSet").Kind,
+				Controller: &[]bool{true}[0],
+			},
+		})
+	}
+	return pod
+}
+
+// MakePodList constructs api.PodList from a list of pod attributes
+func makePodList(pods []map[string]string) []runtime.Object {
+	var list []runtime.Object
+	for _, pod := range pods {
+		p := dummyPod(pod)
+		list = append(list, &p)
+	}
+	return list
+}
+
+func TestRollingUpdateDaemonSetMixedPods(t *testing.T) {
+	objects := makePodList(
+		[]map[string]string{
+			{
+				"name":           "pod1",
+				"ready":          "true",
+				"phase":          string(v1.PodRunning),
+				"controller":     "DaemonSet",
+				"controllerName": "ds1",
+			},
+			{
+				"name":  "pod2",
+				"ready": "true",
+				"phase": string(v1.PodRunning),
+			},
+		},
+	)
+	ds := dummyDaemonSet("ds1")
+	objects = append(objects, &ds)
+	k8sClient := getTestSetup(objects)
+	helper := Helper{
+		Client:              k8sClient,
+		Force:               true,
+		IgnoreAllDaemonSets: false,
+		DeleteLocalData:     true,
+	}
+
+	podList, _ := helper.GetPodsForDeletion("node1")
+	assert.True(t, podList.Warnings() != "")
+	assert.True(t, len(podList.errors()) == 0)
+	assert.True(t, podList.items[0].status.priority == podDeletionPriorityHighest)
+	assert.True(t, podList.items[1].status.priority == podDeletionPriorityLowest)
+	assert.NotNil(t, podList)
+}
+
+func TestRollingUpdateNoDaemonSets(t *testing.T) {
+	objects := makePodList(
+		[]map[string]string{
+			{
+				"name":  "pod1",
+				"ready": "true",
+				"phase": string(v1.PodRunning),
+			},
+			{
+				"name":  "pod2",
+				"ready": "true",
+				"phase": string(v1.PodRunning),
+			},
+		},
+	)
+	k8sClient := getTestSetup(objects)
+	helper := Helper{
+		Client:              k8sClient,
+		Force:               true,
+		IgnoreAllDaemonSets: false,
+		DeleteLocalData:     true,
+	}
+
+	podList, _ := helper.GetPodsForDeletion("node1")
+	assert.True(t, podList.Warnings() != "")
+	assert.True(t, len(podList.errors()) == 0)
+	assert.True(t, podList.items[0].status.priority == podDeletionPriorityHighest)
+	assert.True(t, podList.items[1].status.priority == podDeletionPriorityHighest)
+	assert.NotNil(t, podList)
+}
+
+func TestRollingUpdateAllDaemonSetPods(t *testing.T) {
+	objects := makePodList(
+		[]map[string]string{
+			{
+				"name":           "pod1",
+				"ready":          "true",
+				"phase":          string(v1.PodRunning),
+				"controller":     "DaemonSet",
+				"controllerName": "ds1",
+			},
+			{
+				"name":           "pod2",
+				"ready":          "true",
+				"phase":          string(v1.PodRunning),
+				"controller":     "DaemonSet",
+				"controllerName": "ds1",
+			},
+		},
+	)
+	ds := dummyDaemonSet("ds1")
+	objects = append(objects, &ds)
+	k8sClient := getTestSetup(objects)
+	helper := Helper{
+		Client:              k8sClient,
+		Force:               true,
+		IgnoreAllDaemonSets: false,
+		DeleteLocalData:     true,
+	}
+
+	podList, _ := helper.GetPodsForDeletion("node1")
+	assert.True(t, podList.Warnings() == "")
+	assert.True(t, len(podList.errors()) == 0)
+	assert.True(t, podList.items[0].status.priority == podDeletionPriorityLowest)
+	assert.True(t, podList.items[1].status.priority == podDeletionPriorityLowest)
+	assert.NotNil(t, podList)
+}