Merge pull request #7 from negz/podfilter

Make pod eviction filters configurable

Author: negz

README.md

@@ -27,30 +27,24 @@ usage: draino [<flags>] [<node-conditions>...]
 Automatically cordons and drains nodes that match the supplied conditions.
 
 Flags:
-      --help                   Show context-sensitive help (also try --help-long
-                               and --help-man).
-  -d, --debug                  Run with debug logging.
-      --listen=":10002"        Address at which to expose /metrics and /healthz.
-      --kubeconfig=KUBECONFIG  Path to kubeconfig file. Leave unset to use
-                               in-cluster config.
-      --master=MASTER          Address of Kubernetes API server. Leave unset to
-                               use in-cluster config.
-      --dry-run                Emit an event without cordoning or draining
-                               matching nodes.
-      --max-grace-period=8m0s  Maximum time evicted pods will be given to
-                               terminate gracefully.
-      --eviction-headroom=30s  Additional time to wait after a pod's termination
-                               grace period for it to have been deleted.
-      --drain-buffer=10m0s     Minimum time between starting each drain. Nodes
-                               are always cordoned immediately.
+      --help                     Show context-sensitive help (also try --help-long and --help-man).
+  -d, --debug                    Run with debug logging.
+      --listen=":10002"          Address at which to expose /metrics and /healthz.
+      --kubeconfig=KUBECONFIG    Path to kubeconfig file. Leave unset to use in-cluster config.
+      --master=MASTER            Address of Kubernetes API server. Leave unset to use in-cluster config.
+      --dry-run                  Emit an event without cordoning or draining matching nodes.
+      --max-grace-period=8m0s    Maximum time evicted pods will be given to terminate gracefully.
+      --eviction-headroom=30s    Additional time to wait after a pod's termination grace period for it to have been deleted.
+      --drain-buffer=10m0s       Minimum time between starting each drain. Nodes are always cordoned immediately.
       --node-label=KEY=VALUE ...
-                               Only nodes with this label will be eligible for
-                               cordoning and draining. May be specified multiple
-                               times.
+                                 Only nodes with this label will be eligible for cordoning and draining. May be specified multiple times.
+      --evict-daemonset-pods     Evict pods that were created by an extant DaemonSet.
+      --evict-emptydir-pods      Evict pods with local storage, i.e. with emptyDir volumes.
+      --evict-unreplicated-pods  Evict pods that were not created by a replication controller.
 
 Args:
