Initial implementation of multi-subnet IPAM

ipam/allocate.go

@@ -14,6 +14,7 @@ type allocate struct {
 	resultChan       chan<- allocateResult
 	hasBeenCancelled func() bool
 	ident            string
+	subnet           address.CIDR
 }
 
 // Try returns true if the request is completed, false if pending
@@ -29,15 +30,17 @@ func (g *allocate) Try(alloc *Allocator) bool {
 		return true
 	}
 
-	if ok, addr := alloc.space.Allocate(); ok {
-		alloc.debugln("Allocated", addr, "for", g.ident)
+	// Respect RFC1122 exclusions of first and last addresses
+	start, end := g.subnet.Start+1, g.subnet.End()-1
+	if ok, addr := alloc.space.Allocate(start, end); ok {
+		alloc.debugln("Allocated", addr, "for", g.ident, "in", g.subnet)
 		alloc.addOwned(g.ident, addr)
 		g.resultChan <- allocateResult{addr, nil}
 		return true
 	}
 
 	// out of space
-	if donor, err := alloc.ring.ChoosePeerToAskForSpace(); err == nil {
+	if donor, err := alloc.ring.ChoosePeerToAskForSpace(start, end); err == nil {
 		alloc.debugln("Decided to ask peer", donor, "for space")
 		alloc.sendRequest(donor, msgSpaceRequest)
 	}

ipam/allocator.go

@@ -3,9 +3,7 @@ package ipam
 import (
 	"bytes"
 	"encoding/gob"
-	"errors"
 	"fmt"
-	"net"
 	"sort"
 	"time"
 
@@ -46,12 +44,11 @@ type operation interface {
 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
+	defaultSubnet    address.CIDR
+	universe         address.CIDR               // superset of all ranges
 	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
+	owned            map[string]address.Address // who owns what address, indexed by container-ID -- fixme - needs to be multi-subnet
 	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
@@ -63,34 +60,16 @@ type Allocator struct {
 }
 
 // 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),
+func NewAllocator(ourName router.PeerName, ourUID router.PeerUID, ourNickname string, universe address.CIDR, quorum uint) *Allocator {
+	return &Allocator{
+		ourName:   ourName,
+		universe:  universe,
+		ring:      ring.New(universe.Start, address.Add(universe.Start, universe.Size()), 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
@@ -197,11 +176,24 @@ func hasBeenCancelled(cancelChan <-chan bool) func() bool {
 
 // Actor client API
 
+func (alloc *Allocator) SetDefaultSubnet(subnet address.CIDR) error {
+	resultChan := make(chan error)
+	if subnet.Size() < 4 {
+		return fmt.Errorf("Allocation range smaller than minimum size 4: %s", subnet)
+	}
+	alloc.actionChan <- func() {
+		// todo: check default subnet is in range
+		alloc.defaultSubnet = subnet
+		resultChan <- nil
+	}
+	return <-resultChan
+}
+
 // 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) {
+func (alloc *Allocator) Allocate(ident string, subnet address.CIDR, cancelChan <-chan bool) (address.Address, error) {
 	resultChan := make(chan allocateResult)
-	op := &allocate{resultChan: resultChan, ident: ident,
+	op := &allocate{subnet: subnet, resultChan: resultChan, ident: ident,
 		hasBeenCancelled: hasBeenCancelled(cancelChan)}
 	alloc.doOperation(op, &alloc.pendingAllocates)
 	result := <-resultChan
@@ -470,20 +462,17 @@ func (alloc *Allocator) actorLoop(actionChan <-chan func()) {
 
 func (alloc *Allocator) string() string {
 	var buf bytes.Buffer
-	fmt.Fprintf(&buf, "Allocator subnet %s/%d", alloc.subnetStart.String(), alloc.prefixLen)
+	fmt.Fprintf(&buf, "Allocator universe %s", alloc.universe)
+	if !alloc.defaultSubnet.Blank() {
+		fmt.Fprintf(&buf, " (default subnet %s)", alloc.defaultSubnet)
+	}
 
 	if alloc.ring.Empty() {
 		if alloc.paxosTicker != nil {
 			fmt.Fprintf(&buf, " awaiting consensus: %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)
-
-		fmt.Fprint(&buf, "Owned Ranges:")
+		fmt.Fprint(&buf, "\nOwned Ranges:")
 		alloc.ring.FprintWithNicknames(&buf, alloc.nicknames)
 	}
 	if len(alloc.pendingAllocates)+len(alloc.pendingClaims) > 0 {

ipam/http.go

@@ -34,24 +34,51 @@ func (alloc *Allocator) HandleHTTP(router *mux.Router) {
 	})
 
 	router.Methods("GET").Path("/ip/{id}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if alloc.defaultSubnet.Blank() {
+			badRequest(w, fmt.Errorf("No subnet specified"))
+			return
+		}
 		addr, err := alloc.Lookup(mux.Vars(r)["id"])
 		if err != nil {
 			http.NotFound(w, r)
 			return
 		}
-		fmt.Fprintf(w, "%s/%d", addr.String(), alloc.prefixLen)
+		fmt.Fprintf(w, "%s/%d", addr.String(), alloc.defaultSubnet.PrefixLen)
+	})
+
+	router.Methods("POST").Path("/ip/{id}/{ip}/{prefixlen}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		closedChan := w.(http.CloseNotifier).CloseNotify()
+		vars := mux.Vars(r)
+		ident := vars["id"]
+		cidrStr := vars["ip"] + "/" + vars["prefixlen"]
+		_, cidr, err := address.ParseCIDR(cidrStr)
+		if err != nil {
+			badRequest(w, err)
+			return
+		}
+		addr, err := alloc.Allocate(ident, cidr, closedChan)
+		if err != nil {
+			badRequest(w, fmt.Errorf("Unable to allocate: %s", err))
+			return
+		}
+
+		fmt.Fprintf(w, "%s/%d", addr.String(), cidr.PrefixLen)
 	})
 
 	router.Methods("POST").Path("/ip/{id}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if alloc.defaultSubnet.Blank() {
+			badRequest(w, fmt.Errorf("No subnet specified"))
+			return
+		}
 		closedChan := w.(http.CloseNotifier).CloseNotify()
 		ident := mux.Vars(r)["id"]
-		newAddr, err := alloc.Allocate(ident, closedChan)
+		newAddr, err := alloc.Allocate(ident, alloc.defaultSubnet, closedChan)
 		if err != nil {
 			badRequest(w, err)
 			return
 		}
 
-		fmt.Fprintf(w, "%s/%d", newAddr.String(), alloc.prefixLen)
+		fmt.Fprintf(w, "%s/%d", newAddr.String(), alloc.defaultSubnet.PrefixLen)
 	})
 
 	router.Methods("DELETE").Path("/ip/{id}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

ipam/ring/ring.go

@@ -409,15 +409,22 @@ func (r *Ring) ReportFree(freespace map[address.Address]address.Offset) {
 }
 
 // ChoosePeerToAskForSpace chooses a weighted-random peer to ask
-// for space.
-func (r *Ring) ChoosePeerToAskForSpace() (result router.PeerName, err error) {
+// for space in the range [start, end).  Assumes start<end.
+func (r *Ring) ChoosePeerToAskForSpace(start, end address.Address) (result router.PeerName, err error) {
 	var (
 		sum               address.Offset
 		totalSpacePerPeer = make(map[router.PeerName]address.Offset) // Compute total free space per peer
 	)
 
 	// iterate through tokens
-	for _, entry := range r.Entries {
+	for i, entry := range r.Entries {
+		// Ignore entries that don't span the range we want
+		if i+1 < len(r.Entries) && r.Entries.entry(i+1).Token-1 < start {
+			continue
+		}
+		if entry.Token >= end {
+			break
+		}
 		// Ignore ranges with no free space
 		if entry.Free <= 0 {
 			continue

ipam/ring/ring_test.go

@@ -345,34 +345,40 @@ func TestGossip(t *testing.T) {
 func TestFindFree(t *testing.T) {
 	ring1 := New(start, end, peer1name)
 
-	_, err := ring1.ChoosePeerToAskForSpace()
+	_, err := ring1.ChoosePeerToAskForSpace(start, end)
 	wt.AssertTrue(t, err == ErrNoFreeSpace, "Expected ErrNoFreeSpace")
 
 	ring1.Entries = []*entry{{Token: start, Peer: peer1name}}
-	_, err = ring1.ChoosePeerToAskForSpace()
+	_, err = ring1.ChoosePeerToAskForSpace(start, end)
 	wt.AssertTrue(t, err == ErrNoFreeSpace, "Expected ErrNoFreeSpace")
 
 	// We shouldn't return outselves
 	ring1.ReportFree(map[address.Address]address.Offset{start: 10})
-	_, err = ring1.ChoosePeerToAskForSpace()
+	_, err = ring1.ChoosePeerToAskForSpace(start, end)
 	wt.AssertTrue(t, err == ErrNoFreeSpace, "Expected ErrNoFreeSpace")
 
 	ring1.Entries = []*entry{{Token: start, Peer: peer1name, Free: 1},
-		{Token: start, Peer: peer1name, Free: 1}}
-	_, err = ring1.ChoosePeerToAskForSpace()
+		{Token: middle, Peer: peer1name, Free: 1}}
+	_, err = ring1.ChoosePeerToAskForSpace(start, end)
 	wt.AssertTrue(t, err == ErrNoFreeSpace, "Expected ErrNoFreeSpace")
+	ring1.assertInvariants()
 
 	// We should return others
 	ring1.Entries = []*entry{{Token: start, Peer: peer2name, Free: 1}}
-	peer, err := ring1.ChoosePeerToAskForSpace()
+	peer, err := ring1.ChoosePeerToAskForSpace(start, end)
 	wt.AssertSuccess(t, err)
 	wt.AssertEquals(t, peer, peer2name)
 
 	ring1.Entries = []*entry{{Token: start, Peer: peer2name, Free: 1},
-		{Token: start, Peer: peer2name, Free: 1}}
-	peer, err = ring1.ChoosePeerToAskForSpace()
+		{Token: middle, Peer: peer3name, Free: 1}}
+	peer, err = ring1.ChoosePeerToAskForSpace(start, middle)
 	wt.AssertSuccess(t, err)
 	wt.AssertEquals(t, peer, peer2name)
+
+	peer, err = ring1.ChoosePeerToAskForSpace(middle, end)
+	wt.AssertSuccess(t, err)
+	wt.AssertEquals(t, peer, peer3name)
+	ring1.assertInvariants()
 }
 
 func TestReportFree(t *testing.T) {

ipam/space/space.go

@@ -37,14 +37,26 @@ func (s *Space) Clear() {
 	s.ours = s.ours[:0]
 }
 
-func (s *Space) Allocate() (bool, address.Address) {
-	if len(s.free) == 0 {
-		return false, 0
+func (s *Space) Allocate(rangeStart, rangeEnd address.Address) (bool, address.Address) {
+	// Need to walk forward to the range, then check free space, then pick an address
+	// Walk down the free spaces, looking for one in range
+	for i := 0; i < len(s.free); i += 2 {
+		start, end := s.free[i], s.free[i+1]
+		if end-1 < rangeStart {
+			continue
+		}
+		if start > rangeEnd {
+			break
+		}
+		res := start
+		if res < rangeStart {
+			res = rangeStart
+		}
+		s.ours = add(s.ours, res, res+1)
+		s.free = subtract(s.free, res, res+1)
+		return true, res
 	}
-	res := s.free[0]
-	s.ours = add(s.ours, res, res+1)
-	s.free = subtract(s.free, res, res+1)
-	return true, res
+	return false, 0
 }
 
 func (s *Space) Claim(addr address.Address) error {

ipam/space/space_test.go

@@ -52,12 +52,12 @@ func TestLowlevel(t *testing.T) {
 
 	s := New()
 	wt.AssertEquals(t, s.NumFreeAddresses(), address.Offset(0))
-	ok, got := s.Allocate()
+	ok, got := s.Allocate(0, 1000)
 	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()
+	ok, got = s.Allocate(0, 1000)
 	wt.AssertTrue(t, ok && got == 100, "allocate")
 	wt.AssertEquals(t, s.NumFreeAddresses(), address.Offset(99))
 	wt.AssertNoErr(t, s.Claim(150))
@@ -86,6 +86,7 @@ func TestLowlevel(t *testing.T) {
 func TestSpaceAllocate(t *testing.T) {
 	const (
 		testAddr1   = "10.0.3.4"
+		testAddr2   = "10.0.3.5"
 		testAddrx   = "10.0.3.19"
 		testAddry   = "10.0.9.19"
 		containerID = "deadbeef"
@@ -95,13 +96,13 @@ func TestSpaceAllocate(t *testing.T) {
 	wt.AssertEquals(t, space1.NumFreeAddresses(), address.Offset(20))
 	space1.assertInvariants()
 
-	_, addr1 := space1.Allocate()
-	wt.AssertEqualString(t, addr1.String(), testAddr1, "address")
+	_, addr1 := space1.Allocate(ip(testAddr2), ip(testAddrx))
+	wt.AssertEqualString(t, addr1.String(), testAddr2, "address")
 	wt.AssertEquals(t, space1.NumFreeAddresses(), address.Offset(19))
 	space1.assertInvariants()
 
-	_, addr2 := space1.Allocate()
-	wt.AssertFalse(t, addr2.String() == testAddr1, "address")
+	_, addr2 := space1.Allocate(ip(testAddr2), ip(testAddrx))
+	wt.AssertFalse(t, addr2.String() == testAddr2, "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))
@@ -132,12 +133,12 @@ func TestSpaceFree(t *testing.T) {
 	wt.AssertTrue(t, space.free[pos] == ip(testAddr1) && size == 20, "Wrong space")
 
 	for i := 0; i < 20; i++ {
-		ok, _ := space.Allocate()
+		ok, _ := space.Allocate(ip(testAddr1), ip(testAddry))
 		wt.AssertTrue(t, ok, "Failed to get address")
 	}
 
 	// Check we are full
-	ok, _ := space.Allocate()
+	ok, _ := space.Allocate(ip(testAddr1), ip(testAddry))
 	wt.AssertTrue(t, !ok, "Should have failed to get address")
 	_, size, ok = space.Donate()
 	wt.AssertTrue(t, size == 0, "Wrong space")
@@ -222,7 +223,7 @@ func TestDonateHard(t *testing.T) {
 	// Fill a fresh space
 	spaceset := makeSpace(start, size)
 	for i := address.Offset(0); i < size; i++ {
-		ok, _ := spaceset.Allocate()
+		ok, _ := spaceset.Allocate(start, address.Add(start, size))
 		wt.AssertTrue(t, ok, "Failed to get IP!")
 	}
 

ipam/testutils_test.go

@@ -128,17 +128,19 @@ func CheckAllExpectedMessagesSent(allocs ...*Allocator) {
 	}
 }
 
-func makeAllocator(name string, cidr string, quorum uint) *Allocator {
+func makeAllocator(name string, cidrStr 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)
+	_, cidr, err := address.ParseCIDR(cidrStr)
 	if err != nil {
 		panic(err)
 	}
+	alloc := NewAllocator(peername, router.PeerUID(rand.Int63()),
+		"nick-"+name, cidr, quorum)
+	alloc.defaultSubnet = cidr
 
 	return alloc
 }
@@ -163,7 +165,7 @@ func (alloc *Allocator) claimRingForTesting(allocs ...*Allocator) {
 func (alloc *Allocator) NumFreeAddresses() address.Offset {
 	resultChan := make(chan address.Offset)
 	alloc.actionChan <- func() {
-		resultChan <- alloc.space.NumFreeAddresses()
+		resultChan <- alloc.space.NumFreeAddressesInRange(alloc.defaultSubnet.Start+1, alloc.defaultSubnet.End()-1)
 	}
 	return <-resultChan
 }

prog/weaver/main.go

@@ -17,6 +17,7 @@ import (
 	. "github.com/weaveworks/weave/common"
 	"github.com/weaveworks/weave/common/updater"
 	"github.com/weaveworks/weave/ipam"
+	"github.com/weaveworks/weave/ipam/address"
 	weavenet "github.com/weaveworks/weave/net"
 	weave "github.com/weaveworks/weave/router"
 )
@@ -51,6 +52,7 @@ func main() {
 		noDiscovery bool
 		httpAddr    string
 		iprangeCIDR string
+		defltSubnet string
 		peerCount   int
 		apiPath     string
 		peers       []string
@@ -71,6 +73,7 @@ func main() {
 	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.StringVar(&defltSubnet, "defaultsubnet", "", "default subnet for IP address allocation, 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()
@@ -138,9 +141,9 @@ func main() {
 
 	var allocator *ipam.Allocator
 	if iprangeCIDR != "" {
-		allocator = createAllocator(router, apiPath, iprangeCIDR, determineQuorum(peerCount, peers))
+		allocator = createAllocator(router, apiPath, iprangeCIDR, defltSubnet, determineQuorum(peerCount, peers))
 	} else if peerCount > 0 {
-		log.Fatal("-initpeercount flag specified without -iprange")
+		log.Fatal("-initpeercount flag specified without -iprange or -ipuniverse")
 	} else {
 		router.NewGossip("IPallocation", &ipam.DummyAllocator{})
 	}
@@ -194,13 +197,26 @@ func logFrameFunc(debug bool) weave.LogFrameFunc {
 	}
 }
 
-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)
+func createAllocator(router *weave.Router, apiPath string, ipUniverse string, defaultSubnet string, quorum uint) *ipam.Allocator {
+	_, universe, err := address.ParseCIDR(ipUniverse)
 	if err != nil {
 		log.Fatal(err)
 	}
+	if universe.Size() < 4 {
+		log.Fatal("IP allocation range too small")
+	}
+	allocator := ipam.NewAllocator(router.Ourself.Peer.Name, router.Ourself.Peer.UID, router.Ourself.Peer.NickName, universe, quorum)
 	allocator.SetInterfaces(router.NewGossip("IPallocation", allocator))
 	allocator.Start()
+	if defaultSubnet != "" {
+		_, subnet, err := address.ParseCIDR(defaultSubnet)
+		if err != nil {
+			log.Fatal(err)
+		}
+		if err = allocator.SetDefaultSubnet(subnet); err != nil {
+			log.Fatal(err)
+		}
+	}
 	err = updater.Start(apiPath, allocator)
 	if err != nil {
 		log.Fatal("Unable to start watcher", err)

router/protocol.go

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

weave

@@ -28,7 +28,7 @@ usage() {
     echo "Usage:"
     echo "weave setup"
     echo "weave launch       [-password <password>] [-nickname <nickname>] [-nodiscovery]"
-    echo "                       [-iprange <cidr>] <peer> ..."
+    echo "                       [-iprange <cidr> [-ipsubnet <cidr>]] <peer> ..."
     echo "weave launch-dns   <cidr>"
     echo "weave launch-proxy [-D <docker_endpoint>] [-H <address>] [--with-dns] [--with-ipam]"
     echo "weave connect      [--replace] [<peer> ...]"