sync node on node status change

pkg/controller/node.go

@@ -19,6 +19,7 @@ package controller
 import (
 	"time"
 
+	apiv1 "k8s.io/api/core/v1"
 	listers "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/ingress-gce/pkg/context"
@@ -51,11 +52,12 @@ func NewNodeController(ctx *context.ControllerContext, cm *ClusterManager) *Node
 	ctx.NodeInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc:    c.queue.Enqueue,
 		DeleteFunc: c.queue.Enqueue,
-		// TODO: watch updates for Nodes going from NotReady to Ready.
-		// Otherwise the controller will wait until the complete
-		// refresh to process the node.
+		UpdateFunc: func(oldObj, newObj interface{}) {
+			if nodeStatusChanged(oldObj.(*apiv1.Node), newObj.(*apiv1.Node)) {
+				c.queue.Enqueue(newObj)
+			}
+		},
 	})
-
 	return c
 }
 

pkg/controller/utils.go

@@ -224,3 +224,13 @@ func getReadyNodeNames(lister listers.NodeLister) ([]string, error) {
 	}
 	return nodeNames, nil
 }
+
+func nodeStatusChanged(old, cur *api_v1.Node) bool {
+	if old.Spec.Unschedulable != cur.Spec.Unschedulable {
+		return true
+	}
+	if utils.NodeIsReady(old) != utils.NodeIsReady(cur) {
+		return true
+	}
+	return false
+}

pkg/controller/utils_test.go

@@ -153,3 +153,73 @@ func TestAddInstanceGroupsAnnotation(t *testing.T) {
 		}
 	}
 }
+
+func TestNodeStatusChanged(t *testing.T) {
+	testCases := []struct {
+		desc   string
+		mutate func(node *api_v1.Node)
+		expect bool
+	}{
+		{
+			"no change",
+			func(node *api_v1.Node) {},
+			false,
+		},
+		{
+			"unSchedulable changes",
+			func(node *api_v1.Node) {
+				node.Spec.Unschedulable = true
+			},
+			true,
+		},
+		{
+			"readiness changes",
+			func(node *api_v1.Node) {
+				node.Status.Conditions[0].Status = api_v1.ConditionFalse
+				node.Status.Conditions[0].LastTransitionTime = meta_v1.NewTime(time.Now())
+			},
+			true,
+		},
+		{
+			"new heartbeat",
+			func(node *api_v1.Node) {
+				node.Status.Conditions[0].LastHeartbeatTime = meta_v1.NewTime(time.Now())
+			},
+			false,
+		},
+	}
+
+	for _, tc := range testCases {
+		node := testNode()
+		tc.mutate(node)
+		res := nodeStatusChanged(testNode(), node)
+		if res != tc.expect {
+			t.Fatalf("Test case %q got: %v, expected: %v", tc.desc, res, tc.expect)
+		}
+	}
+}
+
+func testNode() *api_v1.Node {
+	return &api_v1.Node{
+		ObjectMeta: meta_v1.ObjectMeta{
+			Namespace: "ns",
+			Name:      "node",
+			Annotations: map[string]string{
+				"key1": "value1",
+			},
+		},
+		Spec: api_v1.NodeSpec{
+			Unschedulable: false,
+		},
+		Status: api_v1.NodeStatus{
+			Conditions: []api_v1.NodeCondition{
+				{
+					Type:               api_v1.NodeReady,
+					Status:             api_v1.ConditionTrue,
+					LastHeartbeatTime:  meta_v1.NewTime(time.Date(2000, 01, 1, 1, 0, 0, 0, time.UTC)),
+					LastTransitionTime: meta_v1.NewTime(time.Date(2000, 01, 1, 1, 0, 0, 0, time.UTC)),
+				},
+			},
+		},
+	}
+}