Merge pull request #2070 from weaveworks/issues/1721-quorum-callback

Use current target list to compute quorum

ipam/allocator.go

@@ -64,6 +64,7 @@ type Allocator struct {
 	ticker            *time.Ticker
 	shuttingDown      bool // to avoid doing any requests while trying to shut down
 	isKnownPeer       func(mesh.PeerName) bool
+	quorum            func() uint
 	now               func() time.Time
 }
 
@@ -73,13 +74,20 @@ type Config struct {
 	OurNickname string
 	Seed        []mesh.PeerName
 	Universe    address.Range
-	Quorum      uint
+	IsObserver  bool
+	Quorum      func() uint
 	Db          db.DB
 	IsKnownPeer func(name mesh.PeerName) bool
 }
 
 // NewAllocator creates and initialises a new Allocator
 func NewAllocator(config Config) *Allocator {
+	var participant paxos.Participant
+	if config.IsObserver {
+		participant = paxos.NewObserver()
+	} else {
+		participant = paxos.NewNode(config.OurName, config.OurUID, 1)
+	}
 	return &Allocator{
 		ourName:     config.OurName,
 		seed:        config.Seed,
@@ -87,9 +95,10 @@ func NewAllocator(config Config) *Allocator {
 		ring:        ring.New(config.Universe.Start, config.Universe.End, config.OurName),
 		owned:       make(map[string][]address.CIDR),
 		db:          config.Db,
-		paxos:       paxos.NewParticipant(config.OurName, config.OurUID, config.Quorum),
+		paxos:       participant,
 		nicknames:   map[mesh.PeerName]string{config.OurName: config.OurNickname},
 		isKnownPeer: config.IsKnownPeer,
+		quorum:      config.Quorum,
 		dead:        make(map[string]time.Time),
 		now:         time.Now,
 	}
@@ -603,6 +612,7 @@ func (alloc *Allocator) establishRing() {
 	}
 
 	alloc.awaitingConsensus = true
+	alloc.paxos.SetQuorum(alloc.quorum())
 	alloc.propose()
 	if ok, cons := alloc.paxos.Consensus(); ok {
 		// If the quorum was 1, then proposing immediately

ipam/paxos/participant.go

@@ -1,25 +1,19 @@
 package paxos
 
-import "github.com/weaveworks/mesh"
-
 type Participant interface {
 	GossipState() GossipState
 	Update(from GossipState) bool
+	SetQuorum(uint)
 	Propose()
 	Think() bool
 	Consensus() (bool, AcceptedValue)
 }
 
-func NewParticipant(name mesh.PeerName, uid mesh.PeerUID, quorum uint) Participant {
-	switch quorum {
-	case 0:
-		return &Observer{}
-	default:
-		return NewNode(name, uid, quorum)
-	}
+type Observer struct {
 }
 
-type Observer struct {
+func NewObserver() Participant {
+	return &Observer{}
 }
 
 func (observer *Observer) GossipState() GossipState {
@@ -33,6 +27,9 @@ func (observer *Observer) Update(from GossipState) bool {
 func (observer *Observer) Propose() {
 }
 
+func (observer *Observer) SetQuorum(uint) {
+}
+
 func (observer *Observer) Think() bool {
 	return false
 }

ipam/paxos/paxos.go

@@ -80,6 +80,10 @@ func NewNode(name mesh.PeerName, uid mesh.PeerUID, quorum uint) *Node {
 	}
 }
 
+func (node *Node) SetQuorum(quorum uint) {
+	node.quorum = quorum
+}
+
 func (node *Node) GossipState() GossipState {
 	return node.knows
 }
@@ -124,6 +128,9 @@ func max(a uint, b uint) uint {
 // simply a matter of gossipping a new proposal that supersedes all
 // others.
 func (node *Node) Propose() {
+	if node.quorum == 0 {
+		panic("Paxos node.Propose() called with no quorum set")
+	}
 	// Find the highest round number around
 	round := uint(0)
 

ipam/testutils_test.go

@@ -144,7 +144,8 @@ func makeAllocator(name string, cidrStr string, quorum uint) (*Allocator, addres
 		OurUID:      mesh.PeerUID(rand.Int63()),
 		OurNickname: "nick-" + name,
 		Universe:    cidr.Range(),
-		Quorum:      quorum,
+		IsObserver:  quorum == 0,
+		Quorum:      func() uint { return quorum },
 		Db:          new(mockDB),
 		IsKnownPeer: func(mesh.PeerName) bool { return true },
 	}), cidr

prog/weaver/main.go

@@ -255,7 +255,7 @@ func main() {
 		defaultSubnet address.CIDR
 	)
 	if ipamConfig.Enabled() {
-		allocator, defaultSubnet = createAllocator(router.Router, ipamConfig, len(peers), db, isKnownPeer)
+		allocator, defaultSubnet = createAllocator(router, ipamConfig, db, isKnownPeer)
 		observeContainers(allocator)
 		ids, err := dockerCli.AllContainerIDs()
 		checkFatal(err)
@@ -379,7 +379,7 @@ func parseAndCheckCIDR(cidrStr string) address.CIDR {
 	return cidr
 }
 
-func createAllocator(router *mesh.Router, config ipamConfig, peerCount int, db db.DB, isKnownPeer func(mesh.PeerName) bool) (*ipam.Allocator, address.CIDR) {
+func createAllocator(router *weave.NetworkRouter, config ipamConfig, db db.DB, isKnownPeer func(mesh.PeerName) bool) (*ipam.Allocator, address.CIDR) {
 	ipRange := parseAndCheckCIDR(config.IPRangeCIDR)
 	seed := parseIPAMSeed(config.Seed)
 	defaultSubnet := ipRange
@@ -396,7 +396,8 @@ func createAllocator(router *mesh.Router, config ipamConfig, peerCount int, db d
 		OurNickname: router.Ourself.Peer.NickName,
 		Seed:        seed,
 		Universe:    ipRange.Range(),
-		Quorum:      determineQuorum(config.Observer, config.PeerCount, peerCount),
+		IsObserver:  config.Observer,
+		Quorum:      func() uint { return determineQuorum(config.PeerCount, router) },
 		Db:          db,
 		IsKnownPeer: isKnownPeer,
 	}
@@ -426,17 +427,14 @@ func createDNSServer(config dnsConfig, router *mesh.Router, isKnownPeer func(mes
 	return ns, dnsserver
 }
 
-// Pick a quorum size heuristically based on the number of peer
-// addresses passed.
-func determineQuorum(observer bool, initPeerCountFlag int, peerCount int) uint {
-	if observer {
-		return uint(0)
-	}
-
+// Pick a quorum size based on the number of peer addresses.
+func determineQuorum(initPeerCountFlag int, router *weave.NetworkRouter) uint {
 	if initPeerCountFlag > 0 {
 		return uint(initPeerCountFlag/2 + 1)
 	}
 
+	peers := router.ConnectionMaker.Targets(true)
+
 	// Guess a suitable quorum size based on the list of peer
 	// addresses.  The peer list might or might not contain an
 	// address for this peer, so the conservative assumption is
@@ -444,7 +442,7 @@ func determineQuorum(observer bool, initPeerCountFlag int, peerCount int) uint {
 	// that resolve to the same peer, in which case the quorum
 	// might be larger than it needs to be.  But the user can
 	// specify it explicitly if that becomes a problem.
-	clusterSize := uint(peerCount + 1)
+	clusterSize := uint(len(peers) + 1)
 	quorum := clusterSize/2 + 1
 	Log.Println("Assuming quorum size of", quorum)
 	return quorum

test/700_status_and_report_test.sh

@@ -19,11 +19,11 @@ check "DefaultSubnet" "{{.IPAM.DefaultSubnet}}"                $UNIVERSE
 check "Domain"        "{{.DNS.Domain}}"                        "weave.local."
 
 assert_raises "weave_on $HOST1 status peers | grep nicknamington"
-weave_on $HOST1 connect 10.2.2.1
-assert "weave_on $HOST1 status targets" "10.2.2.1"
-assert "weave_on $HOST1 status connections | tr -s ' ' | cut -d ' ' -f 2" "10.2.2.1:6783"
 start_container $HOST1 --name test
 assert "weave_on $HOST1 status dns         | tr -s ' ' | cut -d ' ' -f 1" "test"
 assert_raises "weave_on $HOST1 report | grep nicknamington"
+weave_on $HOST1 connect 10.2.2.1
+assert "weave_on $HOST1 status targets" "10.2.2.1"
+assert "weave_on $HOST1 status connections | tr -s ' ' | cut -d ' ' -f 2" "10.2.2.1:6783"
 
 end_suite