Provide option for adjusting node timeouts, set it for tests.

There's no reason to have 10-20 seconds of delays for testing this,
because in testing, we're running things on the local machine and don't
need to worry about significant network lag. Make retry count and delay
settable options, and set them lower. Moves the Replica2 test in
server/server_test.go from ~21s to ~2s.

cluster.go

@@ -62,9 +62,8 @@ const (
 	resizeJobActionAdd    = "ADD"
 	resizeJobActionRemove = "REMOVE"
 
-	confirmDownRetries = 10
-	confirmDownSleep   = 1
-	confirmDownTimeout = 2
+	defaultConfirmDownRetries = 10
+	defaultConfirmDownSleep   = 1 * time.Second
 )
 
 // Node represents a node in the cluster.
@@ -239,6 +238,9 @@ type cluster struct { // nolint: maligned
 	logger logger.Logger
 
 	InternalClient InternalClient
+
+	confirmDownRetries int
+	confirmDownSleep   time.Duration
 }
 
 // newCluster returns a new instance of Cluster with defaults.
@@ -258,6 +260,9 @@ func newCluster() *cluster {
 		InternalClient: newNopInternalClient(),
 
 		logger: logger.NopLogger,
+
+		confirmDownRetries: defaultConfirmDownRetries,
+		confirmDownSleep:   defaultConfirmDownSleep,
 	}
 }
 
