Gossip协议及Consul中的实现

// Schedule is used to ensure the Tick is performed periodically. This
// function is safe to call multiple times. If the memberlist is already
// scheduled, then it won't do anything.
func (m *Memberlist) schedule() {
  m.tickerLock.Lock()
  defer m.tickerLock.Unlock()

  // If we already have tickers, then don't do anything, since we're
  // scheduled
  if len(m.tickers) > 0 {
    return
  }

  // Create the stop tick channel, a blocking channel. We close this
  // when we should stop the tickers.
  stopCh := make(chan struct{})

  // Create a new probeTicker
  if m.config.ProbeInterval > 0 {
    t := time.NewTicker(m.config.ProbeInterval)
    go m.triggerFunc(m.config.ProbeInterval, t.C, stopCh, m.probe)
    m.tickers = append(m.tickers, t)
  }

  // Create a push pull ticker if needed
  if m.config.PushPullInterval > 0 {
    go m.pushPullTrigger(stopCh)
  }

  // Create a gossip ticker if needed
  if m.config.GossipInterval > 0 && m.config.GossipNodes > 0 {
    t := time.NewTicker(m.config.GossipInterval)
    go m.triggerFunc(m.config.GossipInterval, t.C, stopCh, m.gossip)
    m.tickers = append(m.tickers, t)
  }

  // If we made any tickers, then record the stopTick channel for
  // later.
  if len(m.tickers) > 0 {
    m.stopTick = stopCh
  }
}

func (m *Memberlist) probe() {
  // Track the number of indexes we've considered probing
  numCheck := 0
START:
  m.nodeLock.RLock()

  // Make sure we don't wrap around infinitely
  if numCheck >= len(m.nodes) {
    m.nodeLock.RUnlock()
    return
  }

  // Handle the wrap around case
  if m.probeIndex >= len(m.nodes) {
    m.nodeLock.RUnlock()
    m.resetNodes()
    m.probeIndex = 0
    numCheck++
    goto START
  }

  // Determine if we should probe this node
  skip := false
  var node nodeState

  node = *m.nodes[m.probeIndex]
  if node.Name == m.config.Name {
    skip = true
  } else if node.DeadOrLeft() {
    skip = true
  }

  // Potentially skip
  m.nodeLock.RUnlock()
  m.probeIndex++
  if skip {
    numCheck++
    goto START
  }

  // Probe the specific node
  m.probeNode(&node)
}

