diff --git a/cluster.go b/cluster.go index b3fc86e88..eacad25f7 100644 --- a/cluster.go +++ b/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 diff --git a/cluster_internal_test.go b/cluster_internal_test.go index de87ad5ae..b9e714326 100644 --- a/cluster_internal_test.go +++ b/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") } } diff --git a/server.go b/server.go index 9fbb30811..131a6971c 100644 --- a/server.go +++ b/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 { diff --git a/test/pilosa.go b/test/pilosa.go index 635e0f3d9..9b5f49a48 100644 --- a/test/pilosa.go +++ b/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...) }