From 2553850ed4a5a409a4f2c78c13c36e6f06ba9122 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Tue, 2 Jun 2015 14:15:37 +0100 Subject: [PATCH] Initial implementation of multi-subnet IPAM --- ipam/allocate.go | 9 ++++-- ipam/allocator.go | 67 +++++++++++++++++----------------------- ipam/http.go | 33 ++++++++++++++++++-- ipam/ring/ring.go | 13 ++++++-- ipam/ring/ring_test.go | 22 ++++++++----- ipam/space/space.go | 26 +++++++++++----- ipam/space/space_test.go | 19 ++++++------ ipam/testutils_test.go | 10 +++--- prog/weaver/main.go | 24 +++++++++++--- router/protocol.go | 2 +- weave | 2 +- 11 files changed, 145 insertions(+), 82 deletions(-) diff --git a/ipam/allocate.go b/ipam/allocate.go index d709784d..9fe2522c 100644 --- a/ipam/allocate.go +++ b/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) } diff --git a/ipam/allocator.go b/ipam/allocator.go index 210eb796..ceea6caf 100644 --- a/ipam/allocator.go +++ b/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 { diff --git a/ipam/http.go b/ipam/http.go index 4e7cd907..925fe5fd 100644 --- a/ipam/http.go +++ b/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) { diff --git a/ipam/ring/ring.go b/ipam/ring/ring.go index ddb2f631..910c87c6 100644 --- a/ipam/ring/ring.go +++ b/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 { + break + } // Ignore ranges with no free space if entry.Free <= 0 { continue diff --git a/ipam/ring/ring_test.go b/ipam/ring/ring_test.go index f3ec1e0f..11aa47f9 100644 --- a/ipam/ring/ring_test.go +++ b/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) { diff --git a/ipam/space/space.go b/ipam/space/space.go index 548d9e55..c4d8ea5f 100644 --- a/ipam/space/space.go +++ b/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 { diff --git a/ipam/space/space_test.go b/ipam/space/space_test.go index d49324b7..11df7b58 100644 --- a/ipam/space/space_test.go +++ b/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!") } diff --git a/ipam/testutils_test.go b/ipam/testutils_test.go index c47273cb..f4a79f18 100644 --- a/ipam/testutils_test.go +++ b/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 } diff --git a/prog/weaver/main.go b/prog/weaver/main.go index e59b4202..809ffd49 100644 --- a/prog/weaver/main.go +++ b/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) diff --git a/router/protocol.go b/router/protocol.go index 5e9c7036..60482621 100644 --- a/router/protocol.go +++ b/router/protocol.go @@ -2,7 +2,7 @@ package router const ( Protocol = "weave" - ProtocolVersion = 18 + ProtocolVersion = 19 ) type ProtocolTag byte diff --git a/weave b/weave index 40f8e2a9..f96e25e4 100755 --- a/weave +++ b/weave @@ -28,7 +28,7 @@ usage() { echo "Usage:" echo "weave setup" echo "weave launch [-password ] [-nickname ] [-nodiscovery]" - echo " [-iprange ] ..." + echo " [-iprange [-ipsubnet ]] ..." echo "weave launch-dns " echo "weave launch-proxy [-D ] [-H
] [--with-dns] [--with-ipam]" echo "weave connect [--replace] [ ...]"