Backport #9487 to release/1.7.x (#9492)

This can happen when one other node in the cluster such as a client is unable to communicate with the leader server and sees it as failed. When that happens its failing status eventually gets propagated to the other servers in the cluster and eventually this can result in RPCs returning “No cluster leader” error.

That error is misleading and unhelpful for determing the root cause of the issue as its not raft stability but rather and client -> server networking issue. Therefore this commit will add a new error that will be returned in that case to differentiate between the two cases.

agent/consul/cluster_test.go

@@ -0,0 +1,100 @@
+package consul
+
+import (
+	"net/rpc"
+	"os"
+	"testing"
+
+	"github.com/hashicorp/consul/testrpc"
+	"github.com/stretchr/testify/require"
+)
+
+type testClusterConfig struct {
+	Datacenter string
+	Servers    int
+	Clients    int
+	ServerConf func(*Config)
+	ClientConf func(*Config)
+
+	ServerWait func(*testing.T, *Server)
+	ClientWait func(*testing.T, *Client)
+}
+
+type testCluster struct {
+	Servers      []*Server
+	ServerCodecs []rpc.ClientCodec
+	Clients      []*Client
+}
+
+func newTestCluster(t *testing.T, conf *testClusterConfig) *testCluster {
+	t.Helper()
+
+	require.NotNil(t, conf)
+	cluster := testCluster{}
+
+	// create the servers
+	for i := 0; i < conf.Servers; i++ {
+		dir, srv := testServerWithConfig(t, func(c *Config) {
+			if conf.Datacenter != "" {
+				c.Datacenter = conf.Datacenter
+			}
+			c.Bootstrap = false
+			c.BootstrapExpect = conf.Servers
+
+			if conf.ServerConf != nil {
+				conf.ServerConf(c)
+			}
+		})
+		t.Cleanup(func() { os.RemoveAll(dir) })
+		t.Cleanup(func() { srv.Shutdown() })
+
+		cluster.Servers = append(cluster.Servers, srv)
+
+		codec := rpcClient(t, srv)
+
+		cluster.ServerCodecs = append(cluster.ServerCodecs, codec)
+		t.Cleanup(func() { codec.Close() })
+
+		if i > 0 {
+			joinLAN(t, srv, cluster.Servers[0])
+		}
+	}
+
+	waitForLeaderEstablishment(t, cluster.Servers...)
+	if conf.ServerWait != nil {
+		for _, srv := range cluster.Servers {
+			conf.ServerWait(t, srv)
+		}
+	}
+
+	// create the clients
+	for i := 0; i < conf.Clients; i++ {
+		dir, client := testClientWithConfig(t, func(c *Config) {
+			if conf.Datacenter != "" {
+				c.Datacenter = conf.Datacenter
+			}
+			if conf.ClientConf != nil {
+				conf.ClientConf(c)
+			}
+		})
+
+		t.Cleanup(func() { os.RemoveAll(dir) })
+		t.Cleanup(func() { client.Shutdown() })
+
+		if len(cluster.Servers) > 0 {
+			joinLAN(t, client, cluster.Servers[0])
+		}
+
+		cluster.Clients = append(cluster.Clients, client)
+	}
+
+	for _, client := range cluster.Clients {
+		if conf.ClientWait != nil {
+			conf.ClientWait(t, client)
+		} else {
+			testrpc.WaitForTestAgent(t, client.RPC, client.config.Datacenter)
+		}
+	}
+
+	return &cluster
+}

agent/consul/rpc.go

@@ -342,15 +342,14 @@ CHECK_LEADER:
 	}
 
 	// Find the leader
-	isLeader, leader := s.getLeader()
+	isLeader, leader, rpcErr := s.getLeader()
 
 	// Handle the case we are the leader
 	if isLeader {
 		return false, nil
 	}
 
 	// Handle the case of a known leader
