[1.14.x] grpc: ensure grpc resolver correctly uses lan/wan addresses …

…on servers (#17286)

The grpc resolver implementation is fed from changes to the
router.Router. Within the router there is a map of various areas storing
the addressing information for servers in those areas. All map entries
are of the WAN variety except a single special entry for the LAN.

Addressing information in the LAN "area" are local addresses intended
for use when making a client-to-server or server-to-server request.

The client agent correctly updates this LAN area when receiving lan serf
events, so by extension the grpc resolver works fine in that scenario.

The server agent only initially populates a single entry in the LAN area
(for itself) on startup, and then never mutates that area map again.
For normal RPCs a different structure is used for LAN routing.

Additionally when selecting a server to contact in the local datacenter
it will randomly select addresses from either the LAN or WAN addressed
entries in the map.

Unfortunately this means that the grpc resolver stack as it exists on
server agents is either broken or only accidentally functions by having
servers dial each other over the WAN-accessible address. If the operator
disables the serf wan port completely likely this incidental functioning
would break.

This PR enforces that local requests for servers (both for stale reads
or leader forwarded requests) exclusively use the LAN "area" information
and also fixes it so that servers keep that area up to date in the
router.

A test for the grpc resolver logic was added, as well as a higher level
full-stack test to ensure the externally perceived bug does not return.

Also backport subtle test fix from #16396

.changelog/17270.txt

@@ -0,0 +1,3 @@
+```release-note:bug
+grpc: ensure grpc resolver correctly uses lan/wan addresses on servers
+```

agent/consul/client_test.go

@@ -499,11 +499,15 @@ func newClient(t *testing.T, config *Config) *Client {
 	return client
 }
 
-func newTestResolverConfig(t *testing.T, suffix string) resolver.Config {
+func newTestResolverConfig(t *testing.T, suffix string, dc, agentType string) resolver.Config {
 	n := t.Name()
 	s := strings.Replace(n, "/", "", -1)
 	s = strings.Replace(s, "_", "", -1)
-	return resolver.Config{Authority: strings.ToLower(s) + "-" + suffix}
+	return resolver.Config{
+		Datacenter: dc,
+		AgentType:  agentType,
+		Authority:  strings.ToLower(s) + "-" + suffix,
+	}
 }
 
 func newDefaultDeps(t *testing.T, c *Config) Deps {
@@ -518,7 +522,7 @@ func newDefaultDeps(t *testing.T, c *Config) Deps {
 	tls, err := tlsutil.NewConfigurator(c.TLSConfig, logger)
 	require.NoError(t, err, "failed to create tls configuration")
 
-	builder := resolver.NewServerResolverBuilder(newTestResolverConfig(t, c.NodeName+"-"+c.Datacenter))
+	builder := resolver.NewServerResolverBuilder(newTestResolverConfig(t, c.NodeName+"-"+c.Datacenter, c.Datacenter, "server"))
 	r := router.NewRouter(logger, c.Datacenter, fmt.Sprintf("%s.%s", c.NodeName, c.Datacenter), builder)
 	resolver.Register(builder)
 

agent/consul/server_serf.go

@@ -20,6 +20,7 @@ import (
 	"github.com/hashicorp/consul/lib"
 	libserf "github.com/hashicorp/consul/lib/serf"
 	"github.com/hashicorp/consul/logging"
+	"github.com/hashicorp/consul/types"
 )
 
 const (
@@ -356,6 +357,7 @@ func (s *Server) lanNodeJoin(me serf.MemberEvent) {
 
 		// Update server lookup
 		s.serverLookup.AddServer(serverMeta)
+		s.router.AddServer(types.AreaLAN, serverMeta)
 
 		// If we're still expecting to bootstrap, may need to handle this.
 		if s.config.BootstrapExpect != 0 {
@@ -377,6 +379,7 @@ func (s *Server) lanNodeUpdate(me serf.MemberEvent) {
 
 		// Update server lookup
 		s.serverLookup.AddServer(serverMeta)
+		s.router.AddServer(types.AreaLAN, serverMeta)
 	}
 }
 
@@ -515,5 +518,6 @@ func (s *Server) lanNodeFailed(me serf.MemberEvent) {
 
 		// Update id to address map
 		s.serverLookup.RemoveServer(serverMeta)
+		s.router.RemoveServer(types.AreaLAN, serverMeta)
 	}
 }

agent/consul/subscribe_backend_test.go

@@ -377,7 +377,10 @@ func newClientWithGRPCResolver(t *testing.T, ops ...func(*Config)) (*Client, *re
 	}
 
 	builder := resolver.NewServerResolverBuilder(newTestResolverConfig(t,
-		"client."+config.Datacenter+"."+string(config.NodeID)))
+		"client."+config.Datacenter+"."+string(config.NodeID),
+		config.Datacenter,
+		"client",
+	))
 
 	resolver.Register(builder)
 	t.Cleanup(func() {

agent/grpc-external/services/peerstream/stream_test.go

@@ -1252,8 +1252,8 @@ func TestStreamResources_Server_DisconnectsOnHeartbeatTimeout(t *testing.T) {
 	})
 
 	testutil.RunStep(t, "stream is disconnected due to heartbeat timeout", func(t *testing.T) {
-		disconnectTime := it.FutureNow(1)
 		retry.Run(t, func(r *retry.R) {
+			disconnectTime := it.StaticNow()
 			status, ok := srv.StreamStatus(testPeerID)
 			require.True(r, ok)
 			require.False(r, status.Connected)

agent/grpc-external/services/peerstream/testing.go

@@ -150,6 +150,16 @@ func (t *incrementalTime) Now() time.Time {
 	return t.base.Add(dur)
 }
 
+// StaticNow returns the current internal clock without advancing it.
+func (t *incrementalTime) StaticNow() time.Time {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	dur := time.Duration(t.next) * time.Second
+
+	return t.base.Add(dur)
+}
+
 // FutureNow will return a given future value of the Now() function.
 // The numerical argument indicates which future Now value you wanted. The
 // value must be > 0.

agent/grpc-internal/client_test.go

@@ -33,8 +33,8 @@ func TestNewDialer_WithTLSWrapper(t *testing.T) {
 	require.NoError(t, err)
 	t.Cleanup(logError(t, lis.Close))
 
-	builder := resolver.NewServerResolverBuilder(newConfig(t))
-	builder.AddServer(types.AreaWAN, &metadata.Server{
+	builder := resolver.NewServerResolverBuilder(newConfig(t, "dc1", "server"))
+	builder.AddServer(types.AreaLAN, &metadata.Server{
 		Name:       "server-1",
 		ID:         "ID1",
 		Datacenter: "dc1",
@@ -84,14 +84,21 @@ func TestNewDialer_WithALPNWrapper(t *testing.T) {
 		p.Wait()
 	}()
 
-	builder := resolver.NewServerResolverBuilder(newConfig(t))
+	builder := resolver.NewServerResolverBuilder(newConfig(t, "dc1", "server"))
 	builder.AddServer(types.AreaWAN, &metadata.Server{
 		Name:       "server-1",
 		ID:         "ID1",
 		Datacenter: "dc1",
 		Addr:       lis1.Addr(),
 		UseTLS:     true,
 	})
+	builder.AddServer(types.AreaLAN, &metadata.Server{
+		Name:       "server-1",
+		ID:         "ID1",
+		Datacenter: "dc1",
+		Addr:       lis1.Addr(),
+		UseTLS:     true,
+	})
 	builder.AddServer(types.AreaWAN, &metadata.Server{
 		Name:       "server-2",
 		ID:         "ID2",
@@ -139,7 +146,7 @@ func TestNewDialer_WithALPNWrapper(t *testing.T) {
 func TestNewDialer_IntegrationWithTLSEnabledHandler(t *testing.T) {
 	// if this test is failing because of expired certificates
 	// use the procedure in test/CA-GENERATION.md
-	res := resolver.NewServerResolverBuilder(newConfig(t))
+	res := resolver.NewServerResolverBuilder(newConfig(t, "dc1", "server"))
 	registerWithGRPC(t, res)
 
 	tlsConf, err := tlsutil.NewConfigurator(tlsutil.Config{
@@ -156,9 +163,17 @@ func TestNewDialer_IntegrationWithTLSEnabledHandler(t *testing.T) {
 	srv := newSimpleTestServer(t, "server-1", "dc1", tlsConf)
 
 	md := srv.Metadata()
-	res.AddServer(types.AreaWAN, md)
+	res.AddServer(types.AreaLAN, md)
 	t.Cleanup(srv.shutdown)
 
+	{
+		// Put a duplicate instance of this on the WAN that will
+		// fail if we accidentally use it.
+		srv := newPanicTestServer(t, hclog.Default(), "server-1", "dc1", nil)
+		res.AddServer(types.AreaWAN, srv.Metadata())
+		t.Cleanup(srv.shutdown)
+	}
+
 	pool := NewClientConnPool(ClientConnPoolConfig{
 		Servers:               res,
 		TLSWrapper:            TLSWrapper(tlsConf.OutgoingRPCWrapper()),
@@ -186,7 +201,7 @@ func TestNewDialer_IntegrationWithTLSEnabledHandler_viaMeshGateway(t *testing.T)
 	// use the procedure in test/CA-GENERATION.md
 	gwAddr := ipaddr.FormatAddressPort("127.0.0.1", freeport.GetOne(t))
 
-	res := resolver.NewServerResolverBuilder(newConfig(t))
+	res := resolver.NewServerResolverBuilder(newConfig(t, "dc2", "server"))
 	registerWithGRPC(t, res)
 
 	tlsConf, err := tlsutil.NewConfigurator(tlsutil.Config{
@@ -261,7 +276,7 @@ func TestNewDialer_IntegrationWithTLSEnabledHandler_viaMeshGateway(t *testing.T)
 
 func TestClientConnPool_IntegrationWithGRPCResolver_Failover(t *testing.T) {
 	count := 4
-	res := resolver.NewServerResolverBuilder(newConfig(t))
+	res := resolver.NewServerResolverBuilder(newConfig(t, "dc1", "server"))
 	registerWithGRPC(t, res)
 	pool := NewClientConnPool(ClientConnPoolConfig{
 		Servers:               res,
@@ -272,9 +287,18 @@ func TestClientConnPool_IntegrationWithGRPCResolver_Failover(t *testing.T) {
 
 	for i := 0; i < count; i++ {
 		name := fmt.Sprintf("server-%d", i)
-		srv := newSimpleTestServer(t, name, "dc1", nil)
-		res.AddServer(types.AreaWAN, srv.Metadata())
-		t.Cleanup(srv.shutdown)
+		{
+			srv := newSimpleTestServer(t, name, "dc1", nil)
+			res.AddServer(types.AreaLAN, srv.Metadata())
+			t.Cleanup(srv.shutdown)
+		}
+		{
+			// Put a duplicate instance of this on the WAN that will
+			// fail if we accidentally use it.
+			srv := newPanicTestServer(t, hclog.Default(), name, "dc1", nil)
+			res.AddServer(types.AreaWAN, srv.Metadata())
+			t.Cleanup(srv.shutdown)
+		}
 	}
 
 	conn, err := pool.ClientConn("dc1")
@@ -287,7 +311,7 @@ func TestClientConnPool_IntegrationWithGRPCResolver_Failover(t *testing.T) {
 	first, err := client.Something(ctx, &testservice.Req{})
 	require.NoError(t, err)
 
-	res.RemoveServer(types.AreaWAN, &metadata.Server{ID: first.ServerName, Datacenter: "dc1"})
+	res.RemoveServer(types.AreaLAN, &metadata.Server{ID: first.ServerName, Datacenter: "dc1"})
 
 	resp, err := client.Something(ctx, &testservice.Req{})
 	require.NoError(t, err)
@@ -296,7 +320,7 @@ func TestClientConnPool_IntegrationWithGRPCResolver_Failover(t *testing.T) {
 
 func TestClientConnPool_ForwardToLeader_Failover(t *testing.T) {
 	count := 3
-	res := resolver.NewServerResolverBuilder(newConfig(t))
+	res := resolver.NewServerResolverBuilder(newConfig(t, "dc1", "server"))
 	registerWithGRPC(t, res)
 	pool := NewClientConnPool(ClientConnPoolConfig{
 		Servers:               res,
@@ -308,10 +332,19 @@ func TestClientConnPool_ForwardToLeader_Failover(t *testing.T) {
 	var servers []testServer
 	for i := 0; i < count; i++ {
 		name := fmt.Sprintf("server-%d", i)
-		srv := newSimpleTestServer(t, name, "dc1", nil)
-		res.AddServer(types.AreaWAN, srv.Metadata())
-		servers = append(servers, srv)
-		t.Cleanup(srv.shutdown)
+		{
+			srv := newSimpleTestServer(t, name, "dc1", nil)
+			res.AddServer(types.AreaLAN, srv.Metadata())
+			servers = append(servers, srv)
+			t.Cleanup(srv.shutdown)
+		}
+		{
+			// Put a duplicate instance of this on the WAN that will
+			// fail if we accidentally use it.
+			srv := newPanicTestServer(t, hclog.Default(), name, "dc1", nil)
+			res.AddServer(types.AreaWAN, srv.Metadata())
+			t.Cleanup(srv.shutdown)
+		}
 	}
 
 	// Set the leader address to the first server.
@@ -338,16 +371,20 @@ func TestClientConnPool_ForwardToLeader_Failover(t *testing.T) {
 	require.Equal(t, resp.ServerName, servers[1].name)
 }
 
-func newConfig(t *testing.T) resolver.Config {
+func newConfig(t *testing.T, dc, agentType string) resolver.Config {
 	n := t.Name()
 	s := strings.Replace(n, "/", "", -1)
 	s = strings.Replace(s, "_", "", -1)
-	return resolver.Config{Authority: strings.ToLower(s)}
+	return resolver.Config{
+		Datacenter: dc,
+		AgentType:  agentType,
+		Authority:  strings.ToLower(s),
+	}
 }
 
 func TestClientConnPool_IntegrationWithGRPCResolver_Rebalance(t *testing.T) {
 	count := 5
-	res := resolver.NewServerResolverBuilder(newConfig(t))
+	res := resolver.NewServerResolverBuilder(newConfig(t, "dc1", "server"))
 	registerWithGRPC(t, res)
 	pool := NewClientConnPool(ClientConnPoolConfig{
 		Servers:               res,
@@ -359,8 +396,13 @@ func TestClientConnPool_IntegrationWithGRPCResolver_Rebalance(t *testing.T) {
 	for i := 0; i < count; i++ {
 		name := fmt.Sprintf("server-%d", i)
 		srv := newSimpleTestServer(t, name, "dc1", nil)
-		res.AddServer(types.AreaWAN, srv.Metadata())
+		res.AddServer(types.AreaLAN, srv.Metadata())
 		t.Cleanup(srv.shutdown)
+		// Put a duplicate instance of this on the WAN that will
+		// fail if we accidentally use it.
+		srvBad := newPanicTestServer(t, hclog.Default(), name, "dc1", nil)
+		res.AddServer(types.AreaWAN, srvBad.Metadata())
+		t.Cleanup(srvBad.shutdown)
 	}
 
 	conn, err := pool.ClientConn("dc1")
@@ -401,7 +443,7 @@ func TestClientConnPool_IntegrationWithGRPCResolver_Rebalance(t *testing.T) {
 func TestClientConnPool_IntegrationWithGRPCResolver_MultiDC(t *testing.T) {
 	dcs := []string{"dc1", "dc2", "dc3"}
 
-	res := resolver.NewServerResolverBuilder(newConfig(t))
+	res := resolver.NewServerResolverBuilder(newConfig(t, "dc1", "server"))
 	registerWithGRPC(t, res)
 	pool := NewClientConnPool(ClientConnPoolConfig{
 		Servers:               res,
@@ -413,7 +455,16 @@ func TestClientConnPool_IntegrationWithGRPCResolver_MultiDC(t *testing.T) {
 	for _, dc := range dcs {
 		name := "server-0-" + dc
 		srv := newSimpleTestServer(t, name, dc, nil)
-		res.AddServer(types.AreaWAN, srv.Metadata())
+		if dc == "dc1" {
+			res.AddServer(types.AreaLAN, srv.Metadata())
+			// Put a duplicate instance of this on the WAN that will
+			// fail if we accidentally use it.
+			srvBad := newPanicTestServer(t, hclog.Default(), name, dc, nil)
+			res.AddServer(types.AreaWAN, srvBad.Metadata())
+			t.Cleanup(srvBad.shutdown)
+		} else {
+			res.AddServer(types.AreaWAN, srv.Metadata())
+		}
 		t.Cleanup(srv.shutdown)
 	}
 

agent/grpc-internal/handler_test.go

@@ -26,11 +26,11 @@ func TestHandler_PanicRecoveryInterceptor(t *testing.T) {
 		Output: &buf,
 	})
 
-	res := resolver.NewServerResolverBuilder(newConfig(t))
+	res := resolver.NewServerResolverBuilder(newConfig(t, "dc1", "server"))
 	registerWithGRPC(t, res)
 
 	srv := newPanicTestServer(t, logger, "server-1", "dc1", nil)
-	res.AddServer(types.AreaWAN, srv.Metadata())
+	res.AddServer(types.AreaLAN, srv.Metadata())
 	t.Cleanup(srv.shutdown)
 
 	pool := NewClientConnPool(ClientConnPoolConfig{