We can have outstanding allocations for different ranges owned by different peers, so we should not short circuit after the first space request.
773 lines
22 KiB
Go
773 lines
22 KiB
Go
package ipam
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/gob"
|
|
"fmt"
|
|
"sort"
|
|
"time"
|
|
|
|
"github.com/weaveworks/weave/common"
|
|
"github.com/weaveworks/weave/ipam/paxos"
|
|
"github.com/weaveworks/weave/ipam/ring"
|
|
"github.com/weaveworks/weave/ipam/space"
|
|
"github.com/weaveworks/weave/mesh"
|
|
"github.com/weaveworks/weave/net/address"
|
|
)
|
|
|
|
// Kinds of message we can unicast to other peers
|
|
const (
|
|
msgSpaceRequest = iota
|
|
msgRingUpdate
|
|
msgSpaceRequestDenied
|
|
|
|
paxosInterval = time.Second * 5
|
|
MinSubnetSize = 4 // first and last addresses are excluded, so 2 would be too small
|
|
)
|
|
|
|
// operation represents something which Allocator wants to do, but
|
|
// which may need to wait until some other message arrives.
|
|
type operation interface {
|
|
// Try attempts this operations and returns false if needs to be tried again.
|
|
Try(alloc *Allocator) bool
|
|
|
|
Cancel()
|
|
|
|
// Does this operation pertain to the given container id?
|
|
// Used for tidying up pending operations when containers die.
|
|
ForContainer(ident string) bool
|
|
}
|
|
|
|
// Allocator brings together Ring and space.Set, and does the
|
|
// 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.Address // 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
|
|
gossip mesh.Gossip // our link to the outside world for sending messages
|
|
paxos *paxos.Node
|
|
paxosTicker *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
|
|
func NewAllocator(ourName mesh.PeerName, ourUID mesh.PeerUID, ourNickname string, universe address.Range, quorum uint, 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),
|
|
paxos: paxos.NewNode(ourName, ourUID, quorum),
|
|
nicknames: map[mesh.PeerName]string{ourName: ourNickname},
|
|
isKnownPeer: isKnownPeer,
|
|
now: time.Now,
|
|
}
|
|
}
|
|
|
|
// Start runs the allocator goroutine
|
|
func (alloc *Allocator) Start() {
|
|
actionChan := make(chan func(), mesh.ChannelSize)
|
|
alloc.actionChan = actionChan
|
|
go alloc.actorLoop(actionChan)
|
|
}
|
|
|
|
// Stop makes the actor routine exit, for test purposes ONLY because any
|
|
// calls after this is processed will hang. Async.
|
|
func (alloc *Allocator) Stop() {
|
|
alloc.actionChan <- nil
|
|
}
|
|
|
|
// Operation life cycle
|
|
|
|
// Given an operation, try it, and add it to the pending queue if it didn't succeed
|
|
func (alloc *Allocator) doOperation(op operation, ops *[]operation) {
|
|
alloc.actionChan <- func() {
|
|
if alloc.shuttingDown {
|
|
op.Cancel()
|
|
return
|
|
}
|
|
if !op.Try(alloc) {
|
|
*ops = append(*ops, op)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Given an operation, remove it from the pending queue
|
|
// Note the op may not be on the queue; it may have
|
|
// already succeeded. If it is on the queue, we call
|
|
// cancel on it, allowing callers waiting for the resultChans
|
|
// to unblock.
|
|
func (alloc *Allocator) cancelOp(op operation, ops *[]operation) {
|
|
for i, op := range *ops {
|
|
if op == op {
|
|
*ops = append((*ops)[:i], (*ops)[i+1:]...)
|
|
op.Cancel()
|
|
break
|
|
}
|
|
}
|
|
}
|
|
|
|
// Cancel all operations in a queue
|
|
func (alloc *Allocator) cancelOps(ops *[]operation) {
|
|
for _, op := range *ops {
|
|
op.Cancel()
|
|
}
|
|
*ops = []operation{}
|
|
}
|
|
|
|
// Cancel all operations for a given container id, returns true
|
|
// if we found any.
|
|
func (alloc *Allocator) cancelOpsFor(ops *[]operation, ident string) bool {
|
|
var found bool
|
|
for i := 0; i < len(*ops); {
|
|
if op := (*ops)[i]; op.ForContainer(ident) {
|
|
found = true
|
|
op.Cancel()
|
|
*ops = append((*ops)[:i], (*ops)[i+1:]...)
|
|
} else {
|
|
i++
|
|
}
|
|
}
|
|
return found
|
|
}
|
|
|
|
// Try all operations in a queue
|
|
func (alloc *Allocator) tryOps(ops *[]operation) {
|
|
for i := 0; i < len(*ops); {
|
|
op := (*ops)[i]
|
|
if !op.Try(alloc) {
|
|
i++
|
|
continue
|
|
}
|
|
*ops = append((*ops)[:i], (*ops)[i+1:]...)
|
|
}
|
|
}
|
|
|
|
// Try all pending operations
|
|
func (alloc *Allocator) tryPendingOps() {
|
|
// Process existing claims before servicing new allocations
|
|
alloc.tryOps(&alloc.pendingClaims)
|
|
alloc.tryOps(&alloc.pendingAllocates)
|
|
}
|
|
|
|
func (alloc *Allocator) spaceRequestDenied(sender mesh.PeerName, r address.Range) {
|
|
for i := 0; i < len(alloc.pendingClaims); {
|
|
claim := alloc.pendingClaims[i].(*claim)
|
|
if r.Contains(claim.addr) {
|
|
claim.deniedBy(alloc, sender)
|
|
alloc.pendingClaims = append(alloc.pendingClaims[:i], alloc.pendingClaims[i+1:]...)
|
|
continue
|
|
}
|
|
i++
|
|
}
|
|
}
|
|
|
|
type errorCancelled struct {
|
|
kind string
|
|
ident string
|
|
}
|
|
|
|
func (e *errorCancelled) Error() string {
|
|
return fmt.Sprintf("%s request for %s cancelled", e.kind, e.ident)
|
|
}
|
|
|
|
// Actor client API
|
|
|
|
// Allocate (Sync) - get new IP address for container with given name in range
|
|
// if there isn't any space in that range we block indefinitely
|
|
func (alloc *Allocator) Allocate(ident string, r address.Range, hasBeenCancelled func() bool) (address.Address, error) {
|
|
resultChan := make(chan allocateResult)
|
|
op := &allocate{resultChan: resultChan, ident: ident, r: r, hasBeenCancelled: hasBeenCancelled}
|
|
alloc.doOperation(op, &alloc.pendingAllocates)
|
|
result := <-resultChan
|
|
return result.addr, result.err
|
|
}
|
|
|
|
// Lookup (Sync) - get existing IP address for container with given name in range
|
|
func (alloc *Allocator) Lookup(ident string, r address.Range) (address.Address, error) {
|
|
resultChan := make(chan allocateResult)
|
|
alloc.actionChan <- func() {
|
|
if addr, found := alloc.lookupOwned(ident, r); found {
|
|
resultChan <- allocateResult{addr: addr}
|
|
return
|
|
}
|
|
resultChan <- allocateResult{err: fmt.Errorf("lookup: no address found for %s in range %s", ident, r)}
|
|
}
|
|
result := <-resultChan
|
|
return result.addr, result.err
|
|
}
|
|
|
|
// Claim an address that we think we should own (Sync)
|
|
func (alloc *Allocator) Claim(ident string, addr address.Address, noErrorOnUnknown bool) error {
|
|
resultChan := make(chan error)
|
|
op := &claim{resultChan: resultChan, ident: ident, addr: addr, noErrorOnUnknown: noErrorOnUnknown}
|
|
alloc.doOperation(op, &alloc.pendingClaims)
|
|
return <-resultChan
|
|
}
|
|
|
|
// ContainerDied is provided to satisfy the updater interface; does a 'Delete' underneath. Sync.
|
|
func (alloc *Allocator) ContainerDied(ident string) {
|
|
if err := alloc.Delete(ident); err == nil {
|
|
alloc.debugln("Container", ident, "died; released addresses")
|
|
}
|
|
}
|
|
|
|
func (alloc *Allocator) ContainerStarted(ident string) {}
|
|
|
|
// Delete (Sync) - release all IP addresses for container with given name
|
|
func (alloc *Allocator) Delete(ident string) error {
|
|
errChan := make(chan error)
|
|
alloc.actionChan <- func() {
|
|
addrs, found := alloc.owned[ident]
|
|
for _, addr := range addrs {
|
|
alloc.space.Free(addr)
|
|
}
|
|
delete(alloc.owned, ident)
|
|
|
|
// Also remove any pending ops
|
|
found = alloc.cancelOpsFor(&alloc.pendingAllocates, ident) || found
|
|
found = alloc.cancelOpsFor(&alloc.pendingClaims, ident) || found
|
|
|
|
if !found {
|
|
errChan <- fmt.Errorf("Delete: no addresses for %s", ident)
|
|
return
|
|
}
|
|
errChan <- nil
|
|
}
|
|
return <-errChan
|
|
}
|
|
|
|
// Free (Sync) - release single IP address for container
|
|
func (alloc *Allocator) Free(ident string, addrToFree address.Address) error {
|
|
errChan := make(chan error)
|
|
alloc.actionChan <- func() {
|
|
addrs := alloc.owned[ident]
|
|
for i, ownedAddr := range addrs {
|
|
if ownedAddr == addrToFree {
|
|
alloc.debugln("Freed", addrToFree, "for", ident)
|
|
if len(addrs) == 1 {
|
|
delete(alloc.owned, ident)
|
|
} else {
|
|
alloc.owned[ident] = append(addrs[:i], addrs[i+1:]...)
|
|
}
|
|
alloc.space.Free(addrToFree)
|
|
errChan <- nil
|
|
return
|
|
}
|
|
}
|
|
|
|
errChan <- fmt.Errorf("Free: address %s not found for %s", addrToFree, ident)
|
|
}
|
|
return <-errChan
|
|
}
|
|
|
|
func (alloc *Allocator) pickPeerFromNicknames(isValid func(mesh.PeerName) bool) mesh.PeerName {
|
|
for name := range alloc.nicknames {
|
|
if name != alloc.ourName && isValid(name) {
|
|
return name
|
|
}
|
|
}
|
|
return mesh.UnknownPeerName
|
|
}
|
|
|
|
func (alloc *Allocator) pickPeerForTransfer() mesh.PeerName {
|
|
// first try alive peers that actively participate in IPAM (i.e. have entries)
|
|
if heir := alloc.ring.PickPeerForTransfer(alloc.isKnownPeer); heir != mesh.UnknownPeerName {
|
|
return heir
|
|
}
|
|
// next try alive peers that have IPAM enabled but have no entries
|
|
if heir := alloc.pickPeerFromNicknames(alloc.isKnownPeer); heir != mesh.UnknownPeerName {
|
|
return heir
|
|
}
|
|
// next try disappeared peers that still have entries
|
|
t := func(mesh.PeerName) bool { return true }
|
|
if heir := alloc.ring.PickPeerForTransfer(t); heir != mesh.UnknownPeerName {
|
|
return heir
|
|
}
|
|
// finally, disappeared peers that that passively participated in IPAM
|
|
return alloc.pickPeerFromNicknames(t)
|
|
}
|
|
|
|
// Shutdown (Sync)
|
|
func (alloc *Allocator) Shutdown() {
|
|
alloc.infof("Shutdown")
|
|
doneChan := make(chan struct{})
|
|
alloc.actionChan <- func() {
|
|
alloc.shuttingDown = true
|
|
alloc.cancelOps(&alloc.pendingClaims)
|
|
alloc.cancelOps(&alloc.pendingAllocates)
|
|
if heir := alloc.pickPeerForTransfer(); heir != mesh.UnknownPeerName {
|
|
alloc.ring.Transfer(alloc.ourName, heir)
|
|
alloc.space.Clear()
|
|
alloc.gossip.GossipBroadcast(alloc.Gossip())
|
|
time.Sleep(100 * time.Millisecond)
|
|
}
|
|
doneChan <- struct{}{}
|
|
}
|
|
<-doneChan
|
|
}
|
|
|
|
// AdminTakeoverRanges (Sync) - take over the ranges owned by a given peer.
|
|
// Only done on adminstrator command.
|
|
func (alloc *Allocator) AdminTakeoverRanges(peerNameOrNickname string) error {
|
|
resultChan := make(chan error)
|
|
alloc.actionChan <- func() {
|
|
peername, err := alloc.lookupPeername(peerNameOrNickname)
|
|
if err != nil {
|
|
resultChan <- fmt.Errorf("Cannot find peer '%s'", peerNameOrNickname)
|
|
return
|
|
}
|
|
|
|
alloc.debugln("AdminTakeoverRanges:", peername)
|
|
if peername == alloc.ourName {
|
|
resultChan <- fmt.Errorf("Cannot take over ranges from yourself!")
|
|
return
|
|
}
|
|
|
|
newRanges, err := alloc.ring.Transfer(peername, alloc.ourName)
|
|
alloc.space.AddRanges(newRanges)
|
|
resultChan <- err
|
|
}
|
|
return <-resultChan
|
|
}
|
|
|
|
// Lookup a PeerName by nickname or stringified PeerName. We can't
|
|
// call into the router for this because we are interested in peers
|
|
// that have gone away but are still in the ring, which is why we
|
|
// maintain our own nicknames map.
|
|
func (alloc *Allocator) lookupPeername(name string) (mesh.PeerName, error) {
|
|
for peername, nickname := range alloc.nicknames {
|
|
if nickname == name {
|
|
return peername, nil
|
|
}
|
|
}
|
|
|
|
return mesh.PeerNameFromString(name)
|
|
}
|
|
|
|
// Restrict the peers in "nicknames" to those in the ring plus peers known to the router
|
|
func (alloc *Allocator) pruneNicknames() {
|
|
ringPeers := alloc.ring.PeerNames()
|
|
for name := range alloc.nicknames {
|
|
if _, ok := ringPeers[name]; !ok && !alloc.isKnownPeer(name) {
|
|
delete(alloc.nicknames, name)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (alloc *Allocator) annotatePeernames(names []mesh.PeerName) []string {
|
|
var res []string
|
|
for _, name := range names {
|
|
if nickname, found := alloc.nicknames[name]; found {
|
|
res = append(res, fmt.Sprint(name, "(", nickname, ")"))
|
|
} else {
|
|
res = append(res, name.String())
|
|
}
|
|
}
|
|
return res
|
|
}
|
|
|
|
func decodeRange(msg []byte) (r address.Range, err error) {
|
|
decoder := gob.NewDecoder(bytes.NewReader(msg))
|
|
return r, decoder.Decode(&r)
|
|
}
|
|
|
|
// OnGossipUnicast (Sync)
|
|
func (alloc *Allocator) OnGossipUnicast(sender mesh.PeerName, msg []byte) error {
|
|
alloc.debugln("OnGossipUnicast from", sender, ": ", len(msg), "bytes")
|
|
resultChan := make(chan error)
|
|
alloc.actionChan <- func() {
|
|
switch msg[0] {
|
|
case msgSpaceRequest:
|
|
// some other peer asked us for space
|
|
r, err := decodeRange(msg[1:])
|
|
if err == nil {
|
|
alloc.donateSpace(r, sender)
|
|
}
|
|
resultChan <- err
|
|
case msgSpaceRequestDenied:
|
|
r, err := decodeRange(msg[1:])
|
|
if err == nil {
|
|
alloc.spaceRequestDenied(sender, r)
|
|
}
|
|
resultChan <- err
|
|
case msgRingUpdate:
|
|
resultChan <- alloc.update(sender, msg[1:])
|
|
}
|
|
}
|
|
return <-resultChan
|
|
}
|
|
|
|
// OnGossipBroadcast (Sync)
|
|
func (alloc *Allocator) OnGossipBroadcast(sender mesh.PeerName, msg []byte) (mesh.GossipData, error) {
|
|
alloc.debugln("OnGossipBroadcast from", sender, ":", len(msg), "bytes")
|
|
resultChan := make(chan error)
|
|
alloc.actionChan <- func() {
|
|
resultChan <- alloc.update(sender, msg)
|
|
}
|
|
return alloc.Gossip(), <-resultChan
|
|
}
|
|
|
|
type gossipState struct {
|
|
// We send a timstamp along with the information to be
|
|
// gossipped in order to detect skewed clocks
|
|
Now int64
|
|
Nicknames map[mesh.PeerName]string
|
|
|
|
Paxos paxos.GossipState
|
|
Ring *ring.Ring
|
|
}
|
|
|
|
func (alloc *Allocator) encode() []byte {
|
|
data := gossipState{
|
|
Now: alloc.now().Unix(),
|
|
Nicknames: alloc.nicknames,
|
|
}
|
|
|
|
// We're only interested in Paxos until we have a Ring.
|
|
if alloc.ring.Empty() {
|
|
data.Paxos = alloc.paxos.GossipState()
|
|
} else {
|
|
data.Ring = alloc.ring
|
|
}
|
|
buf := new(bytes.Buffer)
|
|
enc := gob.NewEncoder(buf)
|
|
if err := enc.Encode(data); err != nil {
|
|
panic(err)
|
|
}
|
|
return buf.Bytes()
|
|
}
|
|
|
|
// Encode (Sync)
|
|
func (alloc *Allocator) Encode() []byte {
|
|
resultChan := make(chan []byte)
|
|
alloc.actionChan <- func() {
|
|
resultChan <- alloc.encode()
|
|
}
|
|
return <-resultChan
|
|
}
|
|
|
|
// OnGossip (Sync)
|
|
func (alloc *Allocator) OnGossip(msg []byte) (mesh.GossipData, error) {
|
|
alloc.debugln("Allocator.OnGossip:", len(msg), "bytes")
|
|
resultChan := make(chan error)
|
|
alloc.actionChan <- func() {
|
|
resultChan <- alloc.update(mesh.UnknownPeerName, msg)
|
|
}
|
|
return nil, <-resultChan // for now, we never propagate updates. TBD
|
|
}
|
|
|
|
// GossipData implementation is trivial - we always gossip the latest
|
|
// data we have at time of sending
|
|
type ipamGossipData struct {
|
|
alloc *Allocator
|
|
}
|
|
|
|
func (d *ipamGossipData) Merge(other mesh.GossipData) mesh.GossipData {
|
|
return d // no-op
|
|
}
|
|
|
|
func (d *ipamGossipData) Encode() [][]byte {
|
|
return [][]byte{d.alloc.Encode()}
|
|
}
|
|
|
|
// Gossip returns a GossipData implementation, which in this case always
|
|
// returns the latest ring state (and does nothing on merge)
|
|
func (alloc *Allocator) Gossip() mesh.GossipData {
|
|
return &ipamGossipData{alloc}
|
|
}
|
|
|
|
// SetInterfaces gives the allocator two interfaces for talking to the outside world
|
|
func (alloc *Allocator) SetInterfaces(gossip mesh.Gossip) {
|
|
alloc.gossip = gossip
|
|
}
|
|
|
|
// ACTOR server
|
|
|
|
func (alloc *Allocator) actorLoop(actionChan <-chan func()) {
|
|
for {
|
|
var tickChan <-chan time.Time
|
|
if alloc.paxosTicker != nil {
|
|
tickChan = alloc.paxosTicker.C
|
|
}
|
|
|
|
select {
|
|
case action := <-actionChan:
|
|
if action == nil {
|
|
return
|
|
}
|
|
action()
|
|
case <-tickChan:
|
|
alloc.propose()
|
|
}
|
|
|
|
alloc.assertInvariants()
|
|
alloc.reportFreeSpace()
|
|
}
|
|
}
|
|
|
|
// Helper functions
|
|
|
|
// Ensure we are making progress towards an established ring
|
|
func (alloc *Allocator) establishRing() {
|
|
if !alloc.ring.Empty() || alloc.paxosTicker != nil {
|
|
return
|
|
}
|
|
|
|
alloc.propose()
|
|
if ok, cons := alloc.paxos.Consensus(); ok {
|
|
// If the quorum was 1, then proposing immediately
|
|
// leads to consensus
|
|
alloc.createRing(cons.Value)
|
|
} else {
|
|
// re-try until we get consensus
|
|
alloc.paxosTicker = time.NewTicker(paxosInterval)
|
|
}
|
|
}
|
|
|
|
func (alloc *Allocator) createRing(peers []mesh.PeerName) {
|
|
alloc.debugln("Paxos consensus:", peers)
|
|
alloc.ring.ClaimForPeers(normalizeConsensus(peers))
|
|
alloc.gossip.GossipBroadcast(alloc.Gossip())
|
|
alloc.ringUpdated()
|
|
}
|
|
|
|
func (alloc *Allocator) ringUpdated() {
|
|
// When we have a ring, we don't need paxos any more
|
|
if alloc.paxos != nil {
|
|
alloc.paxos = nil
|
|
|
|
if alloc.paxosTicker != nil {
|
|
alloc.paxosTicker.Stop()
|
|
alloc.paxosTicker = nil
|
|
}
|
|
}
|
|
|
|
alloc.space.UpdateRanges(alloc.ring.OwnedRanges())
|
|
alloc.tryPendingOps()
|
|
}
|
|
|
|
// For compatibility with sort.Interface
|
|
type peerNames []mesh.PeerName
|
|
|
|
func (a peerNames) Len() int { return len(a) }
|
|
func (a peerNames) Less(i, j int) bool { return a[i] < a[j] }
|
|
func (a peerNames) Swap(i, j int) { a[i], a[j] = a[j], a[i] }
|
|
|
|
// When we get a consensus from Paxos, the peer names are not in a
|
|
// defined order and may contain duplicates. This function sorts them
|
|
// and de-dupes.
|
|
func normalizeConsensus(consensus []mesh.PeerName) []mesh.PeerName {
|
|
if len(consensus) == 0 {
|
|
return nil
|
|
}
|
|
|
|
peers := make(peerNames, len(consensus))
|
|
copy(peers, consensus)
|
|
sort.Sort(peers)
|
|
|
|
dst := 0
|
|
for src := 1; src < len(peers); src++ {
|
|
if peers[dst] != peers[src] {
|
|
dst++
|
|
peers[dst] = peers[src]
|
|
}
|
|
}
|
|
|
|
return peers[:dst+1]
|
|
}
|
|
|
|
func (alloc *Allocator) propose() {
|
|
alloc.debugf("Paxos proposing")
|
|
alloc.paxos.Propose()
|
|
alloc.gossip.GossipBroadcast(alloc.Gossip())
|
|
}
|
|
|
|
func encodeRange(r address.Range) []byte {
|
|
buf := new(bytes.Buffer)
|
|
enc := gob.NewEncoder(buf)
|
|
if err := enc.Encode(r); err != nil {
|
|
panic(err)
|
|
}
|
|
return buf.Bytes()
|
|
}
|
|
|
|
func (alloc *Allocator) sendSpaceRequest(dest mesh.PeerName, r address.Range) error {
|
|
msg := append([]byte{msgSpaceRequest}, encodeRange(r)...)
|
|
return alloc.gossip.GossipUnicast(dest, msg)
|
|
}
|
|
|
|
func (alloc *Allocator) sendSpaceRequestDenied(dest mesh.PeerName, r address.Range) error {
|
|
msg := append([]byte{msgSpaceRequestDenied}, encodeRange(r)...)
|
|
return alloc.gossip.GossipUnicast(dest, msg)
|
|
}
|
|
|
|
func (alloc *Allocator) sendRingUpdate(dest mesh.PeerName) {
|
|
msg := append([]byte{msgRingUpdate}, alloc.encode()...)
|
|
alloc.gossip.GossipUnicast(dest, msg)
|
|
}
|
|
|
|
func (alloc *Allocator) update(sender mesh.PeerName, msg []byte) error {
|
|
reader := bytes.NewReader(msg)
|
|
decoder := gob.NewDecoder(reader)
|
|
var data gossipState
|
|
var err error
|
|
|
|
if err := decoder.Decode(&data); err != nil {
|
|
return err
|
|
}
|
|
|
|
deltat := time.Unix(data.Now, 0).Sub(alloc.now())
|
|
if deltat > time.Hour || -deltat > time.Hour {
|
|
return fmt.Errorf("clock skew of %v detected, ignoring update", deltat)
|
|
}
|
|
|
|
// Merge nicknames
|
|
for peer, nickname := range data.Nicknames {
|
|
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 err = alloc.ring.Merge(*data.Ring); err {
|
|
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))
|
|
case ring.ErrDifferentRange:
|
|
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
|
|
alloc.sendRingUpdate(sender)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (alloc *Allocator) donateSpace(r address.Range, to mesh.PeerName) {
|
|
// No matter what we do, we'll send a unicast gossip
|
|
// of our ring back to tha chap who asked for space.
|
|
// This serves to both tell him of any space we might
|
|
// have given him, or tell him where he might find some
|
|
// more.
|
|
defer alloc.sendRingUpdate(to)
|
|
|
|
alloc.debugln("Peer", to, "asked me for space")
|
|
chunk, ok := alloc.space.Donate(r)
|
|
if !ok {
|
|
free := alloc.space.NumFreeAddressesInRange(r)
|
|
common.Assert(free == 0)
|
|
alloc.debugln("No space to give to peer", to)
|
|
// separate message maintains backwards-compatibility:
|
|
// down-level peers will ignore this and still get the ring update.
|
|
alloc.sendSpaceRequestDenied(to, r)
|
|
return
|
|
}
|
|
alloc.debugln("Giving range", chunk, "to", to)
|
|
alloc.ring.GrantRangeToHost(chunk.Start, chunk.End, to)
|
|
alloc.sendRingUpdate(to)
|
|
}
|
|
|
|
func (alloc *Allocator) assertInvariants() {
|
|
// We need to ensure all ranges the ring thinks we own have
|
|
// a corresponding space in the space set, and vice versa
|
|
checkSpace := space.New()
|
|
checkSpace.AddRanges(alloc.ring.OwnedRanges())
|
|
ranges := checkSpace.OwnedRanges()
|
|
spaces := alloc.space.OwnedRanges()
|
|
|
|
common.Assert(len(ranges) == len(spaces))
|
|
|
|
for i := 0; i < len(ranges); i++ {
|
|
r := ranges[i]
|
|
s := spaces[i]
|
|
common.Assert(s.Start == r.Start && s.End == r.End)
|
|
}
|
|
}
|
|
|
|
func (alloc *Allocator) reportFreeSpace() {
|
|
ranges := alloc.ring.OwnedRanges()
|
|
if len(ranges) == 0 {
|
|
return
|
|
}
|
|
|
|
freespace := make(map[address.Address]address.Offset)
|
|
for _, r := range ranges {
|
|
freespace[r.Start] = alloc.space.NumFreeAddressesInRange(r)
|
|
}
|
|
alloc.ring.ReportFree(freespace)
|
|
}
|
|
|
|
// Owned addresses
|
|
|
|
// NB: addr must not be owned by ident already
|
|
func (alloc *Allocator) addOwned(ident string, addr address.Address) {
|
|
alloc.owned[ident] = append(alloc.owned[ident], addr)
|
|
}
|
|
|
|
func (alloc *Allocator) lookupOwned(ident string, r address.Range) (address.Address, bool) {
|
|
for _, addr := range alloc.owned[ident] {
|
|
if r.Contains(addr) {
|
|
return addr, true
|
|
}
|
|
}
|
|
return 0, false
|
|
}
|
|
|
|
func (alloc *Allocator) findOwner(addr address.Address) string {
|
|
for ident, addrs := range alloc.owned {
|
|
for _, candidate := range addrs {
|
|
if candidate == addr {
|
|
return ident
|
|
}
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// Logging
|
|
|
|
func (alloc *Allocator) infof(fmt string, args ...interface{}) {
|
|
common.Log.Infof("[allocator %s] "+fmt, append([]interface{}{alloc.ourName}, args...)...)
|
|
}
|
|
func (alloc *Allocator) debugln(args ...interface{}) {
|
|
common.Log.Debugln(append([]interface{}{fmt.Sprintf("[allocator %s]:", alloc.ourName)}, args...)...)
|
|
}
|
|
func (alloc *Allocator) debugf(fmt string, args ...interface{}) {
|
|
common.Log.Debugf("[allocator %s] "+fmt, append([]interface{}{alloc.ourName}, args...)...)
|
|
}
|