IP Allocation Management

Using the Ring CRDT and Weave gossip, provides an interface
for the weave script to allocate IP addresses in `weave run`,
`weave start` and `weave expose`.

See docs/ipam.md for implementation details, and site/features.md
for user-visible changes.

Makefile

@@ -41,7 +41,7 @@ $(WEAVER_EXE) $(WEAVEDNS_EXE): common/*.go
 		false; \
 	}
 
-$(WEAVER_EXE): router/*.go weaver/main.go
+$(WEAVER_EXE): router/*.go ipam/*.go ipam/*/*.go weaver/main.go
 $(WEAVEDNS_EXE): nameserver/*.go weavedns/main.go
 
 # Sigproxy needs separate rule as it fails the netgo check in the main

common/errors.go

@@ -11,3 +11,10 @@ func CheckWarn(e error) {
 		Warning.Println(e)
 	}
 }
+
+// Assert test is true, panic otherwise
+func Assert(test bool) {
+	if !test {
+		panic("Assertion failure")
+	}
+}

docs/ipam.md

@@ -0,0 +1,214 @@
+See the [requirements](https://github.com/zettio/weave/wiki/IP-allocation-requirements).
+
+At its highest level, the idea is that we start with a certain IP
+address space, known to all peers, and divide it up between the
+peers. This allows peers to allocate and free individual IPs locally
+until they run out.
+
+We use a CRDT to represent shared knowledge about the space,
+transmitted over the Weave Gossip mechanism, together with
+point-to-point messages for one peer to request more space from
+another.
+
+The allocator running at each peer also has an http interface which
+the container infrastructure (e.g. the Weave script, or a Docker
+plugin) uses to request IP addresses.
+
+![Schematic of IP allocation in operation](https://docs.google.com/drawings/d/1-EUIRKYxwfKTpBJ7v_LMcdvSpodIMSz4lT3wgEfWKl4/pub?w=701&h=310)
+
+## Commands
+
+The commands supported by the allocator via the http interface are:
+
+- Allocate: request one IP address for a container
+- Free: return an IP address that is currently allocated
+- Claim: request a specific IP address for a container (e.g. because
+  it is already using that address)
+
+The allocator also watches via the Docker event mechanism: if a
+container dies then all IP addresses allocated to that container are
+freed.
+
+## Definitions
+
+1. Allocations. We use the word 'allocation' to refer to a specific
+   IP address being assigned, e.g. so it can be given to a container.
+
+2. Range. Most of the time, instead of dealing with individual IP
+   addresses, we operate on them in contiguous groups, for which we
+   use the word "range".
+
+3. Ring. We consider the address space as a ring, so ranges wrap
+   around from the highest address to the lowest address.
+
+4. Peer. A Peer is a node on the Weave network. It can own zero or
+   more ranges.
+
+### The Allocation Process
+
+When a peer owns some range(s), it can allocate freely from within
+those ranges to containers on the same machine. If it runs out of
+space (all owned ranges are full), it will ask another peer for space:
+  - it picks a peer to ask at random, weighted by the amount of space
+    owned by each peer
+    - if the target peer decides to give up space, it unicasts a message
+      back to the asker with the newly-updated ring.
+    - if the target peer has no space, it unicasts a message back to the
+      asker with its current copy of the ring, on the basis that the
+      requestor must have acted on out-of-date information.
+  - it will continue to ask for space until it receives some, or its
+    copy of the ring tells it all peers are full.
+
+### Claiming an address
+
+If a Weave process is restarted, in most cases it will hear from
+another peer which ranges it used to own, but it needs to know which
+individual IP addresses are assigned to containers in order to avoid
+giving the same address out in subsequent allocation requests. The
+weave script invokes the `claim` command in `populate_ipam` to do
+this.
+
+When the Allocator is told to claim an address, there are four
+possibilities:
+  - the address is outside of the space managed by this Allocator, in
+    which case we ignore the request.
+  - the address is owned by this peer, in which case we record the
+    address as being assigned to a particular container and return
+    success.
+  - the address is owned by a different peer, in which case we return
+    failure.
+  - we have not yet heard of any address ranges being owned by anyone,
+    in which case we wait until we do hear.
+
+This approach fails if the peer does not hear from another peer about
+the ranges it used to own, e.g. if all peers in a network partition
+are restarted at once.
+
+### The Ring CRDT
+
+We use a Convergent Replicated Data Type - a CRDT - so that peers can
+make changes concurrently and communicate them without central
+coordination. To achieve this, we arrange that peers only make changes
+to the data structure in ranges that they own (except under
+administrator command - see later).
+
+The data structure is a set of tokens, each containing the name of an
+owning peer. A peer can own many ranges. Each token is placed at the
+start address of a range, and the set is kept ordered so each range
+goes from one token to the next. Each range on the ring includes the
+start address but does not include the end address (which is the start
+of the next range).  Ranges wrap, so the 'next' token after the last
+one is the first token.
+
+![Tokens on the Ring](https://docs.google.com/drawings/d/1hp--q2vmxbBAnPjhza4Kqjr1ugrw2iS1M1GerhH-IKY/pub?w=960&h=288)
+
+In more detail:
+- Each token is a tuple {peer name, version}, placed
+  at an IP address.
+- Peer names are taken from Weave: they are unique and survive across restarts.
+- The contents of a token can only be updated by the owning peer, and
+  when this is done the version is incremented
+- The ring data structure is always gossiped in its entirety
+- The merge operation when a peer receives a ring via gossip is:
+  - Tokens with unique addresses are just copied into the combined ring
+  - For tokens at the same address, pick the one with the highest
+    version number
+- The data gossiped about the ring also includes the amount of free
+  space in each range: this is not essential but it improves the
+  selection of which peer to ask for space.
+- When a peer is asked for space, there are four scenarios:
+  1. It has an empty range; it can change the peer associated with
+     the token at the beginning of the range and increment the version.
+  2. It has a range which can be subdivided by a single token to form
+     a free range.  It inserts said token, owned by the peer requesting
+     space.
+  3. It has a 'hole' in the middle of a range; an empty range can be
+     created by inserting two tokens, one at the beginning of the hole
+     owned by the peer requesting the space, and one at the end of the
+     hole owned by the requestee.
+  4. It has no space.
+
+## Initialisation
+
+The previous sections describe how peers coordinate changes to the
+ring.  But how is the initial state of the ring established?  If a new
+peer joins a long-standing cluster, it can learn about the ring state
+from other peers.  But in a freshly started cluster, the initial ring
+state must be established from a clean slate.  And it must be
+consistent across all peers.  This is a distributed consensus problem,
+and we solve it using the Paxos algorithm.
+
+Although Paxos has a reputation for being hard to understand and to
+implement, the implementation used for ring initialization is
+relatively straightforward:
+
+- We only need to establish consensus for a single value, rather than
+  a succession of values (as in a replicated transaction log).  Thus
+  we implement basic Paxos, rather than multi-Paxos, and the
+  implementation closely follows the outline of basic Paxos described
+  [on
+  wikipedia](http://en.wikipedia.org/wiki/Paxos_%28computer_science%29#Basic_Paxos).
+
+- All peers play all the roles described in basic Paxos: Proposer,
+  acceptor and listener.
+
+- Paxos is usually described in terms of unicast messages from one
+  agent to another, although three of the four messages are fanned out
+  to all agents in a certain role ("prepare" and "accept request" from
+  a proposer to all acceptors; "accepted" from an acceptor to all
+  learners).  So most implementations involve a communications layer
+  to implement the required communication patterns.  But our
+  implementation is built on top of the weave gossip layer, which
+  gives us a broadcast medium.  Using broadcast for the one truly
+  unicast message ("promise") may seem wasteful, but as we only
+  establish consensus for a single value, it is not a major concern.
+
+The value the peers obtain consensus on is the set of peers that will
+be represented in the initial ring (with each peer getting an equal
+share of the address space).  When a proposing peer gets to choose the
+value, it includes all the peers it has heard from during the Paxos
+phase, which is at least a quorum.
+
+Once a consensus is reached, it is used to initialize the ring.  If a
+peer hears about an initialized ring via gossip, that implies that a
+consensus was reached, so it will stop participating in Paxos.
+
+Paxos requires that we know the quorum size for the cluster.  As weave
+clusters are a dynamic entity, there is no fixed quorum size.  But the
+Paxos phase used to initialize the ring should normally be
+short-lived, so we just need to know the initial cluster size.  A
+heuristic is used to derive this from the list of addresses passed to
+`weave launch`, but the user can set it explicitly.
+
+## Peer shutdown
+
+When a peer leaves (a `weave reset` command), it grants all its own
+tokens to another peer, then broadcasts the updated ring.
+
+After sending the message, the peer terminates - it does not wait for
+any response.
+
+Failures:
+- message lost
+  - the space will be unusable by other peers because it will still be
+    seen as owned.
+
+To cope with the situation where a peer has left or died without
+managing to tell its peers, an administrator may go to any other peer
+and command that it take over the dead peer's tokens (with `weave
+rmpeer`).  This information will then be gossipped out to the network.
+
+(This operation is not guaranteed to be safe - if the dead peer made
+some transfers which have are known to other peers but have not
+reached the peer that does the `rmpeer`, then we get an inconsistent
+ring.  We could make it safe, up to partitions, by inquiring of every
+live peer what its view of the dead peer is, and making sure we use
+the latest info.)
+
+## Limitations
+
+Nothing is persisted. If one peer restarts it should receive gossip
+from other peers showing what it used to own, and in that case it can
+recover quite well. If, however, there are no peers left alive, or
+this peer cannot establish network communication with those that are,
+then it cannot recover.

ipam/address/address.go

@@ -0,0 +1,54 @@
+package address
+
+import (
+	"github.com/weaveworks/weave/common"
+	"net"
+)
+
+// Using 32-bit integer to represent IPv4 address
+type Address uint32
+type Offset uint32
+
+type Range struct {
+	Start, End Address // [Start, End); Start <= End
+}
+
+func ParseIP(s string) (Address, error) {
+	if ip := net.ParseIP(s); ip == nil {
+		return 0, &net.ParseError{Type: "IP Address", Text: s}
+	} else {
+		return FromIP4(ip), nil
+	}
+}
+
+// FromIP4 converts an ipv4 address to our integer address type
+func FromIP4(ip4 net.IP) (r Address) {
+	for _, b := range ip4.To4() {
+		r <<= 8
+		r |= Address(b)
+	}
+	return
+}
+
+// IP4 converts our integer address type to an ipv4 address
+func (addr Address) IP4() (r net.IP) {
+	r = make([]byte, net.IPv4len)
+	for i := 3; i >= 0; i-- {
+		r[i] = byte(addr)
+		addr >>= 8
+	}
+	return
+}
+
+func (addr Address) String() string {
+	return addr.IP4().String()
+}
+
+func Add(addr Address, i Offset) Address {
+	return addr + Address(i)
+}
+
+func Subtract(a, b Address) Offset {
+	common.Assert(a >= b)
+	return Offset(a - b)
+}

ipam/allocate.go

@@ -0,0 +1,58 @@
+package ipam
+
+import (
+	"fmt"
+	"github.com/weaveworks/weave/ipam/address"
+)
+
+type allocateResult struct {
+	addr address.Address
+	err  error
+}
+
+type allocate struct {
+	resultChan       chan<- allocateResult
+	hasBeenCancelled func() bool
+	ident            string
+}
+
+// Try returns true if the request is completed, false if pending
+func (g *allocate) Try(alloc *Allocator) bool {
+	if g.hasBeenCancelled() {
+		g.Cancel()
+		return true
+	}
+
+	// If we have previously stored an address for this container, return it.
+	if addr, found := alloc.owned[g.ident]; found {
+		g.resultChan <- allocateResult{addr, nil}
+		return true
+	}
+
+	if ok, addr := alloc.space.Allocate(); ok {
+		alloc.debugln("Allocated", addr, "for", g.ident)
+		alloc.addOwned(g.ident, addr)
+		g.resultChan <- allocateResult{addr, nil}
+		return true
+	}
+
+	// out of space
+	if donor, err := alloc.ring.ChoosePeerToAskForSpace(); err == nil {
+		alloc.debugln("Decided to ask peer", donor, "for space")
+		alloc.sendRequest(donor, msgSpaceRequest)
+	}
+
+	return false
+}
+
+func (g *allocate) Cancel() {
+	g.resultChan <- allocateResult{0, fmt.Errorf("Request Cancelled")}
+}
+
+func (g *allocate) String() string {
+	return fmt.Sprintf("Allocate for %s", g.ident)
+}
+
+func (g *allocate) ForContainer(ident string) bool {
+	return g.ident == ident
+}

ipam/allocator.go

@@ -0,0 +1,694 @@
+package ipam
+
+import (
+	"bytes"
+	"encoding/gob"
+	"errors"
+	"fmt"
+	"net"
+	"sort"
+	"time"
+
+	"github.com/weaveworks/weave/common"
+	"github.com/weaveworks/weave/ipam/address"
+	"github.com/weaveworks/weave/ipam/paxos"
+	"github.com/weaveworks/weave/ipam/ring"
+	"github.com/weaveworks/weave/ipam/space"
+	"github.com/weaveworks/weave/router"
+)
+
+// Kinds of message we can unicast to other peers
+const (
+	msgSpaceRequest = iota
+	msgRingUpdate
+
+	paxosInterval = time.Second * 5
+)
+
+// operation represents something which Allocator wants to do, but
+// which may need to wait until some other message arrives.
+type operation interface {
+	// Try attempts this operations and returns false if needs to be tried again.
+	Try(alloc *Allocator) bool
+
+	Cancel()
+
+	String() string
+
+	// Does this operation pertain to the given container id?
+	// Used for tidying up pending operations when containers die.
+	ForContainer(ident string) bool
+}
+
+// Allocator brings together Ring and space.Set, and does the
+// necessary plumbing.  Runs as a single-threaded Actor, so no locks
+// are used around data structures.
+type Allocator struct {
+	actionChan       chan<- func()
+	ourName          router.PeerName
+	subnetStart      address.Address            // start address of space all peers are allocating from
+	subnetSize       address.Offset             // length of space all peers are allocating from
+	prefixLen        int                        // network prefix length, e.g. 24 for a /24 network
+	ring             *ring.Ring                 // information on ranges owned by all peers
+	space            space.Space                // more detail on ranges owned by us
+	owned            map[string]address.Address // who owns what address, indexed by container-ID
+	nicknames        map[router.PeerName]string // so we can map nicknames for rmpeer
+	pendingAllocates []operation                // held until we get some free space
+	pendingClaims    []operation                // held until we know who owns the space
+	gossip           router.Gossip              // our link to the outside world for sending messages
+	paxos            *paxos.Node
+	paxosTicker      *time.Ticker
+	shuttingDown     bool // to avoid doing any requests while trying to shut down
+	now              func() time.Time
+}
+
+// NewAllocator creates and initialises a new Allocator
+func NewAllocator(ourName router.PeerName, ourUID router.PeerUID, ourNickname string, subnetCIDR string, quorum uint) (*Allocator, error) {
+	_, subnet, err := net.ParseCIDR(subnetCIDR)
+	if err != nil {
+		return nil, err
+	}
+	if subnet.IP.To4() == nil {
+		return nil, errors.New("Non-IPv4 address not supported")
+	}
+	// Get the size of the network from the mask
+	ones, bits := subnet.Mask.Size()
+	var subnetSize address.Offset = 1 << uint(bits-ones)
+	if subnetSize < 4 {
+		return nil, errors.New("Allocation subnet too small")
+	}
+	subnetStart := address.FromIP4(subnet.IP)
+	alloc := &Allocator{
+		ourName:     ourName,
+		subnetStart: subnetStart,
+		subnetSize:  subnetSize,
+		prefixLen:   ones,
+		// per RFC 1122, don't allocate the first and last address in the subnet
+		ring:      ring.New(address.Add(subnetStart, 1), address.Add(subnetStart, subnetSize-1), ourName),
+		owned:     make(map[string]address.Address),
+		paxos:     paxos.NewNode(ourName, ourUID, quorum),
+		nicknames: map[router.PeerName]string{ourName: ourNickname},
+		now:       time.Now,
+	}
+	return alloc, nil
+}
+
+// Start runs the allocator goroutine
+func (alloc *Allocator) Start() {
+	actionChan := make(chan func(), router.ChannelSize)
+	alloc.actionChan = actionChan
+	go alloc.actorLoop(actionChan)
+}
+
+// Stop makes the actor routine exit, for test purposes ONLY because any
+// calls after this is processed will hang. Async.
+func (alloc *Allocator) Stop() {
+	alloc.actionChan <- nil
+}
+
+// Operation life cycle
+
+// Given an operation, try it, and add it to the pending queue if it didn't succeed
+func (alloc *Allocator) doOperation(op operation, ops *[]operation) {
+	alloc.actionChan <- func() {
+		if alloc.shuttingDown {
+			op.Cancel()
+			return
+		}
+		alloc.establishRing()
+		if !op.Try(alloc) {
+			*ops = append(*ops, op)
+		}
+	}
+}
+
+// Given an operation, remove it from the pending queue
+//  Note the op may not be on the queue; it may have
+//  already succeeded.  If it is on the queue, we call
+//  cancel on it, allowing callers waiting for the resultChans
+//  to unblock.
+func (alloc *Allocator) cancelOp(op operation, ops *[]operation) {
+	for i, op := range *ops {
+		if op == op {
+			*ops = append((*ops)[:i], (*ops)[i+1:]...)
+			op.Cancel()
+			break
+		}
+	}
+}
+
+// Cancel all operations in a queue
+func (alloc *Allocator) cancelOps(ops *[]operation) {
+	for _, op := range *ops {
+		op.Cancel()
+	}
+	*ops = []operation{}
+}
+
+// Cancel all operations for a given container id, returns true
+// if we found any.
+func (alloc *Allocator) cancelOpsFor(ops *[]operation, ident string) bool {
+	var found bool
+	for i, op := range *ops {
+		if op.ForContainer(ident) {
+			found = true
+			op.Cancel()
+			*ops = append((*ops)[:i], (*ops)[i+1:]...)
+		}
+	}
+	return found
+}
+
+// Try all pending operations
+func (alloc *Allocator) tryPendingOps() {
+	// The slightly different semantics requires us to operate on 'claims' and
+	// 'allocates' separately:
+	// Claims must be tried before Allocates
+	for i := 0; i < len(alloc.pendingClaims); {
+		op := alloc.pendingClaims[i]
+		if !op.Try(alloc) {
+			i++
+			continue
+		}
+		alloc.pendingClaims = append(alloc.pendingClaims[:i], alloc.pendingClaims[i+1:]...)
+	}
+
+	// When the first Allocate fails, bail - no need to
+	// send too many begs for space.
+	for i := 0; i < len(alloc.pendingAllocates); {
+		op := alloc.pendingAllocates[i]
+		if !op.Try(alloc) {
+			break
+		}
+		alloc.pendingAllocates = append(alloc.pendingAllocates[:i], alloc.pendingAllocates[i+1:]...)
+	}
+}
+
+func hasBeenCancelled(cancelChan <-chan bool) func() bool {
+	return func() bool {
+		select {
+		case <-cancelChan:
+			return true
+		default:
+			return false
+		}
+	}
+}
+
+// Actor client API
+
+// Allocate (Sync) - get IP address for container with given name
+// if there isn't any space we block indefinitely
+func (alloc *Allocator) Allocate(ident string, cancelChan <-chan bool) (address.Address, error) {
+	resultChan := make(chan allocateResult)
+	op := &allocate{resultChan: resultChan, ident: ident,
+		hasBeenCancelled: hasBeenCancelled(cancelChan)}
+	alloc.doOperation(op, &alloc.pendingAllocates)
+	result := <-resultChan
+	return result.addr, result.err
+}
+
+// Claim an address that we think we should own (Sync)
+func (alloc *Allocator) Claim(ident string, addr address.Address, cancelChan <-chan bool) error {
+	resultChan := make(chan error)
+	op := &claim{resultChan: resultChan, ident: ident, addr: addr,
+		hasBeenCancelled: hasBeenCancelled(cancelChan)}
+	alloc.doOperation(op, &alloc.pendingClaims)
+	return <-resultChan
+}
+
+// Free (Sync) - release IP address for container with given name
+func (alloc *Allocator) Free(ident string) error {
+	return alloc.free(ident)
+}
+
+// ContainerDied is provided to satisfy the updater interface; does a free underneath.  Async.
+func (alloc *Allocator) ContainerDied(ident string) error {
+	alloc.debugln("Container", ident, "died; releasing addresses")
+	return alloc.free(ident)
+}
+
+func (alloc *Allocator) free(ident string) error {
+	errChan := make(chan error)
+	alloc.actionChan <- func() {
+		addr, found := alloc.owned[ident]
+		if found {
+			alloc.space.Free(addr)
+		}
+		delete(alloc.owned, ident)
+
+		// Also remove any pending ops
+		found = alloc.cancelOpsFor(&alloc.pendingAllocates, ident) || found
+		found = alloc.cancelOpsFor(&alloc.pendingClaims, ident) || found
+
+		if !found {
+			errChan <- fmt.Errorf("No addresses for %s", ident)
+			return
+		}
+		errChan <- nil
+	}
+	return <-errChan
+}
+
+// Sync.
+func (alloc *Allocator) String() string {
+	resultChan := make(chan string)
+	alloc.actionChan <- func() {
+		resultChan <- alloc.string()
+	}
+	return <-resultChan
+}
+
+// Shutdown (Sync)
+func (alloc *Allocator) Shutdown() {
+	alloc.infof("Shutdown")
+	doneChan := make(chan struct{})
+	alloc.actionChan <- func() {
+		alloc.shuttingDown = true
+		alloc.cancelOps(&alloc.pendingClaims)
+		alloc.cancelOps(&alloc.pendingAllocates)
+		if heir := alloc.ring.PickPeerForTransfer(); heir != router.UnknownPeerName {
+			alloc.ring.Transfer(alloc.ourName, heir)
+			alloc.space.Clear()
+			alloc.gossip.GossipBroadcast(alloc.Gossip())
+			time.Sleep(100 * time.Millisecond)
+		}
+		doneChan <- struct{}{}
+	}
+	<-doneChan
+}
+
+// AdminTakeoverRanges (Sync) - take over the ranges owned by a given peer.
+// Only done on adminstrator command.
+func (alloc *Allocator) AdminTakeoverRanges(peerNameOrNickname string) error {
+	resultChan := make(chan error)
+	alloc.actionChan <- func() {
+		peername, err := alloc.lookupPeername(peerNameOrNickname)
+		if err != nil {
+			resultChan <- fmt.Errorf("Cannot find peer '%s'", peerNameOrNickname)
+			return
+		}
+
+		alloc.debugln("AdminTakeoverRanges:", peername)
+		if peername == alloc.ourName {
+			resultChan <- fmt.Errorf("Cannot take over ranges from yourself!")
+			return
+		}
+
+		delete(alloc.nicknames, peername)
+		err, newRanges := alloc.ring.Transfer(peername, alloc.ourName)
+		alloc.space.AddRanges(newRanges)
+		resultChan <- err
+	}
+	return <-resultChan
+}
+
+// Lookup a PeerName by nickname or stringified PeerName.  We can't
+// call into the router for this because we are interested in peers
+// that have gone away but are still in the ring, which is why we
+// maintain our own nicknames map.
+func (alloc *Allocator) lookupPeername(name string) (router.PeerName, error) {
+	for peername, nickname := range alloc.nicknames {
+		if nickname == name {
+			return peername, nil
+		}
+	}
+
+	return router.PeerNameFromString(name)
+}
+
+// Restrict the peers in "nicknames" to those in the ring and our own
+func (alloc *Allocator) pruneNicknames() {
+	ringPeers := alloc.ring.PeerNames()
+	for name, _ := range alloc.nicknames {
+		if _, ok := ringPeers[name]; !ok && name != alloc.ourName {
+			delete(alloc.nicknames, name)
+		}
+	}
+}
+
+// OnGossipUnicast (Sync)
+func (alloc *Allocator) OnGossipUnicast(sender router.PeerName, msg []byte) error {
+	alloc.debugln("OnGossipUnicast from", sender, ": ", len(msg), "bytes")
+	resultChan := make(chan error)
+	alloc.actionChan <- func() {
+		switch msg[0] {
+		case msgSpaceRequest:
+			// some other peer asked us for space
+			alloc.donateSpace(sender)
+			resultChan <- nil
+		case msgRingUpdate:
+			resultChan <- alloc.update(msg[1:])
+		}
+	}
+	return <-resultChan
+}
+
+// OnGossipBroadcast (Sync)
+func (alloc *Allocator) OnGossipBroadcast(msg []byte) (router.GossipData, error) {
+	alloc.debugln("OnGossipBroadcast:", len(msg), "bytes")
+	resultChan := make(chan error)
+	alloc.actionChan <- func() {
+		resultChan <- alloc.update(msg)
+	}
+	return alloc.Gossip(), <-resultChan
+}
+
+type gossipState struct {
+	// We send a timstamp along with the information to be
+	// gossipped in order to do detect skewed clocks
+	Now       int64
+	Nicknames map[router.PeerName]string
+
+	Paxos paxos.GossipState
+	Ring  *ring.Ring
+}
+
+func (alloc *Allocator) encode() []byte {
+	data := gossipState{
+		Now:       alloc.now().Unix(),
+		Nicknames: alloc.nicknames,
+	}
+
+	// We're only interested in Paxos until we have a Ring.
+	if alloc.ring.Empty() {
+		data.Paxos = alloc.paxos.GossipState()
+	} else {
+		data.Ring = alloc.ring
+	}
+	buf := new(bytes.Buffer)
+	enc := gob.NewEncoder(buf)
+	if err := enc.Encode(data); err != nil {
+		panic(err)
+	}
+	return buf.Bytes()
+}
+
+// Encode (Sync)
+func (alloc *Allocator) Encode() []byte {
+	resultChan := make(chan []byte)
+	alloc.actionChan <- func() {
+		resultChan <- alloc.encode()
+	}
+	return <-resultChan
+}
+
+// OnGossip (Sync)
+func (alloc *Allocator) OnGossip(msg []byte) (router.GossipData, error) {
+	alloc.debugln("Allocator.OnGossip:", len(msg), "bytes")
+	resultChan := make(chan error)
+	alloc.actionChan <- func() {
+		resultChan <- alloc.update(msg)
+	}
+	return nil, <-resultChan // for now, we never propagate updates. TBD
+}
+
+// GossipData implementation is trivial - we always gossip the latest
+// data we have at time of sending
+type ipamGossipData struct {
+	alloc *Allocator
+}
+
+func (d *ipamGossipData) Merge(other router.GossipData) {
+	// no-op
+}
+
+func (d *ipamGossipData) Encode() []byte {
+	return d.alloc.Encode()
+}
+
+// Gossip returns a GossipData implementation, which in this case always
+// returns the latest ring state (and does nothing on merge)
+func (alloc *Allocator) Gossip() router.GossipData {
+	return &ipamGossipData{alloc}
+}
+
+// SetInterfaces gives the allocator two interfaces for talking to the outside world
+func (alloc *Allocator) SetInterfaces(gossip router.Gossip) {
+	alloc.gossip = gossip
+}
+
+// ACTOR server
+
+func (alloc *Allocator) actorLoop(actionChan <-chan func()) {
+	for {
+		var tickChan <-chan time.Time
+		if alloc.paxosTicker != nil {
+			tickChan = alloc.paxosTicker.C
+		}
+
+		select {
+		case action := <-actionChan:
+			if action == nil {
+				return
+			}
+			action()
+		case <-tickChan:
+			alloc.propose()
+		}
+
+		alloc.assertInvariants()
+		alloc.reportFreeSpace()
+	}
+}
+
+// Helper functions
+
+func (alloc *Allocator) string() string {
+	var buf bytes.Buffer
+	fmt.Fprintf(&buf, "Allocator subnet %s+%d\n", alloc.subnetStart.String(), alloc.subnetSize)
+
+	if alloc.ring.Empty() {
+		fmt.Fprintf(&buf, "Paxos: %s", alloc.paxos.String())
+	} else {
+		localFreeSpace := alloc.space.NumFreeAddresses()
+		remoteFreeSpace := alloc.ring.TotalRemoteFree()
+		percentFree := 100 * float64(localFreeSpace+remoteFreeSpace) / float64(alloc.subnetSize)
+		fmt.Fprintf(&buf, "  Free IPs: ~%.1f%%, %d local, ~%d remote\n",
+			percentFree, localFreeSpace, remoteFreeSpace)
+
+		alloc.ring.FprintWithNicknames(&buf, alloc.nicknames)
+		fmt.Fprintf(&buf, alloc.space.String())
+		if len(alloc.pendingAllocates)+len(alloc.pendingClaims) > 0 {
+			fmt.Fprintf(&buf, "\nPending requests for ")
+		}
+		for _, op := range alloc.pendingAllocates {
+			fmt.Fprintf(&buf, "%s, ", op.String())
+		}
+		for _, op := range alloc.pendingClaims {
+			fmt.Fprintf(&buf, "%s, ", op.String())
+		}
+	}
+	return buf.String()
+}
+
+func (alloc *Allocator) consensus() []router.PeerName {
+	_, val := alloc.paxos.Consensus()
+	return val.Value
+}
+
+// Ensure we are making progress towards an established ring
+func (alloc *Allocator) establishRing() {
+	if !alloc.ring.Empty() || alloc.paxosTicker != nil {
+		return
+	}
+
+	alloc.propose()
+	if cons := alloc.consensus(); cons != nil {
+		// If the quorum was 1, then proposing immediately
+		// leads to consensus
+		alloc.createRing(cons)
+	} else {
+		// re-try until we get consensus
+		alloc.paxosTicker = time.NewTicker(paxosInterval)
+	}
+}
+
+func (alloc *Allocator) createRing(peers []router.PeerName) {
+	alloc.debugln("Paxos consensus:", peers)
+	alloc.ring.ClaimForPeers(normalizeConsensus(peers))
+	alloc.gossip.GossipBroadcast(alloc.Gossip())
+	alloc.ringUpdated()
+}
+
+func (alloc *Allocator) ringUpdated() {
+	// When we have a ring, we don't need paxos any more
+	if alloc.paxos != nil {
+		alloc.paxos = nil
+
+		if alloc.paxosTicker != nil {
+			alloc.paxosTicker.Stop()
+			alloc.paxosTicker = nil
+		}
+	}
+
+	alloc.space.UpdateRanges(alloc.ring.OwnedRanges())
+	alloc.tryPendingOps()
+}
+
+// For compatibility with sort.Interface
+type peerNames []router.PeerName
+
+func (a peerNames) Len() int           { return len(a) }
+func (a peerNames) Less(i, j int) bool { return a[i] < a[j] }
+func (a peerNames) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
+
+// When we get a consensus from Paxos, the peer names are not in a
+// defined order and may contain duplicates.  This function sorts them
+// and de-dupes.
+func normalizeConsensus(consensus []router.PeerName) []router.PeerName {
+	if len(consensus) == 0 {
+		return nil
+	}
+
+	peers := make(peerNames, len(consensus))
+	copy(peers, consensus)
+	sort.Sort(peers)
+
+	dst := 0
+	for src := 1; src < len(peers); src++ {
+		if peers[dst] != peers[src] {
+			dst++
+			peers[dst] = peers[src]
+		}
+	}
+
+	return peers[:dst+1]
+}
+
+func (alloc *Allocator) propose() {
+	alloc.debugf("Paxos proposing")
+	alloc.paxos.Propose()
+	alloc.gossip.GossipBroadcast(alloc.Gossip())
+}
+
+func (alloc *Allocator) sendRequest(dest router.PeerName, kind byte) {
+	msg := router.Concat([]byte{kind}, alloc.encode())
+	alloc.gossip.GossipUnicast(dest, msg)
+}
+
+func (alloc *Allocator) update(msg []byte) error {
+	reader := bytes.NewReader(msg)
+	decoder := gob.NewDecoder(reader)
+	var data gossipState
+	var err error
+
+	if err := decoder.Decode(&data); err != nil {
+		return err
+	}
+
+	deltat := time.Unix(data.Now, 0).Sub(alloc.now())
+	if deltat > time.Hour || -deltat > time.Hour {
+		return fmt.Errorf("clock skew of %v detected, ignoring update", deltat)
+	}
+
+	// Merge nicknames
+	for peer, nickname := range data.Nicknames {
+		alloc.nicknames[peer] = nickname
+	}
+
+	// only one of Ring and Paxos should be present.  And we
+	// shouldn't get updates for a empty Ring. But tolerate
+	// them just in case.
+	if data.Ring != nil {
+		err = alloc.ring.Merge(*data.Ring)
+		if !alloc.ring.Empty() {
+			alloc.pruneNicknames()
+			alloc.ringUpdated()
+		}
+		return err
+	}
+
+	if data.Paxos != nil && alloc.ring.Empty() {
+		if alloc.paxos.Update(data.Paxos) {
+			if alloc.paxos.Think() {
+				// If something important changed, broadcast
+				alloc.gossip.GossipBroadcast(alloc.Gossip())
+			}
+
+			if cons := alloc.consensus(); cons != nil {
+				alloc.createRing(cons)
+			}
+		}
+	}
+
+	return nil
+}
+
+func (alloc *Allocator) donateSpace(to router.PeerName) {
+	// No matter what we do, we'll send a unicast gossip
+	// of our ring back to tha chap who asked for space.
+	// This serves to both tell him of any space we might
+	// have given him, or tell him where he might find some
+	// more.
+	defer alloc.sendRequest(to, msgRingUpdate)
+
+	alloc.debugln("Peer", to, "asked me for space")
+	start, size, ok := alloc.space.Donate()
+	if !ok {
+		free := alloc.space.NumFreeAddresses()
+		common.Assert(free == 0)
+		alloc.debugln("No space to give to peer", to)
+		return
+	}
+	end := address.Add(start, size)
+	alloc.debugln("Giving range", start, end, size, "to", to)
+	alloc.ring.GrantRangeToHost(start, end, to)
+}
+
+func (alloc *Allocator) assertInvariants() {
+	// We need to ensure all ranges the ring thinks we own have
+	// a corresponding space in the space set, and vice versa
+	checkSpace := space.New()
+	checkSpace.AddRanges(alloc.ring.OwnedRanges())
+	ranges := checkSpace.OwnedRanges()
+	spaces := alloc.space.OwnedRanges()
+
+	common.Assert(len(ranges) == len(spaces))
+
+	for i := 0; i < len(ranges); i++ {
+		r := ranges[i]
+		s := spaces[i]
+		common.Assert(s.Start == r.Start && s.End == r.End)
+	}
+}
+
+func (alloc *Allocator) reportFreeSpace() {
+	ranges := alloc.ring.OwnedRanges()
+	if len(ranges) == 0 {
+		return
+	}
+
+	freespace := make(map[address.Address]address.Offset)
+	for _, r := range ranges {
+		freespace[r.Start] = alloc.space.NumFreeAddressesInRange(r.Start, r.End)
+	}
+	alloc.ring.ReportFree(freespace)
+}
+
+// Owned addresses
+
+func (alloc *Allocator) addOwned(ident string, addr address.Address) {
+	alloc.owned[ident] = addr
+}
+
+func (alloc *Allocator) findOwner(addr address.Address) string {
+	for ident, candidate := range alloc.owned {
+		if candidate == addr {
+			return ident
+		}
+	}
+	return ""
+}
+
+// Logging
+
+func (alloc *Allocator) infof(fmt string, args ...interface{}) {
+	common.Info.Printf("[allocator %s] "+fmt, append([]interface{}{alloc.ourName}, args...)...)
+}
+func (alloc *Allocator) debugln(args ...interface{}) {
+	common.Debug.Println(append([]interface{}{fmt.Sprintf("[allocator %s]:", alloc.ourName)}, args...)...)
+}
+func (alloc *Allocator) debugf(fmt string, args ...interface{}) {
+	common.Debug.Printf("[allocator %s] "+fmt, append([]interface{}{alloc.ourName}, args...)...)
+}

ipam/allocator_test.go

@@ -0,0 +1,453 @@
+package ipam
+
+import (
+	"fmt"
+	"math/rand"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/weaveworks/weave/common"
+	"github.com/weaveworks/weave/ipam/address"
+	"github.com/weaveworks/weave/router"
+	wt "github.com/weaveworks/weave/testing"
+)
+
+const (
+	testStart1 = "10.0.1.0"
+	testStart2 = "10.0.2.0"
+	testStart3 = "10.0.3.0"
+)
+
+func TestAllocFree(t *testing.T) {
+	const (
+		container1 = "abcdef"
+		container2 = "baddf00d"
+		container3 = "b01df00d"
+		universe   = "10.0.3.0/28"
+		testAddr1  = "10.0.3.1" // first address allocated should be .1 because .0 is network addr
+		spaceSize  = 14         // 16 IP addresses in /28, minus .0 and .15
+	)
+
+	alloc := makeAllocatorWithMockGossip(t, "01:00:00:01:00:00", universe, 1)
+	defer alloc.Stop()
+
+	alloc.claimRingForTesting()
+	addr1, _ := alloc.Allocate(container1, nil)
+	wt.AssertEqualString(t, addr1.String(), testAddr1, "address")
+
+	// Ask for another address for a different container and check it's different
+	addr2, _ := alloc.Allocate(container2, nil)
+	if addr2.String() == testAddr1 {
+		t.Fatalf("Expected different address but got %s", addr2.String())
+	}
+
+	// Ask for the first container again and we should get the same address again
+	addr1a, _ := alloc.Allocate(container1, nil)
+	wt.AssertEqualString(t, addr1a.String(), testAddr1, "address")
+
+	// Now free the first one, and we should get it back when we ask
+	wt.AssertSuccess(t, alloc.Free(container1))
+	addr3, _ := alloc.Allocate(container3, nil)
+	wt.AssertEqualString(t, addr3.String(), testAddr1, "address")
+
+	alloc.ContainerDied(container2)
+	alloc.ContainerDied(container3)
+	alloc.String() // force sync-up after async call
+	wt.AssertEquals(t, alloc.space.NumFreeAddresses(), address.Offset(spaceSize))
+}
+
+func TestBootstrap(t *testing.T) {
+	common.InitDefaultLogging(false)
+	const (
+		donateSize     = 5
+		donateStart    = "10.0.1.7"
+		ourNameString  = "01:00:00:01:00:00"
+		peerNameString = "02:00:00:02:00:00"
+	)
+
+	alloc1 := makeAllocatorWithMockGossip(t, ourNameString, testStart1+"/22", 2)
+	defer alloc1.Stop()
+
+	// Simulate another peer on the gossip network
+	alloc2 := makeAllocatorWithMockGossip(t, peerNameString, testStart1+"/22", 2)
+	defer alloc2.Stop()
+
+	alloc1.OnGossipBroadcast(alloc2.encode())
+
+	alloc1.tryPendingOps()
+
+	ExpectBroadcastMessage(alloc1, nil) // alloc1 will try to form consensus
+	done := make(chan bool)
+	go func() {
+		alloc1.Allocate("somecontainer", nil)
+		done <- true
+	}()
+	time.Sleep(100 * time.Millisecond)
+	AssertNothingSent(t, done)
+
+	CheckAllExpectedMessagesSent(alloc1, alloc2)
+
+	alloc1.tryPendingOps()
+	AssertNothingSent(t, done)
+
+	CheckAllExpectedMessagesSent(alloc1, alloc2)
+
+	// alloc2 receives paxos update and broadcasts its reply
+	ExpectBroadcastMessage(alloc2, nil)
+	alloc2.OnGossipBroadcast(alloc1.encode())
+
+	ExpectBroadcastMessage(alloc1, nil)
+	alloc1.OnGossipBroadcast(alloc2.encode())
+
+	// both nodes will get consensus now so initialize the ring
+	ExpectBroadcastMessage(alloc2, nil)
+	ExpectBroadcastMessage(alloc2, nil)
+	alloc2.OnGossipBroadcast(alloc1.encode())
+
+	CheckAllExpectedMessagesSent(alloc1, alloc2)
+
+	alloc1.OnGossipBroadcast(alloc2.encode())
+	// now alloc1 should have space
+
+	AssertSent(t, done)
+
+	CheckAllExpectedMessagesSent(alloc1, alloc2)
+}
+
+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
+	)
+
+	alloc := makeAllocatorWithMockGossip(t, "01:00:00:01:00:00", universe, 1)
+	defer alloc.Stop()
+
+	alloc.claimRingForTesting()
+	addr1, _ := alloc.Allocate(container1, nil)
+	alloc.Allocate(container2, nil)
+
+	// Now free the first one, and try to claim it
+	wt.AssertSuccess(t, alloc.Free(container1))
+	t.Log(alloc)
+	err := alloc.Claim(container3, addr1, nil)
+	wt.AssertNoErr(t, err)
+	addr3, _ := alloc.Allocate(container3, nil)
+	wt.AssertEqualString(t, addr3.String(), testAddr1, "address")
+}
+
+func (alloc *Allocator) pause() func() {
+	paused := make(chan struct{})
+	alloc.actionChan <- func() {
+		paused <- struct{}{}
+		<-paused
+	}
+	<-paused
+	return func() {
+		close(paused)
+	}
+}
+
+func TestCancel(t *testing.T) {
+	common.InitDefaultLogging(false)
+	const (
+		CIDR = "10.0.1.7/26"
+	)
+
+	router := TestGossipRouter{make(map[router.PeerName]chan gossipMessage), 0.0}
+
+	alloc1 := makeAllocator("01:00:00:02:00:00", CIDR, 2)
+	alloc1.SetInterfaces(router.connect(alloc1.ourName, alloc1))
+
+	alloc2 := makeAllocator("02:00:00:02:00:00", CIDR, 2)
+	alloc2.SetInterfaces(router.connect(alloc2.ourName, alloc2))
+	alloc1.claimRingForTesting(alloc1, alloc2)
+	alloc2.claimRingForTesting(alloc1, alloc2)
+
+	alloc1.Start()
+	alloc2.Start()
+
+	// tell peers about each other
+	alloc1.OnGossipBroadcast(alloc2.encode())
+
+	// Get some IPs, so each allocator has some space
+	res1, _ := alloc1.Allocate("foo", nil)
+	common.Debug.Printf("res1 = %s", res1.String())
+	res2, _ := alloc2.Allocate("bar", nil)
+	common.Debug.Printf("res2 = %s", res2.String())
+	if res1 == res2 {
+		wt.Fatalf(t, "Error: got same ips!")
+	}
+
+	// Now we're going to pause alloc2 and ask alloc1
+	// for an allocation
+	unpause := alloc2.pause()
+
+	// Use up all the IPs that alloc1 owns, so the allocation after this will prompt a request to alloc2
+	for i := 0; alloc1.space.NumFreeAddresses() > 0; i++ {
+		alloc1.Allocate(fmt.Sprintf("tmp%d", i), nil)
+	}
+	cancelChan := make(chan bool, 1)
+	doneChan := make(chan bool)
+	go func() {
+		_, ok := alloc1.Allocate("baz", cancelChan)
+		doneChan <- ok == nil
+	}()
+
+	AssertNothingSent(t, doneChan)
+	time.Sleep(100 * time.Millisecond)
+	AssertNothingSent(t, doneChan)
+
+	cancelChan <- true
+	unpause()
+	if <-doneChan {
+		wt.Fatalf(t, "Error: got result from Allocate")
+	}
+}
+
+func TestGossipShutdown(t *testing.T) {
+	const (
+		container1 = "abcdef"
+		container2 = "baddf00d"
+		universe   = "10.0.3.0/30"
+		testAddr1  = "10.0.3.1" // first address allocated should be .1 because .0 is network addr
+	)
+
+	alloc := makeAllocatorWithMockGossip(t, "01:00:00:01:00:00", universe, 1)
+	defer alloc.Stop()
+
+	alloc.claimRingForTesting()
+	addr1, _ := alloc.Allocate(container1, nil)
+	wt.AssertEqualString(t, addr1.String(), testAddr1, "address")
+
+	alloc.Shutdown()
+
+	_, err := alloc.Allocate(container2, nil) // trying to allocate after shutdown should fail
+	wt.AssertFalse(t, err == nil, "no address")
+
+	CheckAllExpectedMessagesSent(alloc)
+}
+
+func TestTransfer(t *testing.T) {
+	const (
+		cidr = "10.0.1.7/22"
+	)
+	allocs, router := makeNetworkOfAllocators(3, cidr)
+	alloc1 := allocs[0]
+	alloc2 := allocs[1]
+	alloc3 := allocs[2] // This will be 'master' and get the first range
+
+	_, err := alloc2.Allocate("foo", nil)
+	wt.AssertTrue(t, err == nil, "Failed to get address")
+
+	_, err = alloc3.Allocate("bar", nil)
+	wt.AssertTrue(t, err == nil, "Failed to get address")
+
+	router.GossipBroadcast(alloc2.Gossip())
+	router.GossipBroadcast(alloc3.Gossip())
+	router.removePeer(alloc2.ourName)
+	router.removePeer(alloc3.ourName)
+	alloc2.Stop()
+	alloc3.Stop()
+	wt.AssertSuccess(t, alloc1.AdminTakeoverRanges(alloc2.ourName.String()))
+	wt.AssertSuccess(t, alloc1.AdminTakeoverRanges(alloc3.ourName.String()))
+
+	wt.AssertEquals(t, alloc1.space.NumFreeAddresses(), address.Offset(1022))
+
+	_, err = alloc1.Allocate("foo", nil)
+	wt.AssertTrue(t, err == nil, "Failed to get address")
+	alloc1.Stop()
+}
+
+func TestFakeRouterSimple(t *testing.T) {
+	common.InitDefaultLogging(false)
+	const (
+		cidr = "10.0.1.7/22"
+	)
+	allocs, _ := makeNetworkOfAllocators(2, cidr)
+	defer stopNetworkOfAllocators(allocs)
+
+	alloc1 := allocs[0]
+	//alloc2 := allocs[1]
+
+	addr, _ := alloc1.Allocate("foo", nil)
+	println("Got addr", addr)
+}
+
+func TestAllocatorFuzz(t *testing.T) {
+	common.InitDefaultLogging(false)
+	const (
+		firstpass    = 1000
+		secondpass   = 5000
+		nodes        = 10
+		maxAddresses = 1000
+		concurrency  = 10
+		cidr         = "10.0.1.7/22"
+	)
+	allocs, _ := makeNetworkOfAllocators(nodes, cidr)
+	defer stopNetworkOfAllocators(allocs)
+
+	// Test state
+	// For each IP issued we store the allocator
+	// that issued it and the name of the container
+	// it was issued to.
+	type result struct {
+		name  string
+		alloc int32
+	}
+	stateLock := sync.Mutex{}
+	state := make(map[string]result)
+	// Keep a list of addresses issued, so we
+	// Can pick random ones
+	var addrs []string
+	numPending := 0
+
+	rand.Seed(0)
+
+	// Remove item from list by swapping it with last
+	// and reducing slice length by 1
+	rm := func(xs []string, i int32) []string {
+		ls := len(xs) - 1
+		xs[i] = xs[ls]
+		return xs[:ls]
+	}
+
+	// Do a Allocate and check the address
+	// is unique.  Needs a unique container
+	// name.
+	allocate := func(name string) {
+		stateLock.Lock()
+		if len(addrs)+numPending >= maxAddresses {
+			stateLock.Unlock()
+			return
+		}
+		numPending++
+		stateLock.Unlock()
+
+		allocIndex := rand.Int31n(nodes)
+		alloc := allocs[allocIndex]
+		//common.Info.Printf("Allocate: asking allocator %d", allocIndex)
+		addr, err := alloc.Allocate(name, nil)
+
+		if err != nil {
+			panic(fmt.Sprintf("Could not allocate addr"))
+		}
+
+		//common.Info.Printf("Allocate: got address %s for name %s", addr, name)
+		addrStr := addr.String()
+
+		stateLock.Lock()
+		defer stateLock.Unlock()
+
+		if res, existing := state[addrStr]; existing {
+			panic(fmt.Sprintf("Dup found for address %s - %s and %s", addrStr,
+				name, res.name))
+		}
+
+		state[addrStr] = result{name, allocIndex}
+		addrs = append(addrs, addrStr)
+		numPending--
+	}
+
+	// Free a random address.
+	free := func() {
+		stateLock.Lock()
+		if len(addrs) == 0 {
+			stateLock.Unlock()
+			return
+		}
+		// Delete an existing allocation
+		// Pick random addr
+		addrIndex := rand.Int31n(int32(len(addrs)))
+		addr := addrs[addrIndex]
+		res := state[addr]
+		addrs = rm(addrs, addrIndex)
+		delete(state, addr)
+		stateLock.Unlock()
+
+		alloc := allocs[res.alloc]
+		//common.Info.Printf("Freeing %s on allocator %d", addr, res.alloc)
+
+		wt.AssertSuccess(t, alloc.Free(res.name))
+	}
+
+	// Do a Allocate on an existing container & allocator
+	// and check we get the right answer.
+	allocateAgain := func() {
+		stateLock.Lock()
+		addrIndex := rand.Int31n(int32(len(addrs)))
+		addr := addrs[addrIndex]
+		res := state[addr]
+		stateLock.Unlock()
+		alloc := allocs[res.alloc]
+
+		//common.Info.Printf("Asking for %s on allocator %d again", addr, res.alloc)
+
+		newAddr, _ := alloc.Allocate(res.name, nil)
+		oldAddr, _ := address.ParseIP(addr)
+		if newAddr != oldAddr {
+			panic(fmt.Sprintf("Got different address for repeat request"))
+		}
+	}
+
+	// Run function _f_ _iterations_ times, in _concurrency_
+	// number of goroutines
+	doConcurrentIterations := func(iterations int, f func(int)) {
+		iterationsPerThread := iterations / concurrency
+
+		wg := sync.WaitGroup{}
+		for i := 0; i < concurrency; i++ {
+			wg.Add(1)
+			go func(j int) {
+				defer wg.Done()
+				for k := 0; k < iterationsPerThread; k++ {
+					f((j * iterationsPerThread) + k)
+				}
+			}(i)
+		}
+		wg.Wait()
+	}
+
+	// First pass, just allocate a bunch of ips
+	doConcurrentIterations(firstpass, func(iteration int) {
+		name := fmt.Sprintf("first%d", iteration)
+		allocate(name)
+	})
+
+	// Second pass, random ask for more allocations,
+	// or remove existing ones, or ask for allocation
+	// again.
+	doConcurrentIterations(secondpass, func(iteration int) {
+		r := rand.Float32()
+		switch {
+		case 0.0 <= r && r < 0.4:
+			// Ask for a new allocation
+			name := fmt.Sprintf("second%d", iteration)
+			allocate(name)
+
+		case (0.4 <= r && r < 0.8):
+			// free a random addr
+			free()
+
+		case 0.8 <= r && r < 1.0:
+			// ask for an existing name again, check we get same ip
+			allocateAgain()
+		}
+	})
+}
+
+func TestGossipSkew(t *testing.T) {
+	alloc1 := makeAllocatorWithMockGossip(t, "01:00:00:01:00:00", "10.0.1.0/22", 2)
+	defer alloc1.Stop()
+	alloc2 := makeAllocatorWithMockGossip(t, "02:00:00:02:00:00", "10.0.1.0/22", 2)
+	alloc2.now = func() time.Time { return time.Now().Add(time.Hour * 2) }
+	defer alloc2.Stop()
+
+	if _, err := alloc1.OnGossipBroadcast(alloc2.encode()); err == nil {
+		t.Fail()
+	}
+}

ipam/claim.go

@@ -0,0 +1,75 @@
+package ipam
+
+import (
+	"fmt"
+
+	"github.com/weaveworks/weave/ipam/address"
+	"github.com/weaveworks/weave/router"
+)
+
+type claim struct {
+	resultChan       chan<- error
+	hasBeenCancelled func() bool
+	ident            string
+	addr             address.Address
+}
+
+// Try returns true for success (or failure), false if we need to try again later
+func (c *claim) Try(alloc *Allocator) bool {
+	if (c.hasBeenCancelled)() {
+		c.Cancel()
+		return true
+	}
+
+	if !alloc.ring.Contains(c.addr) {
+		// Address not within our universe; assume user knows what they are doing
+		alloc.infof("Ignored address %s claimed by %s - not in our universe\n", c.addr, c.ident)
+		c.resultChan <- nil
+		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).
+	owner := alloc.ring.Owner(c.addr)
+	if owner == router.UnknownPeerName {
+		alloc.infof("Ring is empty; will try later.\n", c.addr, owner)
+		return false
+	}
+	if owner != alloc.ourName {
+		c.resultChan <- fmt.Errorf("Address %s is owned by other peer %s", c.addr.String(), owner)
+		return true
+	}
+	// We are the owner, check we haven't given it to another container
+	existingIdent := alloc.findOwner(c.addr)
+	if existingIdent == c.ident {
+		// same identifier is claiming same address; that's OK
+		c.resultChan <- nil
+		return true
+	}
+	if existingIdent == "" {
+		err := alloc.space.Claim(c.addr)
+		if err != nil {
+			c.resultChan <- err
+			return true
+		}
+		alloc.addOwned(c.ident, c.addr)
+		c.resultChan <- nil
+		return true
+	}
+	// Addr already owned by container on this machine
+	c.resultChan <- fmt.Errorf("Claimed address %s is already owned by %s", c.addr.String(), existingIdent)
+	return true
+}
+
+func (c *claim) Cancel() {
+	c.resultChan <- fmt.Errorf("Operation cancelled.")
+}
+
+func (c *claim) String() string {
+	return fmt.Sprintf("Claim %s -> %s", c.ident, c.addr.String())
+}
+
+func (c *claim) ForContainer(ident string) bool {
+	return c.ident == ident
+}

ipam/dummy.go

@@ -0,0 +1,36 @@
+package ipam
+
+import (
+	"errors"
+
+	"github.com/weaveworks/weave/router"
+)
+
+// Exists solely for the purpose of detecting a network in which some
+// peers are configured with IPAM on and some with IPAM off.
+type DummyAllocator struct {
+}
+
+var (
+	ErrMixedConfig = errors.New("Mixed IPAM/non-IPAM configuration not supported")
+)
+
+func (alloc *DummyAllocator) OnGossipUnicast(sender router.PeerName, msg []byte) error {
+	return ErrMixedConfig
+}
+
+func (alloc *DummyAllocator) OnGossipBroadcast(msg []byte) (router.GossipData, error) {
+	return nil, ErrMixedConfig
+}
+
+func (alloc *DummyAllocator) OnGossip(msg []byte) (router.GossipData, error) {
+	return nil, ErrMixedConfig
+}
+
+func (alloc *DummyAllocator) Gossip() router.GossipData {
+	return nil
+}
+
+func (alloc *DummyAllocator) Encode() []byte {
+	return nil
+}

ipam/http.go

@@ -0,0 +1,72 @@
+package ipam
+
+import (
+	"fmt"
+	"net/http"
+
+	"github.com/gorilla/mux"
+
+	"github.com/weaveworks/weave/common"
+	"github.com/weaveworks/weave/ipam/address"
+)
+
+func badRequest(w http.ResponseWriter, err error) {
+	http.Error(w, err.Error(), http.StatusBadRequest)
+	common.Warning.Println(err.Error())
+}
+
+// HandleHTTP wires up ipams HTTP endpoints to the provided mux.
+func (alloc *Allocator) HandleHTTP(router *mux.Router) {
+	router.Methods("PUT").Path("/ip/{id}/{ip}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		closedChan := w.(http.CloseNotifier).CloseNotify()
+		vars := mux.Vars(r)
+		ident := vars["id"]
+		ipStr := vars["ip"]
+		if ip, err := address.ParseIP(ipStr); err != nil {
+			badRequest(w, err)
+			return
+		} else if err := alloc.Claim(ident, ip, closedChan); err != nil {
+			badRequest(w, fmt.Errorf("Unable to claim IP address %s: %s", ip, err))
+			return
+		}
+
+		w.WriteHeader(204)
+	})
+
+	router.Methods("POST").Path("/ip/{id}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		closedChan := w.(http.CloseNotifier).CloseNotify()
+		ident := mux.Vars(r)["id"]
+		newAddr, err := alloc.Allocate(ident, closedChan)
+		if err != nil {
+			badRequest(w, err)
+			return
+		}
+
+		fmt.Fprintf(w, "%s/%d", newAddr.String(), alloc.prefixLen)
+	})
+
+	router.Methods("DELETE").Path("/ip/{id}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		ident := mux.Vars(r)["id"]
+		if err := alloc.Free(ident); err != nil {
+			badRequest(w, err)
+			return
+		}
+
+		w.WriteHeader(204)
+	})
+
+	router.Methods("DELETE").Path("/peer").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		alloc.Shutdown()
+		w.WriteHeader(204)
+	})
+
+	router.Methods("DELETE").Path("/peer/{id}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		ident := mux.Vars(r)["id"]
+		if err := alloc.AdminTakeoverRanges(ident); err != nil {
+			badRequest(w, err)
+			return
+		}
+
+		w.WriteHeader(204)
+	})
+}

ipam/http_test.go

@@ -0,0 +1,164 @@
+package ipam
+
+import (
+	"fmt"
+	"io"
+	"io/ioutil"
+	"math/rand"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/gorilla/mux"
+
+	"github.com/weaveworks/weave/common"
+	wt "github.com/weaveworks/weave/testing"
+)
+
+func HTTPPost(t *testing.T, url string) string {
+	resp, err := http.Post(url, "", nil)
+	wt.AssertNoErr(t, err)
+	wt.AssertStatus(t, resp.StatusCode, http.StatusOK, "http response")
+	defer resp.Body.Close()
+	body, _ := ioutil.ReadAll(resp.Body)
+	return string(body)
+}
+
+func doHTTP(method string, url string) (resp *http.Response, err error) {
+	req, _ := http.NewRequest(method, url, nil)
+	return http.DefaultClient.Do(req)
+}
+
+func listenHTTP(port int, alloc *Allocator) {
+	router := mux.NewRouter()
+	router.HandleFunc("/status", func(w http.ResponseWriter, r *http.Request) {
+		io.WriteString(w, fmt.Sprintln(alloc))
+	})
+	alloc.HandleHTTP(router)
+
+	srv := &http.Server{
+		Addr:    fmt.Sprintf(":%d", port),
+		Handler: router,
+	}
+	if err := srv.ListenAndServe(); err != nil {
+		common.Error.Fatal("Unable to create http listener: ", err)
+	}
+}
+
+func allocURL(port int, containerID string) string {
+	return fmt.Sprintf("http://localhost:%d/ip/%s", port, containerID)
+}
+
+func TestHttp(t *testing.T) {
+	var (
+		containerID = "deadbeef"
+		container2  = "baddf00d"
+		container3  = "b01df00d"
+		testAddr1   = "10.0.3.9"
+		netSuffix   = "/29"
+		testCIDR1   = "10.0.3.8" + netSuffix
+	)
+
+	alloc := makeAllocatorWithMockGossip(t, "08:00:27:01:c3:9a", testCIDR1, 1)
+	port := rand.Intn(10000) + 32768
+	fmt.Println("Http test on port", port)
+	go listenHTTP(port, alloc)
+
+	time.Sleep(100 * time.Millisecond) // Allow for http server to get going
+
+	alloc.claimRingForTesting()
+	// Ask the http server for a new address
+	cidr1 := HTTPPost(t, allocURL(port, containerID))
+	wt.AssertEqualString(t, cidr1, testAddr1+netSuffix, "address")
+
+	// Ask the http server for another address and check it's different
+	cidr2 := HTTPPost(t, allocURL(port, container2))
+	wt.AssertNotEqualString(t, cidr2, testAddr1+netSuffix, "address")
+
+	// Ask for the first container again and we should get the same address again
+	cidr1a := HTTPPost(t, allocURL(port, containerID))
+	wt.AssertEqualString(t, cidr1a, testAddr1+netSuffix, "address")
+
+	// Now free the first one, and we should get it back when we ask
+	doHTTP("DELETE", allocURL(port, containerID))
+	cidr3 := HTTPPost(t, allocURL(port, container3))
+	wt.AssertEqualString(t, cidr3, testAddr1+netSuffix, "address")
+
+	// Would like to shut down the http server at the end of this test
+	// but it's complicated.
+	// See https://groups.google.com/forum/#!topic/golang-nuts/vLHWa5sHnCE
+}
+
+func TestBadHttp(t *testing.T) {
+	var (
+		containerID = "deadbeef"
+		testCIDR1   = "10.0.0.0/8"
+	)
+
+	alloc := makeAllocatorWithMockGossip(t, "08:00:27:01:c3:9a", testCIDR1, 1)
+	defer alloc.Stop()
+	port := rand.Intn(10000) + 32768
+	fmt.Println("BadHttp test on port", port)
+	go listenHTTP(port, alloc)
+
+	alloc.claimRingForTesting()
+	cidr1 := HTTPPost(t, allocURL(port, containerID))
+	parts := strings.Split(cidr1, "/")
+	testAddr1 := parts[0]
+	// Verb that's not handled
+	resp, err := doHTTP("HEAD", fmt.Sprintf("http://localhost:%d/ip/%s/%s", port, containerID, testAddr1))
+	wt.AssertNoErr(t, err)
+	wt.AssertStatus(t, resp.StatusCode, http.StatusNotFound, "http response")
+	// Mis-spelled URL
+	resp, err = doHTTP("POST", fmt.Sprintf("http://localhost:%d/xip/%s/", port, containerID))
+	wt.AssertNoErr(t, err)
+	wt.AssertStatus(t, resp.StatusCode, http.StatusNotFound, "http response")
+	// Malformed URL
+	resp, err = doHTTP("POST", fmt.Sprintf("http://localhost:%d/ip/%s/foo/bar/baz", port, containerID))
+	wt.AssertNoErr(t, err)
+	wt.AssertStatus(t, resp.StatusCode, http.StatusNotFound, "http response")
+}
+
+func TestHTTPCancel(t *testing.T) {
+	wt.RunWithTimeout(t, 2*time.Second, func() {
+		impTestHTTPCancel(t)
+	})
+}
+
+func impTestHTTPCancel(t *testing.T) {
+	common.InitDefaultLogging(true)
+	var (
+		containerID = "deadbeef"
+		testCIDR1   = "10.0.3.5/29"
+	)
+
+	alloc := makeAllocatorWithMockGossip(t, "08:00:27:01:c3:9a", testCIDR1, 1)
+	defer alloc.Stop()
+	alloc.claimRingForTesting()
+	port := rand.Intn(10000) + 32768
+	fmt.Println("Http test on port", port)
+	go listenHTTP(port, alloc)
+
+	time.Sleep(100 * time.Millisecond) // Allow for http server to get going
+
+	// Stop the alloc so nothing actually works
+	unpause := alloc.pause()
+
+	// Ask the http server for a new address
+	done := make(chan *http.Response)
+	req, _ := http.NewRequest("POST", allocURL(port, containerID), nil)
+	go func() {
+		res, _ := http.DefaultClient.Do(req)
+		done <- res
+	}()
+
+	time.Sleep(100 * time.Millisecond)
+	fmt.Println("Cancelling allocate")
+	http.DefaultTransport.(*http.Transport).CancelRequest(req)
+	unpause()
+	res := <-done
+	if res != nil {
+		wt.Fatalf(t, "Error: Allocate returned non-nil")
+	}
+}

ipam/space/space.go

@@ -0,0 +1,248 @@
+package space
+
+import (
+	"bytes"
+	"fmt"
+	"sort"
+
+	"github.com/weaveworks/weave/common"
+	"github.com/weaveworks/weave/ipam/address"
+)
+
+type Addr address.Address
+
+type Space struct {
+	// ours and free represent a set of addresses as a sorted
+	// sequences of ranges.  Even elements give the inclusive
+	// starting points of ranges, and odd elements give the
+	// exclusive ending points.  Ranges in an array do not
+	// overlap, and neighbouring ranges are always coalesced if
+	// possible, so the arrays consist of sorted Addrs without
+	// repetition.
+	ours []address.Address
+	free []address.Address
+}
+
+func New() *Space {
+	return &Space{}
+}
+
+func (s *Space) Add(start address.Address, size address.Offset) {
+	s.free = add(s.free, start, address.Add(start, size))
+}
+
+// Clear removes all spaces from this space set.  Used during node shutdown.
+func (s *Space) Clear() {
+	s.free = s.free[:0]
+	s.ours = s.ours[:0]
+}
+
+func (s *Space) Allocate() (bool, address.Address) {
+	if len(s.free) == 0 {
+		return false, 0
+	} else {
+		res := s.free[0]
+		s.ours = add(s.ours, res, res+1)
+		s.free = subtract(s.free, res, res+1)
+		return true, res
+	}
+}
+
+func (s *Space) Claim(addr address.Address) error {
+	if !contains(s.free, addr) {
+		return fmt.Errorf("Address %v is not free to claim", addr)
+	}
+
+	s.ours = add(s.ours, addr, addr+1)
+	s.free = subtract(s.free, addr, addr+1)
+	return nil
+}
+
+func (s *Space) NumFreeAddresses() address.Offset {
+	res := address.Offset(0)
+	for i := 0; i < len(s.free); i += 2 {
+		res += address.Subtract(s.free[i+1], s.free[i])
+	}
+	return res
+}
+
+func (s *Space) NumFreeAddressesInRange(start, end address.Address) address.Offset {
+	res := address.Offset(0)
+	for i := 0; i < len(s.free); i += 2 {
+		s, e := s.free[i], s.free[i+1]
+		if s < start {
+			s = start
+		}
+		if e > end {
+			e = end
+		}
+		if s >= e {
+			continue
+		}
+		res += address.Subtract(e, s)
+	}
+	return res
+}
+
+func (s *Space) Free(addr address.Address) error {
+	if !contains(s.ours, addr) {
+		return fmt.Errorf("Address %v is not ours", addr)
+	}
+	if contains(s.free, addr) {
+		return fmt.Errorf("Address %v is already free", addr)
+	}
+
+	s.ours = subtract(s.ours, addr, addr+1)
+	s.free = add(s.free, addr, addr+1)
+	return nil
+}
+
+func (s *Space) biggestFreeRange() (int, address.Offset) {
+	pos := -1
+	biggest := address.Offset(0)
+
+	for i := 0; i < len(s.free); i += 2 {
+		size := address.Subtract(s.free[i+1], s.free[i])
+		if size >= biggest {
+			pos = i
+			biggest = size
+		}
+	}
+	return pos, biggest
+}
+
+func (s *Space) Donate() (address.Address, address.Offset, bool) {
+	if len(s.free) == 0 {
+		return 0, 0, false
+	}
+
+	pos, biggest := s.biggestFreeRange()
+
+	// Donate half of that biggest free range, rounding up so
+	// that the donation can't be empty
+	end := s.free[pos+1]
+	start := end - address.Address((biggest+1)/2)
+
+	s.ours = subtract(s.ours, start, end)
+	s.free = subtract(s.free, start, end)
+	return start, address.Subtract(end, start), true
+}
+
+func firstGreater(a []address.Address, x address.Address) int {
+	return sort.Search(len(a), func(i int) bool { return a[i] > x })
+}
+
+func firstGreaterOrEq(a []address.Address, x address.Address) int {
+	return sort.Search(len(a), func(i int) bool { return a[i] >= x })
+}
+
+// Do the ranges contain the given address?
+func contains(addrs []address.Address, addr address.Address) bool {
+	return firstGreater(addrs, addr)&1 != 0
+}
+
+// Take the union of the range [start, end) with the ranges in the array
+func add(addrs []address.Address, start address.Address, end address.Address) []address.Address {
+	return addSub(addrs, start, end, 0)
+}
+
+// Subtract the range [start, end) from the ranges in the array
+func subtract(addrs []address.Address, start address.Address, end address.Address) []address.Address {
+	return addSub(addrs, start, end, 1)
+}
+
+func addSub(addrs []address.Address, start address.Address, end address.Address, sense int) []address.Address {
+	start_pos := firstGreaterOrEq(addrs, start)
+	end_pos := firstGreater(addrs[start_pos:], end) + start_pos
+
+	// Boundaries up to start_pos are unaffected
+	res := make([]address.Address, start_pos, len(addrs)+2)
+	copy(res, addrs)
+
+	// Include start and end as new boundaries if they lie
+	// outside/inside existing ranges (according to sense).
+	if start_pos&1 == sense {
+		res = append(res, start)
+	}
+
+	if end_pos&1 == sense {
+		res = append(res, end)
+	}
+
+	// Boundaries after end_pos are unaffected
+	return append(res, addrs[end_pos:]...)
+}
+
+func (s *Space) String() string {
+	var buf bytes.Buffer
+	if len(s.ours) > 0 {
+		fmt.Fprint(&buf, "owned:")
+		for i := 0; i < len(s.ours); i += 2 {
+			fmt.Fprintf(&buf, " %s+%d ", s.ours[i], s.ours[i+1]-s.ours[i])
+		}
+	}
+	if len(s.free) > 0 {
+		fmt.Fprintf(&buf, "free:")
+		for i := 0; i < len(s.free); i += 2 {
+			fmt.Fprintf(&buf, " %s+%d ", s.free[i], s.free[i+1]-s.free[i])
+		}
+	}
+	if len(s.ours) == 0 && len(s.free) == 0 {
+		fmt.Fprintf(&buf, "No address ranges owned")
+	}
+	return buf.String()
+}
+
+type addressSlice []address.Address
+
+func (p addressSlice) Len() int           { return len(p) }
+func (p addressSlice) Less(i, j int) bool { return p[i] < p[j] }
+func (p addressSlice) Swap(i, j int)      { p[i], p[j] = p[j], p[i] }
+
+func (s *Space) assertInvariants() {
+	common.Assert(sort.IsSorted(addressSlice(s.ours)))
+	common.Assert(sort.IsSorted(addressSlice(s.free)))
+}
+
+// Return a slice representing everything we own, whether it is free or not
+func (s *Space) everything() []address.Address {
+	a := make([]address.Address, len(s.ours))
+	copy(a, s.ours)
+	for i := 0; i < len(s.free); i += 2 {
+		a = add(a, s.free[i], s.free[i+1])
+	}
+	return a
+}
+
+// OwnedRanges returns slice of Ranges, ordered by IP, gluing together
+// contiguous sequences of owned and free addresses
+func (s *Space) OwnedRanges() []address.Range {
+	everything := s.everything()
+	result := make([]address.Range, len(everything)/2)
+	for i := 0; i < len(everything); i += 2 {
+		result[i/2] = address.Range{Start: everything[i], End: everything[i+1]}
+	}
+	return result
+}
+
+// Create a Space that has free space in all the supplied Ranges.
+func (s *Space) AddRanges(ranges []address.Range) {
+	for _, r := range ranges {
+		s.free = add(s.free, r.Start, r.End)
+	}
+}
+
+// Taking ranges to be a set of all space we should own, add in any excess as free space
+func (s *Space) UpdateRanges(ranges []address.Range) {
+	new := []address.Address{}
+	for _, r := range ranges {
+		new = add(new, r.Start, r.End)
+	}
+	current := s.everything()
+	for i := 0; i < len(current); i += 2 {
+		new = subtract(new, current[i], current[i+1])
+	}
+	for i := 0; i < len(new); i += 2 {
+		s.free = add(s.free, new[i], new[i+1])
+	}
+}

ipam/space/space_test.go

@@ -0,0 +1,249 @@
+package space
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/weaveworks/weave/ipam/address"
+	wt "github.com/weaveworks/weave/testing"
+)
+
+func makeSpace(start address.Address, size address.Offset) *Space {
+	s := New()
+	s.Add(start, size)
+	return s
+}
+
+func ip(s string) address.Address {
+	addr, _ := address.ParseIP(s)
+	return addr
+}
+
+func TestLowlevel(t *testing.T) {
+	a := []address.Address{}
+	a = add(a, 100, 200)
+	wt.AssertEquals(t, a, []address.Address{100, 200})
+	wt.AssertTrue(t, !contains(a, 99), "")
+	wt.AssertTrue(t, contains(a, 100), "")
+	wt.AssertTrue(t, contains(a, 199), "")
+	wt.AssertTrue(t, !contains(a, 200), "")
+	a = add(a, 700, 800)
+	wt.AssertEquals(t, a, []address.Address{100, 200, 700, 800})
+	a = add(a, 300, 400)
+	wt.AssertEquals(t, a, []address.Address{100, 200, 300, 400, 700, 800})
+	a = add(a, 400, 500)
+	wt.AssertEquals(t, a, []address.Address{100, 200, 300, 500, 700, 800})
+	a = add(a, 600, 700)
+	wt.AssertEquals(t, a, []address.Address{100, 200, 300, 500, 600, 800})
+	a = add(a, 500, 600)
+	wt.AssertEquals(t, a, []address.Address{100, 200, 300, 800})
+	a = subtract(a, 500, 600)
+	wt.AssertEquals(t, a, []address.Address{100, 200, 300, 500, 600, 800})
+	a = subtract(a, 600, 700)
+	wt.AssertEquals(t, a, []address.Address{100, 200, 300, 500, 700, 800})
+	a = subtract(a, 400, 500)
+	wt.AssertEquals(t, a, []address.Address{100, 200, 300, 400, 700, 800})
+	a = subtract(a, 300, 400)
+	wt.AssertEquals(t, a, []address.Address{100, 200, 700, 800})
+	a = subtract(a, 700, 800)
+	wt.AssertEquals(t, a, []address.Address{100, 200})
+	a = subtract(a, 100, 200)
+	wt.AssertEquals(t, a, []address.Address{})
+
+	s := New()
+	wt.AssertEquals(t, s.NumFreeAddresses(), address.Offset(0))
+	ok, got := s.Allocate()
+	wt.AssertFalse(t, ok, "allocate in empty space should fail")
+
+	s.Add(100, 100)
+	wt.AssertEquals(t, s.NumFreeAddresses(), address.Offset(100))
+	ok, got = s.Allocate()
+	wt.AssertTrue(t, ok && got == 100, "allocate")
+	wt.AssertEquals(t, s.NumFreeAddresses(), address.Offset(99))
+	wt.AssertNoErr(t, s.Claim(150))
+	wt.AssertEquals(t, s.NumFreeAddresses(), address.Offset(98))
+	wt.AssertNoErr(t, s.Free(100))
+	wt.AssertEquals(t, s.NumFreeAddresses(), address.Offset(99))
+	wt.AssertErrorInterface(t, s.Free(0), (*error)(nil), "free not allocated")
+	wt.AssertErrorInterface(t, s.Free(100), (*error)(nil), "double free")
+
+	start, size, ok := s.Donate()
+	wt.AssertTrue(t, ok && start == 125 && size == 25, "donate")
+
+	// test Donate when addresses are scarce
+	s = New()
+	start, size, ok = s.Donate()
+	wt.AssertTrue(t, !ok, "donate on empty space should fail")
+	s.Add(0, 3)
+	wt.AssertNoErr(t, s.Claim(0))
+	wt.AssertNoErr(t, s.Claim(2))
+	start, size, ok = s.Donate()
+	wt.AssertTrue(t, ok && start == 1 && size == 1, "donate")
+	start, size, ok = s.Donate()
+	wt.AssertTrue(t, !ok, "donate should fail")
+}
+
+func TestSpaceAllocate(t *testing.T) {
+	const (
+		testAddr1   = "10.0.3.4"
+		testAddrx   = "10.0.3.19"
+		testAddry   = "10.0.9.19"
+		containerID = "deadbeef"
+	)
+
+	space1 := makeSpace(ip(testAddr1), 20)
+	wt.AssertEquals(t, space1.NumFreeAddresses(), address.Offset(20))
+	space1.assertInvariants()
+
+	_, addr1 := space1.Allocate()
+	wt.AssertEqualString(t, addr1.String(), testAddr1, "address")
+	wt.AssertEquals(t, space1.NumFreeAddresses(), address.Offset(19))
+	space1.assertInvariants()
+
+	_, addr2 := space1.Allocate()
+	wt.AssertNotEqualString(t, addr2.String(), testAddr1, "address")
+	wt.AssertEquals(t, space1.NumFreeAddresses(), address.Offset(18))
+	wt.AssertEquals(t, space1.NumFreeAddressesInRange(ip(testAddr1), ip(testAddrx)), address.Offset(13))
+	wt.AssertEquals(t, space1.NumFreeAddressesInRange(ip(testAddr1), ip(testAddry)), address.Offset(18))
+	space1.assertInvariants()
+
+	space1.Free(addr2)
+	space1.assertInvariants()
+
+	wt.AssertErrorInterface(t, space1.Free(addr2), (*error)(nil), "double free")
+	wt.AssertErrorInterface(t, space1.Free(ip(testAddrx)), (*error)(nil), "address not allocated")
+	wt.AssertErrorInterface(t, space1.Free(ip(testAddry)), (*error)(nil), "wrong out of range")
+
+	space1.assertInvariants()
+}
+
+func TestSpaceFree(t *testing.T) {
+	const (
+		testAddr1   = "10.0.3.4"
+		testAddrx   = "10.0.3.19"
+		testAddry   = "10.0.9.19"
+		containerID = "deadbeef"
+	)
+
+	space := makeSpace(ip(testAddr1), 20)
+
+	// Check we are prepared to give up the entire space
+	pos, size := space.biggestFreeRange()
+	wt.AssertTrue(t, space.free[pos] == ip(testAddr1) && size == 20, "Wrong space")
+
+	for i := 0; i < 20; i++ {
+		ok, _ := space.Allocate()
+		wt.AssertTrue(t, ok, "Failed to get address")
+	}
+
+	// Check we are full
+	ok, _ := space.Allocate()
+	wt.AssertTrue(t, !ok, "Should have failed to get address")
+	_, size, ok = space.Donate()
+	wt.AssertTrue(t, size == 0, "Wrong space")
+
+	// Free in the middle
+	wt.AssertSuccess(t, space.Free(ip("10.0.3.13")))
+	pos, size = space.biggestFreeRange()
+	wt.AssertTrue(t, space.free[pos] == ip("10.0.3.13") && size == 1, "Wrong space")
+
+	// Free one at the end
+	wt.AssertSuccess(t, space.Free(ip("10.0.3.23")))
+	pos, size = space.biggestFreeRange()
+	wt.AssertTrue(t, space.free[pos] == ip("10.0.3.23") && size == 1, "Wrong space")
+
+	// Now free a few at the end
+	wt.AssertSuccess(t, space.Free(ip("10.0.3.22")))
+	wt.AssertSuccess(t, space.Free(ip("10.0.3.21")))
+
+	wt.AssertEquals(t, space.NumFreeAddresses(), address.Offset(4))
+
+	// Now get the biggest free space; should be 3.21
+	pos, size = space.biggestFreeRange()
+	wt.AssertTrue(t, space.free[pos] == ip("10.0.3.21") && size == 3, "Wrong space")
+
+	// Now free a few in the middle
+	wt.AssertSuccess(t, space.Free(ip("10.0.3.12")))
+	wt.AssertSuccess(t, space.Free(ip("10.0.3.11")))
+	wt.AssertSuccess(t, space.Free(ip("10.0.3.10")))
+
+	wt.AssertEquals(t, space.NumFreeAddresses(), address.Offset(7))
+
+	// Now get the biggest free space; should be 3.21
+	pos, size = space.biggestFreeRange()
+	wt.AssertTrue(t, space.free[pos] == ip("10.0.3.10") && size == 4, "Wrong space")
+
+	wt.AssertEquals(t, space.OwnedRanges(), []address.Range{address.Range{Start: ip("10.0.3.4"), End: ip("10.0.3.24")}})
+}
+
+func (s1 *Space) Equal(s2 *Space) bool {
+	return reflect.DeepEqual(s1.ours, s2.ours) && reflect.DeepEqual(s1.free, s2.free)
+}
+
+func TestDonateSimple(t *testing.T) {
+	const (
+		testAddr1 = "10.0.1.0"
+		testAddr2 = "10.0.1.32"
+	)
+
+	var (
+		ipAddr1 = ip(testAddr1)
+	)
+
+	ps1 := makeSpace(ipAddr1, 48)
+
+	// Empty space set should split in two and give me the second half
+	start, numGivenUp, ok := ps1.Donate()
+	wt.AssertBool(t, ok, true, "Donate result")
+	wt.AssertEqualString(t, start.String(), "10.0.1.24", "Invalid start")
+	wt.AssertEquals(t, numGivenUp, address.Offset(24))
+	wt.AssertEquals(t, ps1.NumFreeAddresses(), address.Offset(24))
+
+	// Now check we can give the rest up.
+	count := 0 // count to avoid infinite loop
+	for ; count < 1000; count++ {
+		_, size, ok := ps1.Donate()
+		if !ok {
+			break
+		}
+		numGivenUp += size
+	}
+	wt.AssertEquals(t, ps1.NumFreeAddresses(), address.Offset(0))
+	wt.AssertEquals(t, numGivenUp, address.Offset(48))
+}
+
+func TestDonateHard(t *testing.T) {
+	//common.InitDefaultLogging(true)
+	var (
+		start                = ip("10.0.1.0")
+		size  address.Offset = 48
+	)
+
+	// Fill a fresh space
+	spaceset := makeSpace(start, size)
+	for i := address.Offset(0); i < size; i++ {
+		ok, _ := spaceset.Allocate()
+		wt.AssertTrue(t, ok, "Failed to get IP!")
+	}
+
+	wt.AssertEquals(t, spaceset.NumFreeAddresses(), address.Offset(0))
+
+	// Now free all but the last address
+	// this will force us to split the free list
+	for i := address.Offset(0); i < size-1; i++ {
+		wt.AssertSuccess(t, spaceset.Free(address.Add(start, i)))
+	}
+
+	// Now split
+	newRange, numGivenUp, ok := spaceset.Donate()
+	wt.AssertBool(t, ok, true, "GiveUpSpace result")
+	wt.AssertEquals(t, newRange, ip("10.0.1.23"))
+	wt.AssertEquals(t, numGivenUp, address.Offset(24))
+	wt.AssertEquals(t, spaceset.NumFreeAddresses(), address.Offset(23))
+
+	//Space set should now have 2 spaces
+	expected := New()
+	expected.Add(start, 23)
+	expected.ours = add(nil, ip("10.0.1.47"), ip("10.0.1.48"))
+	wt.AssertEquals(t, spaceset, expected)
+}

ipam/testutils_test.go

@@ -0,0 +1,286 @@
+package ipam
+
+import (
+	"fmt"
+	"math/rand"
+	"testing"
+	"time"
+
+	"github.com/weaveworks/weave/common"
+	"github.com/weaveworks/weave/router"
+	wt "github.com/weaveworks/weave/testing"
+)
+
+type mockMessage struct {
+	dst     router.PeerName
+	msgType byte
+	buf     []byte
+}
+
+func (m *mockMessage) String() string {
+	return fmt.Sprintf("-> %s [%x]", m.dst, m.buf)
+}
+
+func toStringArray(messages []mockMessage) []string {
+	out := make([]string, len(messages))
+	for i := range out {
+		out[i] = messages[i].String()
+	}
+	return out
+}
+
+type mockGossipComms struct {
+	t        *testing.T
+	name     string
+	messages []mockMessage
+}
+
+func (m *mockGossipComms) String() string {
+	return fmt.Sprintf("[mockGossipComms %s]", m.name)
+}
+
+// Note: this style of verification, using equalByteBuffer, requires
+// that the contents of messages are never re-ordered.  Which, for instance,
+// requires they are not based off iterating through a map.
+
+func (m *mockGossipComms) GossipBroadcast(update router.GossipData) error {
+	buf := []byte{}
+	if len(m.messages) == 0 {
+		m.Fatalf("%s: Gossip broadcast message unexpected: \n%x", m.name, buf)
+	} else if msg := m.messages[0]; msg.dst != router.UnknownPeerName {
+		m.Fatalf("%s: Expected Gossip message to %s but got broadcast", m.name, msg.dst)
+	} else if msg.buf != nil && !equalByteBuffer(msg.buf, buf) {
+		m.Fatalf("%s: Gossip message not sent as expected: \nwant: %x\ngot : %x", m.name, msg.buf, buf)
+	} else {
+		// Swallow this message
+		m.messages = m.messages[1:]
+	}
+	return nil
+}
+
+func equalByteBuffer(a, b []byte) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i := 0; i < len(a); i++ {
+		if a[i] != b[i] {
+			return false
+		}
+	}
+	return true
+}
+
+func (m *mockGossipComms) Fatalf(format string, args ...interface{}) {
+	// this sometimes hangs: wt.Fatalf(m.t, args...)
+	panic(fmt.Sprintf(format, args...))
+}
+
+func (m *mockGossipComms) GossipUnicast(dstPeerName router.PeerName, buf []byte) error {
+	if len(m.messages) == 0 {
+		m.Fatalf("%s: Gossip message to %s unexpected: \n%s", m.name, dstPeerName, buf)
+	} else if msg := m.messages[0]; msg.dst == router.UnknownPeerName {
+		m.Fatalf("%s: Expected Gossip broadcast message but got dest %s", m.name, dstPeerName)
+	} else if msg.dst != dstPeerName {
+		m.Fatalf("%s: Expected Gossip message to %s but got dest %s", m.name, msg.dst, dstPeerName)
+	} else if buf[0] != msg.msgType {
+		m.Fatalf("%s: Expected Gossip message of type %d but got type %d", m.name, msg.msgType, buf[0])
+	} else if msg.buf != nil && !equalByteBuffer(msg.buf, buf[1:]) {
+		m.Fatalf("%s: Gossip message not sent as expected: \nwant: %x\ngot : %x", m.name, msg.buf, buf[1:])
+	} else {
+		// Swallow this message
+		m.messages = m.messages[1:]
+	}
+	return nil
+}
+
+func ExpectMessage(alloc *Allocator, dst string, msgType byte, buf []byte) {
+	m := alloc.gossip.(*mockGossipComms)
+	dstPeerName, _ := router.PeerNameFromString(dst)
+	m.messages = append(m.messages, mockMessage{dstPeerName, msgType, buf})
+}
+
+func ExpectBroadcastMessage(alloc *Allocator, buf []byte) {
+	m := alloc.gossip.(*mockGossipComms)
+	m.messages = append(m.messages, mockMessage{router.UnknownPeerName, 0, buf})
+}
+
+func CheckAllExpectedMessagesSent(allocs ...*Allocator) {
+	for _, alloc := range allocs {
+		m := alloc.gossip.(*mockGossipComms)
+		if len(m.messages) > 0 {
+			wt.Fatalf(m.t, "%s: Gossip message(s) not sent as expected: \n%x", m.name, m.messages)
+		}
+	}
+}
+
+func makeAllocator(name string, cidr string, quorum uint) *Allocator {
+	peername, err := router.PeerNameFromString(name)
+	if err != nil {
+		panic(err)
+	}
+
+	alloc, err := NewAllocator(peername, router.PeerUID(rand.Int63()),
+		"nick-"+name, cidr, quorum)
+	if err != nil {
+		panic(err)
+	}
+
+	return alloc
+}
+
+func makeAllocatorWithMockGossip(t *testing.T, name string, universeCIDR string, quorum uint) *Allocator {
+	alloc := makeAllocator(name, universeCIDR, quorum)
+	gossip := &mockGossipComms{t: t, name: name}
+	alloc.SetInterfaces(gossip)
+	alloc.Start()
+	return alloc
+}
+
+func (alloc *Allocator) claimRingForTesting(allocs ...*Allocator) {
+	peers := []router.PeerName{alloc.ourName}
+	for _, alloc2 := range allocs {
+		peers = append(peers, alloc2.ourName)
+	}
+	alloc.ring.ClaimForPeers(normalizeConsensus(peers))
+	alloc.space.AddRanges(alloc.ring.OwnedRanges())
+}
+
+// Check whether or not something was sent on a channel
+func AssertSent(t *testing.T, ch <-chan bool) {
+	timeout := time.After(10 * time.Second)
+	select {
+	case <-ch:
+		// This case is ok
+	case <-timeout:
+		wt.Fatalf(t, "Nothing sent on channel")
+	}
+}
+
+func AssertNothingSent(t *testing.T, ch <-chan bool) {
+	select {
+	case val := <-ch:
+		wt.Fatalf(t, "Unexpected value on channel: %t", val)
+	default:
+		// no message received
+	}
+}
+
+func AssertNothingSentErr(t *testing.T, ch <-chan error) {
+	select {
+	case val := <-ch:
+		wt.Fatalf(t, "Unexpected value on channel: %t", val)
+	default:
+		// no message received
+	}
+}
+
+// Router to convey gossip from one gossiper to another, for testing
+type gossipMessage struct {
+	isUnicast bool
+	sender    *router.PeerName
+	buf       []byte
+	exitChan  chan bool
+}
+
+type TestGossipRouter struct {
+	gossipChans map[router.PeerName]chan gossipMessage
+	loss        float32 // 0.0 means no loss
+}
+
+func (grouter *TestGossipRouter) GossipBroadcast(update router.GossipData) error {
+	for _, gossipChan := range grouter.gossipChans {
+		select {
+		case gossipChan <- gossipMessage{buf: update.(*ipamGossipData).alloc.encode()}:
+		default: // drop the message if we cannot send it
+		}
+	}
+	return nil
+}
+
+func (grouter *TestGossipRouter) removePeer(peer router.PeerName) {
+	gossipChan := grouter.gossipChans[peer]
+	resultChan := make(chan bool)
+	gossipChan <- gossipMessage{exitChan: resultChan}
+	<-resultChan
+	delete(grouter.gossipChans, peer)
+}
+
+type TestGossipRouterClient struct {
+	router *TestGossipRouter
+	sender router.PeerName
+}
+
+func (grouter *TestGossipRouter) connect(sender router.PeerName, gossiper router.Gossiper) router.Gossip {
+	gossipChan := make(chan gossipMessage, 100)
+
+	go func() {
+		gossipTimer := time.Tick(10 * time.Second)
+		for {
+			select {
+			case message := <-gossipChan:
+				if message.exitChan != nil {
+					message.exitChan <- true
+					return
+				}
+
+				if rand.Float32() > (1.0 - grouter.loss) {
+					continue
+				}
+
+				if message.isUnicast {
+					err := gossiper.OnGossipUnicast(*message.sender, message.buf)
+					if err != nil {
+						panic(fmt.Sprintf("Error doing gossip unicast to %s: %s", sender, err))
+					}
+				} else {
+					_, err := gossiper.OnGossipBroadcast(message.buf)
+					if err != nil {
+						panic(fmt.Sprintf("Error doing gossip broadcast to %s: %s", sender, err))
+					}
+				}
+			case <-gossipTimer:
+				grouter.GossipBroadcast(gossiper.Gossip())
+			}
+		}
+	}()
+
+	grouter.gossipChans[sender] = gossipChan
+	return TestGossipRouterClient{grouter, sender}
+}
+
+func (client TestGossipRouterClient) GossipUnicast(dstPeerName router.PeerName, buf []byte) error {
+	select {
+	case client.router.gossipChans[dstPeerName] <- gossipMessage{isUnicast: true, sender: &client.sender, buf: buf}:
+	default: // drop the message if we cannot send it
+		common.Error.Printf("Dropping message")
+	}
+	return nil
+}
+
+func (client TestGossipRouterClient) GossipBroadcast(update router.GossipData) error {
+	return client.router.GossipBroadcast(update)
+}
+
+func makeNetworkOfAllocators(size int, cidr string) ([]*Allocator, TestGossipRouter) {
+
+	gossipRouter := TestGossipRouter{make(map[router.PeerName]chan gossipMessage), 0.0}
+	allocs := make([]*Allocator, size)
+
+	for i := 0; i < size; i++ {
+		alloc := makeAllocator(fmt.Sprintf("%02d:00:00:02:00:00", i),
+			cidr, uint(size/2+1))
+		alloc.SetInterfaces(gossipRouter.connect(alloc.ourName, alloc))
+		alloc.Start()
+		allocs[i] = alloc
+	}
+
+	gossipRouter.GossipBroadcast(allocs[size-1].Gossip())
+	time.Sleep(1000 * time.Millisecond)
+	return allocs, gossipRouter
+}
+
+func stopNetworkOfAllocators(allocs []*Allocator) {
+	for _, alloc := range allocs {
+		alloc.Stop()
+	}
+}

ipam/utils/utils.go

@@ -1,49 +0,0 @@
-package utils
-
-import (
-	"net"
-)
-
-// We shouldn't ever get any errors on *encoding*, but if we do, this will make sure we get to hear about them.
-func panicOnError(err error) {
-	if err != nil {
-		panic(err)
-	}
-}
-
-// IP4int converts an ipv4 address to a uint32
-func IP4int(ip4 net.IP) (r uint32) {
-	for _, b := range ip4.To4() {
-		r <<= 8
-		r |= uint32(b)
-	}
-	return
-}
-
-// IntIP4 converts a uint32 to an ipv4 address
-func IntIP4(key uint32) (r net.IP) {
-	r = make([]byte, net.IPv4len)
-	for i := 3; i >= 0; i-- {
-		r[i] = byte(key)
-		key >>= 8
-	}
-	return
-}
-
-// Add - convert to 32-bit unsigned integer, add, and convert back
-func Add(addr net.IP, i uint32) net.IP {
-	sum := IP4int(addr) + i
-	return IntIP4(sum)
-}
-
-// Subtract - convert to 32-bit unsigned integer, subtract, and convert back
-func Subtract(a, b net.IP) int64 {
-	return int64(IP4int(a)) - int64(IP4int(b))
-}
-
-// Assert test is true, panic otherwise
-func Assert(test bool, message string) {
-	if !test {
-		panic(message)
-	}
-}

router/protocol.go

@@ -2,7 +2,7 @@ package router
 
 const (
 	Protocol        = "weave"
-	ProtocolVersion = 16
+	ProtocolVersion = 102
 )
 
 type ProtocolTag byte

site/features.md

@@ -9,6 +9,7 @@ Weave has a few more features beyond those illustrated by the [basic
 example](https://github.com/weaveworks/weave#example):
 
  * [Virtual ethernet switch](#virtual-ethernet-switch)
+ * [IP Address Management](#ipam)
  * [Application isolation](#application-isolation)
  * [Dynamic network attachment](#dynamic-network-attachment)
  * [Security](#security)
@@ -53,6 +54,53 @@ now re-use the same tools and techniques when deploying applications
 as containers as we would have done when deploying them 'on metal' in
 our data centre.
 
+### <a name="ipam"></a>IP Address Management
+
+Weave can assign unique IP addresses to each container across the
+network.  To make this work, weave must be told what range of
+addresses to allocate from, for example:
+
+    host1# weave launch -iprange 10.2.3.0/24
+
+The `run`, `start`, `attach`, `expose` and `hide` commands will then fetch
+an address automatically if none is specified, i.e.:
+
+    host1# D=$(weave run -t -i ubuntu)
+
+Weave will automatically learn when a container has exited
+and hence can release its IP address.
+
+Weave shares the IP range across all peers, dynamically according to
+their needs.  If a group of peers becomes isolated from the rest (a
+partition), they can continue to work with the IP ranges they had
+before isolation, and can be re-connected to the rest of the network
+and carry on. Note that you must specify the same range with
+`-iprange` on each host, and you cannot mix weaves started with and
+without -iprange.
+
+Just once, when you start up the whole network, weave needs a majority
+of peers to agree in order to avoid isolated groups starting off
+inconsistently. Therefore, you must either supply the list of all
+peers in the network to `weave launch` or add the `-initpeercount`
+flag to specify how many peers there will be.  It isn't a problem to
+over-estimate by a bit, but if you supply a number that is too small
+then multiple independent groups may form.
+
+You may wish to `weave stop` and re-launch to change some config or to
+upgrade to a new version; provided the underlying protocol hasn't
+changed it will pick up where it left off and learn from peers in the
+network which address ranges it was previously using. If, however, you
+run `weave reset` this will remove the peer from the network so
+if Weave is run again on that node it will start from scratch.
+
+For failed peers, the `weave rmpeer` command can be used to
+permanently remove the ranges allocated to said peer.  This will allow
+other peers to allocate IPs in the ranges previously owner by the rm'd
+peer, and as such should be used with extreme caution - if the rm'd
+peer had transferred some range of IP addresses to another peer but
+this is not known to the whole network, or if it later rejoins
+the Weave network, the same IP address may be allocated twice.
+
 ### <a name="application-isolation"></a>Application isolation
 
 A single weave network can host multiple, isolated applications, with

test/120_ipam_test.sh

@@ -0,0 +1,18 @@
+#! /bin/bash
+
+. ./config.sh
+
+UNIVERSE=10.2.1.0/24
+
+start_suite "Ping over cross-host weave network with IPAM"
+
+weave_on $HOST1 launch -iprange $UNIVERSE
+weave_on $HOST2 launch -iprange $UNIVERSE $HOST1
+
+weave_on $HOST1 run -t --name=c1 gliderlabs/alpine /bin/sh
+weave_on $HOST2 run -t --name=c2 gliderlabs/alpine /bin/sh
+# Note can't use weave_on here because it echoes the command
+C2IP=$(DOCKER_HOST=tcp://$HOST2:2375 $WEAVE ps | grep -o -E '[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}')
+assert_raises "exec_on $HOST1 c1 ping -q -c 4 $C2IP"
+
+end_suite

test/140_ipam_expose_test.sh

@@ -0,0 +1,63 @@
+#! /bin/bash
+
+. ./config.sh
+
+R1=10.2.1.0/24
+C3=10.2.2.34
+C4=10.2.2.37
+
+PING="ping -nq -W 1 -c 1"
+
+weave_on1() {
+    assert_raises "weave_on $HOST1 $@"
+}
+
+run_on1() {
+    assert_raises "run_on   $HOST1 $@"
+}
+
+exec_on1() {
+    assert_raises "exec_on  $HOST1 $@"
+}
+
+check_container_connectivity() {
+    exec_on1 "c2 $PING $C1"
+    exec_on1 "c3 $PING $C4"
+    # fails due to #620
+    # exec_on1 "c3 ! $PING $C1"
+}
+
+start_suite "exposing weave network to host"
+
+weave_on $HOST1 launch -iprange $R1
+
+weave_on $HOST1 run -t --name=c1 gliderlabs/alpine /bin/sh
+weave_on $HOST1 run -t --name=c2 gliderlabs/alpine /bin/sh
+weave_on $HOST1 run $C3/24 -t --name=c3 gliderlabs/alpine /bin/sh
+weave_on $HOST1 run $C4/24 -t --name=c4 gliderlabs/alpine /bin/sh
+
+# Note can't use weave_on here because it echoes the command
+C1=$(DOCKER_HOST=tcp://$HOST1:2375 $WEAVE ps c1 | grep -o -E '[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}')
+
+# absence of host connectivity by default
+run_on1 "! $PING $C1"
+check_container_connectivity
+
+# host connectivity after 'expose'
+weave_on1 "expose"
+run_on1   "  $PING $C1"
+run_on1   "! $PING $C3"
+check_container_connectivity
+
+# idempotence of 'expose'
+weave_on1 "expose"
+run_on1   "  $PING $C1"
+
+# no host connectivity after 'hide'
+weave_on1 "hide"
+run_on1   "! $PING $C1"
+
+# idempotence of 'hide'
+weave_on1 "hide"
+
+end_suite

test/250_dns_ipam_test.sh

@@ -0,0 +1,27 @@
+#! /bin/bash
+
+. ./config.sh
+
+UNIVERSE=10.2.3.0/24
+
+start_suite "Resolve names over cross-host weave network with IPAM"
+
+weave_on $HOST1 launch -iprange $UNIVERSE
+weave_on $HOST2 launch -iprange $UNIVERSE $HOST1
+
+weave_on $HOST1 launch-dns 10.2.254.1/24
+weave_on $HOST2 launch-dns 10.2.254.2/24
+
+weave_on $HOST2 run -t --name=c2 -h seetwo.weave.local gliderlabs/alpine /bin/sh
+weave_on $HOST1 run --with-dns --name=c1 -t aanand/docker-dnsutils /bin/sh
+
+# Note can't use weave_on here because it echoes the command
+C2IP=$(DOCKER_HOST=tcp://$HOST2:2375 $WEAVE ps c2 | grep -o -E '[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}')
+
+ok=$(docker -H tcp://$HOST1:2375 exec -i c1 dig +short seetwo.weave.local)
+assert "echo $ok" "$C2IP"
+
+ok=$(docker -H tcp://$HOST1:2375 exec -i c1 dig +short -x $C2IP)
+assert "echo $ok" "seetwo.weave.local."
+
+end_suite

weave

@@ -25,27 +25,29 @@ CIDR_REGEXP="$IP_REGEXP/[0-9]{1,2}"
 usage() {
     echo "Usage:"
     echo "weave setup"
-    echo "weave launch     [-password <password>] [-nickname <nickname>] <peer> ..."
+    echo "weave launch     [-password <password>] [-nickname <nickname>] [-iprange <cidr>] <peer> ..."
     echo "weave launch-dns <cidr>"
     echo "weave connect    <peer>"
     echo "weave forget     <peer>"
-    echo "weave run        [--with-dns] <cidr> [<cidr> ...] <docker run args> ..."
-    echo "weave start      <cidr> [<cidr> ...] <container_id>"
-    echo "weave attach     <cidr> [<cidr> ...] <container_id>"
+    echo "weave run        [--with-dns] [<cidr> ...] <docker run args> ..."
+    echo "weave start      [<cidr> ...] <container_id>"
+    echo "weave attach     [<cidr> ...] <container_id>"
     echo "weave detach     <cidr> [<cidr> ...] <container_id>"
     echo "weave dns-add    <ip_address> [<ip_address> ...] <container_id> [-h <fqdn>]"
     echo "weave dns-remove <ip_address> [<ip_address> ...] <container_id>"
-    echo "weave expose     <cidr> [<cidr> ...] [-h <fqdn>]"
-    echo "weave hide       <cidr> [<cidr> ...]"
+    echo "weave expose     [<cidr> ...] [-h <fqdn>]"
+    echo "weave hide       [<cidr> ...]"
     echo "weave ps         [<container_id> ...]"
     echo "weave status"
     echo "weave version"
     echo "weave stop"
     echo "weave stop-dns"
     echo "weave reset"
+    echo "weave rmpeer     <peerid>"
     echo
     echo "where <peer> is of the form <ip_address_or_fqdn>[:<port>], and"
     echo "      <cidr> is of the form <ip_address>/<routing_prefix_length>"
+    echo "      <peerid> is a nickname or weave internal peer ID"
     exit 1
 }
 
@@ -98,9 +100,6 @@ collect_cidr_args() {
         CIDR_COUNT=$((CIDR_COUNT + 1))
         shift 1
     done
-    if [ $CIDR_COUNT -eq 0 ]; then
-        [ $# -gt 0 ] && validate_cidr "$1" || usage
-    fi
 }
 
 is_ip() {
@@ -532,6 +531,28 @@ wait_for_status() {
     return 1
 }
 
+# Call $1 for all containers, passing container ID, all MACs and all IPs
+with_container_addresses() {
+    COMMAND=$1
+    shift 1
+    for CONTAINER_ID in "$@" ; do
+        if [ "$CONTAINER_ID" = "weave:expose" ]; then
+            ADDRS_CMD="ip addr show dev $BRIDGE"
+        else
+            ADDRS_CMD="with_container_netns $CONTAINER_ID container_weave_addrs"
+        fi
+        if CONTAINER_ADDRS=$($ADDRS_CMD 2>&1) ; then
+            CONTAINER_MAC=$(echo "$CONTAINER_ADDRS" | grep -o 'link/ether .*' | cut -d ' ' -f 2)
+            CONTAINER_IPS=$(echo "$CONTAINER_ADDRS" | grep -o 'inet .*'       | cut -d ' ' -f 2)
+            $COMMAND "$CONTAINER_ID" "$CONTAINER_MAC" "$CONTAINER_IPS"
+        fi
+    done
+}
+
+echo_addresses() {
+    echo $1 $2 $3
+}
+
 ######################################################################
 # weaveDNS helpers
 ######################################################################
@@ -614,6 +635,53 @@ populate_dns() {
     done
 }
 
+######################################################################
+# IP Allocation Management helpers
+######################################################################
+
+populate_1_ipam() {
+    for CIDR in $3; do
+        # NB: CONTAINER_IP is the IP of the weave container; it is set by wait_for_status.
+        http_call_ip $CONTAINER_IP $HTTP_PORT PUT /ip/$1/${CIDR%/*}
+    done
+}
+
+# Tell the newly-started weave IP allocator about existing weave IPs
+populate_ipam() {
+    wait_for_status $CONTAINER_NAME $HTTP_PORT
+    with_container_addresses populate_1_ipam weave:expose $(docker ps -q)
+}
+
+# If we don't already have any CIDRs passed from the command-line, get one from IPAM
+# $1 is the name of the container and $2, if set, is the ID to use inside IPAM
+get_ipam_cidr() {
+    if [ -z $CIDR_ARGS ] ; then
+        if [ -z $2 ] ; then
+            # get the long form of the container ID
+            if ! status=$(docker inspect --format='{{.State.Running}} {{.Id}}' $1 2>/dev/null); then
+                echo "Error: No such container: $1" >&2
+                return 1
+            fi
+            case "$status" in
+                true*)
+                    CONTAINER="${status#true }"
+                    ;;
+                *)
+                    echo "$1 container is not running." >&2
+                    return 1
+                    ;;
+            esac
+        else
+            CONTAINER="$2"
+        fi
+        CIDR_ARGS="$(http_call $CONTAINER_NAME $HTTP_PORT POST /ip/$CONTAINER)"
+        if [ "$CIDR_ARGS" = "404 page not found" ] ; then
+            echo "No IP address supplied (use the -iprange option on 'weave launch'\nto enable IP allocation)" >&2
+            return 1
+        fi
+    fi
+}
+
 ##########################################################################################
 
 [ $(id -u) = 0 ] || {
@@ -697,6 +765,12 @@ case "$COMMAND" in
                     CONTAINER_PORT="$2"
                     shift 2
                     ;;
+                -iprange)
+                    [ $# -gt 1 ] || usage
+                    validate_cidr $2
+                    IPRANGE="-iprange $2"
+                    shift 2
+                    ;;
                 *)
                     break
                     ;;
@@ -706,10 +780,10 @@ case "$COMMAND" in
         # additional parameters, such as resource limits, to docker
         # when launching the weave container.
         CONTAINER=$(docker run --privileged -d --name=$CONTAINER_NAME \
-            -p $PORT:$CONTAINER_PORT/tcp -p $PORT:$CONTAINER_PORT/udp -e WEAVE_PASSWORD \
-            $WEAVE_DOCKER_ARGS $IMAGE -iface $CONTAINER_IFNAME -port $CONTAINER_PORT -name "$PEERNAME" -nickname "$(hostname)" "$@")
+            -p $PORT:$CONTAINER_PORT/tcp -p $PORT:$CONTAINER_PORT/udp -e WEAVE_PASSWORD -v /var/run/docker.sock:/var/run/docker.sock \
+            $WEAVE_DOCKER_ARGS $IMAGE -iface $CONTAINER_IFNAME -port $CONTAINER_PORT -name "$PEERNAME" -nickname "$(hostname)" $IPRANGE "$@")
         with_container_netns $CONTAINER launch >/dev/null
-        wait_for_status $CONTAINER_NAME $HTTP_PORT
+        [ -z "$IPRANGE" ] && wait_for_status $CONTAINER_NAME $HTTP_PORT || populate_ipam
         echo $CONTAINER
         ;;
     launch-dns)
@@ -745,18 +819,7 @@ case "$COMMAND" in
         ;;
     ps)
         [ $# -eq 0 ] && CONTAINERS="weave:expose $(docker ps -q)" || CONTAINERS="$@"
-        for CONTAINER_ID in $CONTAINERS ; do
-            if [ "$CONTAINER_ID" = "weave:expose" ]; then
-                ADDRS_CMD="ip addr show dev $BRIDGE"
-            else
-                ADDRS_CMD="with_container_netns $CONTAINER_ID container_weave_addrs"
-            fi
-            if CONTAINER_ADDRS=$($ADDRS_CMD 2>&1) ; then
-                CONTAINER_MAC=$(echo "$CONTAINER_ADDRS" | grep -o 'link/ether .*' | cut -d ' ' -f 2)
-                CONTAINER_IPS=$(echo "$CONTAINER_ADDRS" | grep -o 'inet .*'       | cut -d ' ' -f 2)
-                echo $CONTAINER_ID $CONTAINER_MAC $CONTAINER_IPS
-            fi
-        done
+        with_container_addresses echo_addresses $CONTAINERS
         ;;
     version)
         [ $# -eq 0 ] || usage
@@ -799,6 +862,7 @@ case "$COMMAND" in
         CONTAINER_ID="$1"
         create_bridge
         CONTAINER=$(docker start $CONTAINER_ID)
+        get_ipam_cidr $CONTAINER_ID $CONTAINER
         with_container_netns $CONTAINER_ID attach $CIDR_ARGS >/dev/null
         tell_dns PUT $CONTAINER_ID $CIDR_ARGS
         echo $CONTAINER
@@ -809,6 +873,7 @@ case "$COMMAND" in
         [ $# -eq 1 ] || usage
         CONTAINER_ID="$1"
         create_bridge
+        get_ipam_cidr $CONTAINER_ID
         with_container_netns $CONTAINER_ID attach $CIDR_ARGS >/dev/null
         tell_dns PUT $CONTAINER_ID $CIDR_ARGS
         ;;
@@ -843,6 +908,7 @@ case "$COMMAND" in
     expose)
         collect_cidr_args "$@"
         shift $CIDR_COUNT
+        [ $CIDR_COUNT -gt 0 ] || get_ipam_cidr weave weave:expose
         if [ $# -eq 0 ]; then
             FQDN=""
         else
@@ -866,7 +932,7 @@ case "$COMMAND" in
     hide)
         collect_cidr_args "$@"
         shift $CIDR_COUNT
-        [ $# -eq 0 ] || usage
+        get_ipam_cidr $CONTAINER_NAME weave:expose
         create_bridge --without-ethtool
         for CIDR in $CIDR_ARGS; do
             if ip addr show dev $BRIDGE | grep -qF $CIDR
@@ -877,6 +943,7 @@ case "$COMMAND" in
                 http_call $DNS_CONTAINER_NAME $DNS_HTTP_PORT DELETE /name/weave:expose/${CIDR%/*} 2>/dev/null || true
             fi
         done
+        http_call $CONTAINER_NAME $HTTP_PORT DELETE /ip/weave:expose > /dev/null && true
         ;;
     stop)
         [ $# -eq 0 ] || usage
@@ -895,6 +962,7 @@ case "$COMMAND" in
         ;;
     reset)
         [ $# -eq 0 ] || usage
+        http_call $CONTAINER_NAME $HTTP_PORT DELETE /peer >/dev/null 2>&1 || true
         docker stop  $CONTAINER_NAME     >/dev/null 2>&1 || true
         docker stop  $DNS_CONTAINER_NAME >/dev/null 2>&1 || true
         docker rm -f $CONTAINER_NAME     >/dev/null 2>&1 || true
@@ -905,6 +973,11 @@ case "$COMMAND" in
             ip link del $LOCAL_IFNAME
         done
         ;;
+    rmpeer)
+        [ $# -eq 1 ] || usage
+        PEER=$1
+        http_call $CONTAINER_NAME $HTTP_PORT DELETE /peer/$PEER
+        ;;
     *)
         echo "Unknown weave command '$COMMAND'" >&2
         usage

weaver/main.go

@@ -8,6 +8,8 @@ import (
 	"github.com/davecheney/profile"
 	"github.com/gorilla/mux"
 	. "github.com/weaveworks/weave/common"
+	"github.com/weaveworks/weave/common/updater"
+	"github.com/weaveworks/weave/ipam"
 	weavenet "github.com/weaveworks/weave/net"
 	weave "github.com/weaveworks/weave/router"
 	"log"
@@ -46,6 +48,9 @@ func main() {
 		peers       []string
 		bufSzMB     int
 		httpAddr    string
+		iprangeCIDR string
+		peerCount   int
+		apiPath     string
 	)
 
 	flag.BoolVar(&justVersion, "version", false, "print version and exit")
@@ -60,6 +65,9 @@ func main() {
 	flag.IntVar(&config.ConnLimit, "connlimit", 30, "connection limit (0 for unlimited)")
 	flag.IntVar(&bufSzMB, "bufsz", 8, "capture buffer size in MB")
 	flag.StringVar(&httpAddr, "httpaddr", fmt.Sprintf(":%d", weave.HTTPPort), "address to bind HTTP interface to (disabled if blank, absolute path indicates unix domain socket)")
+	flag.StringVar(&iprangeCIDR, "iprange", "", "IP address range to allocate within, in CIDR notation")
+	flag.IntVar(&peerCount, "initpeercount", 0, "number of peers in network (for IP address allocation)")
+	flag.StringVar(&apiPath, "api", "unix:///var/run/docker.sock", "Path to Docker API socket")
 	flag.Parse()
 	peers = flag.Args()
 
@@ -121,12 +129,23 @@ func main() {
 
 	router := weave.NewRouter(config, name, nickName)
 	log.Println("Our name is", router.Ourself)
-	router.Start()
-	initiateConnections(router, peers)
+	var allocator *ipam.Allocator
 	if httpAddr != "" {
-		go handleHTTP(router, httpAddr)
+		if iprangeCIDR != "" {
+			allocator = createAllocator(router, apiPath, iprangeCIDR, determineQuorum(peerCount, peers))
+		} else {
+			if peerCount > 0 {
+				log.Fatal("-initpeercount flag specified without -iprange")
+			}
+		}
+	}
+	if httpAddr == "" || iprangeCIDR == "" {
+		router.NewGossip("IPallocation", &ipam.DummyAllocator{})
 	}
 
+	router.Start()
+	initiateConnections(router, peers)
+	go handleHTTP(router, httpAddr, allocator)
 	SignalHandlerLoop(router)
 }
 
@@ -164,7 +183,41 @@ func initiateConnections(router *weave.Router, peers []string) {
 	}
 }
 
-func handleHTTP(router *weave.Router, httpAddr string) {
+func createAllocator(router *weave.Router, apiPath string, iprangeCIDR string, quorum uint) *ipam.Allocator {
+	allocator, err := ipam.NewAllocator(router.Ourself.Peer.Name, router.Ourself.Peer.UID, router.Ourself.Peer.NickName, iprangeCIDR, quorum)
+	if err != nil {
+		log.Fatal(err)
+	}
+	allocator.SetInterfaces(router.NewGossip("IPallocation", allocator))
+	allocator.Start()
+	err = updater.Start(apiPath, allocator)
+	if err != nil {
+		log.Fatal("Unable to start watcher", err)
+	}
+	return allocator
+}
+
+// Pick a quorum size heuristically based on the number of peer
+// addresses passed.
+func determineQuorum(initPeerCountFlag int, peers []string) uint {
+	if initPeerCountFlag > 0 {
+		return uint(initPeerCountFlag/2 + 1)
+	}
+
+	// Guess a suitable quorum size based on the list of peer
+	// addresses.  The peer list might or might not contain an
+	// address for this peer, so the conservative assumption is
+	// that it doesn't.  The list might contain multiple addresses
+	// that resolve to the same peer, in which case the quorum
+	// might be larger than it needs to be.  But the user can
+	// specify it explicitly if that becomes a problem.
+	clusterSize := uint(len(peers) + 1)
+	quorum := clusterSize/2 + 1
+	log.Println("Assuming quorum size of", quorum)
+	return quorum
+}
+
+func handleHTTP(router *weave.Router, httpAddr string, allocator *ipam.Allocator) {
 	encryption := "off"
 	if router.UsingPassword() {
 		encryption = "on"
@@ -172,10 +225,17 @@ func handleHTTP(router *weave.Router, httpAddr string) {
 
 	muxRouter := mux.NewRouter()
 
+	if allocator != nil {
+		allocator.HandleHTTP(muxRouter)
+	}
+
 	muxRouter.Methods("GET").Path("/status").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		fmt.Fprintln(w, "weave router", version)
 		fmt.Fprintln(w, "Encryption", encryption)
 		fmt.Fprintln(w, router.Status())
+		if allocator != nil {
+			fmt.Fprintln(w, allocator.String())
+		}
 	})
 
 	muxRouter.Methods("GET").Path("/status-json").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {