func (s *Server) handleReapMember(member serf.Member) error {
	return s.handleDeregisterMember("reaped", member)
}
handleReapMember is used to handle members that have been
reaped after a prolonged failure. They are deregistered.
package consul
import (
	"fmt"
	"net"
	"strconv"
	"strings"
	"time"
	"github.com/socketplane/socketplane/Godeps/_workspace/src/github.com/armon/go-metrics"
	"github.com/socketplane/socketplane/Godeps/_workspace/src/github.com/hashicorp/consul/consul/structs"
	"github.com/socketplane/socketplane/Godeps/_workspace/src/github.com/hashicorp/raft"
	"github.com/socketplane/socketplane/Godeps/_workspace/src/github.com/hashicorp/serf/serf"
)
const (
	SerfCheckID           = "serfHealth"
	SerfCheckName         = "Serf Health Status"
	SerfCheckAliveOutput  = "Agent alive and reachable"
	SerfCheckFailedOutput = "Agent not live or unreachable"
	ConsulServiceID       = "consul"
	ConsulServiceName     = "consul"
	newLeaderEvent        = "consul:new-leader"
)
func (s *Server) monitorLeadership() {
	leaderCh := s.raft.LeaderCh()
	var stopCh chan struct{}
	for {
		select {
		case isLeader := <-leaderCh:
			if isLeader {
				stopCh = make(chan struct{})
				go s.leaderLoop(stopCh)
				s.logger.Printf("[INFO] consul: cluster leadership acquired")
			} else if stopCh != nil {
				close(stopCh)
				stopCh = nil
				s.logger.Printf("[INFO] consul: cluster leadership lost")
			}
		case <-s.shutdownCh:
			return
		}
	}
}
func (s *Server) leaderLoop(stopCh chan struct{}) {
	defer s.revokeLeadership()
	payload := []byte(s.config.NodeName)
	if err := s.serfLAN.UserEvent(newLeaderEvent, payload, false); err != nil {
		s.logger.Printf("[WARN] consul: failed to broadcast new leader event: %v", err)
	}
	var reconcileCh chan serf.Member
	establishedLeader := false
RECONCILE:
	reconcileCh = nil
	interval := time.After(s.config.ReconcileInterval)
	start := time.Now()
	barrier := s.raft.Barrier(0)
	if err := barrier.Error(); err != nil {
		s.logger.Printf("[ERR] consul: failed to wait for barrier: %v", err)
		goto WAIT
	}
	metrics.MeasureSince([]string{"consul", "leader", "barrier"}, start)
	if !establishedLeader {
		if err := s.establishLeadership(); err != nil {
			s.logger.Printf("[ERR] consul: failed to establish leadership: %v",
				err)
			goto WAIT
		}
		establishedLeader = true
	}
	if err := s.reconcile(); err != nil {
		s.logger.Printf("[ERR] consul: failed to reconcile: %v", err)
		goto WAIT
	}
	reconcileCh = s.reconcileCh
WAIT:
	for {
		select {
		case <-stopCh:
			return
		case <-s.shutdownCh:
			return
		case <-interval:
			goto RECONCILE
		case member := <-reconcileCh:
			s.reconcileMember(member)
		case index := <-s.tombstoneGC.ExpireCh():
			go s.reapTombstones(index)
		}
	}
}
func (s *Server) establishLeadership() error {
	s.tombstoneGC.SetEnabled(true)
	lastIndex := s.raft.LastIndex()
	s.tombstoneGC.Hint(lastIndex)
	s.logger.Printf("[DEBUG] consul: reset tombstone GC to index %d", lastIndex)
	if err := s.initializeACL(); err != nil {
		s.logger.Printf("[ERR] consul: ACL initialization failed: %v", err)
		return err
	}
	if err := s.initializeSessionTimers(); err != nil {
		s.logger.Printf("[ERR] consul: Session Timers initialization failed: %v",
			err)
		return err
	}
	return nil
}
func (s *Server) revokeLeadership() error {
	s.tombstoneGC.SetEnabled(false)
	if err := s.clearAllSessionTimers(); err != nil {
		s.logger.Printf("[ERR] consul: Clearing session timers failed: %v", err)
		return err
	}
	return nil
}
func (s *Server) initializeACL() error {
	authDC := s.config.ACLDatacenter
	if len(authDC) == 0 || authDC != s.config.Datacenter {
		return nil
	}
	s.aclAuthCache.Purge()
	state := s.fsm.State()
	_, acl, err := state.ACLGet(anonymousToken)
	if err != nil {
		return fmt.Errorf("failed to get anonymous token: %v", err)
	}
	if acl == nil {
		req := structs.ACLRequest{
			Datacenter: authDC,
			Op:         structs.ACLSet,
			ACL: structs.ACL{
				ID:   anonymousToken,
				Name: "Anonymous Token",
				Type: structs.ACLTypeClient,
			},
		}
		_, err := s.raftApply(structs.ACLRequestType, &req)
		if err != nil {
			return fmt.Errorf("failed to create anonymous token: %v", err)
		}
	}
	master := s.config.ACLMasterToken
	if len(master) == 0 {
		return nil
	}
	_, acl, err = state.ACLGet(master)
	if err != nil {
		return fmt.Errorf("failed to get master token: %v", err)
	}
	if acl == nil {
		req := structs.ACLRequest{
			Datacenter: authDC,
			Op:         structs.ACLSet,
			ACL: structs.ACL{
				ID:   master,
				Name: "Master Token",
				Type: structs.ACLTypeManagement,
			},
		}
		_, err := s.raftApply(structs.ACLRequestType, &req)
		if err != nil {
			return fmt.Errorf("failed to create master token: %v", err)
		}
	}
	return nil
}
func (s *Server) reconcile() (err error) {
	defer metrics.MeasureSince([]string{"consul", "leader", "reconcile"}, time.Now())
	members := s.serfLAN.Members()
	knownMembers := make(map[string]struct{})
	for _, member := range members {
		if err := s.reconcileMember(member); err != nil {
			return err
		}
		knownMembers[member.Name] = struct{}{}
	}
	return s.reconcileReaped(knownMembers)
}
func (s *Server) reconcileReaped(known map[string]struct{}) error {
	state := s.fsm.State()
	_, checks := state.ChecksInState(structs.HealthAny)
	for _, check := range checks {
		if check.CheckID != SerfCheckID {
			continue
		}
		if _, ok := known[check.Node]; ok {
			continue
		}
		member := serf.Member{
			Name: check.Node,
			Tags: map[string]string{
				"dc":   s.config.Datacenter,
				"role": "node",
			},
		}
		_, services := state.NodeServices(check.Node)
		serverPort := 0
		for _, service := range services.Services {
			if service.ID == ConsulServiceID {
				serverPort = service.Port
				break
			}
		}
		if serverPort > 0 {
			member.Tags["role"] = "consul"
			member.Tags["port"] = strconv.FormatUint(uint64(serverPort), 10)
		}
		if err := s.handleReapMember(member); err != nil {
			return err
		}
	}
	return nil
}
func (s *Server) reconcileMember(member serf.Member) error {
	if !s.shouldHandleMember(member) {
		s.logger.Printf("[WARN] consul: skipping reconcile of node %v", member)
		return nil
	}
	defer metrics.MeasureSince([]string{"consul", "leader", "reconcileMember"}, time.Now())
	var err error
	switch member.Status {
	case serf.StatusAlive:
		err = s.handleAliveMember(member)
	case serf.StatusFailed:
		err = s.handleFailedMember(member)
	case serf.StatusLeft:
		err = s.handleLeftMember(member)
	case StatusReap:
		err = s.handleReapMember(member)
	}
	if err != nil {
		s.logger.Printf("[ERR] consul: failed to reconcile member: %v: %v",
			member, err)
		if strings.Contains(err.Error(), permissionDenied) {
			return nil
		}
		return err
	}
	return nil
}
func (s *Server) shouldHandleMember(member serf.Member) bool {
	if valid, dc := isConsulNode(member); valid && dc == s.config.Datacenter {
		return true
	}
	if valid, parts := isConsulServer(member); valid && parts.Datacenter == s.config.Datacenter {
		return true
	}
	return false
}
func (s *Server) handleAliveMember(member serf.Member) error {
	state := s.fsm.State()
	var service *structs.NodeService
	if valid, parts := isConsulServer(member); valid {
		service = &structs.NodeService{
			ID:      ConsulServiceID,
			Service: ConsulServiceName,
			Port:    parts.Port,
		}
		if err := s.joinConsulServer(member, parts); err != nil {
			return err
		}
	}
	_, found, addr := state.GetNode(member.Name)
	if found && addr == member.Addr.String() {
		if service != nil {
			match := false
			_, services := state.NodeServices(member.Name)
			if services != nil {
				for id, _ := range services.Services {
					if id == service.ID {
						match = true
					}
				}
			}
			if !match {
				goto AFTER_CHECK
			}
		}
		_, checks := state.NodeChecks(member.Name)
		for _, check := range checks {
			if check.CheckID == SerfCheckID && check.Status == structs.HealthPassing {
				return nil
			}
		}
	}
AFTER_CHECK:
	s.logger.Printf("[INFO] consul: member '%s' joined, marking health alive", member.Name)
	req := structs.RegisterRequest{
		Datacenter: s.config.Datacenter,
		Node:       member.Name,
		Address:    member.Addr.String(),
		Service:    service,
		Check: &structs.HealthCheck{
			Node:    member.Name,
			CheckID: SerfCheckID,
			Name:    SerfCheckName,
			Status:  structs.HealthPassing,
			Output:  SerfCheckAliveOutput,
		},
		WriteRequest: structs.WriteRequest{Token: s.config.ACLToken},
	}
	var out struct{}
	return s.endpoints.Catalog.Register(&req, &out)
}
func (s *Server) handleFailedMember(member serf.Member) error {
	state := s.fsm.State()
	_, found, addr := state.GetNode(member.Name)
	if found && addr == member.Addr.String() {
		_, checks := state.NodeChecks(member.Name)
		for _, check := range checks {
			if check.CheckID == SerfCheckID && check.Status == structs.HealthCritical {
				return nil
			}
		}
	}
	s.logger.Printf("[INFO] consul: member '%s' failed, marking health critical", member.Name)
	req := structs.RegisterRequest{
		Datacenter: s.config.Datacenter,
		Node:       member.Name,
		Address:    member.Addr.String(),
		Check: &structs.HealthCheck{
			Node:    member.Name,
			CheckID: SerfCheckID,
			Name:    SerfCheckName,
			Status:  structs.HealthCritical,
			Output:  SerfCheckFailedOutput,
		},
		WriteRequest: structs.WriteRequest{Token: s.config.ACLToken},
	}
	var out struct{}
	return s.endpoints.Catalog.Register(&req, &out)
}
func (s *Server) handleLeftMember(member serf.Member) error {
	return s.handleDeregisterMember("left", member)
}