Consolidate server lookup into one place and replace usages of localConsuls.

agent/consul/rpc.go

@@ -238,9 +238,7 @@ func (s *Server) getLeader() (bool, *metadata.Server) {
 	}
 
 	// Lookup the server
-	s.localLock.RLock()
+	server, _ := s.serverLookup.GetServer(leader)
-	server := s.localConsuls[leader]
-	s.localLock.RUnlock()
 
 	// Server could be nil
 	return false, server

agent/consul/serf.go

@@ -125,18 +125,14 @@ func (s *Server) localEvent(event serf.UserEvent) {
 // lanNodeJoin is used to handle join events on the LAN pool.
 func (s *Server) lanNodeJoin(me serf.MemberEvent) {
 	for _, m := range me.Members {
-		ok, parts := metadata.IsConsulServer(m)
+		ok, serverMeta := metadata.IsConsulServer(m)
 		if !ok {
 			continue
 		}
-		s.logger.Printf("[INFO] consul: Adding LAN server %s", parts)
+		s.logger.Printf("[INFO] consul: Adding LAN server %s", serverMeta)
 
-		// See if it's configured as part of our DC.
+		// Update server lookup
-		if parts.Datacenter == s.config.Datacenter {
+		s.serverLookup.AddServer(serverMeta)
-			s.localLock.Lock()
-			s.localConsuls[raft.ServerAddress(parts.Addr.String())] = parts
-			s.localLock.Unlock()
-		}
 
 		// If we're still expecting to bootstrap, may need to handle this.
 		if s.config.BootstrapExpect != 0 {
@@ -144,7 +140,7 @@ func (s *Server) lanNodeJoin(me serf.MemberEvent) {
 		}
 
 		// Update id to address map
-		s.serverAddressLookup.AddServer(parts.ID, parts.Addr.String())
+		s.serverLookup.AddServer(serverMeta)
 
 		// Kick the join flooders.
 		s.FloodNotify()
@@ -274,11 +270,7 @@ func (s *Server) lanNodeFailed(me serf.MemberEvent) {
 		}
 		s.logger.Printf("[INFO] consul: Removing LAN server %s", parts)
 
-		s.localLock.Lock()
-		delete(s.localConsuls, raft.ServerAddress(parts.Addr.String()))
-		s.localLock.Unlock()
-
 		// Update id to address map
-		s.serverAddressLookup.RemoveServer(parts.ID)
+		s.serverLookup.RemoveServer(parts)
 	}
 }

agent/consul/server.go

@@ -123,11 +123,6 @@ type Server struct {
 	// strong consistency.
 	fsm *consulFSM
 
-	// localConsuls is used to track the known consuls
-	// in the local datacenter. Used to do leader forwarding.
-	localConsuls map[raft.ServerAddress]*metadata.Server
-	localLock    sync.RWMutex
-
 	// Logger uses the provided LogOutput
 	logger *log.Logger
 
@@ -171,8 +166,8 @@ type Server struct {
 	// which SHOULD only consist of Consul servers
 	serfWAN *serf.Serf
 
-	// fast lookup from id to server address to provide to the raft transport layer
+	// serverLookup provides fast and thread-safe lookup by id and address
-	serverAddressLookup *ServerAddressLookup
+	serverLookup *ServerLookup
 
 	// floodLock controls access to floodCh.
 	floodLock sync.RWMutex
@@ -298,7 +293,6 @@ func NewServerLogger(config *Config, logger *log.Logger, tokens *token.Store) (*
 		connPool:              connPool,
 		eventChLAN:            make(chan serf.Event, 256),
 		eventChWAN:            make(chan serf.Event, 256),
-		localConsuls:          make(map[raft.ServerAddress]*metadata.Server),
 		logger:                logger,
 		reconcileCh:           make(chan serf.Member, 32),
 		router:                router.NewRouter(logger, config.Datacenter),
@@ -307,7 +301,7 @@ func NewServerLogger(config *Config, logger *log.Logger, tokens *token.Store) (*
 		reassertLeaderCh:      make(chan chan error),
 		sessionTimers:         NewSessionTimers(),
 		tombstoneGC:           gc,
-		serverAddressLookup:   NewServerAddressLookup(),
+		serverLookup:          NewServerLookup(),
 		shutdownCh:            shutdownCh,
 	}
 
@@ -502,7 +496,7 @@ func (s *Server) setupRaft() error {
 		Stream:                s.raftLayer,
 		MaxPool:               3,
 		Timeout:               10 * time.Second,
-		ServerAddressProvider: s.serverAddressLookup,
+		ServerAddressProvider: s.serverLookup,
 	}
 
 	trans := raft.NewNetworkTransportWithConfig(transConfig)
@@ -705,9 +699,7 @@ func (s *Server) setupRPC(tlsWrap tlsutil.DCWrapper) error {
 			return true
 		}
 
-		s.localLock.RLock()
+		server, ok := s.serverLookup.GetServer(address)
-		server, ok := s.localConsuls[address]
-		s.localLock.RUnlock()
 
 		if !ok {
 			return false

agent/consul/server_address_lookup.go

@@ -4,31 +4,53 @@ import (
 	"fmt"
 	"sync"
 
+	"github.com/hashicorp/consul/agent/metadata"
 	"github.com/hashicorp/raft"
 )
 
-// serverIdToAddress is a map from id to address for servers in the LAN pool.
+// ServerLookup encapsulates looking up servers by id and address
-// used for fast lookup to satisfy the ServerAddressProvider interface
+type ServerLookup struct {
-type ServerAddressLookup struct {
+	lock            sync.RWMutex
-	serverIdToAddress sync.Map
+	addressToServer map[raft.ServerAddress]*metadata.Server
+	IdToServer      map[raft.ServerID]*metadata.Server
 }
 
-func NewServerAddressLookup() *ServerAddressLookup {
+func NewServerLookup() *ServerLookup {
-	return &ServerAddressLookup{}
+	return &ServerLookup{addressToServer: make(map[raft.ServerAddress]*metadata.Server), IdToServer: make(map[raft.ServerID]*metadata.Server)}
 }
 
-func (sa *ServerAddressLookup) AddServer(id string, address string) {
+func (sa *ServerLookup) AddServer(server *metadata.Server) {
-	sa.serverIdToAddress.Store(id, address)
+	sa.lock.Lock()
+	defer sa.lock.Unlock()
+	sa.addressToServer[raft.ServerAddress(server.Addr.String())] = server
+	sa.IdToServer[raft.ServerID(server.ID)] = server
 }
 
-func (sa *ServerAddressLookup) RemoveServer(id string) {
+func (sa *ServerLookup) RemoveServer(server *metadata.Server) {
-	sa.serverIdToAddress.Delete(id)
+	sa.lock.Lock()
+	defer sa.lock.Unlock()
+	delete(sa.addressToServer, raft.ServerAddress(server.Addr.String()))
+	delete(sa.IdToServer, raft.ServerID(server.ID))
 }
 
-func (sa *ServerAddressLookup) ServerAddr(id raft.ServerID) (raft.ServerAddress, error) {
+// Implements the ServerAddressProvider interface
-	val, ok := sa.serverIdToAddress.Load(string(id))
+func (sa *ServerLookup) ServerAddr(id raft.ServerID) (raft.ServerAddress, error) {
+	sa.lock.RLock()
+	defer sa.lock.RUnlock()
+	svr, ok := sa.IdToServer[id]
 	if !ok {
 		return "", fmt.Errorf("Could not find address for server id %v", id)
 	}
-	return raft.ServerAddress(val.(string)), nil
+	return raft.ServerAddress(svr.Addr.String()), nil
+}
+
+// GetServer looks up the server by address, returns a boolean if not found
+func (sa *ServerLookup) GetServer(addr raft.ServerAddress) (*metadata.Server, bool) {
+	sa.lock.RLock()
+	defer sa.lock.RUnlock()
+	svr, ok := sa.addressToServer[addr]
+	if !ok {
+		return nil, false
+	}
+	return svr, true
 }

agent/consul/server_address_lookup_test.go

@@ -3,14 +3,32 @@ package consul
 import (
 	"fmt"
 	"testing"
+
+	"github.com/hashicorp/consul/agent/metadata"
+	"github.com/hashicorp/raft"
 )
 
-func TestServerAddressLookup(t *testing.T) {
+type testAddr struct {
-	lookup := NewServerAddressLookup()
+	addr string
-	addr := "72.0.0.17:8300"
+}
-	lookup.AddServer("1", addr)
 
-	got, err := lookup.ServerAddr("1")
+func (ta *testAddr) Network() string {
+	return "tcp"
+}
+
+func (ta *testAddr) String() string {
+	return ta.addr
+}
+
+func TestServerLookup(t *testing.T) {
+	lookup := NewServerLookup()
+	addr := "72.0.0.17:8300"
+	id := "1"
+
+	svr := &metadata.Server{ID: id, Addr: &testAddr{addr}}
+	lookup.AddServer(svr)
+
+	got, err := lookup.ServerAddr(raft.ServerID(id))
 	if err != nil {
 		t.Fatalf("Unexpected error:%v", err)
 	}
@@ -18,7 +36,15 @@ func TestServerAddressLookup(t *testing.T) {
 		t.Fatalf("Expected %v but got %v", addr, got)
 	}
 
-	lookup.RemoveServer("1")
+	server, ok := lookup.GetServer(raft.ServerAddress(addr))
+	if !ok {
+		t.Fatalf("Expected lookup to return true")
+	}
+	if server.Addr.String() != addr {
+		t.Fatalf("Expected lookup to return address %v but got %v", addr, server.Addr)
+	}
+
+	lookup.RemoveServer(svr)
 
 	got, err = lookup.ServerAddr("1")
 	expectedErr := fmt.Errorf("Could not find address for server id 1")
@@ -26,5 +52,7 @@ func TestServerAddressLookup(t *testing.T) {
 		t.Fatalf("Unexpected error, got %v wanted %v", err, expectedErr)
 	}
 
-	lookup.RemoveServer("3")
+	svr2 := &metadata.Server{ID: "2", Addr: &testAddr{"123.4.5.6"}}
+	lookup.RemoveServer(svr2)
+
 }

agent/consul/server_test.go

@@ -342,7 +342,7 @@ func TestServer_JoinSeparateLanAndWanAddresses(t *testing.T) {
 		if len(s2.router.GetDatacenters()) != 2 {
 			r.Fatalf("remote consul missing")
 		}
-		if len(s2.localConsuls) != 2 {
+		if len(s2.serverLookup.addressToServer) != 2 {
 			r.Fatalf("local consul fellow s3 for s2 missing")
 		}
 	})
@@ -666,14 +666,12 @@ func testVerifyRPC(s1, s2 *Server, t *testing.T) (bool, error) {
 	retry.Run(t, func(r *retry.R) { r.Check(wantPeers(s2, 2)) })
 
 	// Have s2 make an RPC call to s1
-	s2.localLock.RLock()
 	var leader *metadata.Server
-	for _, server := range s2.localConsuls {
+	for _, server := range s2.serverLookup.addressToServer {
 		if server.Name == s1.config.NodeName {
 			leader = server
 		}
 	}
-	s2.localLock.RUnlock()
 	if leader == nil {
 		t.Fatal("no leader")
 	}