open-consul/consul/serf.go

133 lines
3.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 (
"github.com/hashicorp/serf/serf"
"net"
2013-12-09 23:29:44 +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:
2014-01-09 23:49:09 +00:00
fallthrough
2013-12-09 23:29:44 +00:00
case serf.EventMemberLeave:
2014-01-09 23:49:09 +00:00
fallthrough
2013-12-09 23:29:44 +00:00
case serf.EventMemberFailed:
2014-01-09 23:49:09 +00:00
s.localMemberEvent(e.(serf.MemberEvent))
2013-12-09 23:29:44 +00:00
case serf.EventUser:
default:
2014-01-10 19:06:11 +00:00
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.remoteJoin(e.(serf.MemberEvent))
case serf.EventMemberLeave:
2013-12-12 00:24:34 +00:00
fallthrough
2013-12-09 23:29:44 +00:00
case serf.EventMemberFailed:
s.remoteFailed(e.(serf.MemberEvent))
case serf.EventUser:
default:
2014-01-10 19:06:11 +00:00
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
}
// Queue the members for reconciliation
2014-01-09 23:49:09 +00:00
for _, m := range me.Members {
select {
case s.reconcileCh <- m:
default:
}
}
2013-12-09 23:29:44 +00:00
}
// remoteJoin is used to handle join events on the wan serf cluster
func (s *Server) remoteJoin(me serf.MemberEvent) {
2013-12-12 00:24:34 +00:00
for _, m := range me.Members {
2014-01-20 23:39:07 +00:00
ok, parts := isConsulServer(m)
2013-12-12 00:24:34 +00:00
if !ok {
2014-01-10 19:06:11 +00:00
s.logger.Printf("[WARN] consul: non-server in WAN pool: %s %s", m.Name)
2013-12-12 00:24:34 +00:00
continue
}
2014-01-20 23:39:07 +00:00
var addr net.Addr = &net.TCPAddr{IP: m.Addr, Port: parts.Port}
s.logger.Printf("[INFO] consul: adding server for datacenter: %s, addr: %s", parts.Datacenter, addr)
2013-12-12 00:24:34 +00:00
// Check if this server is known
found := false
s.remoteLock.Lock()
2014-01-20 23:39:07 +00:00
existing := s.remoteConsuls[parts.Datacenter]
2013-12-12 00:24:34 +00:00
for _, e := range existing {
if e.String() == addr.String() {
found = true
break
}
}
2013-12-09 23:29:44 +00:00
2013-12-12 00:24:34 +00:00
// Add ot the list if not known
if !found {
2014-01-20 23:39:07 +00:00
s.remoteConsuls[parts.Datacenter] = append(existing, addr)
2013-12-12 00:24:34 +00:00
}
s.remoteLock.Unlock()
}
2013-12-09 23:29:44 +00:00
}
// remoteFailed is used to handle fail events on the wan serf cluster
func (s *Server) remoteFailed(me serf.MemberEvent) {
2013-12-12 00:24:34 +00:00
for _, m := range me.Members {
2014-01-20 23:39:07 +00:00
ok, parts := isConsulServer(m)
2013-12-12 00:24:34 +00:00
if !ok {
continue
}
2014-01-20 23:39:07 +00:00
var addr net.Addr = &net.TCPAddr{IP: m.Addr, Port: parts.Port}
s.logger.Printf("[INFO] consul: removing server for datacenter: %s, addr: %s", parts.Datacenter, addr)
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].String() == addr.String() {
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
}