diff --git a/ipam/allocator.go b/ipam/allocator.go index 8137cb33..dcd673b5 100644 --- a/ipam/allocator.go +++ b/ipam/allocator.go @@ -19,6 +19,7 @@ import ( const ( msgSpaceRequest = iota msgRingUpdate + msgSpaceRequestDenied paxosInterval = time.Second * 5 MinSubnetSize = 4 // first and last addresses are excluded, so 2 would be too small @@ -163,6 +164,18 @@ func (alloc *Allocator) tryPendingOps() { } } +func (alloc *Allocator) spaceRequestDenied(sender router.PeerName, r address.Range) { + for i := 0; i < len(alloc.pendingClaims); { + claim := alloc.pendingClaims[i].(*claim) + if r.Contains(claim.addr) { + claim.DeniedBy(alloc, sender) + alloc.pendingClaims = append(alloc.pendingClaims[:i], alloc.pendingClaims[i+1:]...) + continue + } + i++ + } +} + func hasBeenCancelled(cancelChan <-chan bool) func() bool { return func() bool { select { @@ -342,6 +355,11 @@ func (alloc *Allocator) pruneNicknames() { } } +func decodeRange(msg []byte) (r address.Range, err error) { + decoder := gob.NewDecoder(bytes.NewReader(msg)) + return r, decoder.Decode(&r) +} + // OnGossipUnicast (Sync) func (alloc *Allocator) OnGossipUnicast(sender router.PeerName, msg []byte) error { alloc.debugln("OnGossipUnicast from", sender, ": ", len(msg), "bytes") @@ -350,14 +368,17 @@ func (alloc *Allocator) OnGossipUnicast(sender router.PeerName, msg []byte) erro switch msg[0] { case msgSpaceRequest: // some other peer asked us for space - decoder := gob.NewDecoder(bytes.NewReader(msg[1:])) - var r address.Range - if err := decoder.Decode(&r); err != nil { - resultChan <- err - return + r, err := decodeRange(msg[1:]) + if err == nil { + alloc.donateSpace(r, sender) } - alloc.donateSpace(r, sender) - resultChan <- nil + resultChan <- err + case msgSpaceRequestDenied: + r, err := decodeRange(msg[1:]) + if err == nil { + alloc.spaceRequestDenied(sender, r) + } + resultChan <- err case msgRingUpdate: resultChan <- alloc.update(sender, msg[1:]) } @@ -574,13 +595,22 @@ func (alloc *Allocator) propose() { alloc.gossip.GossipBroadcast(alloc.Gossip()) } -func (alloc *Allocator) sendSpaceRequest(dest router.PeerName, r address.Range) error { +func encodeRange(r address.Range) []byte { buf := new(bytes.Buffer) enc := gob.NewEncoder(buf) if err := enc.Encode(r); err != nil { panic(err) } - msg := router.Concat([]byte{msgSpaceRequest}, buf.Bytes()) + return buf.Bytes() +} + +func (alloc *Allocator) sendSpaceRequest(dest router.PeerName, r address.Range) error { + msg := router.Concat([]byte{msgSpaceRequest}, encodeRange(r)) + return alloc.gossip.GossipUnicast(dest, msg) +} + +func (alloc *Allocator) sendSpaceRequestDenied(dest router.PeerName, r address.Range) error { + msg := router.Concat([]byte{msgSpaceRequestDenied}, encodeRange(r)) return alloc.gossip.GossipUnicast(dest, msg) } @@ -657,10 +687,14 @@ func (alloc *Allocator) donateSpace(r address.Range, to router.PeerName) { free := alloc.space.NumFreeAddressesInRange(r) common.Assert(free == 0) alloc.debugln("No space to give to peer", to) + // separate message maintains backwards-compatibility: + // down-level peers will ignore this and still get the ring update. + alloc.sendSpaceRequestDenied(to, r) return } alloc.debugln("Giving range", chunk, "to", to) alloc.ring.GrantRangeToHost(chunk.Start, chunk.End, to) + alloc.sendRingUpdate(to) } func (alloc *Allocator) assertInvariants() { diff --git a/ipam/allocator_test.go b/ipam/allocator_test.go index 930dc437..34222593 100644 --- a/ipam/allocator_test.go +++ b/ipam/allocator_test.go @@ -131,23 +131,45 @@ func TestBootstrap(t *testing.T) { func TestAllocatorClaim(t *testing.T) { const ( container1 = "abcdef" - container2 = "baddf00d" container3 = "b01df00d" - universe = "10.0.3.0/30" - testAddr1 = "10.0.3.1" // first address allocated should be .1 because .0 is network addr + universe = "10.0.3.0/24" + testAddr1 = "10.0.3.2" + testAddr2 = "10.0.4.2" ) - alloc, subnet := makeAllocatorWithMockGossip(t, "01:00:00:01:00:00", universe, 1) + allocs, _, subnet := makeNetworkOfAllocators(2, universe) + alloc := allocs[1] defer alloc.Stop() - - alloc.claimRingForTesting() addr1, _ := address.ParseIP(testAddr1) + // First claim should trigger "dunno, I'm going to wait" err := alloc.Claim(container3, addr1) require.NoError(t, err) + + // Do one allocate to ensure paxos is all done + alloc.Allocate("unused", subnet, nil) + // Do an allocate on the other peer, which we will try to claim later + addrx, err := allocs[0].Allocate(container1, subnet, nil) + + // Now try the claim again + err = alloc.Claim(container3, addr1) + require.NoError(t, err) // Check we get this address back if we try an allocate addr3, _ := alloc.Allocate(container3, subnet, nil) require.Equal(t, testAddr1, addr3.String(), "address") + // one more claim should still work + err = alloc.Claim(container3, addr1) + require.NoError(t, err) + // claim for a different container should fail + err = alloc.Claim(container1, addr1) + require.Error(t, err) + // claiming the address allocated on the other peer should fail + err = alloc.Claim(container1, addrx) + require.Error(t, err, "claiming address allocated on other peer should fail") + // Check an address outside of our universe + addr2, _ := address.ParseIP(testAddr2) + err = alloc.Claim(container1, addr2) + require.NoError(t, err) } func (alloc *Allocator) pause() func() { @@ -326,27 +348,18 @@ func TestAllocatorFuzz(t *testing.T) { return xs[:ls] } - // Do a Allocate and check the address - // is unique. Needs a unique container - // name. - allocate := func(name string) { + bumpPending := func() bool { stateLock.Lock() if len(addrs)+numPending >= maxAddresses { stateLock.Unlock() - return + return false } numPending++ stateLock.Unlock() + return true + } - allocIndex := rand.Int31n(nodes) - alloc := allocs[allocIndex] - //common.Log.Infof("Allocate: asking allocator %d", allocIndex) - addr, err := alloc.Allocate(name, subnet, nil) - - if err != nil { - panic(fmt.Sprintf("Could not allocate addr")) - } - + noteAllocation := func(allocIndex int32, name string, addr address.Address) { //common.Log.Infof("Allocate: got address %s for name %s", addr, name) addrStr := addr.String() @@ -363,6 +376,26 @@ func TestAllocatorFuzz(t *testing.T) { numPending-- } + // Do a Allocate and check the address + // is unique. Needs a unique container + // name. + allocate := func(name string) { + if !bumpPending() { + return + } + + allocIndex := rand.Int31n(nodes) + alloc := allocs[allocIndex] + //common.Log.Infof("Allocate: asking allocator %d", allocIndex) + addr, err := alloc.Allocate(name, subnet, nil) + + if err != nil { + panic(fmt.Sprintf("Could not allocate addr")) + } + + noteAllocation(allocIndex, name, addr) + } + // Free a random address. free := func() { stateLock.Lock() @@ -423,6 +456,21 @@ func TestAllocatorFuzz(t *testing.T) { stateLock.Unlock() } + // Claim a random address for a unique container name - may not succeed + claim := func(name string) { + if !bumpPending() { + return + } + allocIndex := rand.Int31n(nodes) + addressIndex := rand.Int31n(int32(subnet.Size())) + alloc := allocs[allocIndex] + addr := address.Add(subnet.Start, address.Offset(addressIndex)) + err := alloc.Claim(name, addr) + if err == nil { + noteAllocation(allocIndex, name, addr) + } + } + // Run function _f_ _iterations_ times, in _concurrency_ // number of goroutines doConcurrentIterations := func(iterations int, f func(int)) { @@ -462,9 +510,13 @@ func TestAllocatorFuzz(t *testing.T) { // free a random addr free() - case 0.8 <= r && r < 1.0: + case 0.8 <= r && r < 0.95: // ask for an existing name again, check we get same ip allocateAgain() + + case 0.95 <= r && r < 1.0: + name := fmt.Sprintf("second%d", iteration) + claim(name) } }) } diff --git a/ipam/claim.go b/ipam/claim.go index 63212cd7..fa35ef8f 100644 --- a/ipam/claim.go +++ b/ipam/claim.go @@ -35,23 +35,22 @@ func (c *claim) Try(alloc *Allocator) bool { return true } - // If our ring doesn't know, it must be empty. We will have initiated the - // bootstrap of the ring, so wait until we find some owner for this - // range (might be us). switch owner := alloc.ring.Owner(c.addr); owner { case alloc.ourName: // success case router.UnknownPeerName: - alloc.infof("Ring is empty; will try later.", c.addr, owner) + // If our ring doesn't know, it must be empty. + alloc.infof("Claim %s for %s: address allocator still initializing; will try later.", c.addr, c.ident) c.sendResult(nil) // don't make the caller wait return false default: - name, found := alloc.nicknames[owner] - if found { - name = " (" + name + ")" + alloc.debugf("requesting address %s from other peer %s", c.addr, owner) + err := alloc.sendSpaceRequest(owner, address.NewRange(c.addr, 1)) + if err != nil { // can't speak to owner right now; figure it out later + alloc.infof("Claim %s for %s: %s; will try later.", c.addr, c.ident, err) + c.sendResult(nil) } - c.sendResult(fmt.Errorf("address %s is owned by other peer %s%s", c.addr.String(), owner, name)) - return true + return false } // We are the owner, check we haven't given it to another container @@ -74,6 +73,14 @@ func (c *claim) Try(alloc *Allocator) bool { return true } +func (c *claim) DeniedBy(alloc *Allocator, owner router.PeerName) { + name, found := alloc.nicknames[owner] + if found { + name = " (" + name + ")" + } + c.sendResult(fmt.Errorf("address %s is owned by other peer %s%s", c.addr.String(), owner, name)) +} + func (c *claim) Cancel() { c.sendResult(fmt.Errorf("Operation cancelled.")) } diff --git a/test/160_claim_test.sh b/test/160_claim_test.sh new file mode 100755 index 00000000..c550f3b7 --- /dev/null +++ b/test/160_claim_test.sh @@ -0,0 +1,40 @@ +#! /bin/bash + +. ./config.sh + +start_suite "Claiming addresses" + +weave_on $HOST1 launch-router +weave_on $HOST2 launch-router $HOST1 + +start_container $HOST1 --name=c1 +start_container $HOST2 --name=c2 +C1=$(container_ip $HOST1 c1) +assert_raises "exec_on $HOST2 c2 $PING $C1" + +stop_router_on $HOST1 +stop_router_on $HOST2 + +# Start hosts in reverse order so c1's address has to be claimed from host2 +weave_on $HOST2 launch-router +weave_on $HOST1 launch-router $HOST2 + +# Start another container on host2, so if it hasn't relinquished c1's +# address it would give that out as the first available. +start_container $HOST2 --name=c3 +C3=$(container_ip $HOST2 c3) +assert_raises "[ $C3 != $C1 ]" + +assert_raises "exec_on $HOST1 c1 $PING $C3" + +stop_router_on $HOST1 +stop_router_on $HOST2 + +# Now make host1 attempt to claim from host2, when host2 is stopped +weave_on $HOST2 launch-router +# Introduce host3 to remember the IPAM CRDT when we stop host2 +weave_on $HOST3 launch-router $HOST2 +stop_router_on $HOST2 +weave_on $HOST1 launch-router --log-level=debug $HOST3 + +end_suite