func (m *Memberlist) probeNode(node *nodeState) {
  defer metrics.MeasureSince([]string{"memberlist", "probeNode"}, time.Now())

  // We use our health awareness to scale the overall probe interval, so we
  // slow down if we detect problems. The ticker that calls us can handle
  // us running over the base interval, and will skip missed ticks.
  probeInterval := m.awareness.ScaleTimeout(m.config.ProbeInterval)
  if probeInterval > m.config.ProbeInterval {
    metrics.IncrCounter([]string{"memberlist", "degraded", "probe"}, 1)
  }

  // Prepare a ping message and setup an ack handler.
  selfAddr, selfPort := m.getAdvertise()
  ping := ping{
    SeqNo:      m.nextSeqNo(),
    Node:       node.Name,
    SourceAddr: selfAddr,
    SourcePort: selfPort,
    SourceNode: m.config.Name,
  }
  ackCh := make(chan ackMessage, m.config.IndirectChecks+1)
  nackCh := make(chan struct{}, m.config.IndirectChecks+1)
  m.setProbeChannels(ping.SeqNo, ackCh, nackCh, probeInterval)

  // Mark the sent time here, which should be after any pre-processing but
  // before system calls to do the actual send. This probably over-reports
  // a bit, but it's the best we can do. We had originally put this right
  // after the I/O, but that would sometimes give negative RTT measurements
  // which was not desirable.
  sent := time.Now()

  // Send a ping to the node. If this node looks like it's suspect or dead,
  // also tack on a suspect message so that it has a chance to refute as
  // soon as possible.
  deadline := sent.Add(probeInterval)
  addr := node.Address()

  // Arrange for our self-awareness to get updated.
  var awarenessDelta int
  defer func() {
    m.awareness.ApplyDelta(awarenessDelta)
  }()
  if node.State == StateAlive {
    if err := m.encodeAndSendMsg(node.FullAddress(), pingMsg, &ping); err != nil {
      m.logger.Printf("[ERR] memberlist: Failed to send ping: %s", err)
      if failedRemote(err) {
        goto HANDLE_REMOTE_FAILURE
      } else {
        return
      }
    }
  } else {
    var msgs [][]byte
    if buf, err := encode(pingMsg, &ping); err != nil {
      m.logger.Printf("[ERR] memberlist: Failed to encode ping message: %s", err)
      return
    } else {
      msgs = append(msgs, buf.Bytes())
    }
    s := suspect{Incarnation: node.Incarnation, Node: node.Name, From: m.config.Name}
    if buf, err := encode(suspectMsg, &s); err != nil {
      m.logger.Printf("[ERR] memberlist: Failed to encode suspect message: %s", err)
      return
    } else {
      msgs = append(msgs, buf.Bytes())
    }

    compound := makeCompoundMessage(msgs)
    if err := m.rawSendMsgPacket(node.FullAddress(), &node.Node, compound.Bytes()); err != nil {
      m.logger.Printf("[ERR] memberlist: Failed to send compound ping and suspect message to %s: %s", addr, err)
      if failedRemote(err) {
        goto HANDLE_REMOTE_FAILURE
      } else {
        return
      }
    }
  }

  // Arrange for our self-awareness to get updated. At this point we've
  // sent the ping, so any return statement means the probe succeeded
  // which will improve our health until we get to the failure scenarios
  // at the end of this function, which will alter this delta variable
  // accordingly.
  awarenessDelta = -1

  // Wait for response or round-trip-time.
  select {
  case v := <-ackCh:
    if v.Complete == true {
      if m.config.Ping != nil {
        rtt := v.Timestamp.Sub(sent)
        m.config.Ping.NotifyPingComplete(&node.Node, rtt, v.Payload)
      }
      return
    }

    // As an edge case, if we get a timeout, we need to re-enqueue it
    // here to break out of the select below.
    if v.Complete == false {
      ackCh <- v
    }
  case <-time.After(m.config.ProbeTimeout):
    // Note that we don't scale this timeout based on awareness and
    // the health score. That's because we don't really expect waiting
    // longer to help get UDP through. Since health does extend the
    // probe interval it will give the TCP fallback more time, which
    // is more active in dealing with lost packets, and it gives more
    // time to wait for indirect acks/nacks.
    m.logger.Printf("[DEBUG] memberlist: Failed ping: %s (timeout reached)", node.Name)
  }
...
}

