open-consul/consul/serf.go

307 lines
8.1 KiB
Go
Raw Normal View History

2013-12-07 01:18:09 +00:00
package consul
2013-12-07 00:54:33 +00:00
2013-12-09 23:29:44 +00:00
import (
"net"
"strings"
2016-03-30 00:39:19 +00:00
"github.com/hashicorp/consul/consul/agent"
"github.com/hashicorp/raft"
"github.com/hashicorp/serf/serf"
2013-12-09 23:29:44 +00:00
)
2014-03-20 19:51:49 +00:00
const (
// StatusReap is used to update the status of a node if we
// are handling a EventMemberReap
StatusReap = serf.MemberStatus(-1)
// userEventPrefix is pre-pended to a user event to distinguish it
userEventPrefix = "consul:event:"
2014-03-20 19:51:49 +00:00
)
2014-08-27 01:50:03 +00:00
// userEventName computes the name of a user event
func userEventName(name string) string {
return userEventPrefix + name
}
// isUserEvent checks if a serf event is a user event
func isUserEvent(name string) bool {
return strings.HasPrefix(name, userEventPrefix)
}
// rawUserEventName is used to get the raw user event name
func rawUserEventName(name string) string {
return strings.TrimPrefix(name, userEventPrefix)
2014-08-27 01:50:03 +00:00
}
2013-12-07 00:54:33 +00:00
// lanEventHandler is used to handle events from the lan Serf cluster
func (s *Server) lanEventHandler() {
for {
select {
case e := <-s.eventChLAN:
2013-12-09 23:29:44 +00:00
switch e.EventType() {
case serf.EventMemberJoin:
s.lanNodeJoin(e.(serf.MemberEvent))
s.localMemberEvent(e.(serf.MemberEvent))
case serf.EventMemberLeave, serf.EventMemberFailed:
s.lanNodeFailed(e.(serf.MemberEvent))
s.localMemberEvent(e.(serf.MemberEvent))
2014-03-20 19:51:49 +00:00
case serf.EventMemberReap:
2014-01-09 23:49:09 +00:00
s.localMemberEvent(e.(serf.MemberEvent))
2013-12-09 23:29:44 +00:00
case serf.EventUser:
s.localEvent(e.(serf.UserEvent))
2014-03-20 19:51:49 +00:00
case serf.EventMemberUpdate: // Ignore
2014-03-12 19:46:14 +00:00
case serf.EventQuery: // Ignore
2013-12-09 23:29:44 +00:00
default:
s.logger.Printf("[WARN] consul: Unhandled LAN Serf Event: %#v", e)
2013-12-09 23:29:44 +00:00
}
2013-12-07 00:54:33 +00:00
case <-s.shutdownCh:
return
}
}
}
// wanEventHandler is used to handle events from the wan Serf cluster
func (s *Server) wanEventHandler() {
for {
select {
case e := <-s.eventChWAN:
2013-12-09 23:29:44 +00:00
switch e.EventType() {
case serf.EventMemberJoin:
s.wanNodeJoin(e.(serf.MemberEvent))
case serf.EventMemberLeave, serf.EventMemberFailed:
s.wanNodeFailed(e.(serf.MemberEvent))
2014-03-12 19:46:14 +00:00
case serf.EventMemberUpdate: // Ignore
case serf.EventMemberReap: // Ignore
2013-12-09 23:29:44 +00:00
case serf.EventUser:
2014-03-12 19:46:14 +00:00
case serf.EventQuery: // Ignore
2013-12-09 23:29:44 +00:00
default:
s.logger.Printf("[WARN] consul: Unhandled WAN Serf Event: %#v", e)
2013-12-09 23:29:44 +00:00
}
2013-12-07 00:54:33 +00:00
case <-s.shutdownCh:
return
}
}
}
2013-12-09 23:29:44 +00:00
2014-01-09 23:49:09 +00:00
// localMemberEvent is used to reconcile Serf events with the strongly
// consistent store if we are the current leader
func (s *Server) localMemberEvent(me serf.MemberEvent) {
// Do nothing if we are not the leader
if !s.IsLeader() {
return
}
2014-03-20 19:51:49 +00:00
// Check if this is a reap event
isReap := me.EventType() == serf.EventMemberReap
// Queue the members for reconciliation
2014-01-09 23:49:09 +00:00
for _, m := range me.Members {
2014-03-20 19:51:49 +00:00
// Change the status if this is a reap event
if isReap {
m.Status = StatusReap
}
select {
case s.reconcileCh <- m:
default:
}
}
2013-12-09 23:29:44 +00:00
}
// localEvent is called when we receive an event on the local Serf
func (s *Server) localEvent(event serf.UserEvent) {
// Handle only consul events
if !strings.HasPrefix(event.Name, "consul:") {
return
}
switch name := event.Name; {
case name == newLeaderEvent:
s.logger.Printf("[INFO] consul: New leader elected: %s", event.Payload)
// Trigger the callback
if s.config.ServerUp != nil {
s.config.ServerUp()
}
case isUserEvent(name):
event.Name = rawUserEventName(name)
s.logger.Printf("[DEBUG] consul: User event: %s", event.Name)
// Trigger the callback
if s.config.UserEventHandler != nil {
s.config.UserEventHandler(event)
}
default:
s.logger.Printf("[WARN] consul: Unhandled local event: %v", event)
}
}
// lanNodeJoin is used to handle join events on the LAN pool.
func (s *Server) lanNodeJoin(me serf.MemberEvent) {
2013-12-12 00:24:34 +00:00
for _, m := range me.Members {
2016-03-30 00:39:19 +00:00
ok, parts := agent.IsConsulServer(m)
2013-12-12 00:24:34 +00:00
if !ok {
continue
}
s.logger.Printf("[INFO] consul: Adding LAN server %s", parts)
// See if it's configured as part of our DC.
if parts.Datacenter == s.config.Datacenter {
s.localLock.Lock()
s.localConsuls[raft.ServerAddress(parts.Addr.String())] = parts
s.localLock.Unlock()
}
// If we still expecting to bootstrap, may need to handle this.
if s.config.BootstrapExpect != 0 {
s.maybeBootstrap()
}
}
}
// wanNodeJoin is used to handle join events on the WAN pool.
func (s *Server) wanNodeJoin(me serf.MemberEvent) {
for _, m := range me.Members {
2016-03-30 00:39:19 +00:00
ok, parts := agent.IsConsulServer(m)
if !ok {
s.logger.Printf("[WARN] consul: Non-server in WAN pool: %s", m.Name)
continue
}
s.logger.Printf("[INFO] consul: Adding WAN server %s", parts)
2013-12-12 00:24:34 +00:00
// Search for this node in our existing remotes.
2013-12-12 00:24:34 +00:00
found := false
s.remoteLock.Lock()
2014-01-20 23:39:07 +00:00
existing := s.remoteConsuls[parts.Datacenter]
for idx, e := range existing {
if e.Name == parts.Name {
existing[idx] = parts
2013-12-12 00:24:34 +00:00
found = true
break
}
}
2013-12-09 23:29:44 +00:00
// Add to the list if not known.
2013-12-12 00:24:34 +00:00
if !found {
s.remoteConsuls[parts.Datacenter] = append(existing, parts)
2013-12-12 00:24:34 +00:00
}
s.remoteLock.Unlock()
}
2013-12-09 23:29:44 +00:00
}
2014-06-18 23:15:28 +00:00
// maybeBootsrap is used to handle bootstrapping when a new consul server joins
func (s *Server) maybeBootstrap() {
// Bootstrap can only be done if there are no committed logs, remove our
// expectations of bootstrapping. This is slightly cheaper than the full
// check that BootstrapCluster will do, so this is a good pre-filter.
2014-06-18 23:15:28 +00:00
index, err := s.raftStore.LastIndex()
if err != nil {
s.logger.Printf("[ERR] consul: Failed to read last raft index: %v", err)
2014-06-18 23:15:28 +00:00
return
}
if index != 0 {
s.config.BootstrapExpect = 0
2014-06-18 23:15:28 +00:00
return
}
// Scan for all the known servers.
2014-06-18 23:15:28 +00:00
members := s.serfLAN.Members()
2015-05-08 18:35:12 +00:00
addrs := make([]string, 0)
2014-06-18 23:15:28 +00:00
for _, member := range members {
2016-03-30 00:39:19 +00:00
valid, p := agent.IsConsulServer(member)
2014-06-18 23:15:28 +00:00
if !valid {
continue
}
if p.Datacenter != s.config.Datacenter {
s.logger.Printf("[ERR] consul: Member %v has a conflicting datacenter, ignoring", member)
continue
}
if p.Expect != 0 && p.Expect != s.config.BootstrapExpect {
2014-06-18 23:15:28 +00:00
s.logger.Printf("[ERR] consul: Member %v has a conflicting expect value. All nodes should expect the same number.", member)
return
}
if p.Bootstrap {
s.logger.Printf("[ERR] consul: Member %v has bootstrap mode. Expect disabled.", member)
return
}
2015-05-08 18:35:12 +00:00
addr := &net.TCPAddr{IP: member.Addr, Port: p.Port}
addrs = append(addrs, addr.String())
2014-06-18 23:15:28 +00:00
}
// Skip if we haven't met the minimum expect count.
if len(addrs) < s.config.BootstrapExpect {
2014-06-18 23:15:28 +00:00
return
}
// Attempt a live bootstrap!
var configuration raft.Configuration
for _, addr := range addrs {
// TODO (slackpad) - This will need to be updated once we support
// node IDs.
server := raft.Server{
ID: raft.ServerID(addr),
Address: raft.ServerAddress(addr),
}
configuration.Servers = append(configuration.Servers, server)
}
s.logger.Printf("[INFO] consul: Found expected number of peers (%s), attempting to bootstrap cluster...",
strings.Join(addrs, ","))
future := s.raft.BootstrapCluster(configuration)
if err := future.Error(); err != nil {
s.logger.Printf("[ERR] consul: Failed to bootstrap cluster: %v", err)
2014-06-18 23:15:28 +00:00
}
// Bootstrapping complete, don't enter this again.
s.config.BootstrapExpect = 0
2014-06-18 23:15:28 +00:00
}
// lanNodeFailed is used to handle fail events on the LAN pool.
func (s *Server) lanNodeFailed(me serf.MemberEvent) {
for _, m := range me.Members {
2016-03-30 00:39:19 +00:00
ok, parts := agent.IsConsulServer(m)
if !ok {
continue
}
s.logger.Printf("[INFO] consul: Removing LAN server %s", parts)
s.localLock.Lock()
delete(s.localConsuls, raft.ServerAddress(parts.Addr.String()))
s.localLock.Unlock()
}
}
// wanNodeFailed is used to handle fail events on the WAN pool.
func (s *Server) wanNodeFailed(me serf.MemberEvent) {
2013-12-12 00:24:34 +00:00
for _, m := range me.Members {
2016-03-30 00:39:19 +00:00
ok, parts := agent.IsConsulServer(m)
2013-12-12 00:24:34 +00:00
if !ok {
continue
}
s.logger.Printf("[INFO] consul: Removing WAN server %s", parts)
2013-12-12 00:24:34 +00:00
// Remove the server if known
s.remoteLock.Lock()
2014-01-20 23:39:07 +00:00
existing := s.remoteConsuls[parts.Datacenter]
2013-12-12 00:24:34 +00:00
n := len(existing)
for i := 0; i < n; i++ {
if existing[i].Name == parts.Name {
2013-12-12 00:24:34 +00:00
existing[i], existing[n-1] = existing[n-1], nil
existing = existing[:n-1]
n--
break
}
}
2013-12-09 23:29:44 +00:00
2013-12-12 00:24:34 +00:00
// Trim the list if all known consuls are dead
if n == 0 {
2014-01-20 23:39:07 +00:00
delete(s.remoteConsuls, parts.Datacenter)
2013-12-12 00:24:34 +00:00
} else {
2014-01-20 23:39:07 +00:00
s.remoteConsuls[parts.Datacenter] = existing
2013-12-12 00:24:34 +00:00
}
s.remoteLock.Unlock()
}
2013-12-09 23:29:44 +00:00
}