Add new Node api watcher which watches for add/remove nodes events.

On add/remove node events, perform refresh of peers to the peers as per the
current set of active nodes. If a node is removed, delete the BGP nieghbor relation.

Fixes #14

Author: Murali Reddy

app/controllers/network_routes_controller.go

@@ -35,6 +35,9 @@ type NetworkRoutingController struct {
 	asnNumber          uint32
 	peerAsnNumber      uint32
 }
+var(
+	activeNodes = make(map[string]bool)
+)
 
 func (nrc *NetworkRoutingController) Run(stopCh <-chan struct{}, wg *sync.WaitGroup) {
 
@@ -61,32 +64,8 @@ func (nrc *NetworkRoutingController) Run(stopCh <-chan struct{}, wg *sync.WaitGr
 	defer t.Stop()
 	defer wg.Done()
 
-	nodes, err := nrc.clientset.Core().Nodes().List(metav1.ListOptions{})
-	if err != nil {
-		glog.Errorf("Failed to list nodes: %s", err.Error())
-		return
-	}
-
 	glog.Infof("Starting network route controller")
 
-	// add the current set of nodes (excluding self) as BGP peers. Nodes form full mesh
-	for _, node := range nodes.Items {
-		nodeIP, _ := getNodeIP(&node)
-		if nodeIP.String() == nrc.nodeIP.String() {
-			continue
-		}
-
-		n := &config.Neighbor{
-			Config: config.NeighborConfig{
-				NeighborAddress: nodeIP.String(),
-				PeerAs:          nrc.asnNumber,
-			},
-		}
-		if err := nrc.bgpServer.AddNeighbor(n); err != nil {
-			panic(err)
-		}
-	}
-
 	// if the global routing peer is configured then peer with it
 	if len(nrc.peerRouter) != 0 {
 		n := &config.Neighbor{
@@ -109,6 +88,9 @@ func (nrc *NetworkRoutingController) Run(stopCh <-chan struct{}, wg *sync.WaitGr
 		default:
 		}
 
+		// add the current set of nodes (excluding self) as BGP peers. Nodes form full mesh
+		nrc.syncPeers()
+
 		// advertise cluster IP for the service to be reachable via host
 		if nrc.advertiseClusterIp {
 			glog.Infof("Advertising cluster ips")
@@ -209,7 +191,108 @@ func (nrc *NetworkRoutingController) injectRoute(path *table.Path) error {
 }
 
 func (nrc *NetworkRoutingController) Cleanup() {
+}
+
+// Refresh the peer relationship rest of the nodes in the cluster. Node add/remove
+// events should ensure peer relationship with only currently active nodes. In case
+// we miss any events from API server this method which is called periodically
+// ensure peer relationship with removed nodes is deleted.
+func (nrc *NetworkRoutingController) syncPeers() {
+
+	// get the current list of the nodes from API server
+	nodes, err := nrc.clientset.Core().Nodes().List(metav1.ListOptions{})
+	if err != nil {
+		glog.Errorf("Failed to list nodes: %s", err.Error())
+		return
+	}
+
+	// establish peer with current set of nodes
+	currentNodes := make([]string, 0)
+	for _, node := range nodes.Items {
+		nodeIP, _ := getNodeIP(&node)
+		if nodeIP.String() == nrc.nodeIP.String() {
+			continue
+		}
+		currentNodes = append(currentNodes, nodeIP.String())
+		activeNodes[nodeIP.String()] = true
+		n := &config.Neighbor{
+			Config: config.NeighborConfig{
+				NeighborAddress: nodeIP.String(),
+				PeerAs:          nrc.asnNumber,
+			},
+		}
+		// TODO: check if a node is alredy added as nieighbour in a better way that add and catch error
+		if err := nrc.bgpServer.AddNeighbor(n); err != nil {
+			if !strings.Contains(err.Error(), "Can't overwrite the existing peer") {
+				glog.Errorf("Failed to add node %s as peer due to %s", nodeIP.String(), err)
+			}
+		}
+	}
+
+	// find the list of the node removed, from the last known list of active nodes
+	removedNodes := make([]string, 0)
+	for ip, _ := range activeNodes {
+		stillActive := false
+		for _, node := range currentNodes {
+			if ip == node {
+				stillActive = true
+				break
+			}
+		}
+		if !stillActive {
+			removedNodes = append(removedNodes, ip)
+		}
+	}
 
+	// delete the neighbor for the node that is removed
+	for _, ip := range removedNodes {
+		n := &config.Neighbor{
+			Config: config.NeighborConfig{
+				NeighborAddress: ip,
+				PeerAs:          nrc.asnNumber,
+			},
+		}
+		if err := nrc.bgpServer.DeleteNeighbor(n); err != nil {
+			glog.Errorf("Failed to remove node %s as peer due to %s", ip, err)
+		}
+		delete(activeNodes, ip)
+	}
+}
+
+// Handle updates from Node watcher. Node watcher calls this method whenever there is
+// new node is added or old node is deleted. So peer up with new node and drop peering
+// from old node
+func (nrc *NetworkRoutingController) OnNodeUpdate(nodeUpdate *watchers.NodeUpdate) {
+	nrc.mu.Lock()
+	defer nrc.mu.Unlock()
+
+	node := nodeUpdate.Node
+	nodeIP, _ := getNodeIP(node)
+	if nodeUpdate.Op == watchers.ADD {
+		glog.Infof("Received node %s added update from watch API so peer with new node", nodeIP)
+		n := &config.Neighbor{
+			Config: config.NeighborConfig{
+				NeighborAddress: nodeIP.String(),
+				PeerAs:          nrc.asnNumber,
+			},
+		}
+		if err := nrc.bgpServer.AddNeighbor(n); err != nil {
+			glog.Errorf("Failed to add node %s as peer due to %s", nodeIP, err)
+		}
+		activeNodes[nodeIP.String()] = true
+	} else if nodeUpdate.Op == watchers.REMOVE {
+		glog.Infof("Received node %s removed update from watch API, so remove node from peer", nodeIP)
+		n := &config.Neighbor{
+			Config: config.NeighborConfig{
+				NeighborAddress: nodeIP.String(),
+				PeerAs:          nrc.asnNumber,
+			},
+		}
+		if err := nrc.bgpServer.DeleteNeighbor(n); err != nil {
+			glog.Errorf("Failed to remove node %s as peer due to %s", nodeIP, err)
+		}
+		delete(activeNodes, nodeIP.String())
+	}
 }
 
 func NewNetworkRoutingController(clientset *kubernetes.Clientset, kubeRouterConfig *options.KubeRouterConfig) (*NetworkRoutingController, error) {
@@ -294,6 +377,7 @@ func NewNetworkRoutingController(clientset *kubernetes.Clientset, kubeRouterConf
 		panic(err)
 	}
 
+	watchers.NodeWatcher.RegisterHandler(&nrc)
 	go nrc.watchBgpUpdates()
 
 	return &nrc, nil

app/server.go

@@ -83,6 +83,11 @@ func (kr *KubeRouter) startApiWatchers() error {
 		return errors.New("Failed to launch service api watcher: " + err.Error())
 	}
 
+	_, err = watchers.StartNodeWatcher(kr.Client, kr.Config.ConfigSyncPeriod)
+	if err != nil {
+		return errors.New("Failed to launch nodes api watcher: " + err.Error())
+	}
+
 	return nil
 }
 
@@ -92,6 +97,7 @@ func (kr *KubeRouter) stopApiWatchers() {
 	watchers.StopNetworkPolicyWatcher()
 	watchers.StopNamespaceWatcher()
 	watchers.StopServiceWatcher()
+	watchers.StopNodeWatcher()
 }
 
 func (kr *KubeRouter) Run() error {

app/watchers/namespace_watcher.go

@@ -49,7 +49,7 @@ func (nsw *namespaceWatcher) namespaceDeleteEventHandler(obj interface{}) {
 	nsw.broadcaster.Notify(&NamespaceUpdate{Op: REMOVE, Namespace: namespace})
 }
 
-func (nsw *namespaceWatcher) namespaceAUpdateEventHandler(oldObj, newObj interface{}) {
+func (nsw *namespaceWatcher) namespaceUpdateEventHandler(oldObj, newObj interface{}) {
 	namespace, ok := newObj.(*api.Namespace)
 	if !ok {
 		return
@@ -81,7 +81,7 @@ func StartNamespaceWatcher(clientset *kubernetes.Clientset, resyncPeriod time.Du
 	eventHandler := cache.ResourceEventHandlerFuncs{
 		AddFunc:    nsw.namespaceAddEventHandler,
 		DeleteFunc: nsw.namespaceDeleteEventHandler,
-		UpdateFunc: nsw.namespaceAUpdateEventHandler,
+		UpdateFunc: nsw.namespaceUpdateEventHandler,
 	}
 
 	nsw.clientset = clientset

app/watchers/node_watcher.go

@@ -0,0 +1,101 @@
+package watchers
+
+import (
+	"time"
+
+	"github.com/cloudnativelabs/kube-router/utils"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/client-go/kubernetes"
+	api "k8s.io/client-go/pkg/api/v1"
+	cache "k8s.io/client-go/tools/cache"
+)
+
+type NodeUpdate struct {
+	Node *api.Node
+	Op  Operation
+}
+
+var (
+	NodeWatcher *nodeWatcher
+)
+
+type nodeWatcher struct {
+	clientset     *kubernetes.Clientset
+	nodeController cache.Controller
+	nodeLister     cache.Indexer
+	broadcaster   *utils.Broadcaster
+}
+
+type NodeUpdatesHandler interface {
+	OnNodeUpdate(nodeUpdate *NodeUpdate)
+}
+
+func (nw *nodeWatcher) nodeAddEventHandler(obj interface{}) {
+	node, ok := obj.(*api.Node)
+	if !ok {
+		return
+	}
+	nw.broadcaster.Notify(&NodeUpdate{Op: ADD, Node: node})
+}
+
+func (nw *nodeWatcher) nodeDeleteEventHandler(obj interface{}) {
+	node, ok := obj.(*api.Node)
+	if !ok {
+		return
+	}
+	nw.broadcaster.Notify(&NodeUpdate{Op: REMOVE, Node: node})
+}
+
+func (nw *nodeWatcher) nodeUpdateEventHandler(oldObj, newObj interface{}) {
+	// we are interested only node add/delete, so skip update
+	return
+}
+
+func (nw *nodeWatcher) RegisterHandler(handler NodeUpdatesHandler) {
+	nw.broadcaster.Add(utils.ListenerFunc(func(instance interface{}) {
+		handler.OnNodeUpdate(instance.(*NodeUpdate))
+	}))
+}
+
+func (nw *nodeWatcher) List() []*api.Node {
+	obj_list := nw.nodeLister.List()
+	node_instances := make([]*api.Node, len(obj_list))
+	for i, ins := range obj_list {
+		node_instances[i] = ins.(*api.Node)
+	}
+	return node_instances
+}
+
+func (nw *nodeWatcher) HasSynced() bool {
+	return nw.nodeController.HasSynced()
+}
+
+var nodewatchStopCh chan struct{}
+
+func StartNodeWatcher(clientset *kubernetes.Clientset, resyncPeriod time.Duration) (*nodeWatcher, error) {
+
+	nw := nodeWatcher{}
+	NodeWatcher = &nw
+	eventHandler := cache.ResourceEventHandlerFuncs{
+		AddFunc:    nw.nodeAddEventHandler,
+		DeleteFunc: nw.nodeDeleteEventHandler,
+		UpdateFunc: nw.nodeUpdateEventHandler,
+	}
+
+	nw.clientset = clientset
+	nw.broadcaster = utils.NewBroadcaster()
+	lw := cache.NewListWatchFromClient(clientset.Core().RESTClient(), "nodes", metav1.NamespaceAll, fields.Everything())
+	nw.nodeLister, nw.nodeController = cache.NewIndexerInformer(
+		lw,
+		&api.Node{}, resyncPeriod, eventHandler,
+		cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
+	)
+	nodewatchStopCh = make(chan struct{})
+	go nw.nodeController.Run(nodewatchStopCh)
+	return &nw, nil
+}
+
+func StopNodeWatcher() {
+	nodewatchStopCh <- struct{}{}
+}