From 9c09e36135f043230930773fa180f04155f51ce3 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Thu, 10 Oct 2019 11:02:01 +0000 Subject: [PATCH 1/3] refactor: extract function to normalise a single range that may span zero --- ipam/ring/ring.go | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/ipam/ring/ring.go b/ipam/ring/ring.go index ce7e1292..cc0f4dc6 100644 --- a/ipam/ring/ring.go +++ b/ipam/ring/ring.go @@ -375,6 +375,11 @@ func (r *Ring) Empty() bool { return len(r.Entries) == 0 } +// convenience function to normalise a single range that may span zero +func (r *Ring) makeRanges(start, end address.Address) []address.Range { + return r.splitRangesOverZero([]address.Range{{Start: start, End: end}}) +} + // Given a slice of ranges which are all in the right order except // possibly the last one spans zero, fix that up and return the slice func (r *Ring) splitRangesOverZero(ranges []address.Range) []address.Range { @@ -425,10 +430,7 @@ type RangeInfo struct { func (r *Ring) AllRangeInfo() (result []RangeInfo) { for i, entry := range r.Entries { - nextEntry := r.Entries.entry(i + 1) - ranges := []address.Range{{Start: entry.Token, End: nextEntry.Token}} - ranges = r.splitRangesOverZero(ranges) - for _, r := range ranges { + for _, r := range r.makeRanges(entry.Token, r.Entries.entry(i+1).Token) { result = append(result, RangeInfo{entry.Peer, r, entry.Version}) } } From d2f0287fa3a76c36fe749c6c8a036e5cdd7de161 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Thu, 10 Oct 2019 11:00:30 +0000 Subject: [PATCH 2/3] Handle range wrap-around when checking for allocations Ring wraps around at zero but Space doesn't know about this, so we have to normalise ranges coming out of Ring. In turn we may need to split one range into two, across the end of the Ring. --- ipam/allocator.go | 12 +++++++++--- ipam/ring/ring.go | 12 ++++++------ ipam/ring/ring_test.go | 20 +++++++++++--------- 3 files changed, 26 insertions(+), 18 deletions(-) diff --git a/ipam/allocator.go b/ipam/allocator.go index ae178452..831b9862 100644 --- a/ipam/allocator.go +++ b/ipam/allocator.go @@ -814,8 +814,14 @@ func (alloc *Allocator) sendRingUpdate(dest mesh.PeerName) { alloc.gossip.GossipUnicast(dest, msg) } -func (alloc *Allocator) checkRangeHasAllocations(r address.Range) bool { - return alloc.space.NumFreeAddressesInRange(r) != r.Size() +func (alloc *Allocator) hasAllocations(rs []address.Range) bool { + for _, r := range rs { + alloc.debugf("hasAllocations: %v %v vs %v", r, alloc.space.NumFreeAddressesInRange(r), r.Size()) + if alloc.space.NumFreeAddressesInRange(r) != r.Size() { + return true + } + } + return false } func (alloc *Allocator) update(sender mesh.PeerName, msg []byte) error { @@ -836,7 +842,7 @@ func (alloc *Allocator) update(sender mesh.PeerName, msg []byte) error { // If someone sent us a ring, merge it into ours. Note this will move us // out of the awaiting-consensus state if we didn't have a ring already. case data.Ring != nil: - updated, err := alloc.ring.Merge(*data.Ring, alloc.checkRangeHasAllocations) + updated, err := alloc.ring.Merge(*data.Ring, alloc.hasAllocations) switch err { case nil: if updated { diff --git a/ipam/ring/ring.go b/ipam/ring/ring.go index cc0f4dc6..1eeee069 100644 --- a/ipam/ring/ring.go +++ b/ipam/ring/ring.go @@ -213,7 +213,7 @@ func (r *Ring) GrantRangeToHost(start, end address.Address, peer mesh.PeerName) // Merge the given ring into this ring and indicate whether this ring // got updated as a result. -func (r *Ring) Merge(gossip Ring, checkRangeHasAllocations func(r address.Range) bool) (bool, error) { +func (r *Ring) Merge(gossip Ring, hasAllocations func(r []address.Range) bool) (bool, error) { r.assertInvariants() defer r.trackUpdates()() @@ -238,7 +238,7 @@ func (r *Ring) Merge(gossip Ring, checkRangeHasAllocations func(r address.Range) return false, ErrDifferentRange } - result, updated, err := r.Entries.merge(gossip.Entries, r.Peer, r, checkRangeHasAllocations) + result, updated, err := r.Entries.merge(gossip.Entries, r.Peer, r, hasAllocations) if err != nil { return false, err @@ -270,7 +270,7 @@ func (r *Ring) Merge(gossip Ring, checkRangeHasAllocations func(r address.Range) // entries belonging to ourPeer. Returns the merged entries and an // indication whether the merge resulted in any changes, i.e. the // result differs from the original. -func (es entries) merge(other entries, ourPeer mesh.PeerName, r *Ring, checkRangeHasAllocations func(r address.Range) bool) (result entries, updated bool, err error) { +func (es entries) merge(other entries, ourPeer mesh.PeerName, r *Ring, hasAllocations func(r []address.Range) bool) (result entries, updated bool, err error) { var mine, theirs *entry var previousOwner *mesh.PeerName addToResult := func(e entry) { result = append(result, &e) } @@ -294,7 +294,7 @@ func (es entries) merge(other entries, ourPeer mesh.PeerName, r *Ring, checkRang // insert, checking that a range owned by us hasn't been split if previousOwner != nil && *previousOwner == ourPeer && theirs.Peer != ourPeer { // check we have no allocations in the range that got split - if checkRangeHasAllocations(address.Range{Start: theirs.Token, End: mine.Token}) { + if hasAllocations(r.makeRanges(theirs.Token, mine.Token)) { err = errEntryInMyRange(theirs) return } @@ -309,7 +309,7 @@ func (es entries) merge(other entries, ourPeer mesh.PeerName, r *Ring, checkRang if mine.Version == theirs.Version && !mine.Equal(theirs) { if mine.Peer == ourPeer { // if we own the entry and has allocations - if checkRangeHasAllocations(address.Range{Start: mine.Token, End: es.entry(i + 1).Token}) { + if hasAllocations(r.makeRanges(theirs.Token, es.entry(i+1).Token)) { err = errInconsistentEntry(mine, theirs) return } @@ -332,7 +332,7 @@ func (es entries) merge(other entries, ourPeer mesh.PeerName, r *Ring, checkRang // in the range effectively given away, or it belongs to our own peer, in // which case we should set our version to the one received plus one, // effectively imposing our existing entry. - if theirs.Peer != ourPeer && !checkRangeHasAllocations(address.Range{Start: mine.Token, End: es.entry(i + 1).Token}) { + if theirs.Peer != ourPeer && !hasAllocations(r.makeRanges(mine.Token, es.entry(i+1).Token)) { addTheirs(*theirs) } else if theirs.Peer == ourPeer { mine.Version = theirs.Version + 1 diff --git a/ipam/ring/ring_test.go b/ipam/ring/ring_test.go index 859a411e..c137de65 100644 --- a/ipam/ring/ring_test.go +++ b/ipam/ring/ring_test.go @@ -38,19 +38,21 @@ func merge(r1, r2 *Ring) error { } return address.Count((r1.End - start) + (end - r1.Start)) } - checkEntryHasAllocations := func(r address.Range) bool { - size := distance(r.Start, r.End) - entry, found := r1.Entries.get(r.Start) - if !found { - return false - } - if entry.Free < size { - return true + check := func(rs []address.Range) bool { + for _, r := range rs { + size := distance(r.Start, r.End) + entry, found := r1.Entries.get(r.Start) + if !found { + continue + } + if entry.Free < size { + return true + } } return false } - _, err := r1.Merge(*r2, checkEntryHasAllocations) + _, err := r1.Merge(*r2, check) return err } From 44f91950cb1b73281e783f0a2078f9408f081af0 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Thu, 10 Oct 2019 11:05:56 +0000 Subject: [PATCH 3/3] Improve logging of ring merge Turn some Sprint() calls into `%v` in the outer format string. Remove one debug log line on duplicate token, since that info is already logged. Add an Info log line when we accept another peer's token, since that should be a rare event, and it's important if we've made a mistake. --- ipam/ring/ring.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/ipam/ring/ring.go b/ipam/ring/ring.go index 1eeee069..43322a7f 100644 --- a/ipam/ring/ring.go +++ b/ipam/ring/ring.go @@ -284,7 +284,7 @@ func (es entries) merge(other entries, ourPeer mesh.PeerName, r *Ring, hasAlloca var i, j int for i < len(es) && j < len(other) { mine, theirs = es[i], other[j] - common.Log.Debugln(fmt.Sprintf("[ring %s]: Merge mine.Token=%s theirs.Token=%s mine.Peer=%s theirs.Peer=%s mine.Version=%s theirs.Version=%s", ourPeer, mine.Token, theirs.Token, mine.Peer, theirs.Peer, fmt.Sprint(mine.Version), fmt.Sprint(theirs.Version))) + common.Log.Debugf("[ring %s]: Merge mine.Token=%s theirs.Token=%s mine.Peer=%s theirs.Peer=%s mine.Version=%v theirs.Version=%v", ourPeer, mine.Token, theirs.Token, mine.Peer, theirs.Peer, mine.Version, theirs.Version) switch { case mine.Token < theirs.Token: addToResult(*mine) @@ -302,7 +302,6 @@ func (es entries) merge(other entries, ourPeer mesh.PeerName, r *Ring, hasAlloca addTheirs(*theirs) j++ case mine.Token == theirs.Token: - common.Log.Debugln(fmt.Sprintf("[ring %s]: Merge token=%s mine.Peer=%s theirs.Peer=%s mine.Version=%s theirs.Version=%s", ourPeer, mine.Token, mine.Peer, theirs.Peer, fmt.Sprint(mine.Version), fmt.Sprint(theirs.Version))) // merge switch { case mine.Version >= theirs.Version: @@ -333,6 +332,7 @@ func (es entries) merge(other entries, ourPeer mesh.PeerName, r *Ring, hasAlloca // which case we should set our version to the one received plus one, // effectively imposing our existing entry. if theirs.Peer != ourPeer && !hasAllocations(r.makeRanges(mine.Token, es.entry(i+1).Token)) { + common.Log.Infof("ring: repair inconsistent data by accepting token %v from %v - no allocations up to %v", theirs.Token, theirs.Peer, es.entry(i+1).Token) addTheirs(*theirs) } else if theirs.Peer == ourPeer { mine.Version = theirs.Version + 1 @@ -528,7 +528,7 @@ func (r *Ring) ReportFree(freespace map[address.Address]address.Count) (updated entries[i].Free = free entries[i].Version++ - common.Log.Debugln(fmt.Sprintf("[ring %s]: ReportFree token=%s peer=%s version=%s", r.Peer, entries[i].Token, entries[i].Peer, fmt.Sprint(entries[i].Version))) + common.Log.Debugf("[ring %s]: ReportFree token=%s peer=%s version=%v", r.Peer, entries[i].Token, entries[i].Peer, entries[i].Version) updated = true } return @@ -612,7 +612,7 @@ func (r *Ring) Transfer(from, to mesh.PeerName) []address.Range { // 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))) + common.Log.Debugf("[ring %s]: Transfer token=%s from=%s to=%s version=%v", r.Peer, entry.Token, from, to, entry.Version) newRanges = append(newRanges, address.Range{Start: entry.Token, End: r.Entries.entry(i + 1).Token}) } }