Merge pull request #1993 from tgruben/confirm-fail

False Positive nodeLeave events put cluster in an unusable state (Starting)

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:

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")
+	}
+
+}

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))"})