diff --git a/cluster.go b/cluster.go index 674a0a267..cf0090c0f 100644 --- a/cluster.go +++ b/cluster.go @@ -21,6 +21,8 @@ import ( "hash/fnv" "io/ioutil" "math/rand" + "net/http" + "net/url" "os" "path/filepath" "sort" @@ -59,6 +61,10 @@ const ( resizeJobActionAdd = "ADD" resizeJobActionRemove = "REMOVE" + + confirmDownRetries = 10 + confirmDownSleep = 1 + confirmDownTimeout = 2 ) // Node represents a node in the cluster. @@ -1687,13 +1693,42 @@ func (c *cluster) considerTopology() error { return nil } +// 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 { + u := url.URL{ + Scheme: uri.Scheme, + Host: uri.HostPort(), + Path: "version", + } + ctx, cancel := context.WithTimeout(context.Background(), confirmDownTimeout*time.Second) + defer cancel() + req, err := http.NewRequest("GET", u.String(), nil) + if err != nil { + log.Printf("bad request:%s %s", u.String(), err) + return false + } + + for i := 0; i < confirmDownRetries; i++ { + resp, err := http.DefaultClient.Do(req.WithContext(ctx)) + if err == nil { + if resp.StatusCode == 200 { + return false + } + } + log.Printf("NodeLeave Timeout with %s %d", uri.HostPort(), i) + time.Sleep(confirmDownSleep * time.Second) + } + return true +} + // ReceiveEvent represents an implementation of EventHandler. func (c *cluster) ReceiveEvent(e *NodeEvent) (err error) { // Ignore events sent from this node. if e.Node.ID == c.Node.ID { return nil } - switch e.Event { case NodeJoin: c.logger.Debugf("nodeJoin of %s on %s", e.Node.URI, c.Node.URI) @@ -1711,11 +1746,15 @@ 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 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 - // equal to or greater than ReplicaN - err = c.unprotectedSetStateAndBroadcast(c.determineClusterState()) + if confirmNodeDown(e.Node.URI, c.logger) { + 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 + // equal to or greater than ReplicaN + err = c.unprotectedSetStateAndBroadcast(c.determineClusterState()) + } + } else { + c.logger.Printf("ignored received node leave: %v", e.Node) } } case NodeUpdate: diff --git a/cluster_internal_test.go b/cluster_internal_test.go index 2a557694d..a30d98271 100644 --- a/cluster_internal_test.go +++ b/cluster_internal_test.go @@ -16,15 +16,24 @@ package pilosa import ( "bytes" + "fmt" "io/ioutil" "math/rand" + "net" + "net/http" + "net/http/httptest" + "net/url" + "os" "reflect" + "strconv" "strings" "testing" "testing/quick" "time" "github.com/davecgh/go-spew/spew" + "github.com/gorilla/mux" + "github.com/pilosa/pilosa/logger" "github.com/pilosa/pilosa/roaring" "github.com/pkg/errors" ) @@ -910,3 +919,71 @@ func TestCluster_UpdateCoordinator(t *testing.T) { } }) } + +func TestCluster_confirmNodeDownUp(t *testing.T) { + r := mux.NewRouter() + r.HandleFunc("/version", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + fmt.Fprintln(w, "ignored") + })) + server := httptest.NewServer(r) + // Close the server when test finishes + defer server.Close() + u, err := url.Parse(server.URL) + if err != nil { + t.Error("bad test setup") + } + uri := URI{} + host, port, _ := net.SplitHostPort(u.Host) + uri.Scheme = u.Scheme + uri.Host = host + iport, err := strconv.ParseUint(port, 0, 16) + if err != nil { + t.Error(err) + } + uri.Port = uint16(iport) + if confirmNodeDown(uri, logger.NewVerboseLogger(os.Stdout)) { + t.Errorf("expected node to be up") + } + +} +func TestCluster_confirmNodeDownTimeout(t *testing.T) { + r := mux.NewRouter() + r.HandleFunc("/version", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + time.Sleep(confirmDownSleep * time.Second * confirmDownRetries) + fmt.Fprintln(w, "ignored") + })) + server := httptest.NewServer(r) + // Close the server when test finishes + defer server.Close() + u, err := url.Parse(server.URL) + if err != nil { + t.Error("bad test setup") + } + uri := URI{} + host, port, _ := net.SplitHostPort(u.Host) + uri.Scheme = u.Scheme + uri.Host = host + iport, err := strconv.ParseUint(port, 0, 16) + if err != nil { + t.Error(err) + } + uri.Port = uint16(iport) + + if !confirmNodeDown(uri, logger.NewVerboseLogger(os.Stdout)) { + t.Errorf("expected node to be down") + } + +} + +func TestCluster_confirmNodeDownDown(t *testing.T) { + uri := URI{} + uri.Scheme = "http" + uri.Host = "DoesntMatter" + uri.Port = 6666 + + if !confirmNodeDown(uri, logger.NewVerboseLogger(os.Stdout)) { + t.Errorf("expected node to be down") + } + +} diff --git a/internal/clustertests/cluster_test.go b/internal/clustertests/cluster_test.go index 612189c1b..5c8f4ed6b 100644 --- a/internal/clustertests/cluster_test.go +++ b/internal/clustertests/cluster_test.go @@ -80,7 +80,7 @@ func TestClusterStuff(t *testing.T) { // TODO change the sleep to wait for status to return to NORMAL - need support in internal client for getting status t.Log("done with pause, waiting for stability") - time.Sleep(time.Second * 3) + time.Sleep(time.Second * 20) t.Log("done waiting for stability") r, err = cli.Query(context.Background(), "testidx", &pilosa.QueryRequest{Index: "testidx", Query: "Count(Row(testf=0))"})