Merge pull request #2341 from weaveworks/issues/2017-awsvpc

Add AWSVPC. Fixes #2017

.gitmodules

@@ -73,3 +73,6 @@
 [submodule "vendor/github.com/j-keck/arping"]
 	path = vendor/github.com/j-keck/arping
 	url = https://github.com/j-keck/arping
+[submodule "vendor/github.com/aws/aws-sdk-go"]
+	path = vendor/github.com/aws/aws-sdk-go
+	url = https://github.com/aws/aws-sdk-go

ipam/allocator.go

@@ -14,6 +14,7 @@ import (
 	"github.com/weaveworks/weave/ipam/paxos"
 	"github.com/weaveworks/weave/ipam/ring"
 	"github.com/weaveworks/weave/ipam/space"
+	"github.com/weaveworks/weave/ipam/tracker"
 	"github.com/weaveworks/weave/net/address"
 )
 
@@ -85,11 +86,14 @@ type Config struct {
 	Quorum      func() uint
 	Db          db.DB
 	IsKnownPeer func(name mesh.PeerName) bool
+	Tracker     tracker.LocalRangeTracker
 }
 
 // NewAllocator creates and initialises a new Allocator
 func NewAllocator(config Config) *Allocator {
 	var participant paxos.Participant
+	var alloc *Allocator
+	var onUpdate ring.OnUpdate
 
 	if config.IsObserver {
 		participant = paxos.NewObserver()
@@ -97,11 +101,19 @@ func NewAllocator(config Config) *Allocator {
 		participant = paxos.NewNode(config.OurName, config.OurUID, 0)
 	}
 
-	return &Allocator{
+	if config.Tracker != nil {
+		onUpdate = func(prev []address.Range, curr []address.Range, local bool) {
+			if err := config.Tracker.HandleUpdate(prev, curr, local); err != nil {
+				alloc.errorf("HandleUpdate failed: %s", err)
+			}
+		}
+	}
+
+	alloc = &Allocator{
 		ourName:     config.OurName,
 		seed:        config.Seed,
 		universe:    config.Universe,
-		ring:        ring.New(config.Universe.Range().Start, config.Universe.Range().End, config.OurName),
+		ring:        ring.New(config.Universe.Range().Start, config.Universe.Range().End, config.OurName, onUpdate),
 		owned:       make(map[string]ownedData),
 		db:          config.Db,
 		paxos:       participant,
@@ -111,6 +123,7 @@ func NewAllocator(config Config) *Allocator {
 		dead:        make(map[string]time.Time),
 		now:         time.Now,
 	}
+	return alloc
 }
 
 func ParseCIDRSubnet(cidrStr string) (cidr address.CIDR, err error) {
@@ -958,7 +971,7 @@ func (alloc *Allocator) loadPersistedData() bool {
 		return false
 	}
 
-	alloc.ring = persistedRing
+	alloc.ring.Restore(persistedRing)
 	alloc.space.UpdateRanges(alloc.ring.OwnedRanges())
 
 	if ownedFound {
@@ -1068,6 +1081,9 @@ func (alloc *Allocator) fatalf(fmt string, args ...interface{}) {
 func (alloc *Allocator) warnf(fmt string, args ...interface{}) {
 	alloc.logf(common.Log.Warnf, fmt, args...)
 }
+func (alloc *Allocator) errorf(fmt string, args ...interface{}) {
+	common.Log.Errorf("[allocator %s] "+fmt, append([]interface{}{alloc.ourName}, args...)...)
+}
 func (alloc *Allocator) infof(fmt string, args ...interface{}) {
 	alloc.logf(common.Log.Infof, fmt, args...)
 }

ipam/http.go

@@ -87,7 +87,7 @@ func (alloc *Allocator) handleHTTPClaim(dockerCli *docker.Client, w http.Respons
 }
 
 // HandleHTTP wires up ipams HTTP endpoints to the provided mux.
-func (alloc *Allocator) HandleHTTP(router *mux.Router, defaultSubnet address.CIDR, dockerCli *docker.Client) {
+func (alloc *Allocator) HandleHTTP(router *mux.Router, defaultSubnet address.CIDR, tracker string, dockerCli *docker.Client) {
 	router.Methods("GET").Path("/ipinfo/defaultsubnet").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		fmt.Fprintf(w, "%s", defaultSubnet)
 	})
@@ -174,4 +174,8 @@ func (alloc *Allocator) HandleHTTP(router *mux.Router, defaultSubnet address.CID
 		transferred := alloc.AdminTakeoverRanges(ident)
 		fmt.Fprintf(w, "%d IPs taken over from %s\n", transferred, ident)
 	})
+
+	router.Methods("GET").Path("/ipinfo/tracker").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprintf(w, tracker)
+	})
 }

ipam/http_test.go

@@ -43,7 +43,7 @@ func listenHTTP(alloc *Allocator, subnet address.CIDR) int {
 	router.HandleFunc("/status", func(w http.ResponseWriter, r *http.Request) {
 		io.WriteString(w, fmt.Sprintln(alloc))
 	})
-	alloc.HandleHTTP(router, subnet, nil)
+	alloc.HandleHTTP(router, subnet, "", nil)
 
 	httpListener, err := net.Listen("tcp", ":0")
 	if err != nil {

ipam/ring/ring.go

@@ -18,12 +18,15 @@ import (
 	"github.com/weaveworks/weave/net/address"
 )
 
+type OnUpdate func(prev, curr []address.Range, local bool)
+
 // Ring represents the ring itself
 type Ring struct {
 	Start, End address.Address // [min, max) tokens in this ring.  Due to wrapping, min == max (effectively)
 	Peer       mesh.PeerName   // name of peer owning this ring instance
 	Entries    entries         // list of entries sorted by token
 	Seeds      []mesh.PeerName // peers with which the ring was seeded
+	onUpdate   OnUpdate
 }
 
 func (r *Ring) assertInvariants() {
@@ -94,15 +97,33 @@ func (r *Ring) checkEntries(entries entries) error {
 	return nil
 }
 
+func (r *Ring) trackUpdates() func() {
+	return r.trackUpdatesOfPeer(r.Peer)
+}
+
+func (r *Ring) trackUpdatesOfPeer(peer mesh.PeerName) func() {
+	if r.onUpdate == nil {
+		return func() {}
+	}
+	ranges := r.OwnedRangesOfPeer(peer)
+	return func() { r.onUpdate(ranges, r.OwnedRangesOfPeer(peer), peer == r.Peer) }
+}
+
 // New creates an empty ring belonging to peer.
-func New(start, end address.Address, peer mesh.PeerName) *Ring {
+func New(start, end address.Address, peer mesh.PeerName, f OnUpdate) *Ring {
 	common.Assert(start < end)
 
-	ring := &Ring{Start: start, End: end, Peer: peer, Entries: make([]*entry, 0)}
+	ring := &Ring{Start: start, End: end, Peer: peer, Entries: make([]*entry, 0), onUpdate: f}
 	ring.updateExportedVariables()
 	return ring
 }
 
+func (r *Ring) Restore(other *Ring) {
+	onUpdate := r.onUpdate
+	*r = *other
+	r.onUpdate = onUpdate
+}
+
 func (r *Ring) Range() address.Range {
 	return address.Range{Start: r.Start, End: r.End}
 }
@@ -126,6 +147,7 @@ func (r *Ring) GrantRangeToHost(start, end address.Address, peer mesh.PeerName)
 	//fmt.Printf("%s GrantRangeToHost [%v,%v) -> %s\n", r.Peer, start, end, peer)
 
 	r.assertInvariants()
+	defer r.trackUpdates()()
 	defer r.assertInvariants()
 	defer r.updateExportedVariables()
 
@@ -193,6 +215,7 @@ func (r *Ring) GrantRangeToHost(start, end address.Address, peer mesh.PeerName)
 // got updated as a result.
 func (r *Ring) Merge(gossip Ring) (bool, error) {
 	r.assertInvariants()
+	defer r.trackUpdates()()
 	defer r.updateExportedVariables()
 
 	// Don't panic when checking the gossiped in ring.
@@ -230,6 +253,7 @@ func (r *Ring) Merge(gossip Ring) (bool, error) {
 		r.Seeds = gossip.Seeds
 	}
 	r.Entries = result
+
 	return updated, nil
 }
 
@@ -336,10 +360,14 @@ func (r *Ring) splitRangesOverZero(ranges []address.Range) []address.Range {
 // ranges are owned by this peer.  Will split ranges which
 // span 0 in the ring.
 func (r *Ring) OwnedRanges() (result []address.Range) {
+	return r.OwnedRangesOfPeer(r.Peer)
+}
+
+func (r *Ring) OwnedRangesOfPeer(peer mesh.PeerName) (result []address.Range) {
 	r.assertInvariants()
 
 	for i, entry := range r.Entries {
-		if entry.Peer == r.Peer {
+		if entry.Peer == peer {
 			nextEntry := r.Entries.entry(i + 1)
 			result = append(result, address.Range{Start: entry.Token, End: nextEntry.Token})
 		}
@@ -368,34 +396,38 @@ func (r *Ring) AllRangeInfo() (result []RangeInfo) {
 }
 
 // ClaimForPeers claims the entire ring for the array of peers passed
-// in.  Only works for empty rings.
+// in.  Only works for empty rings. Each claimed range is CIDR-aligned.
 func (r *Ring) ClaimForPeers(peers []mesh.PeerName) {
 	common.Assert(r.Empty())
+
+	defer r.trackUpdates()()
 	defer r.assertInvariants()
 	defer r.updateExportedVariables()
+	defer func() {
+		e := r.Entries[len(r.Entries)-1]
+		common.Assert(address.Add(e.Token, address.Offset(e.Free)) == r.End)
+	}()
 
-	totalSize := r.distance(r.Start, r.End)
-	share := totalSize/address.Count(len(peers)) + 1
-	remainder := totalSize % address.Count(len(peers))
-	pos := r.Start
-
-	for i, peer := range peers {
-		if address.Count(i) == remainder {
-			share--
-			if share == 0 {
-				break
-			}
-		}
-
-		r.Entries.insert(entry{Token: pos, Peer: peer, Free: share})
-		pos += address.Address(share)
-	}
-
-	common.Assert(pos == r.End)
-
+	r.subdivide(r.Start, r.End, peers)
 	r.Seeds = peers
 }
 
+// subdivide subdivides the [from,to) CIDR for the given peers into
+// CIDR-aligned subranges.
+func (r *Ring) subdivide(from, to address.Address, peers []mesh.PeerName) {
+	share := address.Length(to, from)
+	if share == 0 {
+		return
+	}
+	if share == 1 || len(peers) == 1 {
+		r.Entries.insert(entry{Token: from, Peer: peers[0], Free: share})
+		return
+	}
+	mid := address.Add(from, address.Offset(share/2))
+	r.subdivide(from, mid, peers[:len(peers)/2])
+	r.subdivide(address.Add(mid, address.Offset(share%2)), to, peers[len(peers)/2:])
+}
+
 func (r *Ring) FprintWithNicknames(w io.Writer, m map[mesh.PeerName]string) {
 	for _, entry := range r.Entries {
 		nickname, found := m[entry.Peer]
@@ -525,6 +557,8 @@ func (r *Ring) PickPeerForTransfer(isValid func(mesh.PeerName) bool) mesh.PeerNa
 // and return ranges indicating the new space we picked up
 func (r *Ring) Transfer(from, to mesh.PeerName) []address.Range {
 	r.assertInvariants()
+	defer r.trackUpdates()()
+	defer r.trackUpdatesOfPeer(from)()
 	defer r.assertInvariants()
 	defer r.updateExportedVariables()
 

ipam/ring/ring_test.go

@@ -36,8 +36,12 @@ func merge(r1, r2 *Ring) error {
 	return err
 }
 
+func NewRing(start, end address.Address, peer mesh.PeerName) *Ring {
+	return New(start, end, peer, nil)
+}
+
 func TestInvariants(t *testing.T) {
-	ring := New(start, end, peer1name)
+	ring := NewRing(start, end, peer1name)
 
 	// Check ring is sorted
 	ring.Entries = []*entry{{Token: dot245, Peer: peer1name}, {Token: dot10, Peer: peer2name}}
@@ -48,7 +52,7 @@ func TestInvariants(t *testing.T) {
 	require.True(t, ring.checkInvariants() == ErrTokenRepeated, "Expected error")
 
 	// Check tokens are in bounds
-	ring = New(dot10, dot245, peer1name)
+	ring = NewRing(dot10, dot245, peer1name)
 	ring.Entries = []*entry{{Token: start, Peer: peer1name}}
 	require.True(t, ring.checkInvariants() == ErrTokenOutOfRange, "Expected error")
 
@@ -57,7 +61,7 @@ func TestInvariants(t *testing.T) {
 }
 
 func TestInsert(t *testing.T) {
-	ring := New(start, end, peer1name)
+	ring := NewRing(start, end, peer1name)
 	ring.Entries = []*entry{{Token: start, Peer: peer1name, Free: 255}}
 
 	require.Panics(t, func() {
@@ -82,7 +86,7 @@ func TestInsert(t *testing.T) {
 }
 
 func TestBetween(t *testing.T) {
-	ring1 := New(start, end, peer1name)
+	ring1 := NewRing(start, end, peer1name)
 	ring1.Entries = []*entry{{Token: start, Peer: peer1name, Free: 255}}
 
 	// First off, in a ring where everything is owned by the peer
@@ -124,8 +128,8 @@ func TestBetween(t *testing.T) {
 }
 
 func TestGrantSimple(t *testing.T) {
-	ring1 := New(start, end, peer1name)
-	ring2 := New(start, end, peer2name)
+	ring1 := NewRing(start, end, peer1name)
+	ring2 := NewRing(start, end, peer2name)
 
 	// Claim everything for peer1
 	ring1.ClaimItAll()
@@ -159,8 +163,8 @@ func TestGrantSimple(t *testing.T) {
 }
 
 func TestGrantSplit(t *testing.T) {
-	ring1 := New(start, end, peer1name)
-	ring2 := New(start, end, peer2name)
+	ring1 := NewRing(start, end, peer1name)
+	ring2 := NewRing(start, end, peer2name)
 
 	// Claim everything for peer1
 	ring1.Entries = []*entry{{Token: start, Peer: peer1name, Free: 256}}
@@ -188,8 +192,8 @@ func TestGrantSplit(t *testing.T) {
 }
 
 func TestMergeSimple(t *testing.T) {
-	ring1 := New(start, end, peer1name)
-	ring2 := New(start, end, peer2name)
+	ring1 := NewRing(start, end, peer1name)
+	ring2 := NewRing(start, end, peer2name)
 
 	// Claim everything for peer1
 	ring1.ClaimItAll()
@@ -217,18 +221,18 @@ func TestMergeSimple(t *testing.T) {
 
 func TestMergeErrors(t *testing.T) {
 	// Cannot Merge in an invalid ring
-	ring1 := New(start, end, peer1name)
-	ring2 := New(start, end, peer2name)
+	ring1 := NewRing(start, end, peer1name)
+	ring2 := NewRing(start, end, peer2name)
 	ring2.Entries = []*entry{{Token: middle, Peer: peer2name}, {Token: start, Peer: peer2name}}
 	require.True(t, merge(ring1, ring2) == ErrNotSorted, "Expected ErrNotSorted")
 
 	// Should Merge two rings for different ranges
-	ring2 = New(start, middle, peer2name)
+	ring2 = NewRing(start, middle, peer2name)
 	ring2.Entries = []*entry{}
 	require.True(t, merge(ring1, ring2) == ErrDifferentRange, "Expected ErrDifferentRange")
 
 	// Cannot Merge newer version of entry I own
-	ring2 = New(start, end, peer2name)
+	ring2 = NewRing(start, end, peer2name)
 	ring1.Entries = []*entry{{Token: start, Peer: peer1name}}
 	ring2.Entries = []*entry{{Token: start, Peer: peer1name, Version: 1}}
 	fmt.Println(merge(ring1, ring2))
@@ -246,8 +250,8 @@ func TestMergeErrors(t *testing.T) {
 }
 
 func TestMergeMore(t *testing.T) {
-	ring1 := New(start, end, peer1name)
-	ring2 := New(start, end, peer2name)
+	ring1 := NewRing(start, end, peer1name)
+	ring2 := NewRing(start, end, peer2name)
 
 	assertRing := func(ring *Ring, entries entries) {
 		require.Equal(t, entries, ring.Entries)
@@ -297,8 +301,8 @@ func TestMergeMore(t *testing.T) {
 }
 
 func TestMergeSplit(t *testing.T) {
-	ring1 := New(start, end, peer1name)
-	ring2 := New(start, end, peer2name)
+	ring1 := NewRing(start, end, peer1name)
+	ring2 := NewRing(start, end, peer2name)
 
 	// Claim everything for peer2
 	ring1.Entries = []*entry{{Token: start, Peer: peer2name, Free: 256}}
@@ -320,8 +324,8 @@ func TestMergeSplit(t *testing.T) {
 }
 
 func TestMergeSplit2(t *testing.T) {
-	ring1 := New(start, end, peer1name)
-	ring2 := New(start, end, peer2name)
+	ring1 := NewRing(start, end, peer1name)
+	ring2 := NewRing(start, end, peer2name)
 
 	// Claim everything for peer2
 	ring1.Entries = []*entry{{Token: start, Peer: peer2name, Free: 250}, {Token: dot250, Peer: peer2name, Free: 5}}
@@ -344,8 +348,8 @@ func TestMergeSplit2(t *testing.T) {
 
 // A simple test, very similar to above, but using the marshalling to byte[]s
 func TestGossip(t *testing.T) {
-	ring1 := New(start, end, peer1name)
-	ring2 := New(start, end, peer2name)
+	ring1 := NewRing(start, end, peer1name)
+	ring2 := NewRing(start, end, peer2name)
 
 	assertRing := func(ring *Ring, entries entries) {
 		require.Equal(t, entries, ring.Entries)
@@ -372,7 +376,7 @@ func assertPeersWithSpace(t *testing.T, ring *Ring, start, end address.Address,
 }
 
 func TestFindFree(t *testing.T) {
-	ring1 := New(start, end, peer1name)
+	ring1 := NewRing(start, end, peer1name)
 
 	assertPeersWithSpace(t, ring1, start, end, 0)
 
@@ -408,8 +412,8 @@ func TestFindFree(t *testing.T) {
 }
 
 func TestReportFree(t *testing.T) {
-	ring1 := New(start, end, peer1name)
-	ring2 := New(start, end, peer2name)
+	ring1 := NewRing(start, end, peer1name)
+	ring2 := NewRing(start, end, peer2name)
 
 	ring1.ClaimItAll()
 	ring1.GrantRangeToHost(middle, end, peer2name)
@@ -423,7 +427,7 @@ func TestReportFree(t *testing.T) {
 }
 
 func TestMisc(t *testing.T) {
-	ring := New(start, end, peer1name)
+	ring := NewRing(start, end, peer1name)
 
 	require.True(t, ring.Empty(), "empty")
 
@@ -432,8 +436,8 @@ func TestMisc(t *testing.T) {
 }
 
 func TestEmptyGossip(t *testing.T) {
-	ring1 := New(start, end, peer1name)
-	ring2 := New(start, end, peer2name)
+	ring1 := NewRing(start, end, peer1name)
+	ring2 := NewRing(start, end, peer2name)
 
 	ring1.ClaimItAll()
 	// This used to panic, and it shouldn't
@@ -441,8 +445,8 @@ func TestEmptyGossip(t *testing.T) {
 }
 
 func TestMergeOldMessage(t *testing.T) {
-	ring1 := New(start, end, peer1name)
-	ring2 := New(start, end, peer2name)
+	ring1 := NewRing(start, end, peer1name)
+	ring2 := NewRing(start, end, peer2name)
 
 	ring1.ClaimItAll()
 	require.NoError(t, merge(ring2, ring1))
@@ -452,8 +456,8 @@ func TestMergeOldMessage(t *testing.T) {
 }
 
 func TestSplitRangeAtBeginning(t *testing.T) {
-	ring1 := New(start, end, peer1name)
-	ring2 := New(start, end, peer2name)
+	ring1 := NewRing(start, end, peer1name)
+	ring2 := NewRing(start, end, peer2name)
 
 	ring1.ClaimItAll()
 	require.NoError(t, merge(ring2, ring1))
@@ -463,7 +467,7 @@ func TestSplitRangeAtBeginning(t *testing.T) {
 }
 
 func TestOwnedRange(t *testing.T) {
-	ring1 := New(start, end, peer1name)
+	ring1 := NewRing(start, end, peer1name)
 	ring1.ClaimItAll()
 
 	require.Equal(t, []address.Range{{Start: start, End: end}}, ring1.OwnedRanges())
@@ -471,7 +475,7 @@ func TestOwnedRange(t *testing.T) {
 	ring1.GrantRangeToHost(middle, end, peer2name)
 	require.Equal(t, []address.Range{{Start: start, End: middle}}, ring1.OwnedRanges())
 
-	ring2 := New(start, end, peer2name)
+	ring2 := NewRing(start, end, peer2name)
 	merge(ring2, ring1)
 	require.Equal(t, []address.Range{{Start: middle, End: end}}, ring2.OwnedRanges())
 
@@ -486,7 +490,7 @@ func TestOwnedRange(t *testing.T) {
 
 func TestTransfer(t *testing.T) {
 	// First test just checks if we can grant some range to a host, when we transfer it, we get it back
-	ring1 := New(start, end, peer1name)
+	ring1 := NewRing(start, end, peer1name)
 	ring1.ClaimItAll()
 	ring1.GrantRangeToHost(middle, end, peer2name)
 	ring1.Transfer(peer2name, peer1name)
@@ -494,7 +498,7 @@ func TestTransfer(t *testing.T) {
 
 	// Second test is what happens when a token exists at the end of a range but is transferred
 	// - does it get resurrected correctly?
-	ring1 = New(start, end, peer1name)
+	ring1 = NewRing(start, end, peer1name)
 	ring1.ClaimItAll()
 	ring1.GrantRangeToHost(middle, end, peer2name)
 	ring1.Transfer(peer2name, peer1name)
@@ -503,7 +507,7 @@ func TestTransfer(t *testing.T) {
 }
 
 func TestOwner(t *testing.T) {
-	ring1 := New(start, end, peer1name)
+	ring1 := NewRing(start, end, peer1name)
 	require.True(t, ring1.Contains(start), "start should be in ring")
 	require.False(t, ring1.Contains(end), "end should not be in ring")
 
@@ -540,11 +544,11 @@ func TestClaimForPeers(t *testing.T) {
 	// Different end to usual so we get a number of addresses that a)
 	// is smaller than the max number of peers, and b) is divisible by
 	// some number of peers. This maximises coverage of edge cases.
-	end := dot10
+	end := dot8
 	peers := makePeers(numPeers)
 	// Test for a range of peer counts
 	for i := 0; i < numPeers; i++ {
-		ring := New(start, end, peers[0])
+		ring := NewRing(start, end, peers[0])
 		ring.ClaimForPeers(peers[:i+1])
 	}
 }
@@ -582,7 +586,7 @@ func TestFuzzRing(t *testing.T) {
 		sort.Sort(addressSlice(tokens))
 
 		peer := peers[rand.Intn(len(peers))]
-		ring := New(start, end, peer)
+		ring := NewRing(start, end, peer)
 		for _, token := range tokens {
 			peer = peers[rand.Intn(len(peers))]
 			ring.Entries = append(ring.Entries, &entry{Token: start + token, Peer: peer})
@@ -616,7 +620,7 @@ func TestFuzzRing(t *testing.T) {
 		}
 
 		peer := peers[rand.Intn(len(peers))]
-		ring := New(start, end, peer)
+		ring := NewRing(start, end, peer)
 		for _, token := range tokens {
 			peer = peers[rand.Intn(len(peers))]
 			ring.Entries = append(ring.Entries, &entry{Token: start + token, Peer: peer})
@@ -654,7 +658,7 @@ func TestFuzzRingHard(t *testing.T) {
 		common.Log.Debugf("%s: Adding peer", peer)
 		nextPeerID++
 		peers = append(peers, peer)
-		rings = append(rings, New(start, end, peer))
+		rings = append(rings, NewRing(start, end, peer))
 	}
 
 	for i := 0; i < numPeers; i++ {

ipam/tracker/awsvpc.go

@@ -0,0 +1,286 @@
+package tracker
+
+// The AWSVPC tracker tracks the IPAM ring changes and accordingly updates
+// the AWS VPC route table and, if the announced change is local
+// (denoted by local=true) to the host, the host route table.
+//
+// During the initialization, the tracker detects AWS VPC route table id which
+// is associated with the host's subnet on the AWS network. If such a table does
+// not exist, the default VPC route table is used.
+//
+// When a host A donates a range to a host B, the necessary route table
+// updates (removal) happen on the host A first, and afterwards on
+// the host B (installation).
+//
+// NB: there is a hard limit for 50 routes within any VPC route table
+// (practically, it is 48, because one route is used by the AWS Internet GW and
+// one by the AWS host subnet), therefore it is suggested to avoid
+// an excessive fragmentation within the IPAM ring which might happen due to
+// the claim operations or uneven distribution of containers across the hosts.
+
+import (
+	"fmt"
+	"net"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/ec2metadata"
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/vishvananda/netlink"
+
+	"github.com/weaveworks/weave/common"
+	wnet "github.com/weaveworks/weave/net"
+	"github.com/weaveworks/weave/net/address"
+)
+
+type AWSVPCTracker struct {
+	ec2          *ec2.EC2
+	instanceID   string // EC2 Instance ID
+	routeTableID string // VPC Route Table ID
+	linkIndex    int    // The weave bridge link index
+}
+
+// NewAWSVPCTracker creates and initialises AWS VPC based tracker.
+func NewAWSVPCTracker() (*AWSVPCTracker, error) {
+	var (
+		err     error
+		session = session.New()
+		t       = &AWSVPCTracker{}
+	)
+
+	// Detect region and instance id
+	meta := ec2metadata.New(session)
+	t.instanceID, err = meta.GetMetadata("instance-id")
+	if err != nil {
+		return nil, fmt.Errorf("cannot detect instance-id: %s", err)
+	}
+	region, err := meta.Region()
+	if err != nil {
+		return nil, fmt.Errorf("cannot detect region: %s", err)
+	}
+
+	t.ec2 = ec2.New(session, aws.NewConfig().WithRegion(region))
+
+	routeTableID, err := t.detectRouteTableID()
+	if err != nil {
+		return nil, err
+	}
+	t.routeTableID = *routeTableID
+
+	// Detect Weave bridge link index
+	link, err := netlink.LinkByName(wnet.WeaveBridgeName)
+	if err != nil {
+		return nil, fmt.Errorf("cannot find \"%s\" interface: %s", wnet.WeaveBridgeName, err)
+	}
+	t.linkIndex = link.Attrs().Index
+
+	t.infof("AWSVPC has been initialized on %s instance for %s route table at %s region",
+		t.instanceID, t.routeTableID, region)
+
+	return t, nil
+}
+
+// HandleUpdate method updates the AWS VPC and the host route tables.
+func (t *AWSVPCTracker) HandleUpdate(prevRanges, currRanges []address.Range, local bool) error {
+	t.debugf("replacing %q by %q; local(%t)", prevRanges, currRanges, local)
+
+	prev, curr := removeCommon(address.NewCIDRs(merge(prevRanges)), address.NewCIDRs(merge(currRanges)))
+
+	// It might make sense to do the removal first and then add entries
+	// because of the 50 routes limit. However, in such case a container might
+	// not be reachable for a short period of time which is not a desired behavior.
+
+	// Add new entries
+	for _, cidr := range curr {
+		cidrStr := cidr.String()
+		t.debugf("adding route %s to %s", cidrStr, t.instanceID)
+		_, err := t.createVPCRoute(cidrStr)
+		if err != nil {
+			return fmt.Errorf("createVPCRoutes failed: %s", err)
+		}
+		if local {
+			err = t.createHostRoute(cidrStr)
+			if err != nil {
+				return fmt.Errorf("createHostRoute failed: %s", err)
+			}
+		}
+	}
+
+	// Remove obsolete entries
+	for _, cidr := range prev {
+		cidrStr := cidr.String()
+		t.debugf("removing %s route", cidrStr)
+		_, err := t.deleteVPCRoute(cidrStr)
+		if err != nil {
+			return fmt.Errorf("deleteVPCRoute failed: %s", err)
+		}
+		if local {
+			err = t.deleteHostRoute(cidrStr)
+			if err != nil {
+				return fmt.Errorf("deleteHostRoute failed: %s", err)
+			}
+		}
+	}
+
+	return nil
+}
+
+func (t *AWSVPCTracker) createVPCRoute(cidr string) (*ec2.CreateRouteOutput, error) {
+	route := &ec2.CreateRouteInput{
+		RouteTableId:         &t.routeTableID,
+		InstanceId:           &t.instanceID,
+		DestinationCidrBlock: &cidr,
+	}
+	return t.ec2.CreateRoute(route)
+}
+
+func (t *AWSVPCTracker) createHostRoute(cidr string) error {
+	dst, err := parseCIDR(cidr)
+	if err != nil {
+		return err
+	}
+	route := &netlink.Route{
+		LinkIndex: t.linkIndex,
+		Dst:       dst,
+		Scope:     netlink.SCOPE_LINK,
+	}
+	return netlink.RouteAdd(route)
+}
+
+func (t *AWSVPCTracker) deleteVPCRoute(cidr string) (*ec2.DeleteRouteOutput, error) {
+	route := &ec2.DeleteRouteInput{
+		RouteTableId:         &t.routeTableID,
+		DestinationCidrBlock: &cidr,
+	}
+	return t.ec2.DeleteRoute(route)
+}
+
+func (t *AWSVPCTracker) deleteHostRoute(cidr string) error {
+	dst, err := parseCIDR(cidr)
+	if err != nil {
+		return err
+	}
+	route := &netlink.Route{
+		LinkIndex: t.linkIndex,
+		Dst:       dst,
+		Scope:     netlink.SCOPE_LINK,
+	}
+	return netlink.RouteDel(route)
+}
+
+// detectRouteTableID detects AWS VPC Route Table ID of the given tracker instance.
+func (t *AWSVPCTracker) detectRouteTableID() (*string, error) {
+	instancesParams := &ec2.DescribeInstancesInput{
+		InstanceIds: []*string{aws.String(t.instanceID)},
+	}
+	instancesResp, err := t.ec2.DescribeInstances(instancesParams)
+	if err != nil {
+		return nil, fmt.Errorf("DescribeInstances failed: %s", err)
+	}
+	if len(instancesResp.Reservations) == 0 ||
+		len(instancesResp.Reservations[0].Instances) == 0 {
+		return nil, fmt.Errorf("cannot find %s instance within reservations", t.instanceID)
+	}
+	vpcID := instancesResp.Reservations[0].Instances[0].VpcId
+	subnetID := instancesResp.Reservations[0].Instances[0].SubnetId
+
+	// First try to find a routing table associated with the subnet of the instance
+	tablesParams := &ec2.DescribeRouteTablesInput{
+		Filters: []*ec2.Filter{
+			{
+				Name:   aws.String("association.subnet-id"),
+				Values: []*string{subnetID},
+			},
+		},
+	}
+	tablesResp, err := t.ec2.DescribeRouteTables(tablesParams)
+	if err != nil {
+		return nil, fmt.Errorf("DescribeRouteTables failed: %s", err)
+	}
+	if len(tablesResp.RouteTables) != 0 {
+		return tablesResp.RouteTables[0].RouteTableId, nil
+	}
+	// Fallback to the default routing table
+	tablesParams = &ec2.DescribeRouteTablesInput{
+		Filters: []*ec2.Filter{
+			{
+				Name:   aws.String("association.main"),
+				Values: []*string{aws.String("true")},
+			},
+			{
+				Name:   aws.String("vpc-id"),
+				Values: []*string{vpcID},
+			},
+		},
+	}
+	tablesResp, err = t.ec2.DescribeRouteTables(tablesParams)
+	if err != nil {
+		return nil, fmt.Errorf("DescribeRouteTables failed: %s", err)
+	}
+	if len(tablesResp.RouteTables) != 0 {
+		return tablesResp.RouteTables[0].RouteTableId, nil
+	}
+
+	return nil, fmt.Errorf("cannot find routetable for %s instance", t.instanceID)
+}
+
+func (t *AWSVPCTracker) debugf(fmt string, args ...interface{}) {
+	common.Log.Debugf("[tracker] "+fmt, args...)
+}
+
+func (t *AWSVPCTracker) infof(fmt string, args ...interface{}) {
+	common.Log.Infof("[tracker] "+fmt, args...)
+}
+
+// Helpers
+
+// merge merges adjacent range entries.
+// The given slice has to be sorted in increasing order.
+func merge(r []address.Range) []address.Range {
+	var merged []address.Range
+
+	for i := range r {
+		if prev := len(merged) - 1; prev >= 0 && merged[prev].End == r[i].Start {
+			merged[prev].End = r[i].End
+		} else {
+			merged = append(merged, r[i])
+		}
+	}
+
+	return merged
+}
+
+// removeCommon filters out CIDR ranges which are contained in both a and b slices.
+// Both slices have to be sorted in increasing order.
+func removeCommon(a, b []address.CIDR) (newA, newB []address.CIDR) {
+	i, j := 0, 0
+
+	for i < len(a) && j < len(b) {
+		switch {
+		case a[i].Start() < b[j].Start() || a[i].End() < b[j].End():
+			newA = append(newA, a[i])
+			i++
+		case a[i].Start() > b[j].Start() || a[i].End() > b[j].End():
+			newB = append(newB, b[j])
+			j++
+		default:
+			i++
+			j++
+		}
+
+	}
+	newA = append(newA, a[i:]...)
+	newB = append(newB, b[j:]...)
+
+	return
+}
+
+func parseCIDR(cidr string) (*net.IPNet, error) {
+	ip, ipnet, err := net.ParseCIDR(cidr)
+	if err != nil {
+		return nil, err
+	}
+	ipnet.IP = ip
+
+	return ipnet, nil
+}

ipam/tracker/awsvpc_test.go

@@ -0,0 +1,52 @@
+package tracker
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/weaveworks/weave/net/address"
+)
+
+var (
+	r0to127     = cidr("10.0.0.0", "10.0.0.127")
+	r128to255   = cidr("10.0.0.128", "10.0.0.255")
+	r0to255     = cidr("10.0.0.0", "10.0.0.255")
+	r1dot0to255 = cidr("10.0.1.0", "10.0.1.255")
+	r2dot0to255 = cidr("10.0.2.0", "10.0.2.255")
+)
+
+func TestRemoveCommon(t *testing.T) {
+	a := []address.CIDR{r0to127, r1dot0to255}
+	b := []address.CIDR{r1dot0to255, r2dot0to255}
+	newA, newB := removeCommon(a, b)
+	require.Equal(t, []address.CIDR{r0to127}, newA)
+	require.Equal(t, []address.CIDR{r2dot0to255}, newB)
+}
+
+func TestMerge(t *testing.T) {
+	ranges := []address.Range{
+		r0to127.Range(),
+		r128to255.Range(),
+		r2dot0to255.Range(),
+	}
+	require.Equal(t, []address.Range{r0to255.Range(), r2dot0to255.Range()}, merge(ranges))
+}
+
+// Helper
+
+// TODO(mp) DRY with helpers of other tests.
+
+func ip(s string) address.Address {
+	addr, _ := address.ParseIP(s)
+	return addr
+}
+
+// [start; end]
+func cidr(start, end string) address.CIDR {
+	c := address.Range{Start: ip(start), End: ip(end) + 1}.CIDRs()
+	if len(c) != 1 {
+		panic("invalid cidr")
+	}
+	return c[0]
+}

ipam/tracker/tracker.go

@@ -0,0 +1,20 @@
+package tracker
+
+import (
+	"github.com/weaveworks/weave/net/address"
+)
+
+// LocalRangeTracker is an interface for tracking changes in the IPAM ring.
+type LocalRangeTracker interface {
+	// HandleUpdate is called whenever an address ring gets updated.
+	//
+	// prevRanges corresponds to ranges which were owned by a peer before
+	// a change in the ring, while currRanges to the ones which are currently
+	// owned by the peer.
+	// Both slices have to be sorted in increasing order.
+	// Adjacent ranges within each slice might appear as separate ranges.
+	//
+	// The local parameter indicates whether the ranges belong to the peer
+	// by which the method is called.
+	HandleUpdate(prevRanges, currRanges []address.Range, local bool) error
+}

net/address/address.go

@@ -73,6 +73,40 @@ type CIDR struct {
 	PrefixLen int
 }
 
+// CIDRs returns a list of CIDR-aligned ranges which cover this range.
+func (r Range) CIDRs() []CIDR {
+	const (
+		fullMask         = ^Address(0)
+		cidrMaxPrefixLen = 32
+	)
+	var cidrs []CIDR
+
+	for start, end := r.Start, r.End-1; end >= start; {
+		mask, prefixLen := fullMask, cidrMaxPrefixLen
+		// Find the smallest mask which would cover some part of [start;end].
+		// Once we found such, apply it by OR'ing
+		for mask > 0 {
+			tmpMask := mask << 1
+			// Check whether mask neither too short nor too long
+			if (start&tmpMask) != start || (start|^tmpMask) > end {
+				break
+			}
+			mask = tmpMask
+			prefixLen--
+		}
+		cidrs = append(cidrs, CIDR{start, prefixLen})
+		// Apply mask
+		start |= ^mask
+		// Check for overflow
+		if start+1 < start {
+			break
+		}
+		start++
+	}
+
+	return cidrs
+}
+
 func ParseIP(s string) (Address, error) {
 	if ip := net.ParseIP(s); ip != nil {
 		return FromIP4(ip), nil
@@ -91,6 +125,22 @@ func ParseCIDR(s string) (CIDR, error) {
 	}
 }
 
+func NewCIDRs(ranges []Range) (cidrs []CIDR) {
+	for _, r := range ranges {
+		cidrs = append(cidrs, r.CIDRs()...)
+	}
+	return cidrs
+}
+
+func (cidr CIDR) Start() Address {
+	return cidr.Addr
+}
+
+// cidr = [Start; End)
+func (cidr CIDR) End() Address {
+	return cidr.Range().End
+}
+
 func (cidr CIDR) IsSubnet() bool {
 	mask := cidr.Size() - 1
 	return Offset(cidr.Addr)&mask == 0

net/address/address_test.go

@@ -39,3 +39,31 @@ func TestBiggestPow2AlignedRange(t *testing.T) {
 	}
 	require.NoError(t, quick.Check(prop, &quick.Config{MaxCount: 1000000}))
 }
+
+func ip(s string) Address {
+	addr, _ := ParseIP(s)
+	return addr
+}
+
+func cidr(s string) CIDR {
+	c, err := ParseCIDR(s)
+	if err != nil {
+		panic(err)
+	}
+	return c
+}
+
+func TestCIDRs(t *testing.T) {
+	start := ip("10.0.0.1")
+	end := ip("10.0.0.9")
+	r := NewRange(start, Subtract(end, start))
+	require.Equal(t,
+		[]CIDR{cidr("10.0.0.1/32"), cidr("10.0.0.2/31"), cidr("10.0.0.4/30"), cidr("10.0.0.8/32")},
+		r.CIDRs())
+}
+
+func TestCIDRStartAndEnd(t *testing.T) {
+	cidr, _ := ParseCIDR("10.0.0.0/24")
+	require.Equal(t, ip("10.0.0.0"), cidr.Start(), "")
+	require.Equal(t, ip("10.0.1.0"), cidr.End(), "")
+}

net/veth.go

@@ -13,7 +13,7 @@ import (
 )
 
 // create and attach a veth to the Weave bridge
-func CreateAndAttachVeth(name, peerName, bridgeName string, mtu int, init func(peer netlink.Link) error) (*netlink.Veth, error) {
+func CreateAndAttachVeth(name, peerName, bridgeName string, mtu int, keepTXOn bool, init func(peer netlink.Link) error) (*netlink.Veth, error) {
 	bridge, err := netlink.LinkByName(bridgeName)
 	if err != nil {
 		return nil, fmt.Errorf(`bridge "%s" not present; did you launch weave?`, bridgeName)
@@ -42,7 +42,7 @@ func CreateAndAttachVeth(name, peerName, bridgeName string, mtu int, init func(p
 		if err := netlink.LinkSetMasterByIndex(veth, bridge.Attrs().Index); err != nil {
 			return cleanup(`unable to set master of %s: %s`, name, err)
 		}
-		if bridgeType == Bridge {
+		if bridgeType == Bridge && !keepTXOn {
 			if err := EthtoolTXOff(peerName); err != nil {
 				return cleanup(`unable to set tx off on %q: %s`, peerName, err)
 			}
@@ -111,14 +111,14 @@ func interfaceExistsInNamespace(ns netns.NsHandle, ifName string) bool {
 	return err == nil
 }
 
-func AttachContainer(ns netns.NsHandle, id, ifName, bridgeName string, mtu int, withMulticastRoute bool, cidrs []*net.IPNet) error {
+func AttachContainer(ns netns.NsHandle, id, ifName, bridgeName string, mtu int, withMulticastRoute bool, cidrs []*net.IPNet, keepTXOn bool) error {
 	if !interfaceExistsInNamespace(ns, ifName) {
 		maxIDLen := IFNAMSIZ - 1 - len(vethPrefix+"pl")
 		if len(id) > maxIDLen {
 			id = id[:maxIDLen] // trim passed ID if too long
 		}
 		name, peerName := vethPrefix+"pl"+id, vethPrefix+"pg"+id
-		_, err := CreateAndAttachVeth(name, peerName, bridgeName, mtu, func(veth netlink.Link) error {
+		_, err := CreateAndAttachVeth(name, peerName, bridgeName, mtu, keepTXOn, func(veth netlink.Link) error {
 			if err := netlink.LinkSetNsFd(veth, int(ns)); err != nil {
 				return fmt.Errorf("failed to move veth to container netns: %s", err)
 			}

plugin/net/cni.go

@@ -92,7 +92,7 @@ func (c *CNIPlugin) CmdAdd(args *skel.CmdArgs) error {
 		id = fmt.Sprintf("%x", data)
 	}
 
-	if err := weavenet.AttachContainer(ns, id, args.IfName, conf.BrName, conf.MTU, false, []*net.IPNet{&result.IP4.IP}); err != nil {
+	if err := weavenet.AttachContainer(ns, id, args.IfName, conf.BrName, conf.MTU, false, []*net.IPNet{&result.IP4.IP}, false); err != nil {
 		return err
 	}
 	if err := weavenet.WithNetNSLink(ns, args.IfName, func(link netlink.Link) error {

plugin/net/driver.go

@@ -134,8 +134,9 @@ func (driver *driver) JoinEndpoint(j *api.JoinRequest) (*api.JoinResponse, error
 	if err != nil {
 		return nil, driver.error("JoinEndpoint", "unable to find network info: %s", err)
 	}
+
 	name, peerName := vethPair(j.EndpointID)
-	if _, err := weavenet.CreateAndAttachVeth(name, peerName, weavenet.WeaveBridgeName, 0, nil); err != nil {
+	if _, err := weavenet.CreateAndAttachVeth(name, peerName, weavenet.WeaveBridgeName, 0, false, nil); err != nil {
 		return nil, driver.error("JoinEndpoint", "%s", err)
 	}
 

prog/weaver/main.go

@@ -22,6 +22,7 @@ import (
 	"github.com/weaveworks/weave/common/docker"
 	"github.com/weaveworks/weave/db"
 	"github.com/weaveworks/weave/ipam"
+	"github.com/weaveworks/weave/ipam/tracker"
 	"github.com/weaveworks/weave/nameserver"
 	weavenet "github.com/weaveworks/weave/net"
 	"github.com/weaveworks/weave/net/address"
@@ -170,6 +171,7 @@ func main() {
 		datapathName       string
 		trustedSubnetStr   string
 		dbPrefix           string
+		isAWSVPC           bool
 
 		defaultDockerHost = "unix:///var/run/docker.sock"
 	)
@@ -208,6 +210,7 @@ func main() {
 	mflag.StringVar(&datapathName, []string{"-datapath"}, "", "ODP datapath name")
 	mflag.StringVar(&trustedSubnetStr, []string{"-trusted-subnets"}, "", "comma-separated list of trusted subnets in CIDR notation")
 	mflag.StringVar(&dbPrefix, []string{"-db-prefix"}, "/weavedb/weave", "pathname/prefix of filename to store data")
+	mflag.BoolVar(&isAWSVPC, []string{"#awsvpc", "-awsvpc"}, false, "use AWS VPC for routing")
 
 	// crude way of detecting that we probably have been started in a
 	// container, with `weave launch` --> suppress misleading paths in
@@ -250,7 +253,7 @@ func main() {
 		networkConfig.PacketLogging = nopPacketLogging{}
 	}
 
-	overlay, bridge := createOverlay(datapathName, ifaceName, config.Host, config.Port, bufSzMB)
+	overlay, bridge := createOverlay(datapathName, ifaceName, isAWSVPC, config.Host, config.Port, bufSzMB)
 	networkConfig.Bridge = bridge
 
 	name := peerName(routerName, bridge.Interface())
@@ -300,9 +303,19 @@ func main() {
 	var (
 		allocator     *ipam.Allocator
 		defaultSubnet address.CIDR
+		trackerName   string
 	)
 	if ipamConfig.Enabled() {
-		allocator, defaultSubnet = createAllocator(router, ipamConfig, db, isKnownPeer)
+		var t tracker.LocalRangeTracker
+		if isAWSVPC {
+			Log.Infoln("Creating AWSVPC LocalRangeTracker")
+			t, err = tracker.NewAWSVPCTracker()
+			if err != nil {
+				Log.Fatalf("Cannot create AWSVPC LocalRangeTracker: %s", err)
+			}
+			trackerName = "awsvpc"
+		}
+		allocator, defaultSubnet = createAllocator(router, ipamConfig, db, t, isKnownPeer)
 		observeContainers(allocator)
 		ids, err := dockerCli.AllContainerIDs()
 		checkFatal(err)
@@ -333,7 +346,7 @@ func main() {
 	if httpAddr != "" {
 		muxRouter := mux.NewRouter()
 		if allocator != nil {
-			allocator.HandleHTTP(muxRouter, defaultSubnet, dockerCli)
+			allocator.HandleHTTP(muxRouter, defaultSubnet, trackerName, dockerCli)
 		}
 		if ns != nil {
 			ns.HandleHTTP(muxRouter, dockerCli)
@@ -388,10 +401,18 @@ func (nopPacketLogging) LogPacket(string, weave.PacketKey) {
 func (nopPacketLogging) LogForwardPacket(string, weave.ForwardPacketKey) {
 }
 
-func createOverlay(datapathName string, ifaceName string, host string, port int, bufSzMB int) (weave.NetworkOverlay, weave.Bridge) {
+func createOverlay(datapathName string, ifaceName string, isAWSVPC bool, host string, port int, bufSzMB int) (weave.NetworkOverlay, weave.Bridge) {
 	overlay := weave.NewOverlaySwitch()
 	var bridge weave.Bridge
+	var ignoreSleeve bool
+
 	switch {
+	case isAWSVPC:
+		vpc := weave.NewAWSVPC()
+		overlay.Add("awsvpc", vpc)
+		bridge = weave.NullBridge{}
+		// Currently, we do not support any overlay with AWSVPC
+		ignoreSleeve = true
 	case datapathName != "" && ifaceName != "":
 		Log.Fatal("At most one of --datapath and --iface must be specified.")
 	case datapathName != "":
@@ -409,13 +430,17 @@ func createOverlay(datapathName string, ifaceName string, host string, port int,
 	default:
 		bridge = weave.NullBridge{}
 	}
-	sleeve := weave.NewSleeveOverlay(host, port)
-	overlay.Add("sleeve", sleeve)
-	overlay.SetCompatOverlay(sleeve)
+
+	if !ignoreSleeve {
+		sleeve := weave.NewSleeveOverlay(host, port)
+		overlay.Add("sleeve", sleeve)
+		overlay.SetCompatOverlay(sleeve)
+	}
+
 	return overlay, bridge
 }
 
-func createAllocator(router *weave.NetworkRouter, config ipamConfig, db db.DB, isKnownPeer func(mesh.PeerName) bool) (*ipam.Allocator, address.CIDR) {
+func createAllocator(router *weave.NetworkRouter, config ipamConfig, db db.DB, track tracker.LocalRangeTracker, isKnownPeer func(mesh.PeerName) bool) (*ipam.Allocator, address.CIDR) {
 	ipRange, err := ipam.ParseCIDRSubnet(config.IPRangeCIDR)
 	checkFatal(err)
 	defaultSubnet := ipRange
@@ -437,6 +462,7 @@ func createAllocator(router *weave.NetworkRouter, config ipamConfig, db db.DB, i
 		Quorum:      func() uint { return determineQuorum(config.PeerCount, router) },
 		Db:          db,
 		IsKnownPeer: isKnownPeer,
+		Tracker:     track,
 	}
 
 	allocator := ipam.NewAllocator(c)

prog/weaveutil/attach.go

@@ -14,13 +14,22 @@ import (
 
 func attach(args []string) error {
 	if len(args) < 4 {
-		cmdUsage("attach-container", "[--no-multicast-route] <container-id> <bridge-name> <mtu> <cidr>...")
+		cmdUsage("attach-container", "[--no-multicast-route] [--keep-tx-on] <container-id> <bridge-name> <mtu> <cidr>...")
 	}
 
+	keepTXOn := false
 	withMulticastRoute := true
-	if args[0] == "--no-multicast-route" {
-		withMulticastRoute = false
-		args = args[1:]
+	for i := 0; i < len(args); {
+		switch args[i] {
+		case "--no-multicast-route":
+			withMulticastRoute = false
+			args = append(args[:i], args[i+1:]...)
+		case "--keep-tx-on":
+			keepTXOn = true
+			args = append(args[:i], args[i+1:]...)
+		default:
+			i++
+		}
 	}
 
 	pid, nsContainer, err := containerPidAndNs(args[0])
@@ -40,7 +49,8 @@ func attach(args []string) error {
 	if err != nil {
 		return err
 	}
-	err = weavenet.AttachContainer(nsContainer, fmt.Sprint(pid), weavenet.VethName, args[1], mtu, withMulticastRoute, cidrs)
+
+	err = weavenet.AttachContainer(nsContainer, fmt.Sprint(pid), weavenet.VethName, args[1], mtu, withMulticastRoute, cidrs, keepTXOn)
 	// If we detected an error but the container has died, tell the user that instead.
 	if err != nil && !processExists(pid) {
 		err = fmt.Errorf("Container %s died", args[0])

router/awsvpc.go

@@ -0,0 +1,75 @@
+package router
+
+// A dummy overlay for the AWSVPC underlay to make `weave status` to return
+// a valid information about peer connections.
+
+import (
+	"github.com/weaveworks/mesh"
+)
+
+// mesh.OverlayConnection
+
+type AWSVPCConnection struct {
+	establishedChan chan struct{}
+	errorChan       chan error
+}
+
+func (conn *AWSVPCConnection) Confirm() {
+	// We close the channel to notify mesh that the connection has been established.
+	close(conn.establishedChan)
+}
+
+func (conn *AWSVPCConnection) EstablishedChannel() <-chan struct{} {
+	return conn.establishedChan
+}
+
+func (conn *AWSVPCConnection) ErrorChannel() <-chan error {
+	return conn.errorChan
+}
+
+func (conn *AWSVPCConnection) Stop() {}
+
+func (conn *AWSVPCConnection) ControlMessage(tag byte, msg []byte) {
+}
+
+func (conn *AWSVPCConnection) DisplayName() string {
+	return "awsvpc"
+}
+
+// OverlayForwarder
+
+func (conn *AWSVPCConnection) Forward(key ForwardPacketKey) FlowOp {
+	return DiscardingFlowOp{}
+}
+
+type AWSVPC struct{}
+
+func NewAWSVPC() AWSVPC {
+	return AWSVPC{}
+}
+
+// mesh.Overlay
+
+func (vpc AWSVPC) AddFeaturesTo(features map[string]string) {}
+
+func (vpc AWSVPC) PrepareConnection(params mesh.OverlayConnectionParams) (mesh.OverlayConnection, error) {
+	conn := &AWSVPCConnection{
+		establishedChan: make(chan struct{}),
+		errorChan:       make(chan error, 1),
+	}
+	return conn, nil
+}
+
+func (vpc AWSVPC) Diagnostics() interface{} {
+	return nil
+}
+
+// NetworkOverlay
+
+func (vpc AWSVPC) InvalidateRoutes() {}
+
+func (vpc AWSVPC) InvalidateShortIDs() {}
+
+func (vpc AWSVPC) StartConsumingPackets(localPeer *mesh.Peer, peers *mesh.Peers, consumer OverlayConsumer) error {
+	return nil
+}

router/bridge.go

@@ -36,7 +36,7 @@ func (NullBridge) Interface() *net.Interface {
 }
 
 func (NullBridge) String() string {
-	return "<no bridge networking>"
+	return "no overlay bridge"
 }
 
 func (NullBridge) Stats() map[string]int {

weave

@@ -181,6 +181,7 @@ exec_remote() {
         -e PROXY_HOST="$PROXY_HOST" \
         -e COVERAGE \
         -e CHECKPOINT_DISABLE \
+        -e AWSVPC   \
         $WEAVEEXEC_DOCKER_ARGS $EXEC_IMAGE --local "$@"
 }
 
@@ -645,7 +646,7 @@ EOF
         fi
     fi
 
-    [ "$1" = "--without-ethtool" ] || ethtool_tx_off_$BRIDGE_TYPE $BRIDGE
+    [ "$1" = "--without-ethtool" -o -n "$AWSVPC" ] || ethtool_tx_off_$BRIDGE_TYPE $BRIDGE
 
     ip link set dev $BRIDGE up
 
@@ -660,6 +661,12 @@ expose_ip() {
         if ! ip addr show dev $BRIDGE | grep -qF $CIDR ; then
             ip addr add dev $BRIDGE $CIDR
             arp_update $BRIDGE $CIDR || true
+            # Remove a default route installed by the kernel, because awsvpc
+            # has installed it as well
+            if [ -n "$AWSVPC" ]; then
+                RCIDR=$(ip route list exact $CIDR proto kernel | head -n1 | cut -d' ' -f1)
+                [ -n "$RCIDR" ] && ip route del dev $BRIDGE proto kernel $RCIDR
+            fi
         fi
         [ -z "$FQDN" ] || when_weave_running put_dns_fqdn_no_check_alive weave:expose $FQDN $CIDR
     done
@@ -886,6 +893,8 @@ container_in_host_ns() {
 attach() {
     ATTACH_ARGS=""
     [ -n "$NO_MULTICAST_ROUTE" ] && ATTACH_ARGS="--no-multicast-route"
+    # Relying on AWSVPC being set in 'ipam_cidrs allocate', except for 'weave restart'
+    [ -n "$AWSVPC" ] && ATTACH_ARGS="--no-multicast-route --keep-tx-on"
     util_op attach-container $ATTACH_ARGS $CONTAINER $BRIDGE $MTU "$@"
 }
 
@@ -1261,6 +1270,12 @@ ipam_reclaim_no_check_alive() {
     done
 }
 
+detect_awsvpc() {
+    # Ignoring errors here: if we cannot detect AWSVPC we will skip the relevant
+    # steps, because "attach" should work without the weave router running.
+    [ "$(call_weave GET /ipinfo/tracker)" != "awsvpc" ] || AWSVPC=1
+}
+
 # Call IPAM as necessary to lookup or allocate addresses
 #
 # $1 is one of 'lookup', 'allocate' or 'allocate_no_check_alive', $2
@@ -1277,6 +1292,11 @@ ipam_cidrs() {
         allocate)
             METHOD=POST
             CHECK_ALIVE="?check-alive=true"
+            detect_awsvpc
+            if [ -n "$AWSVPC" -a $# -gt 2 ] ; then
+                echo "Error: no IP addresses or subnets may be specified in AWSVPC mode" >&2
+                return 1
+            fi
             ;;
         allocate_no_check_alive)
             METHOD=POST
@@ -1501,22 +1521,19 @@ launch_router() {
     LAUNCHING_ROUTER=1
     check_forwarding_rules
     enforce_docker_bridge_addr_assign_type
-    create_bridge
-    docker_bridge_ip
-    # We set the router name to the bridge MAC, which in turn is
-    # derived from the system UUID (if available), and thus stable
-    # across reboots.
-    PEERNAME=$(cat /sys/class/net/$BRIDGE/address)
+
     # backward compatibility...
     if is_cidr "$1" ; then
         echo "WARNING: $1 parameter ignored; 'weave launch' no longer takes a CIDR as the first parameter" >&2
         shift 1
     fi
+
     CONTAINER_PORT=$PORT
     ARGS=
     IPRANGE=
     IPRANGE_SPECIFIED=
 
+    docker_bridge_ip
     DNS_ROUTER_OPTS="--dns-listen-address $DOCKER_BRIDGE_IP:53"
     NO_DNS_OPT=
 
@@ -1558,6 +1575,10 @@ launch_router() {
             --no-restart)
                 RESTART_POLICY=
                 ;;
+            --awsvpc)
+                AWSVPC_ARGS="--awsvpc"
+                AWSVPC=1
+                ;;
             *)
                 ARGS="$ARGS '$(echo "$1" | sed "s|'|'\"'\"'|g")'"
                 ;;
@@ -1565,6 +1586,13 @@ launch_router() {
         shift
     done
     eval "set -- $ARGS"
+
+    create_bridge
+    # We set the router name to the bridge MAC, which in turn is
+    # derived from the system UUID (if available), and thus stable
+    # across reboots.
+    PEERNAME=$(cat /sys/class/net/$BRIDGE/address)
+
     if [ -z "$IPRANGE_SPECIFIED" ] ; then
         IPRANGE="10.32.0.0/12"
         if ! check_overlap $IPRANGE ; then
@@ -1601,11 +1629,23 @@ launch_router() {
         --ipalloc-range "$IPRANGE" \
         --dns-effective-listen-address $DOCKER_BRIDGE_IP \
         $DNS_ROUTER_OPTS $NO_DNS_OPT \
+        $AWSVPC_ARGS \
         --http-addr $HTTP_ADDR \
         "$@")
     setup_router_iface_$BRIDGE_TYPE
     wait_for_status $CONTAINER_NAME http_call $HTTP_ADDR
     populate_router
+    if [ -n "$AWSVPC" ]; then
+        expose_ip
+        # Set proxy_arp on the bridge, so that it could accept packets destined
+        # to containers within the same subnet but running on remote hosts.
+        # Without it, exact routes on each container are required.
+        echo 1 >/proc/sys/net/ipv4/conf/$BRIDGE/proxy_arp
+        # Avoid delaying the first ARP request. Also, setting it to 0 avoids
+        # placing the request into a bounded queue as it can be seen:
+        # https://git.kernel.org/cgit/linux/kernel/git/stable/linux-stable.git/tree/net/ipv4/arp.c?id=refs/tags/v4.6.1#n819
+        echo 0 >/proc/sys/net/ipv4/neigh/$BRIDGE/proxy_delay
+    fi
 }
 
 # Recreate the parameter values that are set when the router is first launched
@@ -1649,6 +1689,8 @@ launch_proxy() {
        docker create -v /w -v /w-noop -v /w-nomcast --name=$VOLUMES_CONTAINER_NAME \
            --label=weavevolumes --entrypoint=/bin/false $EXEC_IMAGE >/dev/null
     fi
+    detect_awsvpc
+    [ -n "$AWSVPC" ] && PROXY_ARGS="$PROXY_ARGS --no-multicast-route"
     PROXY_CONTAINER=$(docker run -d --name=$PROXY_CONTAINER_NAME \
         $(docker_run_options) \
         $RESTART_POLICY \
@@ -2070,6 +2112,7 @@ EOF
         for CIDR in $ALL_CIDRS ; do
             call_weave PUT /ip/$CONTAINER/$CIDR?check-alive=true
         done
+        detect_awsvpc
         do_or_die $CONTAINER attach $ALL_CIDRS
         when_weave_running with_container_fqdn $CONTAINER put_dns_fqdn $ALL_CIDRS
         echo $RES