func (m *Memberlist) probeNode(node *nodeState) {
...
HANDLE_REMOTE_FAILURE:
  // Get some random live nodes.
  m.nodeLock.RLock()
  kNodes := kRandomNodes(m.config.IndirectChecks, m.nodes, func(n *nodeState) bool {
    return n.Name == m.config.Name ||
      n.Name == node.Name ||
      n.State != StateAlive
  })
  m.nodeLock.RUnlock()

  // Attempt an indirect ping.
  expectedNacks := 0
  selfAddr, selfPort = m.getAdvertise()
  ind := indirectPingReq{
    SeqNo:      ping.SeqNo,
    Target:     node.Addr,
    Port:       node.Port,
    Node:       node.Name,
    SourceAddr: selfAddr,
    SourcePort: selfPort,
    SourceNode: m.config.Name,
  }
  for _, peer := range kNodes {
    // We only expect nack to be sent from peers who understand
    // version 4 of the protocol.
    if ind.Nack = peer.PMax >= 4; ind.Nack {
      expectedNacks++
    }

    if err := m.encodeAndSendMsg(peer.FullAddress(), indirectPingMsg, &ind); err != nil {
      m.logger.Printf("[ERR] memberlist: Failed to send indirect ping: %s", err)
    }
  }

  // Also make an attempt to contact the node directly over TCP. This
  // helps prevent confused clients who get isolated from UDP traffic
  // but can still speak TCP (which also means they can possibly report
  // misinformation to other nodes via anti-entropy), avoiding flapping in
  // the cluster.
  //
  // This is a little unusual because we will attempt a TCP ping to any
  // member who understands version 3 of the protocol, regardless of
  // which protocol version we are speaking. That's why we've included a
  // config option to turn this off if desired.
  fallbackCh := make(chan bool, 1)

  disableTcpPings := m.config.DisableTcpPings ||
    (m.config.DisableTcpPingsForNode != nil && m.config.DisableTcpPingsForNode(node.Name))
  if (!disableTcpPings) && (node.PMax >= 3) {
    go func() {
      defer close(fallbackCh)
      didContact, err := m.sendPingAndWaitForAck(node.FullAddress(), ping, deadline)
      if err != nil {
        m.logger.Printf("[ERR] memberlist: Failed fallback ping: %s", err)
      } else {
        fallbackCh <- didContact
      }
    }()
  } else {
    close(fallbackCh)
  }

  // Wait for the acks or timeout. Note that we don't check the fallback
  // channel here because we want to issue a warning below if that's the
  // *only* way we hear back from the peer, so we have to let this time
  // out first to allow the normal UDP-based acks to come in.
  select {
  case v := <-ackCh:
    if v.Complete == true {
      return
    }
  }

  // Finally, poll the fallback channel. The timeouts are set such that
  // the channel will have something or be closed without having to wait
  // any additional time here.
  for didContact := range fallbackCh {
    if didContact {
      m.logger.Printf("[WARN] memberlist: Was able to connect to %s but other probes failed, network may be misconfigured", node.Name)
      return
    }
  }

  // Update our self-awareness based on the results of this failed probe.
  // If we don't have peers who will send nacks then we penalize for any
  // failed probe as a simple health metric. If we do have peers to nack
  // verify, then we can use that as a more sophisticated measure of self-
  // health because we assume them to be working, and they can help us
  // decide if the probed node was really dead or if it was something wrong
  // with ourselves.
  awarenessDelta = 0
  if expectedNacks > 0 {
    if nackCount := len(nackCh); nackCount < expectedNacks {
      awarenessDelta += (expectedNacks - nackCount)
    }
  } else {
    awarenessDelta += 1
  }

  // No acks received from target, suspect it as failed.
  m.logger.Printf("[INFO] memberlist: Suspect %s has failed, no acks received", node.Name)
  s := suspect{Incarnation: node.Incarnation, Node: node.Name, From: m.config.Name}
  m.suspectNode(&s)
}

func (m *Memberlist) pushPull() {
  // Get a random live node
  m.nodeLock.RLock()
  nodes := kRandomNodes(1, m.nodes, func(n *nodeState) bool {
    return n.Name == m.config.Name ||
      n.State != StateAlive
  })
  m.nodeLock.RUnlock()

  // If no nodes, bail
  if len(nodes) == 0 {
    return
  }
  node := nodes[0]

  // Attempt a push pull
  if err := m.pushPullNode(node.FullAddress(), false); err != nil {
    m.logger.Printf("[ERR] memberlist: Push/Pull with %s failed: %s", node.Name, err)
  }
}

// pushPullNode does a complete state exchange with a specific node.
func (m *Memberlist) pushPullNode(a Address, join bool) error {
  defer metrics.MeasureSince([]string{"memberlist", "pushPullNode"}, time.Now())

  // Attempt to send and receive with the node
  remote, userState, err := m.sendAndReceiveState(a, join)
  if err != nil {
    return err
  }

  if err := m.mergeRemoteState(join, remote, userState); err != nil {
    return err
  }
  return nil
}

localNodes := make([]pushNodeState, len(m.nodes))
for idx, n := range m.nodes {
  localNodes[idx].Name = n.Name
  localNodes[idx].Addr = n.Addr
  localNodes[idx].Port = n.Port
  localNodes[idx].Incarnation = n.Incarnation
  localNodes[idx].State = n.State
  localNodes[idx].Meta = n.Meta
  localNodes[idx].Vsn = []uint8{
    n.PMin, n.PMax, n.PCur,
    n.DMin, n.DMax, n.DCur,
  }
}
m.nodeLock.RUnlock()

// Get the delegate state
var userData []byte
if m.config.Delegate != nil {
  userData = m.config.Delegate.LocalState(join)
}

// Create a bytes buffer writer
bufConn := bytes.NewBuffer(nil)

// Send our node state
header := pushPullHeader{Nodes: len(localNodes), UserStateLen: len(userData), Join: join}
hd := codec.MsgpackHandle{}
enc := codec.NewEncoder(bufConn, &hd)

