Improve config locking by using optimistic locking strategy

Loop using jitter until the update succeeds

prog/kube-peers/annotations.go

@@ -2,12 +2,15 @@ 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"
 )
@@ -50,6 +53,9 @@ func (pl *peerList) add(peerName string, name string) {
 }
 
 const (
+	retryPeriod  = time.Second * 2
+	jitterFactor = 1.0
+
 	// KubePeersAnnotationKey is the default annotation key
 	KubePeersAnnotationKey = "kube-peers.weave.works/peers"
 )
@@ -99,7 +105,6 @@ func (cml *configMapAnnotations) GetPeerList() (*peerList, error) {
 	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")
@@ -108,7 +113,32 @@ func (cml *configMapAnnotations) UpdatePeerList(list peerList) error {
 	if err != nil {
 		return err
 	}
-	cml.cm.Annotations[KubePeersAnnotationKey] = string(recordBytes)
-	cml.cm, err = cml.Client.ConfigMaps(cml.Namespace).Update(cml.cm)
+	cm := cml.cm
+	cm.Annotations[KubePeersAnnotationKey] = string(recordBytes)
+	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

@@ -56,23 +56,25 @@ const (
 // 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)
+	var list *peerList
+	err := cml.LoopUpdate(func() error {
+		var err error
+		list, err = cml.GetPeerList()
 		if err != nil {
-			return nil, err
+			return err
 		}
-	}
-	return list, nil
+		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 err
+			}
+		}
+		return nil
+	})
+	return list, err
 }
 
 func main() {