-  [<node-conditions>]  Nodes for which any of these conditions are true will be
-                       cordoned and drained.
+  [<node-conditions>]  Nodes for which any of these conditions are true will be cordoned and drained.
+
 ```
 
 ## Considerations
@@ -59,9 +53,6 @@ Keep the following in mind before deploying Draino:
 * Always run Draino in `--dry-run` mode first to ensure it would drain the nodes
   you expect it to. In dry run mode Draino will emit logs, metrics, and events
   but will not actually cordon or drain nodes.
-* Draino will not evict [DaemonSet](https://kubernetes.io/docs/concepts/workloads/controllers/daemonset/)
-  or [mirror](https://kubernetes.io/docs/tasks/administer-cluster/static-pod/)
-  pods. It _will_ evict pods with local storage, and unreplicated pods.
 * Draino immediately cordons nodes that match its configured labels and node
   conditions, but will wait a configurable amount of time (10 minutes by default)
   between draining nodes. i.e. If two nodes begin exhibiting a node condition

cmd/draino/draino.go

@@ -38,6 +38,10 @@ func main() {
 		drainBuffer      = app.Flag("drain-buffer", "Minimum time between starting each drain. Nodes are always cordoned immediately.").Default(kubernetes.DefaultDrainBuffer.String()).Duration()
 		nodeLabels       = app.Flag("node-label", "Only nodes with this label will be eligible for cordoning and draining. May be specified multiple times.").PlaceHolder("KEY=VALUE").StringMap()
 
+		evictDaemonSetPods    = app.Flag("evict-daemonset-pods", "Evict pods that were created by an extant DaemonSet.").Bool()
+		evictLocalStoragePods = app.Flag("evict-emptydir-pods", "Evict pods with local storage, i.e. with emptyDir volumes.").Bool()
+		evictUnreplicatedPods = app.Flag("evict-unreplicated-pods", "Evict pods that were not created by a replication controller.").Bool()
+
 		conditions = app.Arg("node-conditions", "Nodes for which any of these conditions are true will be cordoned and drained.").Strings()
 	)
 	kingpin.MustParse(app.Parse(os.Args[1:]))
@@ -82,11 +86,20 @@ func main() {
 	cs, err := client.NewForConfig(c)
 	kingpin.FatalIfError(err, "cannot create Kubernetes client")
 
+	pf := []kubernetes.PodFilterFunc{kubernetes.MirrorPodFilter}
+	switch {
+	case !*evictLocalStoragePods:
+		pf = append(pf, kubernetes.LocalStoragePodFilter)
+	case !*evictUnreplicatedPods:
+		pf = append(pf, kubernetes.UnreplicatedPodFilter)
+	case !*evictDaemonSetPods:
+		pf = append(pf, kubernetes.NewDaemonSetPodFilter(cs))
+	}
 	var h cache.ResourceEventHandler = kubernetes.NewDrainingResourceEventHandler(
 		kubernetes.NewAPICordonDrainer(cs,
 			kubernetes.MaxGracePeriod(*maxGracePeriod),
 			kubernetes.EvictionHeadroom(*evictionHeadroom),
-			kubernetes.WithPodFilter(kubernetes.NewPodFilters(kubernetes.MirrorPodFilter, kubernetes.NewDaemonSetPodFilter(cs)))),
+			kubernetes.WithPodFilter(kubernetes.NewPodFilters(pf...))),
 		kubernetes.NewEventRecorder(cs),
 		kubernetes.WithLogger(log),
 		kubernetes.WithDrainBuffer(*drainBuffer))

internal/kubernetes/drainer.go

@@ -38,54 +38,6 @@ func IsTimeout(err error) bool {
 	return ok
 }
 
-// A FilterFunc returns true if the supplied pod passes the filter.
-type FilterFunc func(p core.Pod) (bool, error)
-
-// MirrorPodFilter returns true if the supplied pod is not a mirror pod, i.e. a
-// pod created by a manifest on the node rather than the API server.
-func MirrorPodFilter(p core.Pod) (bool, error) {
-	_, mirrorPod := p.GetAnnotations()[core.MirrorPodAnnotationKey]
-	return !mirrorPod, nil
-}
-
-// NewDaemonSetPodFilter returns a FilterFunc that returns true if the supplied
-// pod is not managed by an extant DaemonSet.
-func NewDaemonSetPodFilter(client kubernetes.Interface) FilterFunc {
-	return func(p core.Pod) (bool, error) {
-		c := meta.GetControllerOf(&p)
-		if c == nil || c.Kind != kindDaemonSet {
-			return true, nil
-		}
-
-		// Pods pass the filter if they were created by a DaemonSet that no
-		// longer exists.
-		if _, err := client.ExtensionsV1beta1().DaemonSets(p.GetNamespace()).Get(c.Name, meta.GetOptions{}); err != nil {
-			if apierrors.IsNotFound(err) {
-				return true, nil
-			}
-			return false, errors.Wrapf(err, "cannot get DaemonSet %s/%s", p.GetNamespace(), c.Name)
-		}
-		return false, nil
-	}
-}
-
-// NewPodFilters returns a FilterFunc that returns true if all of the supplied
-// FilterFuncs return true.
-func NewPodFilters(filters ...FilterFunc) FilterFunc {
-	return func(p core.Pod) (bool, error) {
-		for _, fn := range filters {
-			passes, err := fn(p)
-			if err != nil {
-				return false, errors.Wrap(err, "cannot apply filters")
-			}
-			if !passes {
-				return false, nil
-			}
-		}
-		return true, nil
-	}
-}
-
 // A Cordoner cordons nodes.
 type Cordoner interface {
 	// Cordon the supplied node. Marks it unschedulable for new pods.
@@ -117,7 +69,7 @@ func (d *NoopCordonDrainer) Drain(n *core.Node) error { return nil }
 type APICordonDrainer struct {
 	c kubernetes.Interface
 
-	filter FilterFunc
+	filter PodFilterFunc
 
 	maxGracePeriod   time.Duration
 	evictionHeadroom time.Duration
@@ -145,7 +97,7 @@ func EvictionHeadroom(h time.Duration) APICordonDrainerOption {
 
 // WithPodFilter configures a filter that may be used to exclude certain pods
 // from eviction when draining.
-func WithPodFilter(f FilterFunc) APICordonDrainerOption {
+func WithPodFilter(f PodFilterFunc) APICordonDrainerOption {
 	return func(d *APICordonDrainer) {
 		d.filter = f
 	}
@@ -154,7 +106,12 @@ func WithPodFilter(f FilterFunc) APICordonDrainerOption {
 // NewAPICordonDrainer returns a CordonDrainer that cordons and drains nodes via
 // the Kubernetes API.
 func NewAPICordonDrainer(c kubernetes.Interface, ao ...APICordonDrainerOption) *APICordonDrainer {
-	d := &APICordonDrainer{c: c, maxGracePeriod: DefaultMaxGracePeriod, evictionHeadroom: DefaultEvictionOverhead}
+	d := &APICordonDrainer{
+		c:                c,
+		filter:           NewPodFilters(),
+		maxGracePeriod:   DefaultMaxGracePeriod,
+		evictionHeadroom: DefaultEvictionOverhead,
+	}
 	for _, o := range ao {
 		o(d)
 	}

internal/kubernetes/drainer_test.go

@@ -10,19 +10,30 @@ import (
 	meta "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/fake"
 	clienttesting "k8s.io/client-go/testing"
 )
 
 const (
-	ns            = "coolNamespace"
-	nodeName      = "coolNode"
-	podName       = "coolPod"
-	daemonsetName = "coolDaemonSet"
+	ns = "coolNamespace"
+
+	nodeName = "coolNode"
+	podName  = "coolPod"
+
+	daemonsetName  = "coolDaemonSet"
+	deploymentName = "coolDeployment"
+	kindDeployment = "Deployment"
+)
+
+var (
+	_ CordonDrainer = (*APICordonDrainer)(nil)
+	_ CordonDrainer = (*NoopCordonDrainer)(nil)
 )
 
 var podGracePeriodSeconds int64 = 10
 var isController = true
+var errExploded = errors.New("kaboom")
 
 type reactor struct {
 	verb        string
@@ -41,6 +52,14 @@ func (r reactor) Fn() clienttesting.ReactionFunc {
 	}
 }
 
+func newFakeClientSet(rs ...reactor) kubernetes.Interface {
+	cs := &fake.Clientset{}
+	for _, r := range rs {
+		cs.AddReactor(r.verb, r.resource, r.Fn())
+	}
+	return cs
+}
+
 func TestCordon(t *testing.T) {
 	cases := []struct {
 		name      string
@@ -274,18 +293,22 @@ func TestDrain(t *testing.T) {
 			},
 		},
 		{
-			name: "FilterMirrorPod",
+			name: "PodDoesNotPassFilter",
 			node: &core.Node{ObjectMeta: meta.ObjectMeta{Name: nodeName}},
+			options: []APICordonDrainerOption{WithPodFilter(func(p core.Pod) (bool, error) {
+				if p.GetName() == "lamePod" {
+					// This pod does not pass the filter.
+					return false, nil
+				}
+				return true, nil
+			})},
 			reactions: []reactor{
 				reactor{
 					verb:     "list",
 					resource: "pods",
 					ret: &core.PodList{Items: []core.Pod{
+						core.Pod{ObjectMeta: meta.ObjectMeta{Name: "lamePod"}},
 						core.Pod{ObjectMeta: meta.ObjectMeta{Name: podName}},
-						core.Pod{ObjectMeta: meta.ObjectMeta{
-							Name:        "mirrorPod",
-							Annotations: map[string]string{core.MirrorPodAnnotationKey: "true"},
-						}},
 					}},
 				},
 				reactor{
@@ -301,23 +324,20 @@ func TestDrain(t *testing.T) {
 			},
 		},
 		{
-			name: "FilterDaemonSetPod",
+			name: "PodFilterErrors",
 			node: &core.Node{ObjectMeta: meta.ObjectMeta{Name: nodeName}},
+			options: []APICordonDrainerOption{WithPodFilter(func(p core.Pod) (bool, error) {
+				if p.GetName() == "explodeyPod" {
+					return false, errExploded
+				}
+				return true, nil
+			})},
 			reactions: []reactor{
 				reactor{
 					verb:     "list",
 					resource: "pods",
 					ret: &core.PodList{Items: []core.Pod{
-						core.Pod{
-							ObjectMeta: meta.ObjectMeta{
-								Name: "daemonsetPod",
-								OwnerReferences: []meta.OwnerReference{meta.OwnerReference{
-									Controller: &isController,
-									Kind:       kindDaemonSet,
-								}},
-							},
-							Spec: core.PodSpec{TerminationGracePeriodSeconds: &podGracePeriodSeconds},
-						},
+						core.Pod{ObjectMeta: meta.ObjectMeta{Name: "explodeyPod"}},
 						core.Pod{ObjectMeta: meta.ObjectMeta{Name: podName}},
 					}},
 				},
@@ -332,74 +352,7 @@ func TestDrain(t *testing.T) {
 					err:      apierrors.NewNotFound(schema.GroupResource{Resource: "pods"}, podName),
 				},
 			},
-		},
-		{
-			name: "EvictOrphanedDaemonSetPod",
-			node: &core.Node{ObjectMeta: meta.ObjectMeta{Name: nodeName}},
-			reactions: []reactor{
-				reactor{
-					verb:     "list",
-					resource: "pods",
-					ret: &core.PodList{Items: []core.Pod{
-						core.Pod{
-							ObjectMeta: meta.ObjectMeta{
-								Name:      "orphanedDaemonsetPod",
-								Namespace: ns,
-								OwnerReferences: []meta.OwnerReference{meta.OwnerReference{
-									Controller: &isController,
-									Kind:       kindDaemonSet,
-									Name:       daemonsetName,
-								}},
-							},
-							Spec: core.PodSpec{TerminationGracePeriodSeconds: &podGracePeriodSeconds},
-						},
-					}},
-				},
-				reactor{
-					verb:     "get",
-					resource: "daemonsets",
-					err:      apierrors.NewNotFound(schema.GroupResource{Resource: "daemonsets"}, daemonsetName),
-				},
-				reactor{
-					verb:        "create",
-					resource:    "pods",
-					subresource: "eviction",
-				},
-				reactor{
-					verb:     "get",
-					resource: "pods",
-					err:      apierrors.NewNotFound(schema.GroupResource{Resource: "pods"}, podName),
-				},
-			},
-		},
-		{
-			name: "ErrorGettingDaemonset",
-			node: &core.Node{ObjectMeta: meta.ObjectMeta{Name: nodeName}},
-			reactions: []reactor{
-				reactor{
-					verb:     "list",
-					resource: "pods",
-					ret: &core.PodList{Items: []core.Pod{
-						core.Pod{
-							ObjectMeta: meta.ObjectMeta{
-								Name:      "orphanedDaemonsetPod",
-								Namespace: ns,
-								OwnerReferences: []meta.OwnerReference{meta.OwnerReference{
-									Controller: &isController,
-									Kind:       kindDaemonSet,
-									Name:       daemonsetName,
-								}},
-							},
-							Spec: core.PodSpec{TerminationGracePeriodSeconds: &podGracePeriodSeconds},
-						},
-					}},
-				},
-				reactor{
-					verb:     "get",
-					resource: "daemonsets",
-					err:      errors.New("nope"),
-				},
-			},
+			errFn: func(err error) bool { return errors.Cause(err) == errExploded },
 		},
 		{
 			name: "ErrorListingPods",
@@ -416,14 +369,8 @@ func TestDrain(t *testing.T) {
 
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
-			c := &fake.Clientset{}
-			for _, r := range tc.reactions {
-				c.AddReactor(r.verb, r.resource, r.Fn())
-			}
-
-			o := tc.options
-			o = append(o, WithPodFilter(NewPodFilters(MirrorPodFilter, NewDaemonSetPodFilter(c))))
-			d := NewAPICordonDrainer(c, o...)
+			c := newFakeClientSet(tc.reactions...)
+			d := NewAPICordonDrainer(c, tc.options...)
 			if err := d.Drain(tc.node); err != nil {
 				for _, r := range tc.reactions {
 					if errors.Cause(err) == r.err {