Apply davidopps comments to TaintController PR

pkg/api/v1/helpers.go

@@ -403,6 +403,9 @@ func DeleteTaint(taints []Taint, taintToDelete *Taint) ([]Taint, bool) {
 
 // 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(taints) == 0 {
+		return true, []Toleration{}
+	}
 	if len(tolerations) == 0 && len(taints) > 0 {
 		return false, []Toleration{}
 	}

pkg/controller/node/cidr_allocator.go

@@ -26,6 +26,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
+	v1core "k8s.io/client-go/kubernetes/typed/core/v1"
 	clientv1 "k8s.io/client-go/pkg/api/v1"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/kubernetes/pkg/api"
@@ -79,6 +80,12 @@ func NewCIDRRangeAllocator(client clientset.Interface, clusterCIDR *net.IPNet, s
 	eventBroadcaster := record.NewBroadcaster()
 	recorder := eventBroadcaster.NewRecorder(api.Scheme, clientv1.EventSource{Component: "cidrAllocator"})
 	eventBroadcaster.StartLogging(glog.Infof)
+	if client != nil {
+		glog.V(0).Infof("Sending events to api server.")
+		eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: v1core.New(client.Core().RESTClient()).Events("")})
+	} else {
+		glog.Fatalf("kubeClient is nil when starting NodeController")
+	}
 
 	ra := &rangeAllocator{
 		client:                client,

pkg/controller/node/cidr_allocator_test.go

@@ -142,6 +142,7 @@ func TestAllocateOrOccupyCIDRSuccess(t *testing.T) {
 				t.Logf("%v: found non-default implementation of CIDRAllocator, skipping white-box test...", tc.description)
 				return
 			}
+			rangeAllocator.recorder = testutil.NewFakeRecorder()
 			if err = rangeAllocator.cidrs.occupy(cidr); err != nil {
 				t.Fatalf("%v: unexpected error when occupying CIDR %v: %v", tc.description, allocated, err)
 			}
@@ -223,6 +224,7 @@ func TestAllocateOrOccupyCIDRFailure(t *testing.T) {
 				t.Logf("%v: found non-default implementation of CIDRAllocator, skipping white-box test...", tc.description)
 				return
 			}
+			rangeAllocator.recorder = testutil.NewFakeRecorder()
 			err = rangeAllocator.cidrs.occupy(cidr)
 			if err != nil {
 				t.Fatalf("%v: unexpected error when occupying CIDR %v: %v", tc.description, allocated, err)
@@ -334,6 +336,7 @@ func TestReleaseCIDRSuccess(t *testing.T) {
 				t.Logf("%v: found non-default implementation of CIDRAllocator, skipping white-box test...", tc.description)
 				return
 			}
+			rangeAllocator.recorder = testutil.NewFakeRecorder()
 			err = rangeAllocator.cidrs.occupy(cidr)
 			if err != nil {
 				t.Fatalf("%v: unexpected error when occupying CIDR %v: %v", tc.description, allocated, err)

pkg/controller/node/controller_utils.go

@@ -26,8 +26,11 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+
+	clientv1 "k8s.io/client-go/pkg/api/v1"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/record"
+
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/v1"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
@@ -261,7 +264,7 @@ func nodeExistsInCloudProvider(cloud cloudprovider.Interface, nodeName types.Nod
 }
 
 func recordNodeEvent(recorder record.EventRecorder, nodeName, nodeUID, eventtype, reason, event string) {
-	ref := &v1.ObjectReference{
+	ref := &clientv1.ObjectReference{
 		Kind:      "Node",
 		Name:      nodeName,
 		UID:       types.UID(nodeUID),
@@ -272,7 +275,7 @@ func recordNodeEvent(recorder record.EventRecorder, nodeName, nodeUID, eventtype
 }
 
 func recordNodeStatusChange(recorder record.EventRecorder, node *v1.Node, new_status string) {
-	ref := &v1.ObjectReference{
+	ref := &clientv1.ObjectReference{
 		Kind:      "Node",
 		Name:      node.Name,
 		UID:       node.UID,

pkg/controller/node/nodecontroller.go

@@ -198,7 +198,7 @@ func NewNodeController(
 		glog.V(0).Infof("Sending events to api server.")
 		eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: v1core.New(kubeClient.Core().RESTClient()).Events("")})
 	} else {
-		glog.V(0).Infof("No api server defined - no events will be sent to API server.")
+		glog.Fatalf("kubeClient is nil when starting NodeController")
 	}
 
 	if kubeClient != nil && kubeClient.Core().RESTClient().GetRateLimiter() != nil {

pkg/controller/node/nodecontroller_test.go

@@ -550,6 +550,7 @@ func TestMonitorNodeStatusEvictPods(t *testing.T) {
 			evictionTimeout, testRateLimiterQPS, testRateLimiterQPS, testLargeClusterThreshold, testUnhealtyThreshold, testNodeMonitorGracePeriod,
 			testNodeStartupGracePeriod, testNodeMonitorPeriod, nil, nil, 0, false)
 		nodeController.now = func() metav1.Time { return fakeNow }
+		nodeController.recorder = testutil.NewFakeRecorder()
 		for _, ds := range item.daemonSets {
 			nodeController.daemonSetInformer.Informer().GetStore().Add(&ds)
 		}
@@ -694,6 +695,7 @@ func TestPodStatusChange(t *testing.T) {
 			evictionTimeout, testRateLimiterQPS, testRateLimiterQPS, testLargeClusterThreshold, testUnhealtyThreshold, testNodeMonitorGracePeriod,
 			testNodeStartupGracePeriod, testNodeMonitorPeriod, nil, nil, 0, false)
 		nodeController.now = func() metav1.Time { return fakeNow }
+		nodeController.recorder = testutil.NewFakeRecorder()
 		if err := syncNodeStore(nodeController, item.fakeNodeHandler); err != nil {
 			t.Errorf("unexpected error: %v", err)
 		}
@@ -1213,6 +1215,7 @@ func TestMonitorNodeStatusEvictPodsWithDisruption(t *testing.T) {
 		nodeController.enterPartialDisruptionFunc = func(nodeNum int) float32 {
 			return testRateLimiterQPS
 		}
+		nodeController.recorder = testutil.NewFakeRecorder()
 		nodeController.enterFullDisruptionFunc = func(nodeNum int) float32 {
 			return testRateLimiterQPS
 		}
@@ -1303,6 +1306,7 @@ func TestCloudProviderNoRateLimit(t *testing.T) {
 		testNodeMonitorPeriod, nil, nil, 0, false)
 	nodeController.cloud = &fakecloud.FakeCloud{}
 	nodeController.now = func() metav1.Time { return metav1.Date(2016, 1, 1, 12, 0, 0, 0, time.UTC) }
+	nodeController.recorder = testutil.NewFakeRecorder()
 	nodeController.nodeExistsInCloudProvider = func(nodeName types.NodeName) (bool, error) {
 		return false, nil
 	}
@@ -1570,6 +1574,7 @@ func TestMonitorNodeStatusUpdateStatus(t *testing.T) {
 			testRateLimiterQPS, testRateLimiterQPS, testLargeClusterThreshold, testUnhealtyThreshold,
 			testNodeMonitorGracePeriod, testNodeStartupGracePeriod, testNodeMonitorPeriod, nil, nil, 0, false)
 		nodeController.now = func() metav1.Time { return fakeNow }
+		nodeController.recorder = testutil.NewFakeRecorder()
 		if err := syncNodeStore(nodeController, item.fakeNodeHandler); err != nil {
 			t.Errorf("unexpected error: %v", err)
 		}
@@ -1803,6 +1808,7 @@ func TestMonitorNodeStatusMarkPodsNotReady(t *testing.T) {
 			testRateLimiterQPS, testRateLimiterQPS, testLargeClusterThreshold, testUnhealtyThreshold,
 			testNodeMonitorGracePeriod, testNodeStartupGracePeriod, testNodeMonitorPeriod, nil, nil, 0, false)
 		nodeController.now = func() metav1.Time { return fakeNow }
+		nodeController.recorder = testutil.NewFakeRecorder()
 		if err := syncNodeStore(nodeController, item.fakeNodeHandler); err != nil {
 			t.Errorf("unexpected error: %v", err)
 		}

pkg/controller/node/testutil/test_utils.go

@@ -29,6 +29,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/watch"
+
 	clientv1 "k8s.io/client-go/pkg/api/v1"
 	"k8s.io/client-go/util/clock"
 	"k8s.io/kubernetes/pkg/api"
@@ -36,6 +37,8 @@ import (
 	"k8s.io/kubernetes/pkg/client/clientset_generated/clientset/fake"
 	v1core "k8s.io/kubernetes/pkg/client/clientset_generated/clientset/typed/core/v1"
 	utilnode "k8s.io/kubernetes/pkg/util/node"
+
+	"github.com/golang/glog"
 )
 
 // FakeNodeHandler is a fake implementation of NodesInterface and NodeInterface. It
@@ -227,6 +230,7 @@ func (m *FakeNodeHandler) Patch(name string, pt types.PatchType, data []byte, su
 
 // FakeRecorder is used as a fake during testing.
 type FakeRecorder struct {
+	sync.Mutex
 	source clientv1.EventSource
 	Events []*clientv1.Event
 	clock  clock.Clock
@@ -247,20 +251,21 @@ func (f *FakeRecorder) PastEventf(obj runtime.Object, timestamp metav1.Time, eve
 }
 
 func (f *FakeRecorder) generateEvent(obj runtime.Object, timestamp metav1.Time, eventtype, reason, message string) {
-	ref, err := v1.GetReference(api.Scheme, obj)
+	f.Lock()
+	defer f.Unlock()
+	ref, err := clientv1.GetReference(api.Scheme, obj)
 	if err != nil {
+		glog.Errorf("Encoutered error while getting reference: %v", err)
 		return
 	}
 	event := f.makeEvent(ref, eventtype, reason, message)
 	event.Source = f.source
 	if f.Events != nil {
-		fmt.Println("write event")
 		f.Events = append(f.Events, event)
 	}
 }
 
-func (f *FakeRecorder) makeEvent(ref *v1.ObjectReference, eventtype, reason, message string) *clientv1.Event {
-	fmt.Println("make event")
+func (f *FakeRecorder) makeEvent(ref *clientv1.ObjectReference, eventtype, reason, message string) *clientv1.Event {
 	t := metav1.Time{Time: f.clock.Now()}
 	namespace := ref.Namespace
 	if namespace == "" {

pkg/controller/node/timed_workers.go

@@ -25,7 +25,7 @@ import (
 	"github.com/golang/glog"
 )
 
-// WorkArgs keeps arguments that will be passed to tha function executed by the worker.
+// WorkArgs keeps arguments that will be passed to the function executed by the worker.
 type WorkArgs struct {
 	NamespacedName types.NamespacedName
 }
@@ -50,7 +50,7 @@ type TimedWorker struct {
 
 // 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())
+	delay := fireAt.Sub(createdAt)
 	if delay <= 0 {
 		go f(args)
 		return nil
@@ -71,9 +71,10 @@ func (w *TimedWorker) Cancel() {
 	}
 }
 
-// TimedWorkerQueue keeps a set of TimedWorkers that still wait for execution.
+// TimedWorkerQueue keeps a set of TimedWorkers that are still wait for execution.
 type TimedWorkerQueue struct {
 	sync.Mutex
+	// map of workers keyed by string returned by 'KeyFromWorkArgs' from the given worker.
 	workers  map[string]*TimedWorker
 	workFunc func(args *WorkArgs) error
 }
@@ -93,6 +94,8 @@ func (q *TimedWorkerQueue) getWrappedWorkerFunc(key string) func(args *WorkArgs)
 		q.Lock()
 		defer q.Unlock()
 		if err == nil {
+			// To avoid duplicated calls we keep the key in the queue, to prevent
+			// subsequent additions.
 			q.workers[key] = nil
 		} else {
 			delete(q.workers, key)
@@ -104,6 +107,7 @@ func (q *TimedWorkerQueue) getWrappedWorkerFunc(key string) func(args *WorkArgs)
 // 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()
+	glog.V(4).Infof("Adding TimedWorkerQueue item %v at %v to be fired at %v", key, createdAt, fireAt)
 
 	q.Lock()
 	defer q.Unlock()
@@ -112,21 +116,24 @@ func (q *TimedWorkerQueue) AddWork(args *WorkArgs, createdAt time.Time, fireAt t
 		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) {
+// CancelWork removes scheduled function execution from the queue. Returns true if work was cancelled.
+func (q *TimedWorkerQueue) CancelWork(key string) bool {
+	glog.V(4).Infof("Cancelling TimedWorkerQueue item %v at %v", key, time.Now())
 	q.Lock()
 	defer q.Unlock()
 	worker, found := q.workers[key]
+	result := false
 	if found {
-		worker.Cancel()
+		if worker != nil {
+			result = true
+			worker.Cancel()
+		}
 		delete(q.workers, key)
 	}
+	return result
 }
 
 // GetWorkerUnsafe returns a TimedWorker corresponding to the given key.

pkg/controller/node/timed_workers_test.go

@@ -26,7 +26,7 @@ import (
 func TestExecute(t *testing.T) {
 	testVal := int32(0)
 	wg := sync.WaitGroup{}
-	wg.Add(10)
+	wg.Add(5)
 	queue := CreateWorkerQueue(func(args *WorkArgs) error {
 		atomic.AddInt32(&testVal, 1)
 		wg.Done()
@@ -38,6 +38,7 @@ func TestExecute(t *testing.T) {
 	queue.AddWork(NewWorkArgs("3", "3"), now, now)
 	queue.AddWork(NewWorkArgs("4", "4"), now, now)
 	queue.AddWork(NewWorkArgs("5", "5"), now, now)
+	// Adding the same thing second time should be no-op
 	queue.AddWork(NewWorkArgs("1", "1"), now, now)
 	queue.AddWork(NewWorkArgs("2", "2"), now, now)
 	queue.AddWork(NewWorkArgs("3", "3"), now, now)
@@ -45,8 +46,8 @@ func TestExecute(t *testing.T) {
 	queue.AddWork(NewWorkArgs("5", "5"), now, now)
 	wg.Wait()
 	lastVal := atomic.LoadInt32(&testVal)
-	if lastVal != 10 {
-		t.Errorf("Espected testVal = 10, got %v", lastVal)
+	if lastVal != 5 {
+		t.Errorf("Espected testVal = 5, got %v", lastVal)
 	}
 }
 
@@ -88,7 +89,7 @@ func TestCancel(t *testing.T) {
 		return nil
 	})
 	now := time.Now()
-	then := now.Add(time.Second)
+	then := now.Add(100 * time.Millisecond)
 	queue.AddWork(NewWorkArgs("1", "1"), now, then)
 	queue.AddWork(NewWorkArgs("2", "2"), now, then)
 	queue.AddWork(NewWorkArgs("3", "3"), now, then)
@@ -118,7 +119,7 @@ func TestCancelAndReadd(t *testing.T) {
 		return nil
 	})
 	now := time.Now()
-	then := now.Add(time.Second)
+	then := now.Add(100 * time.Millisecond)
 	queue.AddWork(NewWorkArgs("1", "1"), now, then)
 	queue.AddWork(NewWorkArgs("2", "2"), now, then)
 	queue.AddWork(NewWorkArgs("3", "3"), now, then)

test/e2e/taints_test.go

@@ -128,7 +128,7 @@ func createPodForTaintsTest(hasToleration bool, tolerationSeconds int, podName,
 	}
 }
 
-// Creates and startes a controller (informer) that watches updates on a pod in given namespace with given name. It puts a new
+// Creates and starts a controller (informer) that watches updates on a pod in given namespace with given name. It puts a new
 // struct into observedDeletion channel for every deletion it sees.
 func createTestController(cs clientset.Interface, observedDeletions chan struct{}, stopCh chan struct{}, podName, ns string) {
 	_, controller := cache.NewInformer(

test/utils/runners.go

@@ -71,7 +71,6 @@ func WaitUntilPodIsScheduled(c clientset.Interface, name, namespace string, time
 }
 
 func RunPodAndGetNodeName(c clientset.Interface, pod *v1.Pod, timeout time.Duration) (string, error) {
-	retries := 5
 	name := pod.Name
 	namespace := pod.Namespace
 	var err error