Merge branch '2.5'

ipam/allocator.go

@@ -910,6 +910,7 @@ func (alloc *Allocator) donateSpace(r address.Range, to mesh.PeerName) {
 	alloc.debugln("Giving range", chunk, "to", to)
 	alloc.ring.GrantRangeToHost(chunk.Start, chunk.End, to)
 	alloc.persistRing()
+	alloc.gossip.GossipBroadcast(alloc.Gossip())
 }
 
 func (alloc *Allocator) assertInvariants() {

ipam/ring/ring.go

@@ -568,7 +568,9 @@ func (r *Ring) Transfer(from, to mesh.PeerName) []address.Range {
 	for i, entry := range r.Entries {
 		if entry.Peer == from {
 			entry.Peer = to
-			entry.Version++
+			// bump version by large value so that transfer of range takes precedence (if there is any conflict in merging entry)
+			// over other opertaions on the entry that increments the version
+			entry.Version = entry.Version + 100
 			common.Log.Debugln(fmt.Sprintf("[ring %s]: Transfer token=%s from=%s to=%s version=%s", r.Peer, entry.Token, from, to, fmt.Sprint(entry.Version)))
 			newRanges = append(newRanges, address.Range{Start: entry.Token, End: r.Entries.entry(i + 1).Token})
 		}

prog/kube-utils/annotations.go

@@ -96,33 +96,33 @@ func (cml *configMapAnnotations) GetAnnotation(key string) (string, bool) {
 }
 
 func (cml *configMapAnnotations) UpdateAnnotation(key, value string) (err error) {
-	if cml.cm == nil {
+	if cml.cm == nil || cml.cm.Annotations == nil {
 		return errors.New("endpoint not initialized, call Init first")
 	}
 	// speculatively change the state, then replace with whatever comes back
-	// from Update(), which will be the latest on the server whatever happened
+	// from Update(), which will be the latest on the server, or nil if error
 	cml.cm.Annotations[cleanKey(key)] = value
 	cml.cm, err = cml.Client.ConfigMaps(cml.Namespace).Update(cml.cm)
 	return err
 }
 
 func (cml *configMapAnnotations) RemoveAnnotation(key string) (err error) {
-	if cml.cm == nil {
+	if cml.cm == nil || cml.cm.Annotations == nil {
 		return errors.New("endpoint not initialized, call Init first")
 	}
 	// speculatively change the state, then replace with whatever comes back
-	// from Update(), which will be the latest on the server whatever happened
+	// from Update(), which will be the latest on the server, or nil if error
 	delete(cml.cm.Annotations, cleanKey(key))
 	cml.cm, err = cml.Client.ConfigMaps(cml.Namespace).Update(cml.cm)
 	return err
 }
 
 func (cml *configMapAnnotations) RemoveAnnotationsWithValue(valueToRemove string) (err error) {
-	if cml.cm == nil {
+	if cml.cm == nil || cml.cm.Annotations == nil {
 		return errors.New("endpoint not initialized, call Init first")
 	}
 	// speculatively change the state, then replace with whatever comes back
-	// from Update(), which will be the latest on the server whatever happened
+	// from Update(), which will be the latest on the server, or nil if error
 	for key, value := range cml.cm.Annotations {
 		if value == valueToRemove {
 			delete(cml.cm.Annotations, key) // don't need to clean this key as it came from the map

prog/kube-utils/main.go

@@ -135,7 +135,7 @@ func reclaimRemovedPeers(kube kubernetes.Interface, cml *configMapAnnotations, m
 				common.Log.Warnln("[kube-peers] not removing myself", peer)
 				continue
 			}
-			changed, err := reclaimPeer(weave, cml, storedPeerList, peer.PeerName, myPeerName)
+			changed, err := reclaimPeer(weave, cml, peer.PeerName, myPeerName)
 			if err != nil {
 				return err
 			}
@@ -155,11 +155,16 @@ func reclaimRemovedPeers(kube kubernetes.Interface, cml *configMapAnnotations, m
 // actions the reclaim.
 // Return a bool to show whether we attempted to change anything,
 // and an error if something went wrong.
-func reclaimPeer(weave weaveClient, cml *configMapAnnotations, storedPeerList *peerList, peerName string, myPeerName string) (changed bool, err error) {
+func reclaimPeer(weave weaveClient, cml *configMapAnnotations, peerName string, myPeerName string) (changed bool, err error) {
 	common.Log.Debugln("[kube-peers] Preparing to remove disappeared peer", peerName)
 	okToRemove := false
 	nonExistentPeer := false
 
+	// Re-read status from Kubernetes; speculative updates may leave it un-set
+	if err := cml.Init(); err != nil {
+		return false, err
+	}
+
 	// 3. Check if there is an existing annotation with key X
 	existingAnnotation, found := cml.GetAnnotation(KubePeersPrefix + peerName)
 	if found {
@@ -168,6 +173,10 @@ func reclaimPeer(weave weaveClient, cml *configMapAnnotations, storedPeerList *p
 		if existingAnnotation == myPeerName {
 			okToRemove = true
 		} else {
+			storedPeerList, err := cml.GetPeerList()
+			if err != nil {
+				return false, err
+			}
 			// handle an edge case where peer claimed to own the action to reclaim but no longer
 			// exists hence lock persists foever
 			if !storedPeerList.contains(existingAnnotation) {

prog/kube-utils/peerlist_test.go

@@ -137,9 +137,10 @@ func TestPeerListFuzz(t *testing.T) {
 		found := storedPeerList.contains(peerName(i))
 		require.True(t, found, "peer %d not found in stored list", i)
 
-		_, err = reclaimPeer(mockWeave{}, cml, storedPeerList, peerName(i), fmt.Sprintf("deleter-%d", i))
+		_, err = reclaimPeer(mockWeave{}, cml, peerName(i), fmt.Sprintf("deleter-%d", i))
 		require.NoError(t, err)
 
+		require.NoError(t, cml.Init())
 		storedPeerList, err = cml.GetPeerList()
 		require.NoError(t, err)
 		stateLock.Lock()

testing/kubernetes/testing/fixture.go

@@ -110,10 +110,6 @@ func ObjectReaction(tracker ObjectTracker) testing.ReactionFunc {
 			}
 			err = tracker.Update(gvr, action.GetObject(), ns)
 			if err != nil {
-				if errors.IsConflict(err) { // return the currently-stored version of the object
-					obj, _ := tracker.Get(gvr, ns, objMeta.GetName())
-					return true, obj, err
-				}
 				return true, nil, err
 			}
 			obj, err := tracker.Get(gvr, ns, objMeta.GetName())