package serf
|
|
import (
|
"bytes"
|
"encoding/base64"
|
"encoding/json"
|
"errors"
|
"fmt"
|
"io/ioutil"
|
"log"
|
"math/rand"
|
"net"
|
"os"
|
"strconv"
|
"sync"
|
"sync/atomic"
|
"time"
|
|
"github.com/armon/go-metrics"
|
"github.com/hashicorp/go-msgpack/codec"
|
"github.com/hashicorp/memberlist"
|
"basic.com/valib/serf.git/coordinate"
|
)
|
|
// These are the protocol versions that Serf can _understand_. These are
|
// Serf-level protocol versions that are passed down as the delegate
|
// version to memberlist below.
|
const (
|
ProtocolVersionMin uint8 = 2
|
ProtocolVersionMax = 5
|
)
|
|
const (
|
// Used to detect if the meta data is tags
|
// or if it is a raw role
|
tagMagicByte uint8 = 255
|
)
|
|
var (
|
// FeatureNotSupported is returned if a feature cannot be used
|
// due to an older protocol version being used.
|
FeatureNotSupported = fmt.Errorf("Feature not supported")
|
)
|
|
func init() {
|
// Seed the random number generator
|
rand.Seed(time.Now().UnixNano())
|
}
|
|
// Serf is a single node that is part of a single cluster that gets
|
// events about joins/leaves/failures/etc. It is created with the Create
|
// method.
|
//
|
// All functions on the Serf structure are safe to call concurrently.
|
type Serf struct {
|
// The clocks for different purposes. These MUST be the first things
|
// in this struct due to Golang issue #599.
|
clock LamportClock
|
eventClock LamportClock
|
queryClock LamportClock
|
|
broadcasts *memberlist.TransmitLimitedQueue
|
config *Config
|
failedMembers []*memberState
|
leftMembers []*memberState
|
memberlist *memberlist.Memberlist
|
memberLock sync.RWMutex
|
members map[string]*memberState
|
|
// recentIntents the lamport time and type of intent for a given node in
|
// case we get an intent before the relevant memberlist event. This is
|
// indexed by node, and always store the latest lamport time / intent
|
// we've seen. The memberLock protects this structure.
|
recentIntents map[string]nodeIntent
|
|
eventBroadcasts *memberlist.TransmitLimitedQueue
|
eventBuffer []*userEvents
|
eventJoinIgnore atomic.Value
|
eventMinTime LamportTime
|
eventLock sync.RWMutex
|
|
queryBroadcasts *memberlist.TransmitLimitedQueue
|
queryBuffer []*queries
|
queryMinTime LamportTime
|
queryResponse map[LamportTime]*QueryResponse
|
queryLock sync.RWMutex
|
|
logger *log.Logger
|
joinLock sync.Mutex
|
stateLock sync.Mutex
|
state SerfState
|
shutdownCh chan struct{}
|
|
snapshotter *Snapshotter
|
keyManager *KeyManager
|
|
coordClient *coordinate.Client
|
coordCache map[string]*coordinate.Coordinate
|
coordCacheLock sync.RWMutex
|
}
|
|
// SerfState is the state of the Serf instance.
|
type SerfState int
|
|
const (
|
SerfAlive SerfState = iota
|
SerfLeaving
|
SerfLeft
|
SerfShutdown
|
)
|
|
func (s SerfState) String() string {
|
switch s {
|
case SerfAlive:
|
return "alive"
|
case SerfLeaving:
|
return "leaving"
|
case SerfLeft:
|
return "left"
|
case SerfShutdown:
|
return "shutdown"
|
default:
|
return "unknown"
|
}
|
}
|
|
// Member is a single member of the Serf cluster.
|
type Member struct {
|
Name string
|
Addr net.IP
|
Port uint16
|
Tags map[string]string
|
Status MemberStatus
|
|
// The minimum, maximum, and current values of the protocol versions
|
// and delegate (Serf) protocol versions that each member can understand
|
// or is speaking.
|
ProtocolMin uint8
|
ProtocolMax uint8
|
ProtocolCur uint8
|
DelegateMin uint8
|
DelegateMax uint8
|
DelegateCur uint8
|
}
|
|
// MemberStatus is the state that a member is in.
|
type MemberStatus int
|
|
const (
|
StatusNone MemberStatus = iota
|
StatusAlive
|
StatusLeaving
|
StatusLeft
|
StatusFailed
|
)
|
|
func (s MemberStatus) String() string {
|
switch s {
|
case StatusNone:
|
return "none"
|
case StatusAlive:
|
return "alive"
|
case StatusLeaving:
|
return "leaving"
|
case StatusLeft:
|
return "left"
|
case StatusFailed:
|
return "failed"
|
default:
|
panic(fmt.Sprintf("unknown MemberStatus: %d", s))
|
}
|
}
|
|
// memberState is used to track members that are no longer active due to
|
// leaving, failing, partitioning, etc. It tracks the member along with
|
// when that member was marked as leaving.
|
type memberState struct {
|
Member
|
statusLTime LamportTime // lamport clock time of last received message
|
leaveTime time.Time // wall clock time of leave
|
}
|
|
// nodeIntent is used to buffer intents for out-of-order deliveries.
|
type nodeIntent struct {
|
// Type is the intent being tracked. Only messageJoinType and
|
// messageLeaveType are tracked.
|
Type messageType
|
|
// WallTime is the wall clock time we saw this intent in order to
|
// expire it from the buffer.
|
WallTime time.Time
|
|
// LTime is the Lamport time, used for cluster-wide ordering of events.
|
LTime LamportTime
|
}
|
|
// userEvent is used to buffer events to prevent re-delivery
|
type userEvent struct {
|
Name string
|
Payload []byte
|
}
|
|
func (ue *userEvent) Equals(other *userEvent) bool {
|
if ue.Name != other.Name {
|
return false
|
}
|
if bytes.Compare(ue.Payload, other.Payload) != 0 {
|
return false
|
}
|
return true
|
}
|
|
// userEvents stores all the user events at a specific time
|
type userEvents struct {
|
LTime LamportTime
|
Events []userEvent
|
}
|
|
// queries stores all the query ids at a specific time
|
type queries struct {
|
LTime LamportTime
|
QueryIDs []uint32
|
}
|
|
const (
|
snapshotSizeLimit = 128 * 1024 // Maximum 128 KB snapshot
|
UserEventSizeLimit = 9 * 10 * 1024 // Maximum 9KB for event name and payload
|
)
|
|
// Create creates a new Serf instance, starting all the background tasks
|
// to maintain cluster membership information.
|
//
|
// After calling this function, the configuration should no longer be used
|
// or modified by the caller.
|
func Create(conf *Config) (*Serf, error) {
|
conf.Init()
|
if conf.ProtocolVersion < ProtocolVersionMin {
|
return nil, fmt.Errorf("Protocol version '%d' too low. Must be in range: [%d, %d]",
|
conf.ProtocolVersion, ProtocolVersionMin, ProtocolVersionMax)
|
} else if conf.ProtocolVersion > ProtocolVersionMax {
|
return nil, fmt.Errorf("Protocol version '%d' too high. Must be in range: [%d, %d]",
|
conf.ProtocolVersion, ProtocolVersionMin, ProtocolVersionMax)
|
}
|
|
if conf.UserEventSizeLimit > UserEventSizeLimit {
|
return nil, fmt.Errorf("user event size limit exceeds limit of %d bytes", UserEventSizeLimit)
|
}
|
|
logger := conf.Logger
|
if logger == nil {
|
logOutput := conf.LogOutput
|
if logOutput == nil {
|
logOutput = os.Stderr
|
}
|
logger = log.New(logOutput, "", log.LstdFlags)
|
}
|
|
serf := &Serf{
|
config: conf,
|
logger: logger,
|
members: make(map[string]*memberState),
|
queryResponse: make(map[LamportTime]*QueryResponse),
|
shutdownCh: make(chan struct{}),
|
state: SerfAlive,
|
}
|
serf.eventJoinIgnore.Store(false)
|
|
// Check that the meta data length is okay
|
if len(serf.encodeTags(conf.Tags)) > memberlist.MetaMaxSize {
|
return nil, fmt.Errorf("Encoded length of tags exceeds limit of %d bytes", memberlist.MetaMaxSize)
|
}
|
|
// Check if serf member event coalescing is enabled
|
if conf.CoalescePeriod > 0 && conf.QuiescentPeriod > 0 && conf.EventCh != nil {
|
c := &memberEventCoalescer{
|
lastEvents: make(map[string]EventType),
|
latestEvents: make(map[string]coalesceEvent),
|
}
|
|
conf.EventCh = coalescedEventCh(conf.EventCh, serf.shutdownCh,
|
conf.CoalescePeriod, conf.QuiescentPeriod, c)
|
}
|
|
// Check if user event coalescing is enabled
|
if conf.UserCoalescePeriod > 0 && conf.UserQuiescentPeriod > 0 && conf.EventCh != nil {
|
c := &userEventCoalescer{
|
events: make(map[string]*latestUserEvents),
|
}
|
|
conf.EventCh = coalescedEventCh(conf.EventCh, serf.shutdownCh,
|
conf.UserCoalescePeriod, conf.UserQuiescentPeriod, c)
|
}
|
|
// Listen for internal Serf queries. This is setup before the snapshotter, since
|
// we want to capture the query-time, but the internal listener does not passthrough
|
// the queries
|
outCh, err := newSerfQueries(serf, serf.logger, conf.EventCh, serf.shutdownCh)
|
if err != nil {
|
return nil, fmt.Errorf("Failed to setup serf query handler: %v", err)
|
}
|
conf.EventCh = outCh
|
|
// Set up network coordinate client.
|
if !conf.DisableCoordinates {
|
serf.coordClient, err = coordinate.NewClient(coordinate.DefaultConfig())
|
if err != nil {
|
return nil, fmt.Errorf("Failed to create coordinate client: %v", err)
|
}
|
}
|
|
// Try access the snapshot
|
var oldClock, oldEventClock, oldQueryClock LamportTime
|
var prev []*PreviousNode
|
if conf.SnapshotPath != "" {
|
eventCh, snap, err := NewSnapshotter(
|
conf.SnapshotPath,
|
snapshotSizeLimit,
|
conf.RejoinAfterLeave,
|
serf.logger,
|
&serf.clock,
|
conf.EventCh,
|
serf.shutdownCh)
|
if err != nil {
|
return nil, fmt.Errorf("Failed to setup snapshot: %v", err)
|
}
|
serf.snapshotter = snap
|
conf.EventCh = eventCh
|
prev = snap.AliveNodes()
|
oldClock = snap.LastClock()
|
oldEventClock = snap.LastEventClock()
|
oldQueryClock = snap.LastQueryClock()
|
serf.eventMinTime = oldEventClock + 1
|
serf.queryMinTime = oldQueryClock + 1
|
}
|
|
// Set up the coordinate cache. We do this after we read the snapshot to
|
// make sure we get a good initial value from there, if we got one.
|
if !conf.DisableCoordinates {
|
serf.coordCache = make(map[string]*coordinate.Coordinate)
|
serf.coordCache[conf.NodeName] = serf.coordClient.GetCoordinate()
|
}
|
|
// Setup the various broadcast queues, which we use to send our own
|
// custom broadcasts along the gossip channel.
|
serf.broadcasts = &memberlist.TransmitLimitedQueue{
|
NumNodes: serf.NumNodes,
|
RetransmitMult: conf.MemberlistConfig.RetransmitMult,
|
}
|
serf.eventBroadcasts = &memberlist.TransmitLimitedQueue{
|
NumNodes: serf.NumNodes,
|
RetransmitMult: conf.MemberlistConfig.RetransmitMult,
|
}
|
serf.queryBroadcasts = &memberlist.TransmitLimitedQueue{
|
NumNodes: serf.NumNodes,
|
RetransmitMult: conf.MemberlistConfig.RetransmitMult,
|
}
|
|
// Create the buffer for recent intents
|
serf.recentIntents = make(map[string]nodeIntent)
|
|
// Create a buffer for events and queries
|
serf.eventBuffer = make([]*userEvents, conf.EventBuffer)
|
serf.queryBuffer = make([]*queries, conf.QueryBuffer)
|
|
// Ensure our lamport clock is at least 1, so that the default
|
// join LTime of 0 does not cause issues
|
serf.clock.Increment()
|
serf.eventClock.Increment()
|
serf.queryClock.Increment()
|
|
// Restore the clock from snap if we have one
|
serf.clock.Witness(oldClock)
|
serf.eventClock.Witness(oldEventClock)
|
serf.queryClock.Witness(oldQueryClock)
|
|
// Modify the memberlist configuration with keys that we set
|
conf.MemberlistConfig.Events = &eventDelegate{serf: serf}
|
conf.MemberlistConfig.Conflict = &conflictDelegate{serf: serf}
|
conf.MemberlistConfig.Delegate = &delegate{serf: serf}
|
conf.MemberlistConfig.DelegateProtocolVersion = conf.ProtocolVersion
|
conf.MemberlistConfig.DelegateProtocolMin = ProtocolVersionMin
|
conf.MemberlistConfig.DelegateProtocolMax = ProtocolVersionMax
|
conf.MemberlistConfig.Name = conf.NodeName
|
conf.MemberlistConfig.ProtocolVersion = ProtocolVersionMap[conf.ProtocolVersion]
|
if !conf.DisableCoordinates {
|
conf.MemberlistConfig.Ping = &pingDelegate{serf: serf}
|
}
|
|
// Setup a merge delegate if necessary
|
if conf.Merge != nil {
|
md := &mergeDelegate{serf: serf}
|
conf.MemberlistConfig.Merge = md
|
conf.MemberlistConfig.Alive = md
|
}
|
|
// Create the underlying memberlist that will manage membership
|
// and failure detection for the Serf instance.
|
memberlist, err := memberlist.Create(conf.MemberlistConfig)
|
if err != nil {
|
return nil, fmt.Errorf("Failed to create memberlist: %v", err)
|
}
|
|
serf.memberlist = memberlist
|
|
// Create a key manager for handling all encryption key changes
|
serf.keyManager = &KeyManager{serf: serf}
|
|
// Start the background tasks. See the documentation above each method
|
// for more information on their role.
|
go serf.handleReap()
|
go serf.handleReconnect()
|
go serf.checkQueueDepth("Intent", serf.broadcasts)
|
go serf.checkQueueDepth("Event", serf.eventBroadcasts)
|
go serf.checkQueueDepth("Query", serf.queryBroadcasts)
|
|
// Attempt to re-join the cluster if we have known nodes
|
if len(prev) != 0 {
|
go serf.handleRejoin(prev)
|
}
|
|
return serf, nil
|
}
|
|
// ProtocolVersion returns the current protocol version in use by Serf.
|
// This is the Serf protocol version, not the memberlist protocol version.
|
func (s *Serf) ProtocolVersion() uint8 {
|
return s.config.ProtocolVersion
|
}
|
|
// EncryptionEnabled is a predicate that determines whether or not encryption
|
// is enabled, which can be possible in one of 2 cases:
|
// - Single encryption key passed at agent start (no persistence)
|
// - Keyring file provided at agent start
|
func (s *Serf) EncryptionEnabled() bool {
|
return s.config.MemberlistConfig.Keyring != nil
|
}
|
|
// KeyManager returns the key manager for the current Serf instance.
|
func (s *Serf) KeyManager() *KeyManager {
|
return s.keyManager
|
}
|
|
// UserEvent is used to broadcast a custom user event with a given
|
// name and payload. If the configured size limit is exceeded and error will be returned.
|
// If coalesce is enabled, nodes are allowed to coalesce this event.
|
// Coalescing is only available starting in v0.2
|
func (s *Serf) UserEvent(name string, payload []byte, coalesce bool) error {
|
payloadSizeBeforeEncoding := len(name) + len(payload)
|
|
// Check size before encoding to prevent needless encoding and return early if it's over the specified limit.
|
if payloadSizeBeforeEncoding > s.config.UserEventSizeLimit {
|
return fmt.Errorf(
|
"user event exceeds configured limit of %d bytes before encoding",
|
s.config.UserEventSizeLimit,
|
)
|
}
|
|
if payloadSizeBeforeEncoding > UserEventSizeLimit {
|
return fmt.Errorf(
|
"user event exceeds sane limit of %d bytes before encoding",
|
UserEventSizeLimit,
|
)
|
}
|
|
// Create a message
|
msg := messageUserEvent{
|
LTime: s.eventClock.Time(),
|
Name: name,
|
Payload: payload,
|
CC: coalesce,
|
}
|
|
// Start broadcasting the event
|
raw, err := encodeMessage(messageUserEventType, &msg)
|
if err != nil {
|
return err
|
}
|
|
// Check the size after encoding to be sure again that
|
// we're not attempting to send over the specified size limit.
|
if len(raw) > s.config.UserEventSizeLimit {
|
return fmt.Errorf(
|
"encoded user event exceeds configured limit of %d bytes after encoding",
|
s.config.UserEventSizeLimit,
|
)
|
}
|
|
if len(raw) > UserEventSizeLimit {
|
return fmt.Errorf(
|
"encoded user event exceeds sane limit of %d bytes before encoding",
|
UserEventSizeLimit,
|
)
|
}
|
|
s.eventClock.Increment()
|
|
// Process update locally
|
s.handleUserEvent(&msg)
|
|
s.eventBroadcasts.QueueBroadcast(&broadcast{
|
msg: raw,
|
})
|
return nil
|
}
|
|
// Query is used to broadcast a new query. The query must be fairly small,
|
// and an error will be returned if the size limit is exceeded. This is only
|
// available with protocol version 4 and newer. Query parameters are optional,
|
// and if not provided, a sane set of defaults will be used.
|
func (s *Serf) Query(name string, payload []byte, params *QueryParam) (*QueryResponse, error) {
|
// Check that the latest protocol is in use
|
if s.ProtocolVersion() < 4 {
|
return nil, FeatureNotSupported
|
}
|
|
// Provide default parameters if none given
|
if params == nil {
|
params = s.DefaultQueryParams()
|
} else if params.Timeout == 0 {
|
params.Timeout = s.DefaultQueryTimeout()
|
}
|
|
// Get the local node
|
local := s.memberlist.LocalNode()
|
|
// Encode the filters
|
filters, err := params.encodeFilters()
|
if err != nil {
|
return nil, fmt.Errorf("Failed to format filters: %v", err)
|
}
|
|
// Setup the flags
|
var flags uint32
|
if params.RequestAck {
|
flags |= queryFlagAck
|
}
|
|
// Create a message
|
q := messageQuery{
|
LTime: s.queryClock.Time(),
|
ID: uint32(rand.Int31()),
|
Addr: local.Addr,
|
Port: local.Port,
|
SourceNode: local.Name,
|
Filters: filters,
|
Flags: flags,
|
RelayFactor: params.RelayFactor,
|
Timeout: params.Timeout,
|
Name: name,
|
Payload: payload,
|
}
|
|
// Encode the query
|
raw, err := encodeMessage(messageQueryType, &q)
|
if err != nil {
|
return nil, err
|
}
|
|
// Check the size
|
if len(raw) > s.config.QuerySizeLimit {
|
return nil, fmt.Errorf("query exceeds limit of %d bytes", s.config.QuerySizeLimit)
|
}
|
|
// Register QueryResponse to track acks and responses
|
resp := newQueryResponse(s.memberlist.NumMembers(), &q)
|
s.registerQueryResponse(params.Timeout, resp)
|
|
// Process query locally
|
s.handleQuery(&q)
|
|
// Start broadcasting the event
|
s.queryBroadcasts.QueueBroadcast(&broadcast{
|
msg: raw,
|
})
|
return resp, nil
|
}
|
|
// registerQueryResponse is used to setup the listeners for the query,
|
// and to schedule closing the query after the timeout.
|
func (s *Serf) registerQueryResponse(timeout time.Duration, resp *QueryResponse) {
|
s.queryLock.Lock()
|
defer s.queryLock.Unlock()
|
|
// Map the LTime to the QueryResponse. This is necessarily 1-to-1,
|
// since we increment the time for each new query.
|
s.queryResponse[resp.lTime] = resp
|
|
// Setup a timer to close the response and deregister after the timeout
|
time.AfterFunc(timeout, func() {
|
s.queryLock.Lock()
|
delete(s.queryResponse, resp.lTime)
|
resp.Close()
|
s.queryLock.Unlock()
|
})
|
}
|
|
// SetTags is used to dynamically update the tags associated with
|
// the local node. This will propagate the change to the rest of
|
// the cluster. Blocks until a the message is broadcast out.
|
func (s *Serf) SetTags(tags map[string]string) error {
|
// Check that the meta data length is okay
|
if len(s.encodeTags(tags)) > memberlist.MetaMaxSize {
|
return fmt.Errorf("Encoded length of tags exceeds limit of %d bytes",
|
memberlist.MetaMaxSize)
|
}
|
|
// Update the config
|
s.config.Tags = tags
|
|
// Trigger a memberlist update
|
return s.memberlist.UpdateNode(s.config.BroadcastTimeout)
|
}
|
|
// Join joins an existing Serf cluster. Returns the number of nodes
|
// successfully contacted. The returned error will be non-nil only in the
|
// case that no nodes could be contacted. If ignoreOld is true, then any
|
// user messages sent prior to the join will be ignored.
|
func (s *Serf) Join(existing []string, ignoreOld bool) (int, error) {
|
// Do a quick state check
|
if s.State() != SerfAlive {
|
return 0, fmt.Errorf("Serf can't Join after Leave or Shutdown")
|
}
|
|
// Hold the joinLock, this is to make eventJoinIgnore safe
|
s.joinLock.Lock()
|
defer s.joinLock.Unlock()
|
|
// Ignore any events from a potential join. This is safe since we hold
|
// the joinLock and nobody else can be doing a Join
|
if ignoreOld {
|
s.eventJoinIgnore.Store(true)
|
defer func() {
|
s.eventJoinIgnore.Store(false)
|
}()
|
}
|
|
// Have memberlist attempt to join
|
num, err := s.memberlist.Join(existing)
|
|
// If we joined any nodes, broadcast the join message
|
if num > 0 {
|
// Start broadcasting the update
|
if err := s.broadcastJoin(s.clock.Time()); err != nil {
|
return num, err
|
}
|
}
|
|
return num, err
|
}
|
|
// broadcastJoin broadcasts a new join intent with a
|
// given clock value. It is used on either join, or if
|
// we need to refute an older leave intent. Cannot be called
|
// with the memberLock held.
|
func (s *Serf) broadcastJoin(ltime LamportTime) error {
|
// Construct message to update our lamport clock
|
msg := messageJoin{
|
LTime: ltime,
|
Node: s.config.NodeName,
|
}
|
s.clock.Witness(ltime)
|
|
// Process update locally
|
s.handleNodeJoinIntent(&msg)
|
|
// Start broadcasting the update
|
if err := s.broadcast(messageJoinType, &msg, nil); err != nil {
|
s.logger.Printf("[WARN] serf: Failed to broadcast join intent: %v", err)
|
return err
|
}
|
return nil
|
}
|
|
// Leave gracefully exits the cluster. It is safe to call this multiple
|
// times.
|
func (s *Serf) Leave() error {
|
// Check the current state
|
s.stateLock.Lock()
|
if s.state == SerfLeft {
|
s.stateLock.Unlock()
|
return nil
|
} else if s.state == SerfLeaving {
|
s.stateLock.Unlock()
|
return fmt.Errorf("Leave already in progress")
|
} else if s.state == SerfShutdown {
|
s.stateLock.Unlock()
|
return fmt.Errorf("Leave called after Shutdown")
|
}
|
s.state = SerfLeaving
|
s.stateLock.Unlock()
|
|
// If we have a snapshot, mark we are leaving
|
if s.snapshotter != nil {
|
s.snapshotter.Leave()
|
}
|
|
// Construct the message for the graceful leave
|
msg := messageLeave{
|
LTime: s.clock.Time(),
|
Node: s.config.NodeName,
|
}
|
s.clock.Increment()
|
|
// Process the leave locally
|
s.handleNodeLeaveIntent(&msg)
|
|
// Only broadcast the leave message if there is at least one
|
// other node alive.
|
if s.hasAliveMembers() {
|
notifyCh := make(chan struct{})
|
if err := s.broadcast(messageLeaveType, &msg, notifyCh); err != nil {
|
return err
|
}
|
|
select {
|
case <-notifyCh:
|
case <-time.After(s.config.BroadcastTimeout):
|
return errors.New("timeout while waiting for graceful leave")
|
}
|
}
|
|
// Attempt the memberlist leave
|
err := s.memberlist.Leave(s.config.BroadcastTimeout)
|
if err != nil {
|
return err
|
}
|
|
// Wait for the leave to propagate through the cluster. The broadcast
|
// timeout is how long we wait for the message to go out from our own
|
// queue, but this wait is for that message to propagate through the
|
// cluster. In particular, we want to stay up long enough to service
|
// any probes from other nodes before they learn about us leaving.
|
time.Sleep(s.config.LeavePropagateDelay)
|
|
// Transition to Left only if we not already shutdown
|
s.stateLock.Lock()
|
if s.state != SerfShutdown {
|
s.state = SerfLeft
|
}
|
s.stateLock.Unlock()
|
return nil
|
}
|
|
// hasAliveMembers is called to check for any alive members other than
|
// ourself.
|
func (s *Serf) hasAliveMembers() bool {
|
s.memberLock.RLock()
|
defer s.memberLock.RUnlock()
|
|
hasAlive := false
|
for _, m := range s.members {
|
// Skip ourself, we want to know if OTHER members are alive
|
if m.Name == s.config.NodeName {
|
continue
|
}
|
|
if m.Status == StatusAlive {
|
hasAlive = true
|
break
|
}
|
}
|
return hasAlive
|
}
|
|
// LocalMember returns the Member information for the local node
|
func (s *Serf) LocalMember() Member {
|
s.memberLock.RLock()
|
defer s.memberLock.RUnlock()
|
return s.members[s.config.NodeName].Member
|
}
|
|
// Members returns a point-in-time snapshot of the members of this cluster.
|
func (s *Serf) Members() []Member {
|
s.memberLock.RLock()
|
defer s.memberLock.RUnlock()
|
|
members := make([]Member, 0, len(s.members))
|
for _, m := range s.members {
|
members = append(members, m.Member)
|
}
|
|
return members
|
}
|
|
// RemoveFailedNode is a backwards compatible form
|
// of forceleave
|
func (s *Serf) RemoveFailedNode(node string) error {
|
return s.forceLeave(node, false)
|
}
|
|
func (s *Serf) RemoveFailedNodePrune(node string) error {
|
return s.forceLeave(node, true)
|
}
|
|
// ForceLeave forcibly removes a failed node from the cluster
|
// immediately, instead of waiting for the reaper to eventually reclaim it.
|
// This also has the effect that Serf will no longer attempt to reconnect
|
// to this node.
|
func (s *Serf) forceLeave(node string, prune bool) error {
|
// Construct the message to broadcast
|
msg := messageLeave{
|
LTime: s.clock.Time(),
|
Node: node,
|
Prune: prune,
|
}
|
s.clock.Increment()
|
|
// Process our own event
|
s.handleNodeLeaveIntent(&msg)
|
|
// If we have no members, then we don't need to broadcast
|
if !s.hasAliveMembers() {
|
return nil
|
}
|
|
// Broadcast the remove
|
notifyCh := make(chan struct{})
|
if err := s.broadcast(messageLeaveType, &msg, notifyCh); err != nil {
|
return err
|
}
|
|
// Wait for the broadcast
|
select {
|
case <-notifyCh:
|
case <-time.After(s.config.BroadcastTimeout):
|
return fmt.Errorf("timed out broadcasting node removal")
|
}
|
|
return nil
|
}
|
|
// Shutdown forcefully shuts down the Serf instance, stopping all network
|
// activity and background maintenance associated with the instance.
|
//
|
// This is not a graceful shutdown, and should be preceded by a call
|
// to Leave. Otherwise, other nodes in the cluster will detect this node's
|
// exit as a node failure.
|
//
|
// It is safe to call this method multiple times.
|
func (s *Serf) Shutdown() error {
|
s.stateLock.Lock()
|
defer s.stateLock.Unlock()
|
|
if s.state == SerfShutdown {
|
return nil
|
}
|
|
if s.state != SerfLeft {
|
s.logger.Printf("[WARN] serf: Shutdown without a Leave")
|
}
|
|
// Wait to close the shutdown channel until after we've shut down the
|
// memberlist and its associated network resources, since the shutdown
|
// channel signals that we are cleaned up outside of Serf.
|
s.state = SerfShutdown
|
err := s.memberlist.Shutdown()
|
if err != nil {
|
return err
|
}
|
close(s.shutdownCh)
|
|
// Wait for the snapshoter to finish if we have one
|
if s.snapshotter != nil {
|
s.snapshotter.Wait()
|
}
|
|
return nil
|
}
|
|
// ShutdownCh returns a channel that can be used to wait for
|
// Serf to shutdown.
|
func (s *Serf) ShutdownCh() <-chan struct{} {
|
return s.shutdownCh
|
}
|
|
// Memberlist is used to get access to the underlying Memberlist instance
|
func (s *Serf) Memberlist() *memberlist.Memberlist {
|
return s.memberlist
|
}
|
|
// State is the current state of this Serf instance.
|
func (s *Serf) State() SerfState {
|
s.stateLock.Lock()
|
defer s.stateLock.Unlock()
|
return s.state
|
}
|
|
// broadcast takes a Serf message type, encodes it for the wire, and queues
|
// the broadcast. If a notify channel is given, this channel will be closed
|
// when the broadcast is sent.
|
func (s *Serf) broadcast(t messageType, msg interface{}, notify chan<- struct{}) error {
|
raw, err := encodeMessage(t, msg)
|
if err != nil {
|
return err
|
}
|
|
s.broadcasts.QueueBroadcast(&broadcast{
|
msg: raw,
|
notify: notify,
|
})
|
return nil
|
}
|
|
// handleNodeJoin is called when a node join event is received
|
// from memberlist.
|
func (s *Serf) handleNodeJoin(n *memberlist.Node) {
|
s.memberLock.Lock()
|
defer s.memberLock.Unlock()
|
|
var oldStatus MemberStatus
|
member, ok := s.members[n.Name]
|
if !ok {
|
oldStatus = StatusNone
|
member = &memberState{
|
Member: Member{
|
Name: n.Name,
|
Addr: net.IP(n.Addr),
|
Port: n.Port,
|
Tags: s.decodeTags(n.Meta),
|
Status: StatusAlive,
|
},
|
}
|
|
// Check if we have a join or leave intent. The intent buffer
|
// will only hold one event for this node, so the more recent
|
// one will take effect.
|
if join, ok := recentIntent(s.recentIntents, n.Name, messageJoinType); ok {
|
member.statusLTime = join
|
}
|
if leave, ok := recentIntent(s.recentIntents, n.Name, messageLeaveType); ok {
|
member.Status = StatusLeaving
|
member.statusLTime = leave
|
}
|
|
s.members[n.Name] = member
|
} else {
|
oldStatus = member.Status
|
deadTime := time.Now().Sub(member.leaveTime)
|
if oldStatus == StatusFailed && deadTime < s.config.FlapTimeout {
|
metrics.IncrCounter([]string{"serf", "member", "flap"}, 1)
|
}
|
|
member.Status = StatusAlive
|
member.leaveTime = time.Time{}
|
member.Addr = net.IP(n.Addr)
|
member.Port = n.Port
|
member.Tags = s.decodeTags(n.Meta)
|
}
|
|
// Update the protocol versions every time we get an event
|
member.ProtocolMin = n.PMin
|
member.ProtocolMax = n.PMax
|
member.ProtocolCur = n.PCur
|
member.DelegateMin = n.DMin
|
member.DelegateMax = n.DMax
|
member.DelegateCur = n.DCur
|
|
// If node was previously in a failed state, then clean up some
|
// internal accounting.
|
// TODO(mitchellh): needs tests to verify not reaped
|
if oldStatus == StatusFailed || oldStatus == StatusLeft {
|
s.failedMembers = removeOldMember(s.failedMembers, member.Name)
|
s.leftMembers = removeOldMember(s.leftMembers, member.Name)
|
}
|
|
// Update some metrics
|
metrics.IncrCounter([]string{"serf", "member", "join"}, 1)
|
|
// Send an event along
|
s.logger.Printf("[INFO] serf: EventMemberJoin: %s %s",
|
member.Member.Name, member.Member.Addr)
|
if s.config.EventCh != nil {
|
s.config.EventCh <- MemberEvent{
|
Type: EventMemberJoin,
|
Members: []Member{member.Member},
|
}
|
}
|
}
|
|
// handleNodeLeave is called when a node leave event is received
|
// from memberlist.
|
func (s *Serf) handleNodeLeave(n *memberlist.Node) {
|
s.memberLock.Lock()
|
defer s.memberLock.Unlock()
|
|
member, ok := s.members[n.Name]
|
if !ok {
|
// We've never even heard of this node that is supposedly
|
// leaving. Just ignore it completely.
|
return
|
}
|
|
switch member.Status {
|
case StatusLeaving:
|
member.Status = StatusLeft
|
member.leaveTime = time.Now()
|
s.leftMembers = append(s.leftMembers, member)
|
case StatusAlive:
|
member.Status = StatusFailed
|
member.leaveTime = time.Now()
|
s.failedMembers = append(s.failedMembers, member)
|
default:
|
// Unknown state that it was in? Just don't do anything
|
s.logger.Printf("[WARN] serf: Bad state when leave: %d", member.Status)
|
return
|
}
|
|
// Send an event along
|
event := EventMemberLeave
|
eventStr := "EventMemberLeave"
|
if member.Status != StatusLeft {
|
event = EventMemberFailed
|
eventStr = "EventMemberFailed"
|
}
|
|
// Update some metrics
|
metrics.IncrCounter([]string{"serf", "member", member.Status.String()}, 1)
|
|
s.logger.Printf("[INFO] serf: %s: %s %s",
|
eventStr, member.Member.Name, member.Member.Addr)
|
if s.config.EventCh != nil {
|
s.config.EventCh <- MemberEvent{
|
Type: event,
|
Members: []Member{member.Member},
|
}
|
}
|
}
|
|
// handleNodeUpdate is called when a node meta data update
|
// has taken place
|
func (s *Serf) handleNodeUpdate(n *memberlist.Node) {
|
s.memberLock.Lock()
|
defer s.memberLock.Unlock()
|
|
member, ok := s.members[n.Name]
|
if !ok {
|
// We've never even heard of this node that is updating.
|
// Just ignore it completely.
|
return
|
}
|
|
// Update the member attributes
|
member.Addr = net.IP(n.Addr)
|
member.Port = n.Port
|
member.Tags = s.decodeTags(n.Meta)
|
|
// Snag the latest versions. NOTE - the current memberlist code will NOT
|
// fire an update event if the metadata (for Serf, tags) stays the same
|
// and only the protocol versions change. If we wake any Serf-level
|
// protocol changes where we want to get this event under those
|
// circumstances, we will need to update memberlist to do a check of
|
// versions as well as the metadata.
|
member.ProtocolMin = n.PMin
|
member.ProtocolMax = n.PMax
|
member.ProtocolCur = n.PCur
|
member.DelegateMin = n.DMin
|
member.DelegateMax = n.DMax
|
member.DelegateCur = n.DCur
|
|
// Update some metrics
|
metrics.IncrCounter([]string{"serf", "member", "update"}, 1)
|
|
// Send an event along
|
s.logger.Printf("[INFO] serf: EventMemberUpdate: %s", member.Member.Name)
|
if s.config.EventCh != nil {
|
s.config.EventCh <- MemberEvent{
|
Type: EventMemberUpdate,
|
Members: []Member{member.Member},
|
}
|
}
|
}
|
|
// handleNodeLeaveIntent is called when an intent to leave is received.
|
func (s *Serf) handleNodeLeaveIntent(leaveMsg *messageLeave) bool {
|
|
// Witness a potentially newer time
|
s.clock.Witness(leaveMsg.LTime)
|
|
s.memberLock.Lock()
|
defer s.memberLock.Unlock()
|
|
member, ok := s.members[leaveMsg.Node]
|
if !ok {
|
// Rebroadcast only if this was an update we hadn't seen before.
|
return upsertIntent(s.recentIntents, leaveMsg.Node, messageLeaveType, leaveMsg.LTime, time.Now)
|
}
|
|
// If the message is old, then it is irrelevant and we can skip it
|
if leaveMsg.LTime <= member.statusLTime {
|
return false
|
}
|
|
// Refute us leaving if we are in the alive state
|
// Must be done in another goroutine since we have the memberLock
|
if leaveMsg.Node == s.config.NodeName && s.state == SerfAlive {
|
s.logger.Printf("[DEBUG] serf: Refuting an older leave intent")
|
go s.broadcastJoin(s.clock.Time())
|
return false
|
}
|
|
// State transition depends on current state
|
switch member.Status {
|
case StatusAlive:
|
member.Status = StatusLeaving
|
member.statusLTime = leaveMsg.LTime
|
|
if leaveMsg.Prune {
|
s.handlePrune(member)
|
}
|
return true
|
case StatusFailed:
|
member.Status = StatusLeft
|
member.statusLTime = leaveMsg.LTime
|
|
// Remove from the failed list and add to the left list. We add
|
// to the left list so that when we do a sync, other nodes will
|
// remove it from their failed list.
|
|
s.failedMembers = removeOldMember(s.failedMembers, member.Name)
|
s.leftMembers = append(s.leftMembers, member)
|
|
// We must push a message indicating the node has now
|
// left to allow higher-level applications to handle the
|
// graceful leave.
|
s.logger.Printf("[INFO] serf: EventMemberLeave (forced): %s %s",
|
member.Member.Name, member.Member.Addr)
|
if s.config.EventCh != nil {
|
s.config.EventCh <- MemberEvent{
|
Type: EventMemberLeave,
|
Members: []Member{member.Member},
|
}
|
}
|
|
if leaveMsg.Prune {
|
s.handlePrune(member)
|
}
|
|
return true
|
|
case StatusLeaving, StatusLeft:
|
if leaveMsg.Prune {
|
s.handlePrune(member)
|
}
|
return true
|
default:
|
return false
|
}
|
}
|
|
// handlePrune waits for nodes that are leaving and then forcibly
|
// erases a member from the list of members
|
func (s *Serf) handlePrune(member *memberState) {
|
if member.Status == StatusLeaving {
|
time.Sleep(s.config.BroadcastTimeout + s.config.LeavePropagateDelay)
|
}
|
|
s.logger.Printf("[INFO] serf: EventMemberReap (forced): %s %s", member.Name, member.Member.Addr)
|
|
//If we are leaving or left we may be in that list of members
|
if member.Status == StatusLeaving || member.Status == StatusLeft {
|
s.leftMembers = removeOldMember(s.leftMembers, member.Name)
|
}
|
s.eraseNode(member)
|
|
}
|
|
// handleNodeJoinIntent is called when a node broadcasts a
|
// join message to set the lamport time of its join
|
func (s *Serf) handleNodeJoinIntent(joinMsg *messageJoin) bool {
|
// Witness a potentially newer time
|
s.clock.Witness(joinMsg.LTime)
|
|
s.memberLock.Lock()
|
defer s.memberLock.Unlock()
|
|
member, ok := s.members[joinMsg.Node]
|
if !ok {
|
// Rebroadcast only if this was an update we hadn't seen before.
|
return upsertIntent(s.recentIntents, joinMsg.Node, messageJoinType, joinMsg.LTime, time.Now)
|
}
|
|
// Check if this time is newer than what we have
|
if joinMsg.LTime <= member.statusLTime {
|
return false
|
}
|
|
// Update the LTime
|
member.statusLTime = joinMsg.LTime
|
|
// If we are in the leaving state, we should go back to alive,
|
// since the leaving message must have been for an older time
|
if member.Status == StatusLeaving {
|
member.Status = StatusAlive
|
}
|
return true
|
}
|
|
// handleUserEvent is called when a user event broadcast is
|
// received. Returns if the message should be rebroadcast.
|
func (s *Serf) handleUserEvent(eventMsg *messageUserEvent) bool {
|
// Witness a potentially newer time
|
s.eventClock.Witness(eventMsg.LTime)
|
|
s.eventLock.Lock()
|
defer s.eventLock.Unlock()
|
|
// Ignore if it is before our minimum event time
|
if eventMsg.LTime < s.eventMinTime {
|
s.logger.Printf("[WARN] serf: eventMsg.LTime: %d < s.eventMinTime:%d",eventMsg.LTime, s.eventMinTime)
|
return false
|
}
|
|
// Check if this message is too old
|
curTime := s.eventClock.Time()
|
if curTime > LamportTime(len(s.eventBuffer)) &&
|
eventMsg.LTime < curTime-LamportTime(len(s.eventBuffer)) {
|
s.logger.Printf(
|
"[WARN] serf: received old event %s from time %d (current: %d)",
|
eventMsg.Name,
|
eventMsg.LTime,
|
s.eventClock.Time())
|
return false
|
}
|
|
// Check if we've already seen this
|
idx := eventMsg.LTime % LamportTime(len(s.eventBuffer))
|
seen := s.eventBuffer[idx]
|
userEvent := userEvent{Name: eventMsg.Name, Payload: eventMsg.Payload}
|
if seen != nil && seen.LTime == eventMsg.LTime {
|
for _, previous := range seen.Events {
|
if previous.Equals(&userEvent) {
|
return false
|
}
|
}
|
} else {
|
seen = &userEvents{LTime: eventMsg.LTime}
|
s.eventBuffer[idx] = seen
|
}
|
|
// Add to recent events
|
seen.Events = append(seen.Events, userEvent)
|
|
// Update some metrics
|
metrics.IncrCounter([]string{"serf", "events"}, 1)
|
metrics.IncrCounter([]string{"serf", "events", eventMsg.Name}, 1)
|
|
if s.config.EventCh != nil {
|
s.logger.Printf("[WARN] serf: s.config.EventCh <- UserEvent")
|
s.config.EventCh <- UserEvent{
|
LTime: eventMsg.LTime,
|
Name: eventMsg.Name,
|
Payload: eventMsg.Payload,
|
Coalesce: eventMsg.CC,
|
}
|
}
|
return true
|
}
|
|
// handleQuery is called when a query broadcast is
|
// received. Returns if the message should be rebroadcast.
|
func (s *Serf) handleQuery(query *messageQuery) bool {
|
// Witness a potentially newer time
|
s.queryClock.Witness(query.LTime)
|
|
s.queryLock.Lock()
|
defer s.queryLock.Unlock()
|
|
// Ignore if it is before our minimum query time
|
if query.LTime < s.queryMinTime {
|
return false
|
}
|
|
// Check if this message is too old
|
curTime := s.queryClock.Time()
|
if curTime > LamportTime(len(s.queryBuffer)) &&
|
query.LTime < curTime-LamportTime(len(s.queryBuffer)) {
|
s.logger.Printf(
|
"[WARN] serf: received old query %s from time %d (current: %d)",
|
query.Name,
|
query.LTime,
|
s.queryClock.Time())
|
return false
|
}
|
|
// Check if we've already seen this
|
idx := query.LTime % LamportTime(len(s.queryBuffer))
|
seen := s.queryBuffer[idx]
|
if seen != nil && seen.LTime == query.LTime {
|
for _, previous := range seen.QueryIDs {
|
if previous == query.ID {
|
// Seen this ID already
|
return false
|
}
|
}
|
} else {
|
seen = &queries{LTime: query.LTime}
|
s.queryBuffer[idx] = seen
|
}
|
|
// Add to recent queries
|
seen.QueryIDs = append(seen.QueryIDs, query.ID)
|
|
// Update some metrics
|
metrics.IncrCounter([]string{"serf", "queries"}, 1)
|
metrics.IncrCounter([]string{"serf", "queries", query.Name}, 1)
|
|
// Check if we should rebroadcast, this may be disabled by a flag
|
rebroadcast := true
|
if query.NoBroadcast() {
|
rebroadcast = false
|
}
|
|
// Filter the query
|
if !s.shouldProcessQuery(query.Filters) {
|
// Even if we don't process it further, we should rebroadcast,
|
// since it is the first time we've seen this.
|
return rebroadcast
|
}
|
|
// Send ack if requested, without waiting for client to Respond()
|
if query.Ack() {
|
ack := messageQueryResponse{
|
LTime: query.LTime,
|
ID: query.ID,
|
From: s.config.NodeName,
|
Flags: queryFlagAck,
|
}
|
raw, err := encodeMessage(messageQueryResponseType, &ack)
|
if err != nil {
|
s.logger.Printf("[ERR] serf: failed to format ack: %v", err)
|
} else {
|
udpAddr := net.UDPAddr{IP: query.Addr, Port: int(query.Port)}
|
addr := memberlist.Address{
|
Addr: udpAddr.String(),
|
Name: query.SourceNode,
|
}
|
if err := s.memberlist.SendToAddress(addr, raw); err != nil {
|
s.logger.Printf("[ERR] serf: failed to send ack: %v", err)
|
}
|
if err := s.relayResponse(query.RelayFactor, udpAddr, query.SourceNode, &ack); err != nil {
|
s.logger.Printf("[ERR] serf: failed to relay ack: %v", err)
|
}
|
}
|
}
|
|
if s.config.EventCh != nil {
|
s.config.EventCh <- &Query{
|
LTime: query.LTime,
|
Name: query.Name,
|
Payload: query.Payload,
|
serf: s,
|
id: query.ID,
|
addr: query.Addr,
|
port: query.Port,
|
sourceNode: query.SourceNode,
|
deadline: time.Now().Add(query.Timeout),
|
relayFactor: query.RelayFactor,
|
}
|
}
|
return rebroadcast
|
}
|
|
// handleResponse is called when a query response is
|
// received.
|
func (s *Serf) handleQueryResponse(resp *messageQueryResponse) {
|
// Look for a corresponding QueryResponse
|
s.queryLock.RLock()
|
query, ok := s.queryResponse[resp.LTime]
|
s.queryLock.RUnlock()
|
if !ok {
|
s.logger.Printf("[WARN] serf: reply for non-running query (LTime: %d, ID: %d) From: %s",
|
resp.LTime, resp.ID, resp.From)
|
return
|
}
|
|
// Verify the ID matches
|
if query.id != resp.ID {
|
s.logger.Printf("[WARN] serf: query reply ID mismatch (Local: %d, Response: %d)",
|
query.id, resp.ID)
|
return
|
}
|
|
// Check if the query is closed
|
if query.Finished() {
|
return
|
}
|
|
// Process each type of response
|
if resp.Ack() {
|
// Exit early if this is a duplicate ack
|
if _, ok := query.acks[resp.From]; ok {
|
metrics.IncrCounter([]string{"serf", "query_duplicate_acks"}, 1)
|
return
|
}
|
|
metrics.IncrCounter([]string{"serf", "query_acks"}, 1)
|
select {
|
case query.ackCh <- resp.From:
|
query.acks[resp.From] = struct{}{}
|
default:
|
s.logger.Printf("[WARN] serf: Failed to deliver query ack, dropping")
|
}
|
} else {
|
// Exit early if this is a duplicate response
|
if _, ok := query.responses[resp.From]; ok {
|
metrics.IncrCounter([]string{"serf", "query_duplicate_responses"}, 1)
|
return
|
}
|
|
metrics.IncrCounter([]string{"serf", "query_responses"}, 1)
|
err := query.sendResponse(NodeResponse{From: resp.From, Payload: resp.Payload})
|
if err != nil {
|
s.logger.Printf("[WARN] %v", err)
|
}
|
}
|
}
|
|
// handleNodeConflict is invoked when a join detects a conflict over a name.
|
// This means two different nodes (IP/Port) are claiming the same name. Memberlist
|
// will reject the "new" node mapping, but we can still be notified.
|
func (s *Serf) handleNodeConflict(existing, other *memberlist.Node) {
|
// Log a basic warning if the node is not us...
|
if existing.Name != s.config.NodeName {
|
s.logger.Printf("[WARN] serf: Name conflict for '%s' both %s:%d and %s:%d are claiming",
|
existing.Name, existing.Addr, existing.Port, other.Addr, other.Port)
|
return
|
}
|
|
// The current node is conflicting! This is an error
|
s.logger.Printf("[ERR] serf: Node name conflicts with another node at %s:%d. Names must be unique! (Resolution enabled: %v)",
|
other.Addr, other.Port, s.config.EnableNameConflictResolution)
|
|
// If automatic resolution is enabled, kick off the resolution
|
if s.config.EnableNameConflictResolution {
|
go s.resolveNodeConflict()
|
}
|
}
|
|
// resolveNodeConflict is used to determine which node should remain during
|
// a name conflict. This is done by running an internal query.
|
func (s *Serf) resolveNodeConflict() {
|
// Get the local node
|
local := s.memberlist.LocalNode()
|
|
// Start a name resolution query
|
qName := internalQueryName(conflictQuery)
|
payload := []byte(s.config.NodeName)
|
resp, err := s.Query(qName, payload, nil)
|
if err != nil {
|
s.logger.Printf("[ERR] serf: Failed to start name resolution query: %v", err)
|
return
|
}
|
|
// Counter to determine winner
|
var responses, matching int
|
|
// Gather responses
|
respCh := resp.ResponseCh()
|
for r := range respCh {
|
// Decode the response
|
if len(r.Payload) < 1 || messageType(r.Payload[0]) != messageConflictResponseType {
|
s.logger.Printf("[ERR] serf: Invalid conflict query response type: %v", r.Payload)
|
continue
|
}
|
var member Member
|
if err := decodeMessage(r.Payload[1:], &member); err != nil {
|
s.logger.Printf("[ERR] serf: Failed to decode conflict query response: %v", err)
|
continue
|
}
|
|
// Update the counters
|
responses++
|
if member.Addr.Equal(local.Addr) && member.Port == local.Port {
|
matching++
|
}
|
}
|
|
// Query over, determine if we should live
|
majority := (responses / 2) + 1
|
if matching >= majority {
|
s.logger.Printf("[INFO] serf: majority in name conflict resolution [%d / %d]",
|
matching, responses)
|
return
|
}
|
|
// Since we lost the vote, we need to exit
|
s.logger.Printf("[WARN] serf: minority in name conflict resolution, quiting [%d / %d]",
|
matching, responses)
|
if err := s.Shutdown(); err != nil {
|
s.logger.Printf("[ERR] serf: Failed to shutdown: %v", err)
|
}
|
}
|
|
//eraseNode takes a node completely out of the member list
|
func (s *Serf) eraseNode(m *memberState) {
|
// Delete from members
|
delete(s.members, m.Name)
|
|
// Tell the coordinate client the node has gone away and delete
|
// its cached coordinates.
|
if !s.config.DisableCoordinates {
|
s.coordClient.ForgetNode(m.Name)
|
|
s.coordCacheLock.Lock()
|
delete(s.coordCache, m.Name)
|
s.coordCacheLock.Unlock()
|
}
|
|
// Send an event along
|
if s.config.EventCh != nil {
|
s.config.EventCh <- MemberEvent{
|
Type: EventMemberReap,
|
Members: []Member{m.Member},
|
}
|
}
|
}
|
|
// handleReap periodically reaps the list of failed and left members, as well
|
// as old buffered intents.
|
func (s *Serf) handleReap() {
|
for {
|
select {
|
case <-time.After(s.config.ReapInterval):
|
s.memberLock.Lock()
|
now := time.Now()
|
s.failedMembers = s.reap(s.failedMembers, now, s.config.ReconnectTimeout)
|
s.leftMembers = s.reap(s.leftMembers, now, s.config.TombstoneTimeout)
|
reapIntents(s.recentIntents, now, s.config.RecentIntentTimeout)
|
s.memberLock.Unlock()
|
case <-s.shutdownCh:
|
return
|
}
|
}
|
}
|
|
// handleReconnect attempts to reconnect to recently failed nodes
|
// on configured intervals.
|
func (s *Serf) handleReconnect() {
|
for {
|
select {
|
case <-time.After(s.config.ReconnectInterval):
|
s.reconnect()
|
case <-s.shutdownCh:
|
return
|
}
|
}
|
}
|
|
// reap is called with a list of old members and a timeout, and removes
|
// members that have exceeded the timeout. The members are removed from
|
// both the old list and the members itself. Locking is left to the caller.
|
func (s *Serf) reap(old []*memberState, now time.Time, timeout time.Duration) []*memberState {
|
n := len(old)
|
for i := 0; i < n; i++ {
|
m := old[i]
|
|
// Skip if the timeout is not yet reached
|
if now.Sub(m.leaveTime) <= timeout {
|
continue
|
}
|
|
// Delete from the list
|
old[i], old[n-1] = old[n-1], nil
|
old = old[:n-1]
|
n--
|
i--
|
|
// Delete from members and send out event
|
s.logger.Printf("[INFO] serf: EventMemberReap: %s", m.Name)
|
s.eraseNode(m)
|
|
}
|
|
return old
|
}
|
|
// reconnect attempts to reconnect to recently fail nodes.
|
func (s *Serf) reconnect() {
|
s.memberLock.RLock()
|
|
// Nothing to do if there are no failed members
|
n := len(s.failedMembers)
|
if n == 0 {
|
s.memberLock.RUnlock()
|
return
|
}
|
|
// Probability we should attempt to reconect is given
|
// by num failed / (num members - num failed - num left)
|
// This means that we probabilistically expect the cluster
|
// to attempt to connect to each failed member once per
|
// reconnect interval
|
numFailed := float32(len(s.failedMembers))
|
numAlive := float32(len(s.members) - len(s.failedMembers) - len(s.leftMembers))
|
if numAlive == 0 {
|
numAlive = 1 // guard against zero divide
|
}
|
prob := numFailed / numAlive
|
if rand.Float32() > prob {
|
s.memberLock.RUnlock()
|
s.logger.Printf("[DEBUG] serf: forgoing reconnect for random throttling")
|
return
|
}
|
|
// Select a random member to try and join
|
idx := rand.Int31n(int32(n))
|
mem := s.failedMembers[idx]
|
s.memberLock.RUnlock()
|
|
// Format the addr
|
addr := net.UDPAddr{IP: mem.Addr, Port: int(mem.Port)}
|
s.logger.Printf("[INFO] serf: attempting reconnect to %v %s", mem.Name, addr.String())
|
|
joinAddr := addr.String()
|
if mem.Name != "" {
|
joinAddr = mem.Name + "/" + addr.String()
|
}
|
|
// Attempt to join at the memberlist level
|
s.memberlist.Join([]string{joinAddr})
|
}
|
|
// getQueueMax will get the maximum queue depth, which might be dynamic depending
|
// on how Serf is configured.
|
func (s *Serf) getQueueMax() int {
|
max := s.config.MaxQueueDepth
|
if s.config.MinQueueDepth > 0 {
|
s.memberLock.RLock()
|
max = 2 * len(s.members)
|
s.memberLock.RUnlock()
|
|
if max < s.config.MinQueueDepth {
|
max = s.config.MinQueueDepth
|
}
|
}
|
return max
|
}
|
|
// checkQueueDepth periodically checks the size of a queue to see if
|
// it is too large
|
func (s *Serf) checkQueueDepth(name string, queue *memberlist.TransmitLimitedQueue) {
|
for {
|
select {
|
case <-time.After(s.config.QueueCheckInterval):
|
numq := queue.NumQueued()
|
metrics.AddSample([]string{"serf", "queue", name}, float32(numq))
|
if numq >= s.config.QueueDepthWarning {
|
s.logger.Printf("[WARN] serf: %s queue depth: %d", name, numq)
|
}
|
if max := s.getQueueMax(); numq > max {
|
s.logger.Printf("[WARN] serf: %s queue depth (%d) exceeds limit (%d), dropping messages!",
|
name, numq, max)
|
queue.Prune(max)
|
}
|
case <-s.shutdownCh:
|
return
|
}
|
}
|
}
|
|
// removeOldMember is used to remove an old member from a list of old
|
// members.
|
func removeOldMember(old []*memberState, name string) []*memberState {
|
for i, m := range old {
|
if m.Name == name {
|
n := len(old)
|
old[i], old[n-1] = old[n-1], nil
|
return old[:n-1]
|
}
|
}
|
|
return old
|
}
|
|
// reapIntents clears out any intents that are older than the timeout. Make sure
|
// the memberLock is held when passing in the Serf instance's recentIntents
|
// member.
|
func reapIntents(intents map[string]nodeIntent, now time.Time, timeout time.Duration) {
|
for node, intent := range intents {
|
if now.Sub(intent.WallTime) > timeout {
|
delete(intents, node)
|
}
|
}
|
}
|
|
// upsertIntent will update an existing intent with the supplied Lamport time,
|
// or create a new entry. This will return true if a new entry was added. The
|
// stamper is used to capture the wall clock time for expiring these buffered
|
// intents. Make sure the memberLock is held when passing in the Serf instance's
|
// recentIntents member.
|
func upsertIntent(intents map[string]nodeIntent, node string, itype messageType,
|
ltime LamportTime, stamper func() time.Time) bool {
|
if intent, ok := intents[node]; !ok || ltime > intent.LTime {
|
intents[node] = nodeIntent{
|
Type: itype,
|
WallTime: stamper(),
|
LTime: ltime,
|
}
|
return true
|
}
|
|
return false
|
}
|
|
// recentIntent checks the recent intent buffer for a matching entry for a given
|
// node, and returns the Lamport time, if an intent is present, indicated by the
|
// returned boolean. Make sure the memberLock is held for read when passing in
|
// the Serf instance's recentIntents member.
|
func recentIntent(intents map[string]nodeIntent, node string, itype messageType) (LamportTime, bool) {
|
if intent, ok := intents[node]; ok && intent.Type == itype {
|
return intent.LTime, true
|
}
|
|
return LamportTime(0), false
|
}
|
|
// handleRejoin attempts to reconnect to previously known alive nodes
|
func (s *Serf) handleRejoin(previous []*PreviousNode) {
|
for _, prev := range previous {
|
// Do not attempt to join ourself
|
if prev.Name == s.config.NodeName {
|
continue
|
}
|
|
joinAddr := prev.Addr
|
if prev.Name != "" {
|
joinAddr = prev.Name + "/" + prev.Addr
|
}
|
|
s.logger.Printf("[INFO] serf: Attempting re-join to previously known node: %s", prev)
|
_, err := s.memberlist.Join([]string{joinAddr})
|
if err == nil {
|
s.logger.Printf("[INFO] serf: Re-joined to previously known node: %s", prev)
|
return
|
}
|
}
|
s.logger.Printf("[WARN] serf: Failed to re-join any previously known node")
|
}
|
|
// encodeTags is used to encode a tag map
|
func (s *Serf) encodeTags(tags map[string]string) []byte {
|
// Support role-only backwards compatibility
|
if s.ProtocolVersion() < 3 {
|
role := tags["role"]
|
return []byte(role)
|
}
|
|
// Use a magic byte prefix and msgpack encode the tags
|
var buf bytes.Buffer
|
buf.WriteByte(tagMagicByte)
|
enc := codec.NewEncoder(&buf, &codec.MsgpackHandle{})
|
if err := enc.Encode(tags); err != nil {
|
panic(fmt.Sprintf("Failed to encode tags: %v", err))
|
}
|
return buf.Bytes()
|
}
|
|
// decodeTags is used to decode a tag map
|
func (s *Serf) decodeTags(buf []byte) map[string]string {
|
tags := make(map[string]string)
|
|
// Backwards compatibility mode
|
if len(buf) == 0 || buf[0] != tagMagicByte {
|
tags["role"] = string(buf)
|
return tags
|
}
|
|
// Decode the tags
|
r := bytes.NewReader(buf[1:])
|
dec := codec.NewDecoder(r, &codec.MsgpackHandle{})
|
if err := dec.Decode(&tags); err != nil {
|
s.logger.Printf("[ERR] serf: Failed to decode tags: %v", err)
|
}
|
return tags
|
}
|
|
// Stats is used to provide operator debugging information
|
func (s *Serf) Stats() map[string]string {
|
toString := func(v uint64) string {
|
return strconv.FormatUint(v, 10)
|
}
|
s.memberLock.RLock()
|
members := toString(uint64(len(s.members)))
|
failed := toString(uint64(len(s.failedMembers)))
|
left := toString(uint64(len(s.leftMembers)))
|
health_score := toString(uint64(s.memberlist.GetHealthScore()))
|
|
s.memberLock.RUnlock()
|
stats := map[string]string{
|
"members": members,
|
"failed": failed,
|
"left": left,
|
"health_score": health_score,
|
"member_time": toString(uint64(s.clock.Time())),
|
"event_time": toString(uint64(s.eventClock.Time())),
|
"query_time": toString(uint64(s.queryClock.Time())),
|
"intent_queue": toString(uint64(s.broadcasts.NumQueued())),
|
"event_queue": toString(uint64(s.eventBroadcasts.NumQueued())),
|
"query_queue": toString(uint64(s.queryBroadcasts.NumQueued())),
|
"encrypted": fmt.Sprintf("%v", s.EncryptionEnabled()),
|
}
|
if !s.config.DisableCoordinates {
|
stats["coordinate_resets"] = toString(uint64(s.coordClient.Stats().Resets))
|
}
|
return stats
|
}
|
|
// WriteKeyringFile will serialize the current keyring and save it to a file.
|
func (s *Serf) writeKeyringFile() error {
|
if len(s.config.KeyringFile) == 0 {
|
return nil
|
}
|
|
keyring := s.config.MemberlistConfig.Keyring
|
keysRaw := keyring.GetKeys()
|
keysEncoded := make([]string, len(keysRaw))
|
|
for i, key := range keysRaw {
|
keysEncoded[i] = base64.StdEncoding.EncodeToString(key)
|
}
|
|
encodedKeys, err := json.MarshalIndent(keysEncoded, "", " ")
|
if err != nil {
|
return fmt.Errorf("Failed to encode keys: %s", err)
|
}
|
|
// Use 0600 for permissions because key data is sensitive
|
if err = ioutil.WriteFile(s.config.KeyringFile, encodedKeys, 0600); err != nil {
|
return fmt.Errorf("Failed to write keyring file: %s", err)
|
}
|
|
// Success!
|
return nil
|
}
|
|
// GetCoordinate returns the network coordinate of the local node.
|
func (s *Serf) GetCoordinate() (*coordinate.Coordinate, error) {
|
if !s.config.DisableCoordinates {
|
return s.coordClient.GetCoordinate(), nil
|
}
|
|
return nil, fmt.Errorf("Coordinates are disabled")
|
}
|
|
// GetCachedCoordinate returns the network coordinate for the node with the given
|
// name. This will only be valid if DisableCoordinates is set to false.
|
func (s *Serf) GetCachedCoordinate(name string) (coord *coordinate.Coordinate, ok bool) {
|
if !s.config.DisableCoordinates {
|
s.coordCacheLock.RLock()
|
defer s.coordCacheLock.RUnlock()
|
if coord, ok = s.coordCache[name]; ok {
|
return coord, true
|
}
|
|
return nil, false
|
}
|
|
return nil, false
|
}
|
|
// NumNodes returns the number of nodes in the serf cluster, regardless of
|
// their health or status.
|
func (s *Serf) NumNodes() (numNodes int) {
|
s.memberLock.RLock()
|
numNodes = len(s.members)
|
s.memberLock.RUnlock()
|
|
return numNodes
|
}
|