@@ -1923,7 +1928,7 @@ func (c *cluster) considerTopology() error {
 // band aid to protect against false nodeLeave events from memberlist
 // the test is the lightest weight endpoint of the node in question /version
 // TODO provide more robust solution to false nodeLeave events
-func confirmNodeDown(uri URI, log logger.Logger) bool {
+func (c *cluster) confirmNodeDown(uri URI) bool {
 	u := url.URL{
 		Scheme: uri.Scheme,
 		Host:   uri.HostPort(),
@@ -1931,11 +1936,11 @@ func confirmNodeDown(uri URI, log logger.Logger) bool {
 	}
 	req, err := http.NewRequest("GET", u.String(), nil)
 	if err != nil {
-		log.Printf("bad request:%s %s", u.String(), err)
+		c.logger.Printf("bad request:%s %s", u.String(), err)
 		return false
 	}
-	for i := 0; i < confirmDownRetries; i++ {
-		ctx, cancel := context.WithTimeout(context.Background(), confirmDownTimeout*time.Second)
+	for i := 0; i < c.confirmDownRetries; i++ {
+		ctx, cancel := context.WithTimeout(context.Background(), c.confirmDownSleep*2)
 		defer cancel()
 		resp, err := http.DefaultClient.Do(req.WithContext(ctx))
 		var bod []byte
@@ -1946,8 +1951,8 @@ func confirmNodeDown(uri URI, log logger.Logger) bool {
 			}
 		}
 
-		log.Printf("NodeLeave confirm with %s %d. err: '%v' bod: '%s'", uri.HostPort(), i, err, bod)
-		time.Sleep(confirmDownSleep * time.Second)
+		c.logger.Printf("NodeLeave confirm with %s %d. err: '%v' bod: '%s'", uri.HostPort(), i, err, bod)
+		time.Sleep(c.confirmDownSleep)
 	}
 	return true
 }
@@ -1975,7 +1980,7 @@ func (c *cluster) ReceiveEvent(e *NodeEvent) (err error) {
 			// not already removed by a removeNode request. We treat this as the
 			// host being temporarily unavailable, and expect it to come back
 			// up.
-			if confirmNodeDown(e.Node.URI, c.logger) {
+			if c.confirmNodeDown(e.Node.URI) {
 				if c.removeNodeBasicSorted(e.Node.ID) {
 					c.Topology.nodeStates[e.Node.ID] = nodeStateDown
 					// put the cluster into STARTING if we've lost a number of nodes

cluster_internal_test.go

@@ -943,18 +943,22 @@ func TestCluster_confirmNodeDownUp(t *testing.T) {
 		t.Error(err)
 	}
 	uri.Port = uint16(iport)
-	if confirmNodeDown(uri, logger.NewVerboseLogger(os.Stdout)) {
+	c := newCluster()
+	c.logger = logger.NewVerboseLogger(os.Stdout)
+	if c.confirmNodeDown(uri) {
 		t.Errorf("expected node to be up")
 	}
 
 }
 func TestCluster_confirmNodeDownTimeout(t *testing.T) {
+	sleep := 50 * time.Millisecond
+	retries := 5
 	if testing.Short() {
 		t.Skip()
 	}
 	r := mux.NewRouter()
 	r.HandleFunc("/version", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		time.Sleep(confirmDownSleep * time.Second * confirmDownRetries)
+		time.Sleep(sleep * time.Duration(retries))
 		fmt.Fprintln(w, "ignored")
 	}))
 	server := httptest.NewServer(r)
@@ -973,8 +977,11 @@ func TestCluster_confirmNodeDownTimeout(t *testing.T) {
 		t.Error(err)
 	}
 	uri.Port = uint16(iport)
-
-	if !confirmNodeDown(uri, logger.NewVerboseLogger(os.Stdout)) {
+	c := newCluster()
+	c.confirmDownSleep = sleep
+	c.confirmDownRetries = retries
+	c.logger = logger.NewVerboseLogger(os.Stdout)
+	if !c.confirmNodeDown(uri) {
 		t.Errorf("expected node to be down")
 	}
 }
@@ -987,8 +994,12 @@ func TestCluster_confirmNodeDownDown(t *testing.T) {
 	uri.Scheme = "http"
 	uri.Host = "DoesntMatter"
 	uri.Port = 6666
+	c := newCluster()
+	c.confirmDownSleep = 50 * time.Millisecond
+	c.confirmDownRetries = 5
+	c.logger = logger.NewVerboseLogger(os.Stdout)
 
-	if !confirmNodeDown(uri, logger.NewVerboseLogger(os.Stdout)) {
+	if !c.confirmNodeDown(uri) {
 		t.Errorf("expected node to be down")
 	}
 }

server.go

@@ -77,6 +77,8 @@ type Server struct { // nolint: maligned
 	metricInterval      time.Duration
 	diagnosticInterval  time.Duration
 	maxWritesPerRequest int
+	confirmDownSleep    time.Duration
+	confirmDownRetries  int
 	isCoordinator       bool
 	syncer              holderSyncer
 
@@ -229,6 +231,17 @@ func OptServerDiagnosticsInterval(dur time.Duration) ServerOption {
 	}
 }
 
+// OptServerNodeDownRetries is a functional option on Server
+// used to specify the retries and sleep duration for node down
+// checks.
+func OptServerNodeDownRetries(retries int, sleep time.Duration) ServerOption {
+	return func(s *Server) error {
+		s.confirmDownRetries = retries
+		s.confirmDownSleep = sleep
+		return nil
+	}
+}
+
 // OptServerURI is a functional option on Server
 // used to set the server URI.
 func OptServerURI(uri *URI) ServerOption {
@@ -330,6 +343,9 @@ func NewServer(opts ...ServerOption) (*Server, error) {
 		metricInterval:      0,
 		diagnosticInterval:  0,
 
+		confirmDownRetries: defaultConfirmDownRetries,
+		confirmDownSleep:   defaultConfirmDownSleep,
+
 		resetTranslationSyncCh: make(chan struct{}),
 
 		logger: logger.NopLogger,
@@ -399,6 +415,8 @@ func NewServer(opts ...ServerOption) (*Server, error) {
 	s.executor.MaxWritesPerRequest = s.maxWritesPerRequest
 	s.cluster.broadcaster = s
 	s.cluster.maxWritesPerRequest = s.maxWritesPerRequest
+	s.cluster.confirmDownRetries = s.confirmDownRetries
+	s.cluster.confirmDownSleep = s.confirmDownSleep
 	s.holder.broadcaster = s
 	err = s.loadAllExtensions()
 	if err != nil {

test/pilosa.go

@@ -92,7 +92,7 @@ func NewCommandNode(isCoordinator bool, opts ...server.CommandOption) *Command {
 	// We want tests to default to using the in-memory translate store, so we
 	// prepend opts with that functional option. If a different translate store
 	// has been specified, it will override this one.
-	opts = prependWithMemStore(opts)
+	opts = prependTestServerOpts(opts)
 	m := newCommand(opts...)
 	m.Config.Cluster.Disabled = false
 	m.Config.Cluster.Coordinator = isCoordinator
@@ -434,25 +434,25 @@ func MustRunCluster(tb testing.TB, size int, opts ...[]server.CommandOption) Clu
 	return c
 }
 
-// prependOpts applies prependWithMemStore to each of the ops (one per
+// prependOpts applies prependTestServerOpts to each of the ops (one per
 // node, or one for the entire cluser).
 func prependOpts(opts [][]server.CommandOption) [][]server.CommandOption {
 	if len(opts) == 0 {
 		opts = [][]server.CommandOption{
-			prependWithMemStore([]server.CommandOption{}),
+			prependTestServerOpts([]server.CommandOption{}),
 		}
 	} else {
 		for i := range opts {
-			opts[i] = prependWithMemStore(opts[i])
+			opts[i] = prependTestServerOpts(opts[i])
 		}
 	}
 	return opts
 }
 
-// prependWithMemStore prepends opts with the OpenInMemTranslateStore.
-func prependWithMemStore(opts []server.CommandOption) []server.CommandOption {
+// prependTestServerOpts prepends opts with the OpenInMemTranslateStore.
+func prependTestServerOpts(opts []server.CommandOption) []server.CommandOption {
 	defaultOpts := []server.CommandOption{
-		server.OptCommandServerOptions(pilosa.OptServerOpenTranslateStore(pilosa.OpenInMemTranslateStore)),
+		server.OptCommandServerOptions(pilosa.OptServerOpenTranslateStore(pilosa.OpenInMemTranslateStore), pilosa.OptServerNodeDownRetries(5, 100*time.Millisecond)),
 	}
 	return append(defaultOpts, opts...)
 }