Taint controller - first commit

cmd/kube-controller-manager/app/controllermanager.go

@@ -433,6 +433,7 @@ func StartControllers(controllers map[string]InitFunc, s *options.CMServer, root
 		serviceCIDR,
 		int(s.NodeCIDRMaskSize),
 		s.AllocateNodeCIDRs,
+		s.EnableTaintManager,
 	)
 	if err != nil {
 		return fmt.Errorf("failed to initialize nodecontroller: %v", err)

cmd/kube-controller-manager/app/options/options.go

@@ -104,6 +104,7 @@ func NewCMServer() *CMServer {
 			ClusterSigningCertFile:   "/etc/kubernetes/ca/ca.pem",
 			ClusterSigningKeyFile:    "/etc/kubernetes/ca/ca.key",
 			ReconcilerSyncLoopPeriod: metav1.Duration{Duration: 5 * time.Second},
+			EnableTaintManager:       true,
 		},
 	}
 	s.LeaderElection.LeaderElect = true
@@ -196,6 +197,7 @@ func (s *CMServer) AddFlags(fs *pflag.FlagSet, allControllers []string, disabled
 	fs.Float32Var(&s.UnhealthyZoneThreshold, "unhealthy-zone-threshold", 0.55, "Fraction of Nodes in a zone which needs to be not Ready (minimum 3) for zone to be treated as unhealthy. ")
 	fs.BoolVar(&s.DisableAttachDetachReconcilerSync, "disable-attach-detach-reconcile-sync", false, "Disable volume attach detach reconciler sync. Disabling this may cause volumes to be mismatched with pods. Use wisely.")
 	fs.DurationVar(&s.ReconcilerSyncLoopPeriod.Duration, "attach-detach-reconcile-sync-period", s.ReconcilerSyncLoopPeriod.Duration, "The reconciler sync wait time between volume attach detach. This duration must be larger than one second, and increasing this value from the default may allow for volumes to be mismatched with pods.")
+	fs.BoolVar(&s.EnableTaintManager, "enable-taint-manager", s.EnableTaintManager, "WARNING: Beta feature. If set to true enables NoExecute Taints and will evict all not-tolerating Pod running on Nodes tainted with this kind of Taints.")
 
 	leaderelection.BindFlags(&s.LeaderElection, fs)
 

hack/verify-flags/known-flags.txt

@@ -168,6 +168,7 @@ enable-garbage-collector
 enable-hostpath-provisioner
 enable-server
 enable-swagger-ui
+enable-taint-manager
 etcd-address
 etcd-cafile
 etcd-certfile

pkg/api/v1/helpers.go

@@ -396,6 +396,28 @@ func DeleteTaint(taints []Taint, taintToDelete *Taint) ([]Taint, bool) {
 	return newTaints, deleted
 }
 
+// Returns true and list of Tolerations matching all Taints if all are tolerated, or false otherwise.
+func GetMatchingTolerations(taints []Taint, tolerations []Toleration) (bool, []Toleration) {
+	if len(tolerations) == 0 && len(taints) > 0 {
+		return false, []Toleration{}
+	}
+	result := []Toleration{}
+	for i := range taints {
+		tolerated := false
+		for j := range tolerations {
+			if tolerations[j].ToleratesTaint(&taints[i]) {
+				result = append(result, tolerations[j])
+				tolerated = true
+				break
+			}
+		}
+		if !tolerated {
+			return false, []Toleration{}
+		}
+	}
+	return true, result
+}
+
 // MatchTaint checks if the taint matches taintToMatch. Taints are unique by key:effect,
 // if the two taints have same key:effect, regard as they match.
 func (t *Taint) MatchTaint(taintToMatch Taint) bool {

pkg/apis/componentconfig/types.go

@@ -790,6 +790,9 @@ type KubeControllerManagerConfiguration struct {
 	// ReconcilerSyncLoopPeriod is the amount of time the reconciler sync states loop
 	// wait between successive executions. Is set to 5 sec by default.
 	ReconcilerSyncLoopPeriod metav1.Duration
+	// If set to true enables NoExecute Taints and will evict all not-tolerating
+	// Pod running on Nodes tainted with this kind of Taints.
+	EnableTaintManager bool
 }
 
 // VolumeConfiguration contains *all* enumerated flags meant to configure all volume

pkg/controller/node/nodecontroller.go

@@ -148,6 +148,8 @@ type NodeController struct {
 
 	// allocate/recycle CIDRs for node if allocateNodeCIDRs == true
 	cidrAllocator CIDRAllocator
+	// manages taints
+	taintManager *NoExecuteTaintManager
 
 	forcefullyDeletePod        func(*v1.Pod) error
 	nodeExistsInCloudProvider  func(types.NodeName) (bool, error)
@@ -160,6 +162,10 @@ type NodeController struct {
 	secondaryEvictionLimiterQPS float32
 	largeClusterThreshold       int32
 	unhealthyZoneThreshold      float32
+
+	// if set to true NodeController will start TaintManager that will evict Pods from
+	// tainted nodes, if they're not tolerated.
+	runTaintManager bool
 }
 
 // NewNodeController returns a new node controller to sync instances from cloudprovider.
@@ -183,7 +189,8 @@ func NewNodeController(
 	clusterCIDR *net.IPNet,
 	serviceCIDR *net.IPNet,
 	nodeCIDRMaskSize int,
-	allocateNodeCIDRs bool) (*NodeController, error) {
+	allocateNodeCIDRs bool,
+	runTaintManager bool) (*NodeController, error) {
 	eventBroadcaster := record.NewBroadcaster()
 	recorder := eventBroadcaster.NewRecorder(api.Scheme, clientv1.EventSource{Component: "controllermanager"})
 	eventBroadcaster.StartLogging(glog.Infof)
@@ -232,14 +239,47 @@ func NewNodeController(
 		largeClusterThreshold:       largeClusterThreshold,
 		unhealthyZoneThreshold:      unhealthyZoneThreshold,
 		zoneStates:                  make(map[string]zoneState),
+		runTaintManager:             runTaintManager,
 	}
 	nc.enterPartialDisruptionFunc = nc.ReducedQPSFunc
 	nc.enterFullDisruptionFunc = nc.HealthyQPSFunc
 	nc.computeZoneStateFunc = nc.ComputeZoneState
 
 	podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc:    nc.maybeDeleteTerminatingPod,
-		UpdateFunc: func(_, obj interface{}) { nc.maybeDeleteTerminatingPod(obj) },
+		AddFunc: func(obj interface{}) {
+			nc.maybeDeleteTerminatingPod(obj)
+			pod := obj.(*v1.Pod)
+			if nc.taintManager != nil {
+				nc.taintManager.PodUpdated(nil, pod)
+			}
+		},
+		UpdateFunc: func(prev, obj interface{}) {
+			nc.maybeDeleteTerminatingPod(obj)
+			prevPod := prev.(*v1.Pod)
+			newPod := obj.(*v1.Pod)
+			if nc.taintManager != nil {
+				nc.taintManager.PodUpdated(prevPod, newPod)
+			}
+		},
+		DeleteFunc: func(obj interface{}) {
+			pod, isPod := obj.(*v1.Pod)
+			// We can get DeletedFinalStateUnknown instead of *v1.Node here and we need to handle that correctly. #34692
+			if !isPod {
+				deletedState, ok := obj.(cache.DeletedFinalStateUnknown)
+				if !ok {
+					glog.Errorf("Received unexpected object: %v", obj)
+					return
+				}
+				pod, ok = deletedState.Obj.(*v1.Pod)
+				if !ok {
+					glog.Errorf("DeletedFinalStateUnknown contained non-Node object: %v", deletedState.Obj)
+					return
+				}
+			}
+			if nc.taintManager != nil {
+				nc.taintManager.PodUpdated(pod, nil)
+			}
+		},
 	})
 	nc.podInformerSynced = podInformer.Informer().HasSynced
 
@@ -279,9 +319,13 @@ func NewNodeController(
 				if err := nc.cidrAllocator.AllocateOrOccupyCIDR(node); err != nil {
 					utilruntime.HandleError(fmt.Errorf("Error allocating CIDR: %v", err))
 				}
+				if nc.taintManager != nil {
+					nc.taintManager.NodeUpdated(nil, node)
+				}
 			},
-			UpdateFunc: func(_, obj interface{}) {
-				node := obj.(*v1.Node)
+			UpdateFunc: func(oldNode, newNode interface{}) {
+				node := newNode.(*v1.Node)
+				prevNode := oldNode.(*v1.Node)
 				// If the PodCIDR is not empty we either:
 				// - already processed a Node that already had a CIDR after NC restarted
 				//   (cidr is marked as used),
@@ -312,6 +356,9 @@ func NewNodeController(
 						utilruntime.HandleError(fmt.Errorf("Error allocating CIDR: %v", err))
 					}
 				}
+				if nc.taintManager != nil {
+					nc.taintManager.NodeUpdated(prevNode, node)
+				}
 			},
 			DeleteFunc: func(originalObj interface{}) {
 				obj, err := api.Scheme.DeepCopy(originalObj)
@@ -334,6 +381,9 @@ func NewNodeController(
 						return
 					}
 				}
+				if nc.taintManager != nil {
+					nc.taintManager.NodeUpdated(node, nil)
+				}
 				if err := nc.cidrAllocator.ReleaseCIDR(node); err != nil {
 					glog.Errorf("Error releasing CIDR: %v", err)
 				}
@@ -341,6 +391,10 @@ func NewNodeController(
 		}
 	}
 
+	if nc.runTaintManager {
+		nc.taintManager = NewNoExecuteTaintManager(kubeClient)
+	}
+
 	nodeInformer.Informer().AddEventHandler(nodeEventHandlerFuncs)
 	nc.nodeLister = nodeInformer.Lister()
 	nc.nodeInformerSynced = nodeInformer.Informer().HasSynced
@@ -368,6 +422,10 @@ func (nc *NodeController) Run() {
 			}
 		}, nc.nodeMonitorPeriod, wait.NeverStop)
 
