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