From 1460756b3f047cb82b0c19283bf20e5bbb6b9306 Mon Sep 17 00:00:00 2001 From: Seebs Date: Fri, 5 Jun 2020 12:24:07 -0500 Subject: [PATCH] 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 | 25 +++++++++++++++---------- cluster_internal_test.go | 21 ++++++++++++++++----- server.go | 18 ++++++++++++++++++ test/pilosa.go | 14 +++++++------- 4 files changed, 56 insertions(+), 22 deletions(-) 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...) }