+		if nc.runTaintManager {
+			go nc.taintManager.Run(wait.NeverStop)
+		}
+
 		// Managing eviction of nodes:
 		// When we delete pods off a node, if the node was not empty at the time we then
 		// queue an eviction watcher. If we hit an error, retry deletion.

pkg/controller/node/nodecontroller_test.go

@@ -99,6 +99,7 @@ func NewNodeControllerFromClient(
 		serviceCIDR,
 		nodeCIDRMaskSize,
 		allocateNodeCIDRs,
+		false,
 	)
 	if err != nil {
 		return nil, err

pkg/controller/node/timed_workers.go

+/*
+Copyright 2015 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 node
+
+import (
+	"sync"
+	"time"
+
+	"k8s.io/apimachinery/pkg/types"
+
+	"github.com/golang/glog"
+)
+
+// WorkArgs keeps arguments that will be passed to tha function executed by the worker.
+type WorkArgs struct {
+	NamespacedName types.NamespacedName
+}
+
+// KeyFromWorkArgs creates a key for the given `WorkArgs`
+func (w *WorkArgs) KeyFromWorkArgs() string {
+	return w.NamespacedName.String()
+}
+
+// NewWorkArgs is a helper function to create new `WorkArgs`
+func NewWorkArgs(name, namespace string) *WorkArgs {
+	return &WorkArgs{types.NamespacedName{Namespace: namespace, Name: name}}
+}
+
+// TimedWorker is a responsible for executing a function no earlier than at FireAt time.
+type TimedWorker struct {
+	WorkItem  *WorkArgs
+	CreatedAt time.Time
+	FireAt    time.Time
+	Timer     *time.Timer
+}
+
+// CreateWorker creates a TimedWorker that will execute `f` not earlier than `fireAt`.
+func CreateWorker(args *WorkArgs, createdAt time.Time, fireAt time.Time, f func(args *WorkArgs) error) *TimedWorker {
+	delay := fireAt.Sub(time.Now())
+	if delay <= 0 {
+		go f(args)
+		return nil
+	}
+	timer := time.AfterFunc(delay, func() { f(args) })
+	return &TimedWorker{
+		WorkItem:  args,
+		CreatedAt: createdAt,
+		FireAt:    fireAt,
+		Timer:     timer,
+	}
+}
+
+// Cancel cancels the execution of function by the `TimedWorker`
+func (w *TimedWorker) Cancel() {
+	if w != nil {
+		w.Timer.Stop()
+	}
+}
+
+// TimedWorkerQueue keeps a set of TimedWorkers that still wait for execution.
+type TimedWorkerQueue struct {
+	sync.Mutex
+	workers  map[string]*TimedWorker
+	workFunc func(args *WorkArgs) error
+}
+
+// CreateWorkerQueue creates a new TimedWorkerQueue for workers that will execute
+// given function `f`.
+func CreateWorkerQueue(f func(args *WorkArgs) error) *TimedWorkerQueue {
+	return &TimedWorkerQueue{
+		workers:  make(map[string]*TimedWorker),
+		workFunc: f,
+	}
+}
+
+func (q *TimedWorkerQueue) getWrappedWorkerFunc(key string) func(args *WorkArgs) error {
+	return func(args *WorkArgs) error {
+		err := q.workFunc(args)
+		q.Lock()
+		defer q.Unlock()
+		if err == nil {
+			q.workers[key] = nil
+		} else {
+			delete(q.workers, key)
+		}
+		return err
+	}
+}
+
+// AddWork adds a work to the WorkerQueue which will be executed not earlier than `fireAt`.
+func (q *TimedWorkerQueue) AddWork(args *WorkArgs, createdAt time.Time, fireAt time.Time) {
+	key := args.KeyFromWorkArgs()
+
+	q.Lock()
+	defer q.Unlock()
+	if _, exists := q.workers[key]; exists {
+		glog.Warningf("Trying to add already existing work for %+v. Skipping.", args)
+		return
+	}
+	worker := CreateWorker(args, createdAt, fireAt, q.getWrappedWorkerFunc(key))
+	if worker == nil {
+		return
+	}
+	q.workers[key] = worker
+}
+
+// CancelWork removes scheduled function execution from the queue.
+func (q *TimedWorkerQueue) CancelWork(key string) {
+	q.Lock()
+	defer q.Unlock()
+	worker, found := q.workers[key]
+	if found {
+		worker.Cancel()
+		delete(q.workers, key)
+	}
+}
+
+// GetWorkerUnsafe returns a TimedWorker corresponding to the given key.
+// Unsafe method - workers have attached goroutines which can fire afater this function is called.
+func (q *TimedWorkerQueue) GetWorkerUnsafe(key string) *TimedWorker {
+	q.Lock()
+	defer q.Unlock()
+	return q.workers[key]
+}

pkg/controller/node/timed_workers_test.go

+/*
+Copyright 2017 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 node
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestExecute(t *testing.T) {
+	testVal := int32(0)
+	wg := sync.WaitGroup{}
+	wg.Add(10)
+	queue := CreateWorkerQueue(func(args *WorkArgs) error {
+		atomic.AddInt32(&testVal, 1)
+		wg.Done()
+		return nil
+	})
+	now := time.Now()
+	queue.AddWork(NewWorkArgs("1", "1"), now, now)
+	queue.AddWork(NewWorkArgs("2", "2"), now, now)
+	queue.AddWork(NewWorkArgs("3", "3"), now, now)
+	queue.AddWork(NewWorkArgs("4", "4"), now, now)
+	queue.AddWork(NewWorkArgs("5", "5"), now, now)
+	queue.AddWork(NewWorkArgs("1", "1"), now, now)
+	queue.AddWork(NewWorkArgs("2", "2"), now, now)
+	queue.AddWork(NewWorkArgs("3", "3"), now, now)
+	queue.AddWork(NewWorkArgs("4", "4"), now, now)
+	queue.AddWork(NewWorkArgs("5", "5"), now, now)
+	wg.Wait()
+	lastVal := atomic.LoadInt32(&testVal)
+	if lastVal != 10 {
+		t.Errorf("Espected testVal = 10, got %v", lastVal)
+	}
+}
+
+func TestExecuteDelayed(t *testing.T) {
+	testVal := int32(0)
+	wg := sync.WaitGroup{}
+	wg.Add(5)
+	queue := CreateWorkerQueue(func(args *WorkArgs) error {
+		atomic.AddInt32(&testVal, 1)
+		wg.Done()
+		return nil
+	})
+	now := time.Now()
+	then := now.Add(time.Second)
+	queue.AddWork(NewWorkArgs("1", "1"), now, then)
+	queue.AddWork(NewWorkArgs("2", "2"), now, then)
+	queue.AddWork(NewWorkArgs("3", "3"), now, then)
+	queue.AddWork(NewWorkArgs("4", "4"), now, then)
+	queue.AddWork(NewWorkArgs("5", "5"), now, then)
+	queue.AddWork(NewWorkArgs("1", "1"), now, then)
+	queue.AddWork(NewWorkArgs("2", "2"), now, then)
+	queue.AddWork(NewWorkArgs("3", "3"), now, then)
+	queue.AddWork(NewWorkArgs("4", "4"), now, then)
+	queue.AddWork(NewWorkArgs("5", "5"), now, then)
+	wg.Wait()
+	lastVal := atomic.LoadInt32(&testVal)
+	if lastVal != 5 {
+		t.Errorf("Espected testVal = 5, got %v", lastVal)
+	}
+}
+
+func TestCancel(t *testing.T) {
+	testVal := int32(0)
+	wg := sync.WaitGroup{}
+	wg.Add(3)
+	queue := CreateWorkerQueue(func(args *WorkArgs) error {
+		atomic.AddInt32(&testVal, 1)
+		wg.Done()
+		return nil
+	})
+	now := time.Now()
+	then := now.Add(time.Second)
+	queue.AddWork(NewWorkArgs("1", "1"), now, then)
+	queue.AddWork(NewWorkArgs("2", "2"), now, then)
+	queue.AddWork(NewWorkArgs("3", "3"), now, then)
+	queue.AddWork(NewWorkArgs("4", "4"), now, then)
+	queue.AddWork(NewWorkArgs("5", "5"), now, then)
+	queue.AddWork(NewWorkArgs("1", "1"), now, then)
+	queue.AddWork(NewWorkArgs("2", "2"), now, then)
+	queue.AddWork(NewWorkArgs("3", "3"), now, then)
+	queue.AddWork(NewWorkArgs("4", "4"), now, then)
+	queue.AddWork(NewWorkArgs("5", "5"), now, then)
+	queue.CancelWork(NewWorkArgs("2", "2").KeyFromWorkArgs())
+	queue.CancelWork(NewWorkArgs("4", "4").KeyFromWorkArgs())
+	wg.Wait()
+	lastVal := atomic.LoadInt32(&testVal)
+	if lastVal != 3 {
+		t.Errorf("Espected testVal = 3, got %v", lastVal)
+	}
+}
+
+func TestCancelAndReadd(t *testing.T) {
+	testVal := int32(0)
+	wg := sync.WaitGroup{}
+	wg.Add(4)
+	queue := CreateWorkerQueue(func(args *WorkArgs) error {
+		atomic.AddInt32(&testVal, 1)
+		wg.Done()
+		return nil
+	})
+	now := time.Now()
+	then := now.Add(time.Second)
+	queue.AddWork(NewWorkArgs("1", "1"), now, then)
+	queue.AddWork(NewWorkArgs("2", "2"), now, then)
+	queue.AddWork(NewWorkArgs("3", "3"), now, then)
+	queue.AddWork(NewWorkArgs("4", "4"), now, then)
+	queue.AddWork(NewWorkArgs("5", "5"), now, then)
+	queue.AddWork(NewWorkArgs("1", "1"), now, then)
+	queue.AddWork(NewWorkArgs("2", "2"), now, then)
+	queue.AddWork(NewWorkArgs("3", "3"), now, then)
+	queue.AddWork(NewWorkArgs("4", "4"), now, then)
+	queue.AddWork(NewWorkArgs("5", "5"), now, then)
+	queue.CancelWork(NewWorkArgs("2", "2").KeyFromWorkArgs())
+	queue.CancelWork(NewWorkArgs("4", "4").KeyFromWorkArgs())
+	queue.AddWork(NewWorkArgs("2", "2"), now, then)
+	wg.Wait()
+	lastVal := atomic.LoadInt32(&testVal)
+	if lastVal != 4 {
+		t.Errorf("Espected testVal = 4, got %v", lastVal)
+	}
+}

test/utils/runners.go

@@ -52,6 +52,46 @@ const (
 	nonExist = "NonExist"
 )
 
+func WaitUntilPodIsScheduled(c clientset.Interface, name, namespace string, timeout time.Duration) (*v1.Pod, error) {
+	// Wait until it's scheduled
+	p, err := c.Core().Pods(namespace).Get(name, metav1.GetOptions{ResourceVersion: "0"})
+	if err == nil && p.Spec.NodeName != "" {
+		return p, nil
+	}
+	pollingPeriod := 200 * time.Millisecond
+	startTime := time.Now()
+	for startTime.Add(timeout).After(time.Now()) {
+		time.Sleep(pollingPeriod)
+		p, err := c.Core().Pods(namespace).Get(name, metav1.GetOptions{ResourceVersion: "0"})
+		if err == nil && p.Spec.NodeName != "" {
+			return p, nil
+		}
+	}
+	return nil, fmt.Errorf("Timed out after %v when waiting for pod %v/%v to start.", timeout, namespace, name)
+}
+
+func RunPodAndGetNodeName(c clientset.Interface, pod *v1.Pod, timeout time.Duration) (string, error) {
+	retries := 5
+	name := pod.Name
+	namespace := pod.Namespace
+	var err error
+	// Create a Pod
+	for i := 0; i < retries; i++ {
+		_, err = c.Core().Pods(namespace).Create(pod)
+		if err == nil || apierrs.IsAlreadyExists(err) {
+			break
+		}
+	}
+	if err != nil && !apierrs.IsAlreadyExists(err) {
+		return "", err
+	}
+	p, err := WaitUntilPodIsScheduled(c, name, namespace, timeout)
+	if err != nil {
+		return "", err
+	}
+	return p.Spec.NodeName, nil
+}
+
 type RunObjectConfig interface {
 	Run() error
 	GetName() string