From 38c6eb28e8217b1af90818e22765a54b356248da Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Fri, 12 Feb 2016 08:23:51 +0000 Subject: [PATCH 1/5] Add basic infrastructure to use BoltDB for persistence --- .gitmodules | 3 ++ Makefile | 2 +- db/boltdb.go | 81 +++++++++++++++++++++++++++++++++++ prog/weaver/main.go | 7 +++ vendor/github.com/boltdb/bolt | 1 + 5 files changed, 93 insertions(+), 1 deletion(-) create mode 100644 db/boltdb.go create mode 160000 vendor/github.com/boltdb/bolt 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/prog/weaver/main.go b/prog/weaver/main.go index 2c1914d1..8fc4a579 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) 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 From ad036499eb3592e2b8e57636186086d5b5bb55f3 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Fri, 12 Feb 2016 08:46:47 +0000 Subject: [PATCH 2/5] Add db infra to IPAM --- ipam/allocator.go | 5 ++++- ipam/testutils_test.go | 7 ++++++- prog/weaver/main.go | 6 +++--- 3 files changed, 13 insertions(+), 5 deletions(-) diff --git a/ipam/allocator.go b/ipam/allocator.go index 45944310..57c01d87 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, 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 8fc4a579..627696db 100644 --- a/prog/weaver/main.go +++ b/prog/weaver/main.go @@ -195,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") @@ -315,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 != "" { @@ -324,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() From f396c6040243a82d55a255dc258cb55610980342 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Fri, 12 Feb 2016 09:00:30 +0000 Subject: [PATCH 3/5] Persist the Ring --- ipam/allocator.go | 23 ++++++++++++++++++ test/162_persist_ipam_2_test.sh | 43 +++++++++++++++++++++++++++++++++ 2 files changed, 66 insertions(+) create mode 100755 test/162_persist_ipam_2_test.sh diff --git a/ipam/allocator.go b/ipam/allocator.go index 57c01d87..7d4d4096 100644 --- a/ipam/allocator.go +++ b/ipam/allocator.go @@ -83,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) @@ -583,6 +584,7 @@ func (alloc *Allocator) ringUpdated() { alloc.paxos = nil } + alloc.persistRing() alloc.space.UpdateRanges(alloc.ring.OwnedRanges()) alloc.tryPendingOps() } @@ -730,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() { @@ -762,6 +765,26 @@ func (alloc *Allocator) reportFreeSpace() { alloc.ring.ReportFree(freespace) } +// Persisting the Ring +const ( + ringIdent = "ring" +) + +func (alloc *Allocator) persistRing() { + if err := alloc.db.Save(ringIdent, alloc.ring); err != nil { + common.Log.Errorf("Error persisting ring data: %s", err) + } +} + +func (alloc *Allocator) loadPersistedRing() { + if err := alloc.db.Load(ringIdent, &alloc.ring); err != nil { + common.Log.Errorf("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 { 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 From 06f3793f0525b6571dd1282d6ceb69189fb347ee Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Fri, 12 Feb 2016 10:19:07 +0000 Subject: [PATCH 4/5] Remove persisted IPAM ring data on peerName change --- ipam/allocator.go | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/ipam/allocator.go b/ipam/allocator.go index 7d4d4096..4d758f36 100644 --- a/ipam/allocator.go +++ b/ipam/allocator.go @@ -768,15 +768,31 @@ func (alloc *Allocator) reportFreeSpace() { // 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 { + common.Log.Errorf("Error persisting ring data: %s", err) + return + } if err := alloc.db.Save(ringIdent, alloc.ring); err != nil { common.Log.Errorf("Error persisting ring data: %s", err) } } func (alloc *Allocator) loadPersistedRing() { + var checkPeerName mesh.PeerName + if err := alloc.db.Load(nameIdent, &checkPeerName); err != nil { + common.Log.Errorf("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 { common.Log.Errorf("Error loading persisted ring data: %s", err) } From 546f9592f9b70bfc52cc23e00b657db2ea2c89aa Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Fri, 12 Feb 2016 13:36:38 +0000 Subject: [PATCH 5/5] Hard-fail on persistence errors, otherwise the network can get confused on a restart. Create `Allocator.fatalf()`, for consistency --- ipam/allocator.go | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/ipam/allocator.go b/ipam/allocator.go index 4d758f36..df349293 100644 --- a/ipam/allocator.go +++ b/ipam/allocator.go @@ -774,18 +774,18 @@ const ( 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 { - common.Log.Errorf("Error persisting ring data: %s", err) + alloc.fatalf("Error persisting ring data: %s", err) return } if err := alloc.db.Save(ringIdent, alloc.ring); err != nil { - common.Log.Errorf("Error persisting ring data: %s", err) + 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 { - common.Log.Errorf("Error loading persisted peer name: %s", err) + alloc.fatalf("Error loading persisted peer name: %s", err) return } if checkPeerName != alloc.ourName { @@ -794,7 +794,7 @@ func (alloc *Allocator) loadPersistedRing() { return } if err := alloc.db.Load(ringIdent, &alloc.ring); err != nil { - common.Log.Errorf("Error loading persisted ring data: %s", err) + alloc.fatalf("Error loading persisted ring data: %s", err) } if alloc.ring != nil { alloc.space.UpdateRanges(alloc.ring.OwnedRanges()) @@ -860,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...)...) }