diff --git a/.gitmodules b/.gitmodules index feedaad8..d03fda92 100644 --- a/.gitmodules +++ b/.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 diff --git a/Makefile b/Makefile index 6efd5b5a..73072389 100644 --- a/Makefile +++ b/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 diff --git a/db/boltdb.go b/db/boltdb.go new file mode 100644 index 00000000..0d6b1b86 --- /dev/null +++ b/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() +} diff --git a/ipam/allocator.go b/ipam/allocator.go index 45944310..df349293 100644 --- a/ipam/allocator.go +++ b/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...)...) } diff --git a/ipam/testutils_test.go b/ipam/testutils_test.go index f817c1ea..76fdd7f4 100644 --- a/ipam/testutils_test.go +++ b/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() } diff --git a/prog/weaver/main.go b/prog/weaver/main.go index 2c1914d1..627696db 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/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() diff --git a/test/162_persist_ipam_2_test.sh b/test/162_persist_ipam_2_test.sh new file mode 100755 index 00000000..e5efac96 --- /dev/null +++ b/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 diff --git a/vendor/github.com/boltdb/bolt b/vendor/github.com/boltdb/bolt new file mode 160000 index 00000000..ee4a0888 --- /dev/null +++ b/vendor/github.com/boltdb/bolt @@ -0,0 +1 @@ +Subproject commit ee4a0888a9abe7eefe5a0992ca4cb06864839873