Maintain a list of peers in a Kubernetes annotation

This is prework for a feature which will remove 'dead' peers

prog/kube-peers/annotations.go

@@ -0,0 +1,114 @@
+package main
+
+import (
+	"encoding/json"
+
+	"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"
+	kubernetes "k8s.io/client-go/kubernetes"
+	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
+)
+
+type configMapAnnotations struct {
+	Name      string
+	Namespace string
+	Client    corev1client.ConfigMapsGetter
+	cm        *v1.ConfigMap
+}
+
+func newConfigMapAnnotations(ns string, name string, client *kubernetes.Clientset) *configMapAnnotations {
+	return &configMapAnnotations{
+		Namespace: ns,
+		Name:      name,
+		Client:    client.CoreV1(),
+	}
+}
+
+type peerList struct {
+	Peers []peerInfo
+}
+
+type peerInfo struct {
+	PeerName string // Weave internal unique ID
+	Name     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, Name: name})
+}
+
+const (
+	// 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.Name, api.GetOptions{})
+		if err != nil {
+			if !kubeErrors.IsNotFound(err) {
+				return errors.Wrapf(err, "Unable to fetch ConfigMap %s/%s", cml.Namespace, cml.Name)
+			}
+			cml.cm, err = cml.Client.ConfigMaps(cml.Namespace).Create(&v1.ConfigMap{
+				ObjectMeta: api.ObjectMeta{
+					Name:      cml.Name,
+					Namespace: cml.Namespace,
+				},
+			})
+			if err != nil {
+				if kubeErrors.IsAlreadyExists(err) {
+					continue
+				}
+				return errors.Wrapf(err, "Unable to create ConfigMap %s/%s", cml.Namespace, cml.Name)
+			}
+		}
+		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
+}
+
+// Update will update and existing annotation on a given resource.
+func (cml *configMapAnnotations) UpdatePeerList(list peerList) error {
+	if cml.cm == nil {
+		return errors.New("endpoint not initialized, call Init first")
+	}
+	recordBytes, err := json.Marshal(list)
+	if err != nil {
+		return err
+	}
+	cml.cm.Annotations[KubePeersAnnotationKey] = string(recordBytes)
+	cml.cm, err = cml.Client.ConfigMaps(cml.Namespace).Update(cml.cm)
+	return err
+}

prog/kube-peers/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"flag"
 	"fmt"
 	"log"
 	"net"
@@ -10,12 +11,18 @@ import (
 	"k8s.io/client-go/rest"
 )
 
-func getKubePeers(c *kubernetes.Clientset) ([]string, error) {
-	nodeList, err := c.Nodes().List(api.ListOptions{})
+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
 	}
-	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 {
@@ -33,15 +40,51 @@ func getKubePeers(c *kubernetes.Clientset) ([]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
 }
 
+const (
+	configMapName      = "weave-net"
+	configMapNamespace = "kube-system"
+)
+
+// update the list of all peers that have gone through this code path
+func addMyselfToPeerList(c *kubernetes.Clientset, peerName, name string) (*peerList, error) {
+	cml := newConfigMapAnnotations(configMapNamespace, configMapName, c)
+	if err := cml.Init(); err != nil {
+		return nil, err
+	}
+	list, err := cml.GetPeerList()
+	if err != nil {
+		return nil, err
+	}
+	log.Println("Fetched existing peer list", list)
+	if !list.contains(peerName) {
+		list.add(peerName, name)
+		log.Println("Storing new peer list", list)
+		err = cml.UpdatePeerList(*list)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return list, nil
+}
+
 func main() {
+	var (
+		justReclaim bool
+		peerName    string
+		nodeName    string
+	)
+	flag.StringVar(&peerName, "peer-name", "unknown", "name of this Weave Net peer")
+	flag.StringVar(&nodeName, "node-name", "unknown", "name of this Kubernetes node")
+	flag.Parse()
+
 	config, err := rest.InClusterConfig()
 	if err != nil {
 		log.Fatalf("Could not get cluster config: %v", err)
@@ -54,7 +97,15 @@ func main() {
 	if err != nil {
 		log.Fatalf("Could not get peers: %v", err)
 	}
-	for _, addr := range peers {
-		fmt.Println(addr)
+	if justReclaim {
+		log.Println("Checking if any peers need to be reclaimed")
+		_, err := addMyselfToPeerList(c, peerName, nodeName)
+		if err != nil {
+			log.Fatalf("Could not get peer list: %v", err)
+		}
+		return
+	}
+	for _, node := range peers {
+		fmt.Println(node.addr)
 	}
 }

prog/weave-kube/launch.sh

@@ -108,6 +108,8 @@ post_start_actions() {
     export HOST_ROOT
     /home/weave/weave --local setup-cni
 
+    /home/weave/kube-peers -reclaim -node-name="$HOSTNAME" -peer-name="$(cat /sys/class/net/weave/address)"
+
     # Expose the weave network so host processes can communicate with pods
     /home/weave/weave --local expose $WEAVE_EXPOSE_IP
 }