[WIP] 2017: AWSVPC

.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

TODO.md

@@ -0,0 +1,17 @@
+# AWSVPC
+
+# Urgent
+
+* Add the subnet checks / keep checksum offloading for the plugin
+* Test with the plugin
+
+* Test with multiple addr from the default subnet
+
+* Create new SSH keys or reuse the same for each test run
+
+# Nice To Have
+
+* What happens if $RCIDR is not found
+* Do fuzz testing
+* Read on VPC routing tables again.
+* create a separate VPC for each test run

api/ipam.go

@@ -53,6 +53,14 @@ func (client *Client) DefaultSubnet() (*net.IPNet, error) {
 	return ipnet, err
 }
 
+func (client *Client) LocalRangeTracker() (string, error) {
+	t, err := client.httpVerb("GET", fmt.Sprintf("/ipinfo/tracker"), nil)
+	if err != nil {
+		return "", err
+	}
+	return t, nil
+}
+
 func parseIP(body string) (*net.IPNet, error) {
 	ip, ipnet, err := net.ParseCIDR(string(body))
 	if err != nil {

circle.yml

@@ -12,6 +12,7 @@ machine:
     SRCDIR: /home/ubuntu/src/github.com/weaveworks/weave
     PATH: $PATH:$HOME/.local/bin:$HOME/google-cloud-sdk/bin/
     CLOUDSDK_CORE_DISABLE_PROMPTS: 1
+    AWS_DEFAULT_REGION: us-east-1
 
 dependencies:
   cache_directories:
@@ -22,7 +23,9 @@ dependencies:
     - bin/setup-circleci-secrets "$SECRET_PASSWORD"
     - "mkdir -p $(dirname $SRCDIR) && cp -r $(pwd)/ $SRCDIR"
     - "cd $SRCDIR; git submodule update --init"
-    - cd $SRCDIR/test; ./gce.sh make_template:
+    #- cd $SRCDIR/test; ./gce.sh make_template:
+    #    parallel: false
+    - cd $SRCDIR/test; ./aws.sh make_template:
         parallel: false
     - "cd $SRCDIR/build; ../tools/rebuild-image weaveworks/weavebuild . Dockerfile build.sh && touch $SRCDIR/.build.uptodate"
 
@@ -36,14 +39,22 @@ test:
         parallel: true
     - cd $SRCDIR; make testrunner:
         parallel: true
-    - cd $SRCDIR/test; ./gce.sh setup && eval $(./gce.sh hosts) && ./setup.sh:
-        parallel: true
-    - cd $SRCDIR/test; eval $(./gce.sh hosts); export COVERAGE=true; ./run_all.sh:
-        parallel: true
+    #- cd $SRCDIR/test; ./gce.sh setup && eval $(./gce.sh hosts) && ./setup.sh:
+    #    parallel: true
+    #- cd $SRCDIR/test; eval $(./gce.sh hosts); export COVERAGE=true; ./run_all.sh:
+    #    parallel: true
+    #    timeout: 300
+    - cd $SRCDIR/test; ./aws.sh setup && eval $(./aws.sh hosts) && ./setup.sh:
+        parallel: false
+    - cd $SRCDIR/test; eval $(./aws.sh hosts); export COVERAGE=true; AWS=1 ./run_all.sh:
+        parallel: false
         timeout: 300
+
   post:
-    - cd $SRCDIR/test; ./gce.sh destroy:
-        parallel: true
+    #- cd $SRCDIR/test; ./gce.sh destroy:
+    #    parallel: true
+    - cd $SRCDIR/test; ./aws.sh destroy:
+        parallel: false
     - test "$CIRCLE_NODE_INDEX" != "0" || (cd $SRCDIR/test; ./gen_coverage_reports.sh):
         parallel: true
 

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,23 +86,34 @@ 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()
 	} else {
 		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/allocator_test.go

@@ -15,7 +15,7 @@ import (
 )
 
 const (
-	testStart1 = "10.0.1.0"
+	testStart1 = "10.0.0.0"
 	testStart2 = "10.0.2.0"
 	testStart3 = "10.0.3.0"
 )
@@ -644,3 +644,156 @@ func cidrRanges(s string) []address.Range {
 	c, _ := address.ParseCIDR(s)
 	return []address.Range{c.Range()}
 }
+
+func TestTracker(t *testing.T) {
+	const (
+		cidr       = "10.0.0.0/30"
+		container1 = "container-1"
+		container2 = "container-2"
+	)
+
+	trackChan := make(chan rangePair, 10)
+	allocs, router, _ := makeNetworkOfAllocatorsWithTracker(2, cidr, newTestTracker(trackChan))
+	defer stopNetworkOfAllocators(allocs, router)
+
+	cidr1, _ := address.ParseCIDR(cidr)
+	addr1, err := allocs[0].Allocate(container1, cidr1, true, returnFalse)
+	require.Equal(t, nil, err, "")
+	require.Equal(t, ip("10.0.0.1"), addr1, "")
+
+	// Check HandleUpdate invocations. 2 of them should be invoked only with new ranges.
+	newPeer1 := false
+	newPeer2 := false
+	for i := 0; i < 3; i++ {
+		pair := <-trackChan
+		switch {
+		case !newPeer1 && pair.new[0].Equal(addrRange("10.0.0.0", "10.0.0.1")):
+			newPeer1 = true
+		case !newPeer2 && pair.new[0].Equal(addrRange("10.0.0.2", "10.0.0.3")):
+			newPeer2 = true
+		default:
+			continue
+		}
+	}
+	require.True(t, newPeer1 && newPeer2, "")
+
+	// The following allocation should trigger request for donation, because
+	// peer1 ran out of space.
+	addr2, err := allocs[0].Allocate(container2, cidr1, true, returnFalse)
+	require.Equal(t, nil, err, "")
+	require.Equal(t, ip("10.0.0.2"), addr2, "")
+
+	// Check whether HandleUpdate is invoked after donation.
+	newDonation1 := false
+	newDonation2 := false
+	for i := 0; i < 2; i++ {
+		pair := <-trackChan
+		switch {
+		case !newDonation1 &&
+			pair.old[0].Equal(addrRange("10.0.0.0", "10.0.0.1")) &&
+			pair.new[0].Equal(addrRange("10.0.0.0", "10.0.0.1")) &&
+			pair.new[1].Equal(addrRange("10.0.0.3", "10.0.0.3")):
+			// peer1
+
+			require.Equal(t, 1, len(pair.old), "")
+			require.Equal(t, 2, len(pair.new), "")
+			newDonation1 = true
+		case !newDonation2 &&
+			pair.old[0].Equal(addrRange("10.0.0.2", "10.0.0.3")) &&
+			pair.new[0].Equal(addrRange("10.0.0.2", "10.0.0.2")):
+			// peer2
+
+			require.Equal(t, 1, len(pair.old), "")
+			require.Equal(t, 1, len(pair.new), "")
+			newDonation2 = true
+		default:
+			continue
+		}
+	}
+	require.True(t, newDonation1 && newDonation2, "")
+}
+
+func TestShutdownWithTracker(t *testing.T) {
+	const (
+		cidr       = "10.0.0.0/30"
+		container1 = "container-1"
+	)
+
+	trackChan := make(chan rangePair, 10)
+	allocs, router, _ := makeNetworkOfAllocatorsWithTracker(2, cidr, newTestTracker(trackChan))
+	defer stopNetworkOfAllocators(allocs, router)
+
+	cidr1, _ := address.ParseCIDR(cidr)
+	_, err := allocs[0].Allocate(container1, cidr1, true, returnFalse)
+	require.NoError(t, err, "")
+
+	time.Sleep(500 * time.Millisecond)
+	flush(trackChan, 2)
+
+	// After allocation (and ring establishment):
+	// * peer1: [10.0.0.0-10.0.0.1]
+	// * peer2: [10.0.0.2-10.0.0.3]
+
+	// Shutdown peer1
+	allocs[0].Shutdown()
+
+	done := false
+	for i := 0; i < 5; i++ {
+		p := <-trackChan
+		switch {
+		// This should uniquely match HandleUpdate invocation on peer2 which
+		// happens after peer1 notified peer2 about the donation due to its
+		// termination:
+		case !done && len(p.old) == 1 && len(p.new) == 2:
+			require.Equal(t, addrRange("10.0.0.2", "10.0.0.3"), p.old[0], "")
+			require.Equal(t, addrRange("10.0.0.0", "10.0.0.1"), p.new[0], "")
+			require.Equal(t, addrRange("10.0.0.2", "10.0.0.3"), p.new[1], "")
+			// peer2 has received peer1's previously owned ranges
+			done = true
+		default:
+			continue
+		}
+	}
+	require.True(t, done)
+
+	// Shutdown peer2
+	allocs[1].Shutdown()
+	p := <-trackChan
+	require.Equal(t,
+		[]address.Range{addrRange("10.0.0.0", "10.0.0.1"), addrRange("10.0.0.2", "10.0.0.3")},
+		p.old)
+	require.Len(t, p.new, 0)
+}
+
+func flush(c chan rangePair, count int) {
+	for i := 0; i < count; i++ {
+		<-c
+	}
+}
+
+type testTracker struct {
+	trackChan chan rangePair
+}
+
+type rangePair struct {
+	old, new []address.Range
+}
+
+func newTestTracker(trackChan chan rangePair) *testTracker {
+	return &testTracker{trackChan}
+}
+
+func (t *testTracker) HandleUpdate(old, new []address.Range, local bool) error {
+	t.trackChan <- rangePair{old, new}
+	return nil
+}
+
+// Creates [start;end] address.Range.
+func addrRange(start, end string) address.Range {
+	return address.NewRange(ip(start), address.Subtract(ip(end)+1, ip(start)))
+}
+
+func ip(addr string) address.Address {
+	retAddr, _ := address.ParseIP(addr)
+	return retAddr
+}

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 {