Handling concurrent updates to configmap and IP reclamation (#3724)

checking if peer to be removed has been removed by another pod

Signed-off-by: mmerrill3 <michael.merrill@vonage.com>

prog/kube-utils/main.go

@@ -94,11 +94,8 @@ type weaveClient interface {
 func reclaimRemovedPeers(kube kubernetes.Interface, cml *configMapAnnotations, myPeerName, myNodeName string) error {
 	weave := weaveapi.NewClient(os.Getenv("WEAVE_HTTP_ADDR"), common.Log)
 	for loopsWhenNothingChanged := 0; loopsWhenNothingChanged < 3; loopsWhenNothingChanged++ {
-		if err := cml.Init(); err != nil {
-			return err
-		}
 		// 1. Compare peers stored in the peerList against all peers reported by k8s now.
-		storedPeerList, err := cml.GetPeerList()
+		storedPeerList, err := getRefreshedPeerList(cml)
 		if err != nil {
 			return err
 		}
@@ -200,10 +197,15 @@ func reclaimPeer(weave weaveClient, cml *configMapAnnotations, peerName string,
 		return false, nil
 	}
 	// 6.   If step 4 or 5 succeeded, rmpeer X
-	result, err := weave.RmPeer(peerName)
-	common.Log.Infof("[kube-peers] rmpeer of %s: %s", peerName, result)
-	if err != nil {
-		return false, err
+	// we will try once more to see if the peerName has been removed from the annotation
+	// by another weave pod
+	storedPeerList, err := getRefreshedPeerList(cml)
+	if storedPeerList.contains(peerName) {
+		result, err := weave.RmPeer(peerName)
+		common.Log.Infof("[kube-peers] rmpeer of %s: %s", peerName, result)
+		if err != nil {
+			return false, err
+		}
 	}
 	err = cml.LoopUpdate(func() error {
 		// 7aa.   Remove any annotations Z* that have contents X
@@ -215,9 +217,13 @@ func reclaimPeer(weave weaveClient, cml *configMapAnnotations, peerName string,
 		if err != nil {
 			return err
 		}
-		storedPeerList.remove(peerName)
-		if err := cml.UpdatePeerList(*storedPeerList); err != nil {
-			return err
+		if storedPeerList.contains(peerName) {
+			storedPeerList.remove(peerName)
+			if err := cml.UpdatePeerList(*storedPeerList); err != nil {
+				return err
+			}
+		} else {
+			common.Log.Infof("[kube-peers] annotation was removed already for %s", peerName)
 		}
 		// 7b.    Remove annotation with key X
 		return cml.RemoveAnnotation(KubePeersPrefix + peerName)
@@ -288,6 +294,18 @@ func registerForNodeUpdates(client *kubernetes.Clientset, stopCh <-chan struct{}
 	informerFactory.Start(stopCh)
 }
 
+// getRefreshedPeerList will fetch the latest peer list from the config map annotation
+func getRefreshedPeerList(cml *configMapAnnotations) (*peerList, error) {
+	if err := cml.Init(); err != nil {
+		return nil, err
+	}
+	storedPeerList, err := cml.GetPeerList()
+	if err != nil {
+		return nil, err
+	}
+	return storedPeerList, nil
+}
+
 func main() {
 	var (
 		justReclaim       bool