diff --git a/Makefile b/Makefile
index 801afd0c..ed060073 100644
--- a/Makefile
+++ b/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
diff --git a/common/errors.go b/common/errors.go
index 23072055..a0a9cc96 100644
--- a/common/errors.go
+++ b/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")
+ }
+}
diff --git a/docs/ipam.md b/docs/ipam.md
new file mode 100644
index 00000000..706ad842
--- /dev/null
+++ b/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.
+
+
+
+## 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.
+
+
+
+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.
diff --git a/ipam/address/address.go b/ipam/address/address.go
new file mode 100644
index 00000000..b9f47ac3
--- /dev/null
+++ b/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)
+}
diff --git a/ipam/allocate.go b/ipam/allocate.go
new file mode 100644
index 00000000..320e8e02
--- /dev/null
+++ 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
+}
diff --git a/ipam/allocator.go b/ipam/allocator.go
new file mode 100644
index 00000000..28f7f6c7
--- /dev/null
+++ b/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...)...)
+}
diff --git a/ipam/allocator_test.go b/ipam/allocator_test.go
new file mode 100644
index 00000000..b3c550c8
--- /dev/null
+++ b/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()
+ }
+}
diff --git a/ipam/claim.go b/ipam/claim.go
new file mode 100644
index 00000000..ec7bc2bf
--- /dev/null
+++ b/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
+}
diff --git a/ipam/dummy.go b/ipam/dummy.go
new file mode 100644
index 00000000..d9761c4d
--- /dev/null
+++ b/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
+}
diff --git a/ipam/http.go b/ipam/http.go
new file mode 100644
index 00000000..39efa892
--- /dev/null
+++ b/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)
+ })
+}
diff --git a/ipam/http_test.go b/ipam/http_test.go
new file mode 100644
index 00000000..710bb4fb
--- /dev/null
+++ b/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")
+ }
+}
diff --git a/ipam/space/space.go b/ipam/space/space.go
new file mode 100644
index 00000000..2848a033
--- /dev/null
+++ b/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])
+ }
+}
diff --git a/ipam/space/space_test.go b/ipam/space/space_test.go
new file mode 100644
index 00000000..fd13c42c
--- /dev/null
+++ b/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)
+}
diff --git a/ipam/testutils_test.go b/ipam/testutils_test.go
new file mode 100644
index 00000000..ee72d8af
--- /dev/null
+++ b/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()
+ }
+}
diff --git a/ipam/utils/utils.go b/ipam/utils/utils.go
deleted file mode 100644
index 18957155..00000000
--- a/ipam/utils/utils.go
+++ /dev/null
@@ -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)
- }
-}
diff --git a/router/protocol.go b/router/protocol.go
index 441c208c..41f0f70c 100644
--- a/router/protocol.go
+++ b/router/protocol.go
@@ -2,7 +2,7 @@ package router
const (
Protocol = "weave"
- ProtocolVersion = 16
+ ProtocolVersion = 102
)
type ProtocolTag byte
diff --git a/site/features.md b/site/features.md
index 11ec05bb..3b75a80d 100644
--- a/site/features.md
+++ b/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.
+### 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.
+
### Application isolation
A single weave network can host multiple, isolated applications, with
diff --git a/test/120_ipam_test.sh b/test/120_ipam_test.sh
new file mode 100755
index 00000000..f7cb15d1
--- /dev/null
+++ b/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
diff --git a/test/140_ipam_expose_test.sh b/test/140_ipam_expose_test.sh
new file mode 100755
index 00000000..01f74df5
--- /dev/null
+++ b/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
diff --git a/test/250_dns_ipam_test.sh b/test/250_dns_ipam_test.sh
new file mode 100755
index 00000000..07bc302c
--- /dev/null
+++ b/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
diff --git a/weave b/weave
index b85d6200..055ef756 100755
--- a/weave
+++ b/weave
@@ -25,27 +25,29 @@ CIDR_REGEXP="$IP_REGEXP/[0-9]{1,2}"
usage() {
echo "Usage:"
echo "weave setup"
- echo "weave launch [-password ] [-nickname ] ..."
+ echo "weave launch [-password ] [-nickname ] [-iprange ] ..."
echo "weave launch-dns "
echo "weave connect "
echo "weave forget "
- echo "weave run [--with-dns] [ ...] ..."
- echo "weave start [ ...] "
- echo "weave attach [ ...] "
+ echo "weave run [--with-dns] [ ...] ..."
+ echo "weave start [ ...] "
+ echo "weave attach [ ...] "
echo "weave detach [ ...] "
echo "weave dns-add [ ...] [-h ]"
echo "weave dns-remove [ ...] "
- echo "weave expose [ ...] [-h ]"
- echo "weave hide [ ...]"
+ echo "weave expose [ ...] [-h ]"
+ echo "weave hide [ ...]"
echo "weave ps [ ...]"
echo "weave status"
echo "weave version"
echo "weave stop"
echo "weave stop-dns"
echo "weave reset"
+ echo "weave rmpeer "
echo
echo "where is of the form [:], and"
echo " is of the form /"
+ echo " 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
diff --git a/weaver/main.go b/weaver/main.go
index 4ee8ef26..85e5e8b8 100644
--- a/weaver/main.go
+++ b/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) {