Merge pull request #1971 from weaveworks/issues/678-persist-ipam2

Persist IPAM ring

.gitmodules

@@ -55,3 +55,6 @@
 [submodule "vendor/github.com/vishvananda/netns"]
 	path = vendor/github.com/vishvananda/netns
 	url = https://github.com/vishvananda/netns.git
+[submodule "vendor/github.com/boltdb/bolt"]
+	path = vendor/github.com/boltdb/bolt
+	url = https://github.com/boltdb/bolt

Makefile

@@ -63,7 +63,7 @@ testrunner: $(RUNNER_EXE) $(TEST_TLS_EXE)
 
 $(EXES): $(BUILD_UPTODATE)
 $(WEAVER_EXE) $(WEAVEPROXY_EXE) $(WEAVEUTIL_EXE): common/*.go common/*/*.go net/*.go net/*/*.go
-$(WEAVER_EXE): router/*.go ipam/*.go ipam/*/*.go nameserver/*.go prog/weaver/*.go
+$(WEAVER_EXE): router/*.go ipam/*.go ipam/*/*.go db/*.go nameserver/*.go prog/weaver/*.go
 $(WEAVEPROXY_EXE): proxy/*.go prog/weaveproxy/*.go
 $(WEAVEUTIL_EXE): prog/weaveutil/*.go
 $(SIGPROXY_EXE): prog/sigproxy/*.go

db/boltdb.go

@@ -0,0 +1,81 @@
+package db
+
+import (
+	"bytes"
+	"encoding/gob"
+	"fmt"
+
+	"github.com/boltdb/bolt"
+)
+
+type DB interface {
+	Load(string, interface{}) error
+	Save(string, interface{}) error
+}
+
+type BoltDB struct {
+	db *bolt.DB
+}
+
+var (
+	versionIdent       = []byte("version")
+	persistenceVersion = []byte{1, 0} // major.minor
+	topBucket          = []byte("top")
+)
+
+func NewBoltDB(dbPathname string) (*BoltDB, error) {
+	db, err := bolt.Open(dbPathname, 0660, nil)
+	if err != nil {
+		return nil, fmt.Errorf("[boltDB] Unable to open %s: %s", dbPathname, err)
+	}
+	// check the persistence version
+	err = db.Update(func(tx *bolt.Tx) error {
+		if top := tx.Bucket(topBucket); top == nil {
+			top, err := tx.CreateBucket(topBucket)
+			if err != nil {
+				return err
+			}
+			if err := top.Put(versionIdent, persistenceVersion); err != nil {
+				return err
+			}
+		} else {
+			if checkVersion := top.Get(versionIdent); checkVersion != nil {
+				if checkVersion[0] != persistenceVersion[0] {
+					return fmt.Errorf("[boltDB] Cannot use persistence file %s - version %x", dbPathname, checkVersion)
+				}
+			}
+		}
+		return nil
+	})
+	return &BoltDB{db: db}, err
+}
+
+func (d *BoltDB) Load(ident string, data interface{}) error {
+	return d.db.View(func(tx *bolt.Tx) error {
+		b := tx.Bucket(topBucket)
+		v := b.Get([]byte(ident))
+		if v == nil {
+			return nil
+		}
+		reader := bytes.NewReader(v)
+		decoder := gob.NewDecoder(reader)
+		err := decoder.Decode(data)
+		return err
+	})
+}
+
+func (d *BoltDB) Save(ident string, data interface{}) error {
+	return d.db.Update(func(tx *bolt.Tx) error {
+		b := tx.Bucket(topBucket)
+		buf := new(bytes.Buffer)
+		enc := gob.NewEncoder(buf)
+		if err := enc.Encode(data); err != nil {
+			return err
+		}
+		return b.Put([]byte(ident), buf.Bytes())
+	})
+}
+
+func (d *BoltDB) Close() error {
+	return d.db.Close()
+}

ipam/allocator.go

@@ -10,6 +10,7 @@ import (
 	"github.com/weaveworks/mesh"
 
 	"github.com/weaveworks/weave/common"
+	"github.com/weaveworks/weave/db"
 	"github.com/weaveworks/weave/ipam/paxos"
 	"github.com/weaveworks/weave/ipam/ring"
 	"github.com/weaveworks/weave/ipam/space"
@@ -54,6 +55,7 @@ type Allocator struct {
 	pendingAllocates []operation                  // held until we get some free space
 	pendingClaims    []operation                  // held until we know who owns the space
 	dead             map[string]time.Time         // containers we heard were dead, and when
+	db               db.DB                        // persistence
 	gossip           mesh.Gossip                  // our link to the outside world for sending messages
 	paxos            *paxos.Node
 	paxosActive      bool
@@ -64,12 +66,13 @@ type Allocator struct {
 }
 
 // NewAllocator creates and initialises a new Allocator
-func NewAllocator(ourName mesh.PeerName, ourUID mesh.PeerUID, ourNickname string, universe address.Range, quorum uint, isKnownPeer func(name mesh.PeerName) bool) *Allocator {
+func NewAllocator(ourName mesh.PeerName, ourUID mesh.PeerUID, ourNickname string, universe address.Range, quorum uint, db db.DB, isKnownPeer func(name mesh.PeerName) bool) *Allocator {
 	return &Allocator{
 		ourName:     ourName,
 		universe:    universe,
 		ring:        ring.New(universe.Start, universe.End, ourName),
 		owned:       make(map[string][]address.Address),
+		db:          db,
 		paxos:       paxos.NewNode(ourName, ourUID, quorum),
 		nicknames:   map[mesh.PeerName]string{ourName: ourNickname},
 		isKnownPeer: isKnownPeer,
@@ -80,6 +83,7 @@ func NewAllocator(ourName mesh.PeerName, ourUID mesh.PeerUID, ourNickname string
 
 // Start runs the allocator goroutine
 func (alloc *Allocator) Start() {
+	alloc.loadPersistedRing()
 	actionChan := make(chan func(), mesh.ChannelSize)
 	alloc.actionChan = actionChan
 	alloc.ticker = time.NewTicker(tickInterval)
@@ -580,6 +584,7 @@ func (alloc *Allocator) ringUpdated() {
 		alloc.paxos = nil
 	}
 
+	alloc.persistRing()
 	alloc.space.UpdateRanges(alloc.ring.OwnedRanges())
 	alloc.tryPendingOps()
 }
@@ -727,6 +732,7 @@ func (alloc *Allocator) donateSpace(r address.Range, to mesh.PeerName) {
 	}
 	alloc.debugln("Giving range", chunk, "to", to)
 	alloc.ring.GrantRangeToHost(chunk.Start, chunk.End, to)
+	alloc.persistRing()
 }
 
 func (alloc *Allocator) assertInvariants() {
@@ -759,6 +765,42 @@ func (alloc *Allocator) reportFreeSpace() {
 	alloc.ring.ReportFree(freespace)
 }
 
+// Persisting the Ring
+const (
+	ringIdent = "ring"
+	nameIdent = "peername"
+)
+
+func (alloc *Allocator) persistRing() {
+	// It would be better if these two Save operations happened in the same transaction
+	if err := alloc.db.Save(nameIdent, alloc.ourName); err != nil {
+		alloc.fatalf("Error persisting ring data: %s", err)
+		return
+	}
+	if err := alloc.db.Save(ringIdent, alloc.ring); err != nil {
+		alloc.fatalf("Error persisting ring data: %s", err)
+	}
+}
+
+func (alloc *Allocator) loadPersistedRing() {
+	var checkPeerName mesh.PeerName
+	if err := alloc.db.Load(nameIdent, &checkPeerName); err != nil {
+		alloc.fatalf("Error loading persisted peer name: %s", err)
+		return
+	}
+	if checkPeerName != alloc.ourName {
+		alloc.infof("Deleting persisted data for peername %s", checkPeerName)
+		alloc.persistRing()
+		return
+	}
+	if err := alloc.db.Load(ringIdent, &alloc.ring); err != nil {
+		alloc.fatalf("Error loading persisted ring data: %s", err)
+	}
+	if alloc.ring != nil {
+		alloc.space.UpdateRanges(alloc.ring.OwnedRanges())
+	}
+}
+
 // Owned addresses
 
 func (alloc *Allocator) allOwned(ident string) []address.Address {
@@ -818,6 +860,9 @@ func (alloc *Allocator) findOwner(addr address.Address) string {
 
 // Logging
 
+func (alloc *Allocator) fatalf(fmt string, args ...interface{}) {
+	common.Log.Fatalf("[allocator %s] "+fmt, append([]interface{}{alloc.ourName}, args...)...)
+}
 func (alloc *Allocator) infof(fmt string, args ...interface{}) {
 	common.Log.Infof("[allocator %s] "+fmt, append([]interface{}{alloc.ourName}, args...)...)
 }

ipam/testutils_test.go

@@ -123,6 +123,11 @@ func CheckAllExpectedMessagesSent(allocs ...*Allocator) {
 	}
 }
 
+type mockDB struct{}
+
+func (d *mockDB) Load(_ string, _ interface{}) error { return nil }
+func (d *mockDB) Save(_ string, _ interface{}) error { return nil }
+
 func makeAllocator(name string, cidrStr string, quorum uint) (*Allocator, address.Range) {
 	peername, err := mesh.PeerNameFromString(name)
 	if err != nil {
@@ -135,7 +140,7 @@ func makeAllocator(name string, cidrStr string, quorum uint) (*Allocator, addres
 	}
 
 	alloc := NewAllocator(peername, mesh.PeerUID(rand.Int63()),
-		"nick-"+name, cidr.Range(), quorum, func(mesh.PeerName) bool { return true })
+		"nick-"+name, cidr.Range(), quorum, new(mockDB), func(mesh.PeerName) bool { return true })
 
 	return alloc, cidr.HostRange()
 }

prog/weaver/main.go

@@ -17,6 +17,7 @@ import (
 
 	. "github.com/weaveworks/weave/common"
 	"github.com/weaveworks/weave/common/docker"
+	"github.com/weaveworks/weave/db"
 	"github.com/weaveworks/weave/ipam"
 	"github.com/weaveworks/weave/nameserver"
 	weavenet "github.com/weaveworks/weave/net"
@@ -67,6 +68,7 @@ func main() {
 		dnsConfig          dnsConfig
 		datapathName       string
 		trustedSubnetStr   string
+		dbPrefix           string
 
 		defaultDockerHost = "unix:///var/run/docker.sock"
 	)
@@ -101,6 +103,7 @@ func main() {
 	mflag.DurationVar(&dnsConfig.ClientTimeout, []string{"-dns-fallback-timeout"}, nameserver.DefaultClientTimeout, "timeout for fallback DNS requests")
 	mflag.StringVar(&dnsConfig.EffectiveListenAddress, []string{"-dns-effective-listen-address"}, "", "address DNS will actually be listening, after Docker port mapping")
 	mflag.StringVar(&datapathName, []string{"-datapath"}, "", "ODP datapath name")
+	mflag.StringVar(&dbPrefix, []string{"-db-prefix"}, "weave", "pathname/prefix of filename to store data")
 
 	mflag.StringVar(&trustedSubnetStr, []string{"-trusted-subnets"}, "", "Command separated list of trusted subnets in CIDR notation")
 
@@ -159,6 +162,10 @@ func main() {
 	config.TrustedSubnets = parseTrustedSubnets(trustedSubnetStr)
 	config.PeerDiscovery = !noDiscovery
 
+	db, err := db.NewBoltDB(dbPrefix + "data.db")
+	checkFatal(err)
+	defer db.Close()
+
 	router := weave.NewNetworkRouter(config, networkConfig, name, nickName, overlay)
 	Log.Println("Our name is", router.Ourself)
 
@@ -188,7 +195,7 @@ func main() {
 		defaultSubnet address.CIDR
 	)
 	if iprangeCIDR != "" {
-		allocator, defaultSubnet = createAllocator(router.Router, iprangeCIDR, ipsubnetCIDR, determineQuorum(peerCount, peers), isKnownPeer)
+		allocator, defaultSubnet = createAllocator(router.Router, iprangeCIDR, ipsubnetCIDR, determineQuorum(peerCount, peers), db, isKnownPeer)
 		observeContainers(allocator)
 	} else if peerCount > 0 {
 		Log.Fatal("--init-peer-count flag specified without --ipalloc-range")
@@ -308,7 +315,7 @@ func parseAndCheckCIDR(cidrStr string) address.CIDR {
 	return cidr
 }
 
-func createAllocator(router *mesh.Router, ipRangeStr string, defaultSubnetStr string, quorum uint, isKnownPeer func(mesh.PeerName) bool) (*ipam.Allocator, address.CIDR) {
+func createAllocator(router *mesh.Router, ipRangeStr string, defaultSubnetStr string, quorum uint, db db.DB, isKnownPeer func(mesh.PeerName) bool) (*ipam.Allocator, address.CIDR) {
 	ipRange := parseAndCheckCIDR(ipRangeStr)
 	defaultSubnet := ipRange
 	if defaultSubnetStr != "" {
@@ -317,7 +324,7 @@ func createAllocator(router *mesh.Router, ipRangeStr string, defaultSubnetStr st
 			Log.Fatalf("IP address allocation default subnet %s does not overlap with allocation range %s", defaultSubnet, ipRange)
 		}
 	}
-	allocator := ipam.NewAllocator(router.Ourself.Peer.Name, router.Ourself.Peer.UID, router.Ourself.Peer.NickName, ipRange.Range(), quorum, isKnownPeer)
+	allocator := ipam.NewAllocator(router.Ourself.Peer.Name, router.Ourself.Peer.UID, router.Ourself.Peer.NickName, ipRange.Range(), quorum, db, isKnownPeer)
 
 	allocator.SetInterfaces(router.NewGossip("IPallocation", allocator))
 	allocator.Start()

test/162_persist_ipam_2_test.sh

@@ -0,0 +1,43 @@
+#! /bin/bash
+
+. ./config.sh
+
+start_suite "Checking persistence of IPAM"
+
+launch_router_with_db() {
+    host=$1
+    shift
+    WEAVE_DOCKER_ARGS="-v /tmp:/db" weave_on $host launch-router --db-prefix=/db/test162- "$@"
+}
+
+# Remove any persisted data from previous runs
+run_on $HOST1 "sudo rm -f /tmp/test162-*"
+run_on $HOST2 "sudo rm -f /tmp/test162-*"
+
+launch_router_with_db $HOST1
+launch_router_with_db $HOST2 $HOST1
+
+start_container $HOST1 --name=c1
+C1=$(container_ip $HOST1 c1)
+start_container $HOST2 --name=c2
+assert_raises "exec_on $HOST2 c2 $PING $C1"
+
+stop_router_on $HOST1
+stop_router_on $HOST2
+
+# Start just HOST2; if nothing persisted it would form its own ring
+launch_router_with_db $HOST2
+start_container $HOST2 --name=c3
+C3=$(container_ip $HOST2 c3)
+assert_raises "[ $C3 != $C1 ]"
+
+stop_router_on $HOST2
+
+# Now start HOST1 with HOST2 down and see if it hangs when we launch a container
+launch_router_with_db $HOST1 $HOST2
+start_container $HOST1 --name=c4
+C4=$(container_ip $HOST1 c4)
+assert_raises "[ $C4 != $C1 ]"
+assert_raises "[ $C4 != $C3 ]"
+
+end_suite