From 085e29543efd143bf7b2af5354657c184a6d14c5 Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Wed, 5 Jun 2019 11:28:41 -0500 Subject: [PATCH 1/5] False Positive nodeLeave events put cluster in an unusable state --- cluster.go | 35 +++++++++++++++++++++++++++++------ 1 file changed, 29 insertions(+), 6 deletions(-) diff --git a/cluster.go b/cluster.go index 204a1016b..fc332df09 100644 --- a/cluster.go +++ b/cluster.go @@ -21,6 +21,7 @@ import ( "hash/fnv" "io/ioutil" "math/rand" + "net/http" "os" "path/filepath" "sort" @@ -59,6 +60,9 @@ const ( resizeJobActionAdd = "ADD" resizeJobActionRemove = "REMOVE" + + confirmDownRetries = 20 + confirmDownSleep = 1 ) // Node represents a node in the cluster. @@ -1687,13 +1691,28 @@ 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 nodeJoin events +func confirmNodeDown(uri URI) bool { + for i := 0; i < confirmDownRetries; i++ { + resp, err := http.Get(uri.Scheme + uri.HostPort() + "/version") + if err == nil { + if resp.StatusCode == 200 { + return false + } + 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 +1730,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) { + 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("received node leave: %v", e.Node) } } case NodeUpdate: From c99071d5bf31c0ca1abc029b4e6ff76600b69631 Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Wed, 5 Jun 2019 14:53:35 -0500 Subject: [PATCH 2/5] timeout handleded incorrectly;added tests --- cluster.go | 28 +++++++++++---- cluster_internal_test.go | 77 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 99 insertions(+), 6 deletions(-) diff --git a/cluster.go b/cluster.go index fc332df09..7e2ff315b 100644 --- a/cluster.go +++ b/cluster.go @@ -22,6 +22,7 @@ import ( "io/ioutil" "math/rand" "net/http" + "net/url" "os" "path/filepath" "sort" @@ -61,8 +62,9 @@ const ( resizeJobActionAdd = "ADD" resizeJobActionRemove = "REMOVE" - confirmDownRetries = 20 + confirmDownRetries = 15 confirmDownSleep = 1 + confirmDownTimeout = 2 ) // Node represents a node in the cluster. @@ -1693,16 +1695,30 @@ 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 nodeJoin events -func confirmNodeDown(uri URI) bool { +// 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.Get(uri.Scheme + uri.HostPort() + "/version") + resp, err := http.DefaultClient.Do(req.WithContext(ctx)) if err == nil { if resp.StatusCode == 200 { return false } - time.Sleep(confirmDownSleep * time.Second) } + log.Printf("NodeLeave Timeout with %s %d", uri.HostPort(), i) + time.Sleep(confirmDownSleep * time.Second) } return true } @@ -1730,7 +1746,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) { + 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 diff --git a/cluster_internal_test.go b/cluster_internal_test.go index a6cf437b2..88eaf3a6e 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" ) @@ -885,3 +894,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") + } + +} From e4dbafd03e2339bb734a8c740bd8bd210dd2a449 Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Wed, 5 Jun 2019 14:58:35 -0500 Subject: [PATCH 3/5] Duplicate log entry --- cluster.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cluster.go b/cluster.go index 4af90413c..391ddb5ca 100644 --- a/cluster.go +++ b/cluster.go @@ -1754,7 +1754,7 @@ func (c *cluster) ReceiveEvent(e *NodeEvent) (err error) { err = c.unprotectedSetStateAndBroadcast(c.determineClusterState()) } } else { - c.logger.Printf("received node leave: %v", e.Node) + c.logger.Printf("ignored received node leave: %v", e.Node) } } case NodeUpdate: From b9ab21dfd2a8978fcd3a26c6c5c7f4c2b8b76a8b Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Wed, 5 Jun 2019 15:13:53 -0500 Subject: [PATCH 4/5] Decreased the number of retries for dead node confirmation --- cluster.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cluster.go b/cluster.go index 391ddb5ca..cf0090c0f 100644 --- a/cluster.go +++ b/cluster.go @@ -62,7 +62,7 @@ const ( resizeJobActionAdd = "ADD" resizeJobActionRemove = "REMOVE" - confirmDownRetries = 15 + confirmDownRetries = 10 confirmDownSleep = 1 confirmDownTimeout = 2 ) From 0b392164cc4c92f8f8e02acdf48af4e7cd7d7223 Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Wed, 5 Jun 2019 18:48:35 -0500 Subject: [PATCH 5/5] Stabilization Time not long enough for new cluster test --- internal/clustertests/cluster_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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))"})