watch for node delete events and call reclaimRemovedPeers

prog/kube-utils/main.go

@@ -8,12 +8,19 @@ package main
 import (
 	"flag"
 	"fmt"
+	"math/rand"
 	"net"
 	"os"
+	"os/signal"
+	"syscall"
+	"time"
 
+	v1core "k8s.io/api/core/v1"
 	api "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/cache"
 
 	weaveapi "github.com/weaveworks/weave/api"
 	"github.com/weaveworks/weave/common"
@@ -194,6 +201,42 @@ func reclaimPeer(weave weaveClient, cml *configMapAnnotations, storedPeerList *p
 	return true, err
 }
 
+// regiesters with Kubernetes API server for node delete events. Node delete event handler
+// invokes reclaimRemovedPeers to remove it from IPAM so that IP space is reclaimed
+func registerForNodeUpdates(client *kubernetes.Clientset, stopCh <-chan struct{}, nodeName, peerName string) {
+	informerFactory := informers.NewSharedInformerFactory(client, 0)
+	nodeInformer := informerFactory.Core().V1().Nodes().Informer()
+	common.Log.Debugln("registering for updates for node delete events")
+	nodeInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		DeleteFunc: func(obj interface{}) {
+			node := obj.(*v1core.Node)
+			common.Log.Debugln("[kube-peers] Nodes deleted:", node.Name)
+			config, err := rest.InClusterConfig()
+			if err != nil {
+				common.Log.Fatalf("[kube-peers] Could not get cluster config: %v", err)
+			}
+			client, err := kubernetes.NewForConfig(config)
+			if err != nil {
+				common.Log.Fatalf("[kube-peers] Could not make Kubernetes connection: %v", err)
+			}
+			cml := newConfigMapAnnotations(configMapNamespace, configMapName, client)
+			weave := weaveapi.NewClient(os.Getenv("WEAVE_HTTP_ADDR"), common.Log)
+
+			// add random delay to avoid all nodes acting on node delete event at the same
+			// time leading to contention to use `weave-net` configmap
+			r := rand.Intn(5000)
+			time.Sleep(time.Duration(r) * time.Millisecond)
+
+			err = reclaimRemovedPeers(client, weave, cml, peerName, nodeName)
+			if err != nil {
+				common.Log.Fatalf("[kube-peers] Error while reclaiming space: %v", err)
+			}
+		},
+	})
+	informerFactory.WaitForCacheSync(stopCh)
+	informerFactory.Start(stopCh)
+}
+
 func main() {
 	var (
 		justReclaim       bool
@@ -202,8 +245,10 @@ func main() {
 		peerName          string
 		nodeName          string
 		logLevel          string
+		runReclaimDaemon  bool
 	)
 	flag.BoolVar(&justReclaim, "reclaim", false, "reclaim IP space from dead peers")
+	flag.BoolVar(&runReclaimDaemon, "run-reclaim-daemon", false, "run background process that reclaim IP space from dead peers ")
 	flag.BoolVar(&justCheck, "check-peer-new", false, "return success if peer name is not stored in annotation")
 	flag.BoolVar(&justSetNodeStatus, "set-node-status", false, "set NodeNetworkUnavailable to false")
 	flag.StringVar(&peerName, "peer-name", "unknown", "name of this Weave Net peer")
@@ -262,4 +307,14 @@ func main() {
 	for _, node := range peers {
 		fmt.Println(node.addr)
 	}
+
+	if runReclaimDaemon {
+		// Handle SIGINT and SIGTERM
+		ch := make(chan os.Signal)
+		signal.Notify(ch, syscall.SIGINT, syscall.SIGTERM)
+		stopCh := make(chan struct{})
+		registerForNodeUpdates(c, stopCh, nodeName, peerName)
+		<-ch
+		close(stopCh)
+	}
 }

prog/weave-kube/launch.sh

@@ -151,6 +151,8 @@ post_start_actions() {
 
     # Mark network as up
     /home/weave/kube-utils -set-node-status -node-name="$HOSTNAME"
+
+    /home/weave/kube-utils -run-reclaim-daemon -node-name="$HOSTNAME" -peer-name="$PEERNAME" -log-level=debug&
 }
 
 post_start_actions &