From b84ef98de84cfa276c0439d314c30f8368572e32 Mon Sep 17 00:00:00 2001 From: Adam Harrison Date: Mon, 22 Feb 2016 16:02:44 +0000 Subject: [PATCH] Support non-electing peers via --observer --- ipam/allocator.go | 117 +++++++++++++++++++------------- ipam/paxos/participant.go | 42 ++++++++++++ ipam/status.go | 6 +- prog/weaver/main.go | 13 +++- site/ipam/overview-init-ipam.md | 11 +++ 5 files changed, 137 insertions(+), 52 deletions(-) create mode 100644 ipam/paxos/participant.go diff --git a/ipam/allocator.go b/ipam/allocator.go index f9d67fe0..83172c97 100644 --- a/ipam/allocator.go +++ b/ipam/allocator.go @@ -45,24 +45,24 @@ type operation interface { // necessary plumbing. Runs as a single-threaded Actor, so no locks // are used around data structures. type Allocator struct { - actionChan chan<- func() - ourName mesh.PeerName - universe address.Range // superset of all ranges - ring *ring.Ring // information on ranges owned by all peers - space space.Space // more detail on ranges owned by us - owned map[string][]address.CIDR // who owns what addresses, indexed by container-ID - nicknames map[mesh.PeerName]string // so we can map nicknames for rmpeer - 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 - ticker *time.Ticker - shuttingDown bool // to avoid doing any requests while trying to shut down - isKnownPeer func(mesh.PeerName) bool - now func() time.Time + actionChan chan<- func() + ourName mesh.PeerName + universe address.Range // superset of all ranges + ring *ring.Ring // information on ranges owned by all peers + space space.Space // more detail on ranges owned by us + owned map[string][]address.CIDR // who owns what addresses, indexed by container-ID + nicknames map[mesh.PeerName]string // so we can map nicknames for rmpeer + 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.Participant + awaitingConsensus bool + ticker *time.Ticker + shuttingDown bool // to avoid doing any requests while trying to shut down + isKnownPeer func(mesh.PeerName) bool + now func() time.Time } // NewAllocator creates and initialises a new Allocator @@ -73,7 +73,7 @@ func NewAllocator(ourName mesh.PeerName, ourUID mesh.PeerUID, ourNickname string ring: ring.New(universe.Start, universe.End, ourName), owned: make(map[string][]address.CIDR), db: db, - paxos: paxos.NewNode(ourName, ourUID, quorum), + paxos: paxos.NewParticipant(ourName, ourUID, quorum), nicknames: map[mesh.PeerName]string{ourName: ourNickname}, isKnownPeer: isKnownPeer, dead: make(map[string]time.Time), @@ -472,6 +472,9 @@ func (alloc *Allocator) encode() []byte { } // We're only interested in Paxos until we have a Ring. + // Non-electing participants (e.g. observers) return + // a nil gossip state in order to provoke a unicast ring + // update from electing peers who have reached consensus. if alloc.ring.Empty() { data.Paxos = alloc.paxos.GossipState() } else { @@ -540,7 +543,7 @@ func (alloc *Allocator) actorLoop(actionChan <-chan func()) { } action() case <-alloc.ticker.C: - if alloc.paxosActive { + if alloc.awaitingConsensus { alloc.propose() } alloc.removeDeadContainers() @@ -556,11 +559,11 @@ func (alloc *Allocator) actorLoop(actionChan <-chan func()) { // Ensure we are making progress towards an established ring func (alloc *Allocator) establishRing() { - if !alloc.ring.Empty() || alloc.paxosActive { + if !alloc.ring.Empty() || alloc.awaitingConsensus { return } - alloc.paxosActive = true + alloc.awaitingConsensus = true alloc.propose() if ok, cons := alloc.paxos.Consensus(); ok { // If the quorum was 1, then proposing immediately @@ -578,8 +581,8 @@ func (alloc *Allocator) createRing(peers []mesh.PeerName) { func (alloc *Allocator) ringUpdated() { // When we have a ring, we don't need paxos any more - if alloc.paxosActive { - alloc.paxosActive = false + if alloc.awaitingConsensus { + alloc.awaitingConsensus = false alloc.paxos = nil } @@ -668,11 +671,16 @@ func (alloc *Allocator) update(sender mesh.PeerName, msg []byte) error { alloc.nicknames[peer] = nickname } - // only one of Ring and Paxos should be present. And we - // shouldn't get updates for a empty Ring. But tolerate - // them just in case. - if data.Ring != nil { + switch { + // If someone sent us a ring, merge it into ours. Note this will move us + // out of the awaiting-consensus state if we didn't have a ring already. + case data.Ring != nil: switch err = alloc.ring.Merge(*data.Ring); err { + case nil: + if !alloc.ring.Empty() { + alloc.pruneNicknames() + alloc.ringUpdated() + } case ring.ErrDifferentSeeds: return fmt.Errorf("IP allocation was seeded by different peers (received: %v, ours: %v)", alloc.annotatePeernames(data.Ring.Seeds), alloc.annotatePeernames(alloc.ring.Seeds)) @@ -680,31 +688,44 @@ func (alloc *Allocator) update(sender mesh.PeerName, msg []byte) error { return fmt.Errorf("Incompatible IP allocation ranges (received: %s, ours: %s)", data.Ring.Range().AsCIDRString(), alloc.ring.Range().AsCIDRString()) default: - if err == nil && !alloc.ring.Empty() { - alloc.pruneNicknames() - alloc.ringUpdated() - } return err } - } - if data.Paxos != nil { - if alloc.ring.Empty() { - if alloc.paxos.Update(data.Paxos) { - if alloc.paxos.Think() { - // If something important changed, broadcast - alloc.gossip.GossipBroadcast(alloc.Gossip()) - } - - if ok, cons := alloc.paxos.Consensus(); ok { - alloc.createRing(cons.Value) - } - } - } else if sender != mesh.UnknownPeerName { - // Sender is trying to initialize a ring, but we have one - // already - send it straight back + // If we reach this point we know the sender is either an elector + // broadcasting a paxos proposal to form a ring or a non-elector + // broadcasting a ring request. If we have a ring already we can just send + // it back regardless. + case !alloc.ring.Empty(): + if sender != mesh.UnknownPeerName { alloc.sendRingUpdate(sender) } + + // Otherwise, we need to react according to whether or not we received a + // paxos proposal. + case data.Paxos != nil: + // Process the proposal (this is a no-op if we're an observer) + if alloc.paxos.Update(data.Paxos) { + if alloc.paxos.Think() { + // If something important changed, broadcast + alloc.gossip.GossipBroadcast(alloc.Gossip()) + } + + if ok, cons := alloc.paxos.Consensus(); ok { + alloc.createRing(cons.Value) + } + } + + // No paxos proposal present, so sender is a non-elector. We don't have a + // ring to send, so attempt to establish one on their behalf. NB we only do + // this: + // + // * On an explicit broadcast request triggered by a remote allocation attempt + // (if we did so on periodic gossip we would force consensus unnecessarily) + // * If we are an elector (to avoid a broadcast storm of ring request messages) + default: + if _, ok := alloc.paxos.(*paxos.Node); ok && sender != mesh.UnknownPeerName { + alloc.establishRing() + } } return nil diff --git a/ipam/paxos/participant.go b/ipam/paxos/participant.go new file mode 100644 index 00000000..64f3a626 --- /dev/null +++ b/ipam/paxos/participant.go @@ -0,0 +1,42 @@ +package paxos + +import "github.com/weaveworks/mesh" + +type Participant interface { + GossipState() GossipState + Update(from GossipState) bool + 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 { +} + +func (observer *Observer) GossipState() GossipState { + return nil +} + +func (observer *Observer) Update(from GossipState) bool { + return false +} + +func (observer *Observer) Propose() { +} + +func (observer *Observer) Think() bool { + return false +} + +func (observer *Observer) Consensus() (bool, AcceptedValue) { + return false, AcceptedValue{} +} diff --git a/ipam/status.go b/ipam/status.go index f205ecee..f98f303d 100644 --- a/ipam/status.go +++ b/ipam/status.go @@ -36,8 +36,10 @@ func NewStatus(allocator *Allocator, defaultSubnet address.CIDR) *Status { } var paxosStatus *paxos.Status - if allocator.paxosActive { - paxosStatus = paxos.NewStatus(allocator.paxos) + if allocator.awaitingConsensus && allocator.paxos != nil { + if node, ok := allocator.paxos.(*paxos.Node); ok { + paxosStatus = paxos.NewStatus(node) + } } resultChan := make(chan *Status) diff --git a/prog/weaver/main.go b/prog/weaver/main.go index 70a3aba5..511d017d 100644 --- a/prog/weaver/main.go +++ b/prog/weaver/main.go @@ -88,6 +88,7 @@ func main() { iprangeCIDR string ipsubnetCIDR string peerCount int + observer bool dockerAPI string peers []string noDNS bool @@ -121,6 +122,7 @@ func main() { mflag.StringVar(&iprangeCIDR, []string{"#iprange", "#-iprange", "-ipalloc-range"}, "", "IP address range reserved for automatic allocation, in CIDR notation") mflag.StringVar(&ipsubnetCIDR, []string{"#ipsubnet", "#-ipsubnet", "-ipalloc-default-subnet"}, "", "subnet to allocate within by default, in CIDR notation") mflag.IntVar(&peerCount, []string{"#initpeercount", "#-initpeercount", "-init-peer-count"}, 0, "number of peers in network (for IP address allocation)") + mflag.BoolVar(&observer, []string{"-observer"}, false, "do not participate in paxos, observe only") mflag.StringVar(&dockerAPI, []string{"#api", "#-api", "-docker-api"}, defaultDockerHost, "Docker API endpoint") mflag.BoolVar(&noDNS, []string{"-no-dns"}, false, "disable DNS server") mflag.StringVar(&dnsConfig.Domain, []string{"-dns-domain"}, nameserver.DefaultDomain, "local domain to server requests for") @@ -225,8 +227,11 @@ func main() { allocator *ipam.Allocator defaultSubnet address.CIDR ) + if observer && peerCount > 0 { + Log.Fatal("At most one of --observer and --init-peer-count must be specified.") + } if iprangeCIDR != "" { - allocator, defaultSubnet = createAllocator(router.Router, iprangeCIDR, ipsubnetCIDR, determineQuorum(peerCount, peers), db, isKnownPeer) + allocator, defaultSubnet = createAllocator(router.Router, iprangeCIDR, ipsubnetCIDR, determineQuorum(observer, peerCount, peers), db, isKnownPeer) observeContainers(allocator) ids, err := dockerCli.AllContainerIDs() checkFatal(err) @@ -388,7 +393,11 @@ func createDNSServer(config dnsConfig, router *mesh.Router, isKnownPeer func(mes // Pick a quorum size heuristically based on the number of peer // addresses passed. -func determineQuorum(initPeerCountFlag int, peers []string) uint { +func determineQuorum(observer bool, initPeerCountFlag int, peers []string) uint { + if observer { + return uint(0) + } + if initPeerCountFlag > 0 { return uint(initPeerCountFlag/2 + 1) } diff --git a/site/ipam/overview-init-ipam.md b/site/ipam/overview-init-ipam.md index 727216c9..f85eb49c 100644 --- a/site/ipam/overview-init-ipam.md +++ b/site/ipam/overview-init-ipam.md @@ -60,6 +60,17 @@ through three states: 'deferred', 'waiting' and 'achieved': themselves successfully * 'achieved' - consensus achieved; allocations proceed normally +Finally, you may launch some peers as election observers using the +`--observer` option: + + host4$ weave launch --observer $HOST3 + +You do not need to specify an initial count to such peers, as they do +not participate in the election - they await its outcome. This can be +useful to add ephemeral peers to an existing fixed cluster (for example +in response to a scale-out event) without worrying about adjusting +initial peer counts accordingly. + #### `--init-peer-count` and How Quorum is Achieved Normally it isn't a problem to over-estimate `--init-peer-count`, but if you supply