Move RPC router from Client/Server and into BaseDeps (#8559)

This will allow it to be a shared component which is needed for AutoConfig

Author: mkeeler

agent/agent.go

@@ -18,6 +18,7 @@ import (
 	"time"
 
 	"github.com/hashicorp/consul/agent/dns"
+	"github.com/hashicorp/consul/agent/router"
 	"github.com/hashicorp/go-connlimit"
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-memdb"
@@ -307,6 +308,9 @@ type Agent struct {
 	// Connection Pool
 	connPool *pool.ConnPool
 
+	// Shared RPC Router
+	router *router.Router
+
 	// enterpriseAgent embeds fields that we only access in consul-enterprise builds
 	enterpriseAgent
 }
@@ -352,6 +356,7 @@ func New(bd BaseDeps) (*Agent, error) {
 		MemSink:         bd.MetricsHandler,
 		connPool:        bd.ConnPool,
 		autoConf:        bd.AutoConfig,
+		router:          bd.Router,
 	}
 
 	a.serviceManager = NewServiceManager(&a)
@@ -463,6 +468,7 @@ func (a *Agent) Start(ctx context.Context) error {
 		consul.WithTokenStore(a.tokens),
 		consul.WithTLSConfigurator(a.tlsConfigurator),
 		consul.WithConnectionPool(a.connPool),
+		consul.WithRouter(a.router),
 	}
 
 	// Setup either the client or the server.

agent/agent_test.go

@@ -4752,3 +4752,33 @@ func TestAgent_AutoEncrypt(t *testing.T) {
 	require.Len(t, x509Cert.URIs, 1)
 	require.Equal(t, id.URI(), x509Cert.URIs[0])
 }
+
+func TestSharedRPCRouter(t *testing.T) {
+	// this test runs both a server and client and ensures that the shared
+	// router is being used. It would be possible for the Client and Server
+	// types to create and use their own routers and for RPCs such as the
+	// ones used in WaitForTestAgent to succeed. However accessing the
+	// router stored on the agent ensures that Serf information from the
+	// Client/Server types are being set in the same shared rpc router.
+
+	srv := NewTestAgent(t, "")
+	defer srv.Shutdown()
+
+	testrpc.WaitForTestAgent(t, srv.RPC, "dc1")
+
+	mgr, server := srv.Agent.router.FindLANRoute()
+	require.NotNil(t, mgr)
+	require.NotNil(t, server)
+
+	client := NewTestAgent(t, `
+		server = false
+		bootstrap = false
+		retry_join = ["`+srv.Config.SerfBindAddrLAN.String()+`"]
+	`)
+
+	testrpc.WaitForTestAgent(t, client.RPC, "dc1")
+
+	mgr, server = client.Agent.router.FindLANRoute()
+	require.NotNil(t, mgr)
+	require.NotNil(t, server)
+}

agent/consul/auto_encrypt.go

@@ -73,7 +73,7 @@ func (c *Client) RequestAutoEncryptCerts(ctx context.Context, servers []string,
 	// Check if we know about a server already through gossip. Depending on
 	// how the agent joined, there might already be one. Also in case this
 	// gets called because the cert expired.
-	server := c.routers.FindServer()
+	server := c.router.FindLANServer()
 	if server != nil {
 		servers = []string{server.Addr.String()}
 	}

agent/consul/client.go

@@ -15,6 +15,7 @@ import (
 	"github.com/hashicorp/consul/lib"
 	"github.com/hashicorp/consul/logging"
 	"github.com/hashicorp/consul/tlsutil"
+	"github.com/hashicorp/consul/types"
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/serf/serf"
 	"golang.org/x/time/rate"
@@ -59,9 +60,9 @@ type Client struct {
 	// Connection pool to consul servers
 	connPool *pool.ConnPool
 
-	// routers is responsible for the selection and maintenance of
+	// router is responsible for the selection and maintenance of
 	// Consul servers this agent uses for RPC requests
-	routers *router.Manager
+	router *router.Router
 
 	// rpcLimiter is used to rate limit the total number of RPCs initiated
 	// from an agent.
@@ -120,12 +121,14 @@ func NewClient(config *Config, options ...ConsulOption) (*Client, error) {
 		}
 	}
 
+	logger := flat.logger.NamedIntercept(logging.ConsulClient)
+
 	// Create client
 	c := &Client{
 		config:          config,
 		connPool:        connPool,
 		eventCh:         make(chan serf.Event, serfEventBacklog),
-		logger:          flat.logger.NamedIntercept(logging.ConsulClient),
+		logger:          logger,
 		shutdownCh:      make(chan struct{}),
 		tlsConfigurator: tlsConfigurator,
 	}
@@ -160,15 +163,22 @@ func NewClient(config *Config, options ...ConsulOption) (*Client, error) {
 		return nil, fmt.Errorf("Failed to start lan serf: %v", err)
 	}
 
-	// Start maintenance task for servers
-	c.routers = router.New(c.logger, c.shutdownCh, c.serf, c.connPool, "")
-	go c.routers.Start()
+	rpcRouter := flat.router
+	if rpcRouter == nil {
+		rpcRouter = router.NewRouter(logger, config.Datacenter, fmt.Sprintf("%s.%s", config.NodeName, config.Datacenter))
+	}
+
+	if err := rpcRouter.AddArea(types.AreaLAN, c.serf, c.connPool); err != nil {
+		c.Shutdown()
+		return nil, fmt.Errorf("Failed to add LAN area to the RPC router: %w", err)
+	}
+	c.router = rpcRouter
 
 	// Start LAN event handlers after the router is complete since the event
 	// handlers depend on the router and the router depends on Serf.
 	go c.lanEventHandler()
 
-	// This needs to happen after initializing c.routers to prevent a race
+	// This needs to happen after initializing c.router to prevent a race
 	// condition where the router manager is used when the pointer is nil
 	if c.acls.ACLsEnabled() {
 		go c.monitorACLMode()
@@ -276,7 +286,7 @@ func (c *Client) RPC(method string, args interface{}, reply interface{}) error {
 	firstCheck := time.Now()
 
 TRY:
-	server := c.routers.FindServer()
+	manager, server := c.router.FindLANRoute()
 	if server == nil {
 		return structs.ErrNoServers
 	}
@@ -301,7 +311,7 @@ TRY:
 		"error", rpcErr,
 	)
 	metrics.IncrCounterWithLabels([]string{"client", "rpc", "failed"}, 1, []metrics.Label{{Name: "server", Value: server.Name}})
-	c.routers.NotifyFailedServer(server)
+	manager.NotifyFailedServer(server)
 	if retry := canRetry(args, rpcErr); !retry {
 		return rpcErr
 	}
@@ -323,7 +333,7 @@ TRY:
 // operation.
 func (c *Client) SnapshotRPC(args *structs.SnapshotRequest, in io.Reader, out io.Writer,
 	replyFn structs.SnapshotReplyFn) error {
-	server := c.routers.FindServer()
+	manager, server := c.router.FindLANRoute()
 	if server == nil {
 		return structs.ErrNoServers
 	}
@@ -339,6 +349,7 @@ func (c *Client) SnapshotRPC(args *structs.SnapshotRequest, in io.Reader, out io
 	var reply structs.SnapshotResponse
 	snap, err := SnapshotRPC(c.connPool, c.config.Datacenter, server.ShortName, server.Addr, args, in, &reply)
 	if err != nil {
+		manager.NotifyFailedServer(server)
 		return err
 	}
 	defer func() {
@@ -367,7 +378,7 @@ func (c *Client) SnapshotRPC(args *structs.SnapshotRequest, in io.Reader, out io
 // Stats is used to return statistics for debugging and insight
 // for various sub-systems
 func (c *Client) Stats() map[string]map[string]string {
-	numServers := c.routers.NumServers()
+	numServers := c.router.GetLANManager().NumServers()
 
 	toString := func(v uint64) string {
 		return strconv.FormatUint(v, 10)

agent/consul/client_serf.go

@@ -9,6 +9,7 @@ import (
 	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/lib"
 	"github.com/hashicorp/consul/logging"
+	"github.com/hashicorp/consul/types"
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/serf/serf"
 )
@@ -115,7 +116,7 @@ func (c *Client) nodeJoin(me serf.MemberEvent) {
 			continue
 		}
 		c.logger.Info("adding server", "server", parts)
-		c.routers.AddServer(parts)
+		c.router.AddServer(types.AreaLAN, parts)
 
 		// Trigger the callback
 		if c.config.ServerUp != nil {
@@ -139,7 +140,7 @@ func (c *Client) nodeUpdate(me serf.MemberEvent) {
 			continue
 		}
 		c.logger.Info("updating server", "server", parts.String())
-		c.routers.AddServer(parts)
+		c.router.AddServer(types.AreaLAN, parts)
 	}
 }
 
@@ -151,7 +152,7 @@ func (c *Client) nodeFail(me serf.MemberEvent) {
 			continue
 		}
 		c.logger.Info("removing server", "server", parts.String())
-		c.routers.RemoveServer(parts)
+		c.router.RemoveServer(types.AreaLAN, parts)
 	}
 }
 

agent/consul/client_test.go

@@ -115,7 +115,7 @@ func TestClient_JoinLAN(t *testing.T) {
 	joinLAN(t, c1, s1)
 	testrpc.WaitForTestAgent(t, c1.RPC, "dc1")
 	retry.Run(t, func(r *retry.R) {
-		if got, want := c1.routers.NumServers(), 1; got != want {
+		if got, want := c1.router.GetLANManager().NumServers(), 1; got != want {
 			r.Fatalf("got %d servers want %d", got, want)
 		}
 		if got, want := len(s1.LANMembers()), 2; got != want {
@@ -153,7 +153,7 @@ func TestClient_LANReap(t *testing.T) {
 
 	// Check the router has both
 	retry.Run(t, func(r *retry.R) {
-		server := c1.routers.FindServer()
+		server := c1.router.FindLANServer()
 		require.NotNil(t, server)
 		require.Equal(t, s1.config.NodeName, server.Name)
 	})
@@ -163,7 +163,7 @@ func TestClient_LANReap(t *testing.T) {
 
 	retry.Run(t, func(r *retry.R) {
 		require.Len(r, c1.LANMembers(), 1)
-		server := c1.routers.FindServer()
+		server := c1.router.FindLANServer()
 		require.Nil(t, server)
 	})
 }
@@ -393,7 +393,7 @@ func TestClient_RPC_ConsulServerPing(t *testing.T) {
 	}
 
 	// Sleep to allow Serf to sync, shuffle, and let the shuffle complete
-	c.routers.ResetRebalanceTimer()
+	c.router.GetLANManager().ResetRebalanceTimer()
 	time.Sleep(time.Second)
 
 	if len(c.LANMembers()) != numServers+numClients {
@@ -409,7 +409,7 @@ func TestClient_RPC_ConsulServerPing(t *testing.T) {
 	var pingCount int
 	for range servers {
 		time.Sleep(200 * time.Millisecond)
-		s := c.routers.FindServer()
+		m, s := c.router.FindLANRoute()
 		ok, err := c.connPool.Ping(s.Datacenter, s.ShortName, s.Addr)
 		if !ok {
 			t.Errorf("Unable to ping server %v: %s", s.String(), err)
@@ -418,7 +418,7 @@ func TestClient_RPC_ConsulServerPing(t *testing.T) {
 
 		// Artificially fail the server in order to rotate the server
 		// list
-		c.routers.NotifyFailedServer(s)
+		m.NotifyFailedServer(s)
 	}
 
 	if pingCount != numServers {
@@ -527,7 +527,7 @@ func TestClient_SnapshotRPC(t *testing.T) {
 
 	// Wait until we've got a healthy server.
 	retry.Run(t, func(r *retry.R) {
-		if got, want := c1.routers.NumServers(), 1; got != want {
+		if got, want := c1.router.GetLANManager().NumServers(), 1; got != want {
 			r.Fatalf("got %d servers want %d", got, want)
 		}
 	})
@@ -562,7 +562,7 @@ func TestClient_SnapshotRPC_RateLimit(t *testing.T) {
 
 	joinLAN(t, c1, s1)
 	retry.Run(t, func(r *retry.R) {
-		if got, want := c1.routers.NumServers(), 1; got != want {
+		if got, want := c1.router.GetLANManager().NumServers(), 1; got != want {
 			r.Fatalf("got %d servers want %d", got, want)
 		}
 	})
@@ -610,7 +610,7 @@ func TestClient_SnapshotRPC_TLS(t *testing.T) {
 		}
 
 		// Wait until we've got a healthy server.
-		if got, want := c1.routers.NumServers(), 1; got != want {
+		if got, want := c1.router.GetLANManager().NumServers(), 1; got != want {
 			r.Fatalf("got %d servers want %d", got, want)
 		}
 	})

agent/consul/coordinate_endpoint.go

@@ -10,6 +10,7 @@ import (
 	"github.com/hashicorp/consul/agent/consul/state"
 	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/logging"
+	"github.com/hashicorp/consul/types"
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-memdb"
 )
@@ -161,23 +162,32 @@ func (c *Coordinate) Update(args *structs.CoordinateUpdateRequest, reply *struct
 
 // ListDatacenters returns the list of datacenters and their respective nodes
 // and the raw coordinates of those nodes (if no coordinates are available for
-// any of the nodes, the node list may be empty).
+// any of the nodes, the node list may be empty). This endpoint will not return
+// information about the LAN network area.
 func (c *Coordinate) ListDatacenters(args *struct{}, reply *[]structs.DatacenterMap) error {
 	maps, err := c.srv.router.GetDatacenterMaps()
 	if err != nil {
 		return err
 	}
 
+	var out []structs.DatacenterMap
+
 	// Strip the datacenter suffixes from all the node names.
-	for i := range maps {
-		suffix := fmt.Sprintf(".%s", maps[i].Datacenter)
-		for j := range maps[i].Coordinates {
-			node := maps[i].Coordinates[j].Node
-			maps[i].Coordinates[j].Node = strings.TrimSuffix(node, suffix)
+	for _, dcMap := range maps {
+		if dcMap.AreaID == types.AreaLAN {
+			continue
+		}
+
+		suffix := fmt.Sprintf(".%s", dcMap.Datacenter)
+		for j := range dcMap.Coordinates {
+			node := dcMap.Coordinates[j].Node
+			dcMap.Coordinates[j].Node = strings.TrimSuffix(node, suffix)
 		}
+
+		out = append(out, dcMap)
 	}
 
-	*reply = maps
+	*reply = out
 	return nil
 }
 

agent/consul/options.go

@@ -2,6 +2,7 @@ package consul
 
 import (
 	"github.com/hashicorp/consul/agent/pool"
+	"github.com/hashicorp/consul/agent/router"
 	"github.com/hashicorp/consul/agent/token"
 	"github.com/hashicorp/consul/tlsutil"
 	"github.com/hashicorp/go-hclog"
@@ -12,6 +13,7 @@ type consulOptions struct {
 	tlsConfigurator *tlsutil.Configurator
 	connPool        *pool.ConnPool
 	tokens          *token.Store
+	router          *router.Router
 }
 
 type ConsulOption func(*consulOptions)
@@ -40,6 +42,12 @@ func WithTokenStore(tokens *token.Store) ConsulOption {
 	}
 }
 
+func WithRouter(router *router.Router) ConsulOption {
+	return func(opt *consulOptions) {
+		opt.router = router
+	}
+}
+
 func flattenConsulOptions(options []ConsulOption) consulOptions {
 	var flat consulOptions
 	for _, opt := range options {