diff --git a/ipam/allocator.go b/ipam/allocator.go index bfff91f0..63d0f658 100644 --- a/ipam/allocator.go +++ b/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() { diff --git a/ipam/ring/ring.go b/ipam/ring/ring.go index bdbd8d8c..d4b2517d 100644 --- a/ipam/ring/ring.go +++ b/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}) } diff --git a/prog/kube-utils/annotations.go b/prog/kube-utils/annotations.go index 1b079365..f83b676a 100644 --- a/prog/kube-utils/annotations.go +++ b/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 diff --git a/prog/kube-utils/main.go b/prog/kube-utils/main.go index f009c5bb..0743d360 100644 --- a/prog/kube-utils/main.go +++ b/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) { diff --git a/prog/kube-utils/peerlist_test.go b/prog/kube-utils/peerlist_test.go index 77866f69..093c575b 100644 --- a/prog/kube-utils/peerlist_test.go +++ b/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() diff --git a/testing/kubernetes/testing/fixture.go b/testing/kubernetes/testing/fixture.go index 60965d79..3df53523 100644 --- a/testing/kubernetes/testing/fixture.go +++ b/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())