Improved Client handling of failed RPCs

client/client.go

@@ -137,9 +137,11 @@ type Client struct {
 	// server for the node event
 	triggerEmitNodeEvent chan *structs.NodeEvent
 
-	// discovered will be ticked whenever Consul discovery completes
-	// successfully
-	serversDiscoveredCh chan struct{}
+	// rpcRetryCh is closed when there an event such as server discovery or a
+	// successful RPC occurring happens such that a retry should happen. Access
+	// should only occur via the getter method
+	rpcRetryCh   chan struct{}
+	rpcRetryLock sync.Mutex
 
 	// allocs maps alloc IDs to their AllocRunner. This map includes all
 	// AllocRunners - running and GC'd - until the server GCs them.
@@ -217,7 +219,6 @@ func NewClient(cfg *config.Config, consulCatalog consul.CatalogAPI, consulServic
 		shutdownCh:           make(chan struct{}),
 		triggerDiscoveryCh:   make(chan struct{}),
 		triggerNodeUpdate:    make(chan struct{}, 8),
-		serversDiscoveredCh:  make(chan struct{}),
 		triggerEmitNodeEvent: make(chan *structs.NodeEvent, 8),
 	}
 
@@ -1154,7 +1155,7 @@ func (c *Client) registerAndHeartbeat() {
 
 	for {
 		select {
-		case <-c.serversDiscoveredCh:
+		case <-c.rpcRetryWatcher():
 		case <-heartbeat:
 		case <-c.shutdownCh:
 			return
@@ -1307,7 +1308,7 @@ func (c *Client) retryRegisterNode() {
 			c.logger.Printf("[ERR] client: registration failure: %v", err)
 		}
 		select {
-		case <-c.serversDiscoveredCh:
+		case <-c.rpcRetryWatcher():
 		case <-time.After(c.retryIntv(registerRetryIntv)):
 		case <-c.shutdownCh:
 			return
@@ -1567,7 +1568,7 @@ OUTER:
 			}
 			retry := c.retryIntv(getAllocRetryIntv)
 			select {
-			case <-c.serversDiscoveredCh:
+			case <-c.rpcRetryWatcher():
 				continue
 			case <-time.After(retry):
 				continue
@@ -1622,7 +1623,7 @@ OUTER:
 				c.logger.Printf("[ERR] client: failed to query updated allocations: %v", err)
 				retry := c.retryIntv(getAllocRetryIntv)
 				select {
-				case <-c.serversDiscoveredCh:
+				case <-c.rpcRetryWatcher():
 					continue
 				case <-time.After(retry):
 					continue
@@ -2090,13 +2091,8 @@ DISCOLOOP:
 	// Notify waiting rpc calls. If a goroutine just failed an RPC call and
 	// isn't receiving on this chan yet they'll still retry eventually.
 	// This is a shortcircuit for the longer retry intervals.
-	for {
-		select {
-		case c.serversDiscoveredCh <- struct{}{}:
-		default:
-			return nil
-		}
-	}
+	c.fireRpcRetryWatcher()
+	return nil
 }
 
 // emitStats collects host resource usage stats periodically

client/client_test.go

@@ -89,6 +89,34 @@ func TestClient_RPC(t *testing.T) {
 	})
 }
 
+func TestClient_RPC_FireRetryWatchers(t *testing.T) {
+	t.Parallel()
+	s1, addr := testServer(t, nil)
+	defer s1.Shutdown()
+
+	c1 := TestClient(t, func(c *config.Config) {
+		c.Servers = []string{addr}
+	})
+	defer c1.Shutdown()
+
+	watcher := c1.rpcRetryWatcher()
+
+	// RPC should succeed
+	testutil.WaitForResult(func() (bool, error) {
+		var out struct{}
+		err := c1.RPC("Status.Ping", struct{}{}, &out)
+		return err == nil, err
+	}, func(err error) {
+		t.Fatalf("err: %v", err)
+	})
+
+	select {
+	case <-watcher:
+	default:
+		t.Fatal("watcher should be fired")
+	}
+}
+
 func TestClient_RPC_Passthrough(t *testing.T) {
 	t.Parallel()
 	s1, _ := testServer(t, nil)

client/rpc.go

@@ -67,6 +67,7 @@ TRY:
 	// Make the request.
 	rpcErr := c.connPool.RPC(c.Region(), server.Addr, c.RPCMajorVersion(), method, args, reply)
 	if rpcErr == nil {
+		c.fireRpcRetryWatcher()
 		return nil
 	}
 
@@ -382,3 +383,27 @@ func (c *Client) Ping(srv net.Addr) error {
 	err := c.connPool.RPC(c.Region(), srv, c.RPCMajorVersion(), "Status.Ping", struct{}{}, &reply)
 	return err
 }
+
+// rpcRetryWatcher returns a channel that will be closed if an event happens
+// such that we expect the next RPC to be successful.
+func (c *Client) rpcRetryWatcher() <-chan struct{} {
+	c.rpcRetryLock.Lock()
+	defer c.rpcRetryLock.Unlock()
+
+	if c.rpcRetryCh == nil {
+		c.rpcRetryCh = make(chan struct{})
+	}
+
+	return c.rpcRetryCh
+}
+
+// fireRpcRetryWatcher causes any RPC retryloops to retry their RPCs because we
+// believe the will be successful.
+func (c *Client) fireRpcRetryWatcher() {
+	c.rpcRetryLock.Lock()
+	defer c.rpcRetryLock.Unlock()
+	if c.rpcRetryCh != nil {
+		close(c.rpcRetryCh)
+		c.rpcRetryCh = nil
+	}
+}

client/servers/manager.go

@@ -74,6 +74,16 @@ func (s *Server) String() string {
 	return s.addr
 }
 
+func (s *Server) Equal(o *Server) bool {
+	if s == nil && o == nil {
+		return true
+	} else if s == nil && o != nil || s != nil && o == nil {
+		return false
+	}
+
+	return s.Addr.String() == o.Addr.String() && s.DC == o.DC
+}
+
 type Servers []*Server
 
 func (s Servers) String() string {
@@ -160,6 +170,9 @@ func (m *Manager) Start() {
 func (m *Manager) SetServers(servers Servers) {
 	m.Lock()
 	defer m.Unlock()
+
+	// Randomize the incoming servers
+	servers.shuffle()
 	m.servers = servers
 }
 
@@ -204,7 +217,7 @@ func (m *Manager) NotifyFailedServer(s *Server) {
 	// If the server being failed is not the first server on the list,
 	// this is a noop.  If, however, the server is failed and first on
 	// the list, move the server to the end of the list.
-	if len(m.servers) > 1 && m.servers[0] == s {
+	if len(m.servers) > 1 && m.servers[0].Equal(s) {
 		m.servers.cycle()
 	}
 }

client/servers/manager_test.go

@@ -10,6 +10,7 @@ import (
 	"testing"
 
 	"github.com/hashicorp/nomad/client/servers"
+	"github.com/hashicorp/nomad/helper/testlog"
 )
 
 type fauxAddr struct {
@@ -32,22 +33,22 @@ func (cp *fauxConnPool) Ping(net.Addr) error {
 	return fmt.Errorf("bad server")
 }
 
-func testManager() (m *servers.Manager) {
-	logger := log.New(os.Stderr, "", log.LstdFlags)
+func testManager(t *testing.T) (m *servers.Manager) {
+	logger := testlog.Logger(t)
 	shutdownCh := make(chan struct{})
 	m = servers.New(logger, shutdownCh, &fauxConnPool{})
 	return m
 }
 
-func testManagerFailProb(failPct float64) (m *servers.Manager) {
-	logger := log.New(os.Stderr, "", log.LstdFlags)
+func testManagerFailProb(t *testing.T, failPct float64) (m *servers.Manager) {
+	logger := testlog.Logger(t)
 	shutdownCh := make(chan struct{})
 	m = servers.New(logger, shutdownCh, &fauxConnPool{failPct: failPct})
 	return m
 }
 
 func TestServers_SetServers(t *testing.T) {
-	m := testManager()
+	m := testManager(t)
 	var num int
 	num = m.NumServers()
 	if num != 0 {
@@ -66,14 +67,10 @@ func TestServers_SetServers(t *testing.T) {
 	if l := len(all); l != 2 {
 		t.Fatalf("expected 2 servers got %d", l)
 	}
-
-	if all[0] == s1 || all[0] == s2 {
-		t.Fatalf("expected a copy, got actual server")
-	}
 }
 
 func TestServers_FindServer(t *testing.T) {
-	m := testManager()
+	m := testManager(t)
 
 	if m.FindServer() != nil {
 		t.Fatalf("Expected nil return")
@@ -105,20 +102,14 @@ func TestServers_FindServer(t *testing.T) {
 		t.Fatalf("Expected two servers")
 	}
 	s1 = m.FindServer()
-	if s1 == nil || s1.String() != "s1" {
-		t.Fatalf("Expected s1 server (still)")
-	}
 
-	m.NotifyFailedServer(s1)
-	s2 := m.FindServer()
-	if s2 == nil || s2.String() != "s2" {
-		t.Fatalf("Expected s2 server")
+	for _, srv := range srvs {
+		m.NotifyFailedServer(srv)
 	}
 
-	m.NotifyFailedServer(s2)
-	s1 = m.FindServer()
-	if s1 == nil || s1.String() != "s1" {
-		t.Fatalf("Expected s1 server")
+	s2 := m.FindServer()
+	if s1.Equal(s2) {
+		t.Fatalf("Expected different server")
 	}
 }
 
@@ -132,7 +123,7 @@ func TestServers_New(t *testing.T) {
 }
 
 func TestServers_NotifyFailedServer(t *testing.T) {
-	m := testManager()
+	m := testManager(t)
 
 	if m.NumServers() != 0 {
 		t.Fatalf("Expected zero servers to start")
@@ -159,32 +150,39 @@ func TestServers_NotifyFailedServer(t *testing.T) {
 		t.Fatalf("Expected two servers")
 	}
 
-	s1 = m.FindServer()
-	if s1 == nil || s1.String() != "s1" {
-		t.Fatalf("Expected s1 server")
+	// Grab a server
+	first := m.FindServer()
+
+	// Find the other server
+	second := s1
+	if first.Equal(s1) {
+		second = s2
 	}
 
-	m.NotifyFailedServer(s2)
-	s1 = m.FindServer()
-	if s1 == nil || s1.String() != "s1" {
-		t.Fatalf("Expected s1 server (still)")
+	// Fail the other server
+	m.NotifyFailedServer(second)
+	next := m.FindServer()
+	if !next.Equal(first) {
+		t.Fatalf("Expected first server (still)")
 	}
 
-	m.NotifyFailedServer(s1)
-	s2 = m.FindServer()
-	if s2 == nil || s2.String() != "s2" {
-		t.Fatalf("Expected s2 server")
+	// Fail the first
+	m.NotifyFailedServer(first)
+	next = m.FindServer()
+	if !next.Equal(second) {
+		t.Fatalf("Expected second server")
 	}
 
-	m.NotifyFailedServer(s2)
-	s1 = m.FindServer()
-	if s1 == nil || s1.String() != "s1" {
-		t.Fatalf("Expected s1 server")
+	// Fail the second
+	m.NotifyFailedServer(second)
+	next = m.FindServer()
+	if !next.Equal(first) {
+		t.Fatalf("Expected first server")
 	}
 }
 
 func TestServers_NumServers(t *testing.T) {
-	m := testManager()
+	m := testManager(t)
 	var num int
 	num = m.NumServers()
 	if num != 0 {
@@ -201,7 +199,7 @@ func TestServers_NumServers(t *testing.T) {
 
 func TestServers_RebalanceServers(t *testing.T) {
 	const failPct = 0.5
-	m := testManagerFailProb(failPct)
+	m := testManagerFailProb(t, failPct)
 	const maxServers = 100
 	const numShuffleTests = 100
 	const uniquePassRate = 0.5