func (m *Memberlist) readRemoteState(bufConn io.Reader, dec *codec.Decoder) (bool, []pushNodeState, []byte, error) {
  // Read the push/pull header
  var header pushPullHeader
  if err := dec.Decode(&header); err != nil {
    return false, nil, nil, err
  }

  // Allocate space for the transfer
  remoteNodes := make([]pushNodeState, header.Nodes)

  // Try to decode all the states
  for i := 0; i < header.Nodes; i++ {
    if err := dec.Decode(&remoteNodes[i]); err != nil {
      return false, nil, nil, err
    }
  }

  // Read the remote user state into a buffer
  var userBuf []byte
  if header.UserStateLen > 0 {
    userBuf = make([]byte, header.UserStateLen)
    bytes, err := io.ReadAtLeast(bufConn, userBuf, header.UserStateLen)
    if err == nil && bytes != header.UserStateLen {
      err = fmt.Errorf(
        "Failed to read full user state (%d / %d)",
        bytes, header.UserStateLen)
    }
    if err != nil {
      return false, nil, nil, err
    }
  }

  // For proto versions < 2, there is no port provided. Mask old
  // behavior by using the configured port
  for idx := range remoteNodes {
    if m.ProtocolVersion() < 2 || remoteNodes[idx].Port == 0 {
      remoteNodes[idx].Port = uint16(m.config.BindPort)
    }
  }

  return header.Join, remoteNodes, userBuf, nil
}

func (m *Memberlist) mergeRemoteState(join bool, remoteNodes []pushNodeState, userBuf []byte) error {
  if err := m.verifyProtocol(remoteNodes); err != nil {
    return err
  }

  // Invoke the merge delegate if any
  if join && m.config.Merge != nil {
    nodes := make([]*Node, len(remoteNodes))
    for idx, n := range remoteNodes {
      nodes[idx] = &Node{
        Name:  n.Name,
        Addr:  n.Addr,
        Port:  n.Port,
        Meta:  n.Meta,
        State: n.State,
        PMin:  n.Vsn[0],
        PMax:  n.Vsn[1],
        PCur:  n.Vsn[2],
        DMin:  n.Vsn[3],
        DMax:  n.Vsn[4],
        DCur:  n.Vsn[5],
      }
    }
    if err := m.config.Merge.NotifyMerge(nodes); err != nil {
      return err
    }
  }

  // Merge the membership state
  m.mergeState(remoteNodes)

  // Invoke the delegate for user state
  if userBuf != nil && m.config.Delegate != nil {
    m.config.Delegate.MergeRemoteState(userBuf, join)
  }
  return nil
}

func (m *Memberlist) gossip() {
  defer metrics.MeasureSince([]string{"memberlist", "gossip"}, time.Now())

  // Get some random live, suspect, or recently dead nodes
  m.nodeLock.RLock()
  kNodes := kRandomNodes(m.config.GossipNodes, m.nodes, func(n *nodeState) bool {
    if n.Name == m.config.Name {
      return true
    }

    switch n.State {
    case StateAlive, StateSuspect:
      return false

    case StateDead:
      return time.Since(n.StateChange) > m.config.GossipToTheDeadTime

    default:
      return true
    }
  })
  m.nodeLock.RUnlock()

  // Compute the bytes available
  bytesAvail := m.config.UDPBufferSize - compoundHeaderOverhead
  if m.config.EncryptionEnabled() {
    bytesAvail -= encryptOverhead(m.encryptionVersion())
  }

  for _, node := range kNodes {
    // Get any pending broadcasts
    msgs := m.getBroadcasts(compoundOverhead, bytesAvail)
    if len(msgs) == 0 {
      return
    }

    addr := node.Address()
    if len(msgs) == 1 {
      // Send single message as is
      if err := m.rawSendMsgPacket(node.FullAddress(), &node.Node, msgs[0]); err != nil {
        m.logger.Printf("[ERR] memberlist: Failed to send gossip to %s: %s", addr, err)
      }
    } else {
      // Otherwise create and send a compound message
      compound := makeCompoundMessage(msgs)
      if err := m.rawSendMsgPacket(node.FullAddress(), &node.Node, compound.Bytes()); err != nil {
        m.logger.Printf("[ERR] memberlist: Failed to send gossip to %s: %s", addr, err)
      }
    }
  }
}