Merge pull request #3149 from weaveworks/fix-2797-loosing-ipam-ips

Fix 2797 on Kubernetes - Cluster goes down because all IPs become unreachable

api/ipam.go

@@ -62,6 +62,12 @@ func (client *Client) ReleaseIPsFor(ID string) error {
 	return err
 }
 
+// release all IP space owned by a peer
+func (client *Client) RmPeer(peerName string) (string, error) {
+	result, err := client.httpVerb("DELETE", fmt.Sprintf("/peer/%s", peerName), nil)
+	return result, err
+}
+
 func (client *Client) DefaultSubnet() (*net.IPNet, error) {
 	cidr, err := client.httpVerb("GET", fmt.Sprintf("/ipinfo/defaultsubnet"), nil)
 	if err != nil {

prog/kube-peers/annotations.go

@@ -0,0 +1,195 @@
+/*
+In order to keep track of active weave peers, we use annotations on the Kubernetes cluster.
+
+Kubernetes uses etcd to distribute and synchronise these annotations so we don't have to.
+
+This module deals with operations on the peerlist backed by Kubernetes' annotation mechanism.
+*/
+package main
+
+import (
+	"encoding/json"
+	"log"
+	"time"
+
+	"github.com/pkg/errors"
+
+	v1 "k8s.io/api/core/v1"
+	kubeErrors "k8s.io/apimachinery/pkg/api/errors"
+	api "k8s.io/apimachinery/pkg/apis/meta/v1"
+	wait "k8s.io/apimachinery/pkg/util/wait"
+	kubernetes "k8s.io/client-go/kubernetes"
+	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
+)
+
+type configMapAnnotations struct {
+	ConfigMapName string
+	Namespace     string
+	Client        corev1client.ConfigMapsGetter
+	cm            *v1.ConfigMap
+}
+
+func newConfigMapAnnotations(ns string, configMapName string, clientset *kubernetes.Clientset) *configMapAnnotations {
+	return &configMapAnnotations{
+		Namespace:     ns,
+		ConfigMapName: configMapName,
+		Client:        clientset.CoreV1(),
+	}
+}
+
+type peerList struct {
+	Peers []peerInfo
+}
+
+type peerInfo struct {
+	PeerName string // Weave internal unique ID
+	NodeName string // Kubernetes node name
+}
+
+func (pl peerList) contains(peerName string) bool {
+	for _, peer := range pl.Peers {
+		if peer.PeerName == peerName {
+			return true
+		}
+	}
+	return false
+}
+
+func (pl *peerList) add(peerName string, name string) {
+	pl.Peers = append(pl.Peers, peerInfo{PeerName: peerName, NodeName: name})
+}
+
+func (pl *peerList) remove(peerNameToRemove string) {
+	for i := 0; i < len(pl.Peers); {
+		if pl.Peers[i].PeerName == peerNameToRemove {
+			pl.Peers = append(pl.Peers[:i], pl.Peers[i+1:]...)
+		} else {
+			i++
+		}
+	}
+}
+
+const (
+	retryPeriod  = time.Second * 2
+	jitterFactor = 1.0
+
+	// KubePeersAnnotationKey is the default annotation key
+	KubePeersAnnotationKey = "kube-peers.weave.works/peers"
+)
+
+func (cml *configMapAnnotations) Init() error {
+	for {
+		// Since it's potentially racy to GET, then CREATE if not found, we wrap in a check loop
+		// so that if the configmap is created after our GET but before or CREATE, we'll gracefully
+		// re-try to get the configmap.
+		var err error
+		cml.cm, err = cml.Client.ConfigMaps(cml.Namespace).Get(cml.ConfigMapName, api.GetOptions{})
+		if err != nil {
+			if !kubeErrors.IsNotFound(err) {
+				return errors.Wrapf(err, "Unable to fetch ConfigMap %s/%s", cml.Namespace, cml.ConfigMapName)
+			}
+			cml.cm, err = cml.Client.ConfigMaps(cml.Namespace).Create(&v1.ConfigMap{
+				ObjectMeta: api.ObjectMeta{
+					Name:      cml.ConfigMapName,
+					Namespace: cml.Namespace,
+				},
+			})
+			if err != nil {
+				if kubeErrors.IsAlreadyExists(err) {
+					continue
+				}
+				return errors.Wrapf(err, "Unable to create ConfigMap %s/%s", cml.Namespace, cml.ConfigMapName)
+			}
+		}
+		break
+	}
+	if cml.cm.Annotations == nil {
+		cml.cm.Annotations = make(map[string]string)
+	}
+	return nil
+}
+
+func (cml *configMapAnnotations) GetPeerList() (*peerList, error) {
+	var record peerList
+	if cml.cm == nil {
+		return nil, errors.New("endpoint not initialized, call Init first")
+	}
+	if recordBytes, found := cml.cm.Annotations[KubePeersAnnotationKey]; found {
+		if err := json.Unmarshal([]byte(recordBytes), &record); err != nil {
+			return nil, err
+		}
+	}
+	return &record, nil
+}
+
+func (cml *configMapAnnotations) UpdatePeerList(list peerList) error {
+	recordBytes, err := json.Marshal(list)
+	if err != nil {
+		return err
+	}
+	return cml.UpdateAnnotation(KubePeersAnnotationKey, string(recordBytes))
+}
+
+func (cml *configMapAnnotations) UpdateAnnotation(key, value string) error {
+	if cml.cm == nil {
+		return errors.New("endpoint not initialized, call Init first")
+	}
+	cm := cml.cm
+	cm.Annotations[key] = value
+	cm, err := cml.Client.ConfigMaps(cml.Namespace).Update(cml.cm)
+	if err == nil {
+		cml.cm = cm
+	}
+	return err
+}
+
+func (cml *configMapAnnotations) RemoveAnnotation(key string) error {
+	if cml.cm == nil {
+		return errors.New("endpoint not initialized, call Init first")
+	}
+	cm := cml.cm
+	delete(cm.Annotations, key)
+	cm, err := cml.Client.ConfigMaps(cml.Namespace).Update(cml.cm)
+	if err == nil {
+		cml.cm = cm
+	}
+	return err
+}
+
+func (cml *configMapAnnotations) RemoveAnnotationsWithValue(valueToRemove string) error {
+	if cml.cm == nil {
+		return errors.New("endpoint not initialized, call Init first")
+	}
+	cm := cml.cm
+	for key, value := range cm.Annotations {
+		if value == valueToRemove {
+			delete(cm.Annotations, key)
+		}
+	}
+	cm, err := cml.Client.ConfigMaps(cml.Namespace).Update(cml.cm)
+	if err == nil {
+		cml.cm = cm
+	}
+	return err
+}
+
+// Loop with jitter, fetching the cml data and calling f() until it
+// doesn't get an optimistic locking conflict.
+// If it succeeds or gets any other kind of error, stop the loop.
+func (cml *configMapAnnotations) LoopUpdate(f func() error) error {
+	stop := make(chan struct{})
+	var err error
+	wait.JitterUntil(func() {
+		if err = cml.Init(); err != nil {
+			close(stop)
+			return
+		}
+		err = f()
+		if err != nil && kubeErrors.IsConflict(err) {
+			log.Printf("Optimistic locking conflict: trying again: %s", err)
+			return
+		}
+		close(stop)
+	}, retryPeriod, jitterFactor, true, stop)
+	return err
+}

prog/kube-peers/main.go

@@ -1,29 +1,36 @@
+/*
+Package main deals with weave-net peers on the cluster.
+
+This involves peer management, such as getting the latest peers or removing defunct peers from the cluster
+*/
 package main
 
 import (
+	"flag"
 	"fmt"
-	"log"
 	"net"
+	"os"
 
 	api "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
+
+	weaveapi "github.com/weaveworks/weave/api"
+	"github.com/weaveworks/weave/common"
 )
 
-func getKubePeers() ([]string, error) {
-	config, err := rest.InClusterConfig()
+type nodeInfo struct {
+	name string
+	addr string
+}
+
+// return the IP addresses of all nodes in the cluster
+func getKubePeers(c *kubernetes.Clientset) ([]nodeInfo, error) {
+	nodeList, err := c.CoreV1().Nodes().List(api.ListOptions{})
 	if err != nil {
 		return nil, err
 	}
-	c, err := kubernetes.NewForConfig(config)
-	if err != nil {
-		return nil, err
-	}
-	nodeList, err := c.Core().Nodes().List(api.ListOptions{})
-	if err != nil {
-		return nil, err
-	}
-	addresses := make([]string, 0, len(nodeList.Items))
+	addresses := make([]nodeInfo, 0, len(nodeList.Items))
 	for _, peer := range nodeList.Items {
 		var internalIP, externalIP string
 		for _, addr := range peer.Status.Addresses {
@@ -41,20 +48,187 @@ func getKubePeers() ([]string, error) {
 
 		// Fallback for cases where a Node has an ExternalIP but no InternalIP
 		if internalIP != "" {
-			addresses = append(addresses, internalIP)
+			addresses = append(addresses, nodeInfo{name: peer.Name, addr: internalIP})
 		} else if externalIP != "" {
-			addresses = append(addresses, externalIP)
+			addresses = append(addresses, nodeInfo{name: peer.Name, addr: externalIP})
 		}
 	}
 	return addresses, nil
 }
 
-func main() {
-	peers, err := getKubePeers()
-	if err != nil {
-		log.Fatalf("Could not get peers: %v", err)
+const (
+	configMapName      = "weave-net"
+	configMapNamespace = "kube-system"
+)
+
+// update the list of all peers that have gone through this code path
+func addMyselfToPeerList(cml *configMapAnnotations, c *kubernetes.Clientset, peerName, name string) (*peerList, error) {
+	var list *peerList
+	err := cml.LoopUpdate(func() error {
+		var err error
+		list, err = cml.GetPeerList()
+		if err != nil {
+			return err
+		}
+		if !list.contains(peerName) {
+			list.add(peerName, name)
+			err = cml.UpdatePeerList(*list)
+			if err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+	common.Log.Infoln("[kube-peers] Added myself to peer list", list)
+	return list, err
+}
+
+func checkIamInPeerList(cml *configMapAnnotations, c *kubernetes.Clientset, peerName string) (bool, error) {
+	if err := cml.Init(); err != nil {
+		return false, err
 	}
-	for _, addr := range peers {
-		fmt.Println(addr)
+	list, err := cml.GetPeerList()
+	if err != nil {
+		return false, err
+	}
+	common.Log.Debugf("[kube-peers] Checking peer %q against list %v", peerName, list)
+	return list.contains(peerName), nil
+}
+
+// For each of those peers that is no longer listed as a node by
+// Kubernetes, remove it from Weave IPAM
+func reclaimRemovedPeers(weave *weaveapi.Client, cml *configMapAnnotations, nodes []nodeInfo, myPeerName string) error {
+	for {
+		// 1. Compare peers stored in the peerList against all peers reported by k8s now.
+		storedPeerList, err := cml.GetPeerList()
+		if err != nil {
+			return err
+		}
+		peerMap := make(map[string]peerInfo, len(storedPeerList.Peers))
+		for _, peer := range storedPeerList.Peers {
+			peerMap[peer.NodeName] = peer
+		}
+		for _, node := range nodes {
+			delete(peerMap, node.name)
+		}
+		common.Log.Debugln("[kube-peers] Nodes that have disappeared:", peerMap)
+		if len(peerMap) == 0 {
+			break
+		}
+		// 2. Loop for each X in the first set and not in the second - we wish to remove X from our data structures
+		for _, peer := range peerMap {
+			common.Log.Debugln("[kube-peers] Preparing to remove disappeared peer", peer)
+			okToRemove := false
+			// 3. Check if there is an existing annotation with key X
+			if existingAnnotation, found := cml.cm.Annotations[peer.PeerName]; found {
+				common.Log.Debugln("[kube-peers] Existing annotation", existingAnnotation)
+				// 4.   If annotation already contains my identity, ok;
+				if existingAnnotation == myPeerName {
+					okToRemove = true
+				}
+			} else {
+				// 5.   If non-existent, write an annotation with key X and contents "my identity"
+				common.Log.Debugln("[kube-peers] Noting I plan to remove ", peer.PeerName)
+				if err := cml.UpdateAnnotation(peer.PeerName, myPeerName); err == nil {
+					okToRemove = true
+				}
+			}
+			if okToRemove {
+				// 6.   If step 4 or 5 succeeded, rmpeer X
+				result, err := weave.RmPeer(peer.PeerName)
+				common.Log.Infoln("[kube-peers] rmpeer of", peer.PeerName, ":", result)
+				if err != nil {
+					return err
+				}
+				cml.LoopUpdate(func() error {
+					// 7aa.   Remove any annotations Z* that have contents X
+					if err := cml.RemoveAnnotationsWithValue(peer.PeerName); err != nil {
+						return err
+					}
+					// 7a.    Remove X from peerList
+					storedPeerList.remove(peer.PeerName)
+					common.Log.Infoln("[kube-peers] Removing peer ", peer.PeerName, ". Expecting to remove linked annotation next.")
+					if err := cml.UpdatePeerList(*storedPeerList); err != nil {
+						return err
+					}
+					common.Log.Infoln("[kube-peers] Removing annotation ", peer.PeerName)
+					// 7b.    Remove annotation with key X
+					return cml.RemoveAnnotation(peer.PeerName)
+				})
+				common.Log.Debugln("[kube-peers] Finished removal of ", peer.PeerName)
+			}
+			// 8.   If step 5 failed due to optimistic lock conflict, stop: someone else is handling X
+
+			// Step 3-5 is to protect against two simultaneous rmpeers of X
+			// Step 4 is to pick up again after a restart between step 5 and step 7b
+			// If the peer doing the reclaim disappears between steps 5 and 7a, then someone will clean it up in step 7aa
+			// If peer doing the reclaim disappears forever between 7a and 7b then we get a dangling annotation
+			// This should be sufficiently rare that we don't care.
+		}
+
+		// 9. Go back to step 1 until there is no difference between the two sets
+	}
+	// Question: Should we narrow step 2 by checking against Weave Net IPAM?
+	// i.e. If peer X owns any address space and is marked unreachable, we want to rmpeer X
+	return nil
+}
+
+func main() {
+	var (
+		justReclaim bool
+		justCheck   bool
+		peerName    string
+		nodeName    string
+		logLevel    string
+	)
+	flag.BoolVar(&justReclaim, "reclaim", false, "reclaim IP space from dead peers")
+	flag.BoolVar(&justCheck, "check-peer-exists", false, "return success if peer name is stored in annotation")
+	flag.StringVar(&peerName, "peer-name", "unknown", "name of this Weave Net peer")
+	flag.StringVar(&nodeName, "node-name", "unknown", "name of this Kubernetes node")
+	flag.StringVar(&logLevel, "log-level", "info", "logging level (debug, info, warning, error)")
+	flag.Parse()
+
+	common.SetLogLevel(logLevel)
+	config, err := rest.InClusterConfig()
+	if err != nil {
+		common.Log.Fatalf("[kube-peers] Could not get cluster config: %v", err)
+	}
+	c, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		common.Log.Fatalf("[kube-peers] Could not make Kubernetes connection: %v", err)
+	}
+	if justCheck {
+		cml := newConfigMapAnnotations(configMapNamespace, configMapName, c)
+		exists, err := checkIamInPeerList(cml, c, peerName)
+		if err != nil {
+			common.Log.Fatalf("[kube-peers] Could not check peer list: %v", err)
+		}
+		if exists {
+			os.Exit(0)
+		} else {
+			os.Exit(9)
+		}
+	}
+	peers, err := getKubePeers(c)
+	if err != nil {
+		common.Log.Fatalf("[kube-peers] Could not get peers: %v", err)
+	}
+	if justReclaim {
+		cml := newConfigMapAnnotations(configMapNamespace, configMapName, c)
+
+		_, err := addMyselfToPeerList(cml, c, peerName, nodeName)
+		if err != nil {
+			common.Log.Fatalf("[kube-peers] Could not update peer list: %v", err)
+		}
+
+		weave := weaveapi.NewClient(os.Getenv("WEAVE_HTTP_ADDR"), common.Log)
+		err = reclaimRemovedPeers(weave, cml, peers, peerName)
+		if err != nil {
+			common.Log.Fatalf("[kube-peers] Error while reclaiming space: %v", err)
+		}
+		return
+	}
+	for _, node := range peers {
+		fmt.Println(node.addr)
 	}
 }

prog/weave-kube/launch.sh

@@ -30,6 +30,7 @@ HTTP_ADDR=${WEAVE_HTTP_ADDR:-127.0.0.1:6784}
 STATUS_ADDR=${WEAVE_STATUS_ADDR:-0.0.0.0:6782}
 HOST_ROOT=${HOST_ROOT:-/host}
 CONN_LIMIT=${CONN_LIMIT:-30}
+DB_PREFIX=${DB_PREFIX:-/weavedb/weave-net}
 
 # Check if the IP range overlaps anything existing on the host
 /usr/bin/weaveutil netcheck $IPALLOC_RANGE weave
@@ -87,6 +88,23 @@ if [ -z "$IPALLOC_INIT" ]; then
     IPALLOC_INIT="consensus=$(peer_count $KUBE_PEERS)"
 fi
 
+# Find out what peer name we will be using
+PEERNAME=$(/home/weave/weaver $EXTRA_ARGS --print-peer-name --host-root=$HOST_ROOT --db-prefix="$DB_PREFIX")
+if [ -z "$PEERNAME" ]; then
+    echo "Unable to get peer name" >&2
+    exit 2
+fi
+
+# If this peer name is not stored in the list, either we are a
+# brand-new peer or we were removed by another peer while temporarily
+# absent. In order to avoid a CRDT clash for the latter case, clean up
+if ! /home/weave/kube-peers -check-peer-exists -peer-name="$PEERNAME" -log-level=debug ; then
+    if [ -f ${DB_PREFIX}data.db ]; then
+        echo "Peer not in list; removing persisted data" >&2
+        rm -f ${DB_PREFIX}data.db
+    fi
+fi
+
 post_start_actions() {
     # Wait for weave process to become responsive
     while true ; do
@@ -108,6 +126,8 @@ post_start_actions() {
     export HOST_ROOT
     /home/weave/weave --local setup-cni
 
+    /home/weave/kube-peers -reclaim -node-name="$HOSTNAME" -peer-name="$PEERNAME" -log-level=debug
+
     # Expose the weave network so host processes can communicate with pods
     /home/weave/weave --local expose $WEAVE_EXPOSE_IP
 }
@@ -115,9 +135,10 @@ post_start_actions() {
 post_start_actions &
 
 /home/weave/weaver $EXTRA_ARGS --port=6783 $(router_bridge_opts) \
+     --name="$PEERNAME" \
      --host-root=$HOST_ROOT \
      --http-addr=$HTTP_ADDR --status-addr=$STATUS_ADDR --docker-api='' --no-dns \
-     --db-prefix="/weavedb/weave-net" \
+     --db-prefix="$DB_PREFIX" \
      --ipalloc-range=$IPALLOC_RANGE $NICKNAME_ARG \
      --ipalloc-init $IPALLOC_INIT \
      --conn-limit=$CONN_LIMIT \

prog/weave-kube/weave-daemonset-k8s-1.6.yaml

@@ -47,6 +47,44 @@ items:
       - kind: ServiceAccount
         name: weave-net
         namespace: kube-system
+  - apiVersion: rbac.authorization.k8s.io/v1beta1
+    kind: Role
+    metadata:
+      name: weave-net-kube-peer
+      namespace: kube-system
+      labels:
+        name: weave-net-kube-peer
+    rules:
+      - apiGroups:
+          - ''
+        resources:
+          - configmaps
+        resourceNames:
+          - weave-net
+        verbs:
+          - get
+          - update
+      - apiGroups:
+          - ''
+        resources:
+          - configmaps
+        verbs:
+          - create
+  - apiVersion: rbac.authorization.k8s.io/v1beta1
+    kind: RoleBinding
+    metadata:
+      name: weave-net-kube-peer
+      namespace: kube-system
+      labels:
+        name: weave-net-kube-peer
+    roleRef:
+      kind: Role
+      name: weave-net-kube-peer
+      apiGroup: rbac.authorization.k8s.io
+    subjects:
+      - kind: ServiceAccount
+        name: weave-net
+        namespace: kube-system
   - apiVersion: extensions/v1beta1
     kind: DaemonSet
     metadata:

prog/weave-kube/weave-daemonset-k8s-1.7.yaml

@@ -55,6 +55,44 @@ items:
       - kind: ServiceAccount
         name: weave-net
         namespace: kube-system
+  - apiVersion: rbac.authorization.k8s.io/v1beta1
+    kind: Role
+    metadata:
+      name: weave-net
+      namespace: kube-system
+      labels:
+        name: weave-net
+    rules:
+      - apiGroups:
+          - ''
+        resources:
+          - configmaps
+        resourceNames:
+          - weave-net
+        verbs:
+          - get
+          - update
+      - apiGroups:
+          - ''
+        resources:
+          - configmaps
+        verbs:
+          - create
+  - apiVersion: rbac.authorization.k8s.io/v1beta1
+    kind: RoleBinding
+    metadata:
+      name: weave-net
+      namespace: kube-system
+      labels:
+        name: weave-net
+    roleRef:
+      kind: Role
+      name: weave-net
+      apiGroup: rbac.authorization.k8s.io
+    subjects:
+      - kind: ServiceAccount
+        name: weave-net
+        namespace: kube-system
   - apiVersion: extensions/v1beta1
     kind: DaemonSet
     metadata:

prog/weaver/main.go

@@ -131,6 +131,7 @@ func main() {
 
 	var (
 		justVersion        bool
+		justPeerName       bool
 		config             mesh.Config
 		bridgeConfig       weavenet.BridgeConfig
 		networkConfig      weave.NetworkConfig
@@ -163,6 +164,7 @@ func main() {
 	)
 
 	mflag.BoolVar(&justVersion, []string{"-version"}, false, "print version and exit")
+	mflag.BoolVar(&justPeerName, []string{"-print-peer-name"}, false, "print peer name and exit")
 	mflag.StringVar(&config.Host, []string{"-host"}, "", "router host")
 	mflag.IntVar(&config.Port, []string{"-port"}, mesh.Port, "router port")
 	mflag.IntVar(&protocolMinVersion, []string{"-min-protocol-version"}, mesh.ProtocolMinVersion, "minimum weave protocol version")
@@ -226,6 +228,11 @@ func main() {
 		fmt.Printf("weave %s\n", version)
 		os.Exit(0)
 	}
+	name := peerName(routerName, bridgeConfig.WeaveBridgeName, dbPrefix, hostRoot)
+	if justPeerName {
+		fmt.Printf("%s\n", name)
+		os.Exit(0)
+	}
 
 	proxyConfig.DockerHost = dockerAPI
 	if bridgeConfig.AWSVPC {
@@ -284,8 +291,6 @@ func main() {
 		}
 	}
 
-	name := peerName(routerName, bridgeConfig.WeaveBridgeName, dbPrefix, hostRoot)
-
 	bridgeConfig.Mac = name.String()
 	bridgeConfig.Port = config.Port
 	bridgeType, err := weavenet.EnsureBridge(procPath, &bridgeConfig, Log)

test/870_weave_recovers_unreachable_ips_on_relaunch_3_test.sh

@@ -0,0 +1,117 @@
+#! /bin/bash
+
+. "$(dirname "$0")/config.sh"
+
+function howmany { echo $#; }
+
+#
+# Test vars
+#
+TOKEN=112233.4455667788990000
+HOST1IP=$($SSH $HOST1 "getent hosts $HOST1 | cut -f 1 -d ' '")
+NUM_HOSTS=$(howmany $HOSTS)
+SUCCESS="$(( $NUM_HOSTS * ($NUM_HOSTS-1) )) established"
+KUBECTL="sudo kubectl --kubeconfig /etc/kubernetes/admin.conf"
+KUBE_PORT=6443
+IMAGE=weaveworks/network-tester:latest
+
+if [ -n "$COVERAGE" ]; then
+    COVERAGE_ARGS="env:\\n                - name: EXTRA_ARGS\\n                  value: \"-test.coverprofile=/home/weave/cover.prof --\""
+else
+    COVERAGE_ARGS="env:"
+fi
+
+#
+# Utility functions
+#
+function teardown_kubernetes_cluster {
+    greyly echo "Tearing down kubernetes cluster"
+    for host in $HOSTS; do
+        run_on $host "sudo kubeadm reset && sudo rm -r -f /opt/cni/bin/*weave*"
+    done
+}
+
+function setup_kubernetes_cluster {
+    teardown_kubernetes_cluster;
+
+    greyly echo "Setting up kubernetes cluster"
+
+    # kubeadm init upgrades to latest Kubernetes version by default, therefore we try to lock the version using the below option:
+    k8s_version="$(run_on $HOST1 "kubelet --version" | grep -oP "(?<=Kubernetes )v[\d\.\-beta]+")"
+    k8s_version_option="$([[ "$k8s_version" > "v1.6" ]] && echo "kubernetes-version" || echo "use-kubernetes-version")"
+
+    for host in $HOSTS; do
+        if [ $host = $HOST1 ] ; then
+        run_on $host "sudo systemctl start kubelet && sudo kubeadm init --$k8s_version_option=$k8s_version --token=$TOKEN"
+        else
+        run_on $host "sudo systemctl start kubelet && sudo kubeadm join --token=$TOKEN $HOST1IP:$KUBE_PORT"
+        fi
+    done
+
+    # Ensure Kubernetes uses locally built container images and inject code coverage environment variable (or do nothing depending on $COVERAGE):
+    sed -e "s%imagePullPolicy: Always%imagePullPolicy: Never%" \
+        -e "s%env:%$COVERAGE_ARGS%" \
+        "$(dirname "$0")/../prog/weave-kube/weave-daemonset-k8s-1.7.yaml" | run_on "$HOST1" "$KUBECTL apply -n kube-system -f -"
+}
+
+function force_drop_node {
+    greyly echo "Dropping node $1 with 'sudo kubectl delete node'"
+    local target=$(echo "$1" | awk -F"." '{print $1}')
+    run_on $HOST1 "$KUBECTL delete node $target"
+}
+
+function weave_connected {
+    run_on $HOST1 "curl -sS http://127.0.0.1:6784/status | grep \"$SUCCESS\""
+}
+
+function unreachable_ip_addresses_count {
+    # Note Well: This will return 0 if weave is not running at all.
+    local host=$1
+    local count=$(get_command_output_on $host "sudo weave status ipam | grep unreachable | wc -l")
+    echo $count
+}
+
+function relaunch_weave_pod {
+    local target=$(echo "$1" | awk -F"." '{print $1}')
+
+    # This is a pretty complex jq query. Is there a simpler way to
+    # get the weave-net pod name back using kubectl templates?
+    local QUERY="$KUBECTL get pods -n kube-system -o json | jq -r \".items | map(select(.spec.nodeName == \\\"$target\\\")) | map(select(.metadata.labels.name == \\\"weave-net\\\")) | map(.metadata.name) | .[]\""
+    local podname=$(get_command_output_on $HOST1 $QUERY) # Run on master node
+    run_on $HOST1 "$KUBECTL get pods -n kube-system \"$podname\" -o yaml | $KUBECTL replace --force -f -"
+}
+
+#
+# Suite
+#
+function main {
+    local IPAM_RECOVER_DELAY=15
+
+    start_suite "Test weave-net deallocates from IPAM on node failure";
+
+    setup_kubernetes_cluster;
+
+    # Need to wait until all pods have come up
+    assert_raises "wait_for_x weave_connected 'pods to be running weave net'"
+
+    greyly echo "Checking unreachable IPs"
+    assert "unreachable_ip_addresses_count $HOST1" "0";
+    assert "unreachable_ip_addresses_count $HOST2" "0";
+    assert "unreachable_ip_addresses_count $HOST3" "0";
+
+    force_drop_node $HOST2;
+
+    relaunch_weave_pod $HOST3;
+
+    sleep $IPAM_RECOVER_DELAY;
+
+    greyly echo "Checking unreachable IPs"
+    assert "unreachable_ip_addresses_count $HOST1" "0";
+    assert "unreachable_ip_addresses_count $HOST3" "0";
+
+    teardown_kubernetes_cluster;
+
+    end_suite;
+}
+
+main

test/config.sh

@@ -104,6 +104,13 @@ run_on() {
     remote $host $SSH $host "$@"
 }
 
+get_command_output_on() {
+    host=$1
+    shift 1
+    [ -z "$DEBUG" ] || greyly echo "Running on $host: $@" >&2
+    $SSH $host "$@"
+}
+
 docker_on() {
     host=$1
     shift 1

test/run-integration-tests.sh

@@ -354,6 +354,13 @@ function main() {
             exit $status
             ;;
 
+        up) # Setup a test environment without actually doing any testing.
+            provision on "$PROVIDER"
+            configure "$ssh_user" "$ssh_hosts" "${ssh_port:-22}" "$ssh_id_file"
+            "$DIR/setup.sh"
+            echo_export_hosts
+            ;;
+
         provision)
             provision on "$PROVIDER"
             echo_export_hosts