Merge pull request #1159 from weaveworks/1150-claim-from-peer

Implement claiming addresses from other peers

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() {

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)
 		}
 	})
 }

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."))
 }

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