diff --git a/prog/kube-utils/main.go b/prog/kube-utils/main.go index 5926394d..bd96f0b7 100644 --- a/prog/kube-utils/main.go +++ b/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) + } } diff --git a/prog/weave-kube/launch.sh b/prog/weave-kube/launch.sh index b82e4067..e5fb393f 100755 --- a/prog/weave-kube/launch.sh +++ b/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 &