-	rpcErr := structs.ErrNoLeader
 	if leader != nil {
 		rpcErr = s.connPool.RPC(s.config.Datacenter, leader.Addr,
 			leader.Version, method, args, reply)
@@ -381,24 +380,38 @@ RETRY:
 
 // getLeader returns if the current node is the leader, and if not then it
 // returns the leader which is potentially nil if the cluster has not yet
-// elected a leader.
-func (s *Server) getLeader() (bool, *metadata.Server) {
+// elected a leader. In the case of not having a leader elected yet
+// then a NoClusterLeader error gets returned. In the case of Raft having
+// a leader but out internal tracking failing to find the leader we
+// return a LeaderNotTracked error. Therefore if the err is nil AND
+// the bool is false then the Server will be non-nil
+func (s *Server) getLeader() (bool, *metadata.Server, error) {
 	// Check if we are the leader
 	if s.IsLeader() {
-		return true, nil
+		return true, nil, nil
 	}
 
 	// Get the leader
 	leader := s.raft.Leader()
 	if leader == "" {
-		return false, nil
+		return false, nil, structs.ErrNoLeader
 	}
 
 	// Lookup the server
 	server := s.serverLookup.Server(leader)
 
-	// Server could be nil
-	return false, server
+	// if server is nil this indicates that while we have a Raft leader
+	// something has caused that node to be considered unhealthy which
+	// cascades into its removal from the serverLookup struct. In this case
+	// we should not report no cluster leader but instead report a different
+	// error so as not to confuse our users as to the what the root cause of
+	// an issue might be.
+	if server == nil {
+		s.logger.Warn("Raft has a leader but other tracking of the node would indicate that the node is unhealthy or does not exist. The network may be misconfigured.", "leader", leader)
+		return false, nil, structs.ErrLeaderNotTracked
+	}
+
+	return false, server, nil
 }
 
 // forwardDC is used to forward an RPC call to a remote DC, or fail if no servers

agent/consul/rpc_test.go

@@ -112,7 +112,7 @@ func TestRPC_NoLeader_Retry(t *testing.T) {
 
 	// This isn't sure-fire but tries to check that we don't have a
 	// leader going into the RPC, so we exercise the retry logic.
-	if ok, _ := s1.getLeader(); ok {
+	if ok, _, _ := s1.getLeader(); ok {
 		t.Fatalf("should not have a leader yet")
 	}
 
@@ -124,6 +124,53 @@ func TestRPC_NoLeader_Retry(t *testing.T) {
 	}
 }
 
+func TestRPC_getLeader_ErrLeaderNotTracked(t *testing.T) {
+	if testing.Short() {
+		t.Skip("too slow for testing.Short")
+	}
+
+	cluster := newTestCluster(t, &testClusterConfig{
+		Datacenter: "dc1",
+		Servers:    3,
+		ServerWait: func(t *testing.T, srv *Server) {
+			// The test cluster waits for a leader to be established
+			// but not for all the RPC tracking of all servers to be updated
+			// so we also want to wait for that here
+			retry.Run(t, func(r *retry.R) {
+				if !srv.IsLeader() {
+					_, _, err := srv.getLeader()
+					require.NoError(r, err)
+				}
+			})
+
+		},
+	})
+
+	// At this point we know we have a cluster with a leader and all followers are tracking that
+	// leader in the serverLookup struct. We need to find a follower to hack its server lookup
+	// to force the error we desire
+
+	var follower *Server
+	for _, srv := range cluster.Servers {
+		if !srv.IsLeader() {
+			follower = srv
+			break
+		}
+	}
+
+	_, leaderMeta, err := follower.getLeader()
+	require.NoError(t, err)
+
+	// now do some behind the scenes trickery on the followers server lookup
+	// to remove the leader from it so that we can force a ErrLeaderNotTracked error
+	follower.serverLookup.RemoveServer(leaderMeta)
+
+	isLeader, meta, err := follower.getLeader()
+	require.Equal(t, structs.ErrLeaderNotTracked, err)
+	require.Nil(t, meta)
+	require.False(t, isLeader)
+}
+
 type MockSink struct {
 	*bytes.Buffer
 	cancel bool

agent/consul/snapshot_endpoint.go

@@ -48,9 +48,9 @@ func (s *Server) dispatchSnapshotRequest(args *structs.SnapshotRequest, in io.Re
 
 	// Perform leader forwarding if required.
 	if !args.AllowStale {
-		if isLeader, server := s.getLeader(); !isLeader {
-			if server == nil {
-				return nil, structs.ErrNoLeader
+		if isLeader, server, err := s.getLeader(); !isLeader {
+			if err != nil {
+				return nil, err
 			}
 			return SnapshotRPC(s.connPool, args.Datacenter, server.Addr, server.UseTLS, args, in, reply)
 		}

agent/structs/errors.go

@@ -14,6 +14,7 @@ const (
 	errSegmentsNotSupported       = "Network segments are not supported in this version of Consul"
 	errRPCRateExceeded            = "RPC rate limit exceeded"
 	errServiceNotFound            = "Service not found: "
+	errLeaderNotTracked           = "Raft leader not found in server lookup mapping"
 )
 
 var (
@@ -24,6 +25,7 @@ var (
 	ErrSegmentsNotSupported       = errors.New(errSegmentsNotSupported)
 	ErrRPCRateExceeded            = errors.New(errRPCRateExceeded)
 	ErrDCNotAvailable             = errors.New(errDCNotAvailable)
+	ErrLeaderNotTracked           = errors.New(errLeaderNotTracked)
 )
 
 func IsErrNoLeader(err error) bool {