From 48af3adc2196135b917fc25c1f03c9a63450f80e Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Mon, 6 Aug 2018 11:57:24 -0500 Subject: [PATCH 1/5] add degraded cluster state and handle node failure cluster is in degraded state when some number of nodes greater than 0 but less than replicaN have failed. This is sort of a hybrid of "STARTING" and "NORMAL" states because we can still respond to queries as in the NORMAL state, but we need to be alert to re-add nodes to the cluster if they come back online which required some changes to the cluster logic. In order to make debugging easier, the test.MustRunCluster functionality now names the nodes in the cluster explicitly as "node0", "node1", etc. "node0" is the coordinator. A number of TODOs are left in the test for scenarios that need to be checked. --- Gopkg.lock | 28 ++++++++++---------- Gopkg.toml | 9 +++++++ api.go | 1 + cluster.go | 60 +++++++++++++++++++++++++++++++++---------- server/server_test.go | 50 ++++++++++++++++++++++++++++++++++++ test/pilosa.go | 6 +++++ 6 files changed, 126 insertions(+), 28 deletions(-) diff --git a/Gopkg.lock b/Gopkg.lock index 7c5513ee6..dc3ddc49a 100644 --- a/Gopkg.lock +++ b/Gopkg.lock @@ -23,7 +23,7 @@ branch = "master" name = "github.com/armon/go-metrics" packages = ["."] - revision = "58588f401c2cc130a7308a52ca3bc6c0a76db04b" + revision = "3c58d8115a78a6879e5df75ae900846768d36895" [[projects]] name = "github.com/boltdb/bolt" @@ -61,8 +61,8 @@ [[projects]] name = "github.com/gogo/protobuf" packages = ["proto"] - revision = "1adfc126b41513cc696b209667c8656ea7aac67c" - version = "v1.0.0" + revision = "636bf0302bc95575d69441b25a2603156ffdddf1" + version = "v1.1.1" [[projects]] name = "github.com/golang/protobuf" @@ -104,7 +104,7 @@ branch = "master" name = "github.com/hashicorp/errwrap" packages = ["."] - revision = "7554cd9344cec97297fa6649b055a8c98c2a1e55" + revision = "d6c0cd88035724dd42e0f335ae30161c20575ecc" [[projects]] branch = "master" @@ -122,7 +122,7 @@ branch = "master" name = "github.com/hashicorp/go-multierror" packages = ["."] - revision = "b7773ae218740a7be65057fc60b366a49b538a44" + revision = "3d5d8f294aa03d8e98859feac328afbdf1ae0703" [[projects]] branch = "master" @@ -154,10 +154,10 @@ revision = "ef8a98b0bbce4a65b5aa4c368430a80ddc533168" [[projects]] + branch = "master" name = "github.com/hashicorp/memberlist" packages = ["."] - revision = "ce8abaa0c60c2d6bee7219f5ddf500e0a1457b28" - version = "v0.1.0" + revision = "2288bf30e9c8d7b5f6549bf62e07120d72fd4b6c" [[projects]] name = "github.com/inconshreveable/mousetrap" @@ -181,7 +181,7 @@ branch = "master" name = "github.com/mitchellh/mapstructure" packages = ["."] - revision = "bb74f1db0675b241733089d5a1faa5dd8b0ef57b" + revision = "f15292f7a699fcc1a38a80977f80a046874ba8ac" [[projects]] name = "github.com/pelletier/go-toml" @@ -217,8 +217,8 @@ "net", "process" ] - revision = "4a180b209f5f494e5923cfce81ea30ba23915877" - version = "v2.18.06" + revision = "8048a2e9c5773235122027dd585cf821b2af1249" + version = "v2.18.07" [[projects]] branch = "master" @@ -272,7 +272,7 @@ "ed25519", "ed25519/internal/edwards25519" ] - revision = "a49355c7e3f8fe157a85be2f77e6e269a0f89602" + revision = "c126467f60eb25f8f27e5a981f32a87e3965053f" [[projects]] branch = "master" @@ -285,7 +285,7 @@ "ipv4", "ipv6" ] - revision = "6f138e0f60713a248abf7046f1014a3ba90f5341" + revision = "22bb95c5e783d192c577a7b310b06637db9f1d94" [[projects]] branch = "master" @@ -300,7 +300,7 @@ "unix", "windows" ] - revision = "1b2967e3c290b7c545b3db0deeda16e9be4f98a2" + revision = "bd9dbc187b6e1dacfdd2722a87e83093c2d7bd6e" [[projects]] name = "golang.org/x/text" @@ -324,6 +324,6 @@ [solve-meta] analyzer-name = "dep" analyzer-version = 1 - inputs-digest = "da6d02118ca77527c4ff00e9522880032fc052fb39bc8efe6c76602857c8c84e" + inputs-digest = "6ae38dc521f55a89507ec21c575e0b65505de9514d1db6255a1b9b99fcf19ab3" solver-name = "gps-cdcl" solver-version = 1 diff --git a/Gopkg.toml b/Gopkg.toml index 377ef0fdc..f5f81ad68 100644 --- a/Gopkg.toml +++ b/Gopkg.toml @@ -16,3 +16,12 @@ # Recommended: the version constraint to enforce for the project. # Only one of "branch", "version" or "revision" can be specified. branch = "master" + +[[constraint]] + name = "github.com/hashicorp/memberlist" + branch = "master" + + +[[constraint]] + name = "github.com/gorilla/handlers" + version = "=1.3.0" diff --git a/api.go b/api.go index 649a99cd2..c9242a9bd 100644 --- a/api.go +++ b/api.go @@ -71,6 +71,7 @@ func NewAPI(opts ...apiOption) (*API, error) { var validAPIMethods = map[string]map[apiMethod]struct{}{ ClusterStateStarting: methodsCommon, ClusterStateNormal: appendMap(methodsCommon, methodsNormal), + ClusterStateDegraded: appendMap(methodsCommon, methodsNormal), ClusterStateResizing: appendMap(methodsCommon, methodsResizing), } diff --git a/cluster.go b/cluster.go index fcc4425c9..12513c14e 100644 --- a/cluster.go +++ b/cluster.go @@ -41,11 +41,13 @@ const ( // ClusterState represents the state returned in the /status endpoint. ClusterStateStarting = "STARTING" + ClusterStateDegraded = "DEGRADED" // cluster is running but we've lost some # of hosts >0 but < replicaN ClusterStateNormal = "NORMAL" ClusterStateResizing = "RESIZING" // NodeState represents the state of a node during startup. nodeStateReady = "READY" + nodeStateDown = "DOWN" // resizeJob states. resizeJobStateRunning = "RUNNING" @@ -396,7 +398,7 @@ func (c *cluster) unprotectedSetState(state string) { var doCleanup bool switch state { - case ClusterStateNormal: + case ClusterStateNormal, ClusterStateDegraded: // If state is RESIZING -> NORMAL then run cleanup. if c.state == ClusterStateResizing { doCleanup = true @@ -451,11 +453,6 @@ func (c *cluster) receiveNodeState(nodeID string, state string) error { return nil } - // This method is really only useful during initial startup. - if c.state != ClusterStateStarting { - return nil - } - c.Topology.mu.Lock() c.Topology.nodeStates[nodeID] = state c.Topology.mu.Unlock() @@ -469,6 +466,20 @@ func (c *cluster) receiveNodeState(nodeID string, state string) error { return nil } +// determineClusterState is unprotected. +func (c *cluster) determineClusterState() (clusterState string) { + if c.state == ClusterStateResizing { + return ClusterStateResizing + } + if c.haveTopologyAgreement() && c.allNodesReady() { + return ClusterStateNormal + } + if len(c.nodeIDs())-len(c.Topology.nodeIDs) >= c.ReplicaN && c.allNodesReady() { + return ClusterStateDegraded + } + return ClusterStateStarting +} + func (c *cluster) status() *ClusterStatus { c.mu.RLock() defer c.mu.RUnlock() @@ -907,12 +918,12 @@ func (c *cluster) haveTopologyAgreement() bool { } // allNodesReady is unprotected. -func (c *cluster) allNodesReady() bool { +func (c *cluster) allNodesReady() (ret bool) { if c.Static { return true } - for _, uri := range c.Topology.nodeIDs { - if c.Topology.nodeStates[uri] != nodeStateReady { + for _, id := range c.nodeIDs() { + if c.Topology.nodeStates[id] != nodeStateReady { return false } } @@ -1564,7 +1575,7 @@ func (c *cluster) considerTopology() error { } // ReceiveEvent represents an implementation of EventHandler. -func (c *cluster) ReceiveEvent(e *NodeEvent) error { +func (c *cluster) ReceiveEvent(e *NodeEvent) (err error) { // Ignore events sent from this node. if e.Node.ID == c.Node.ID { return nil @@ -1579,14 +1590,31 @@ func (c *cluster) ReceiveEvent(e *NodeEvent) error { } return c.nodeJoin(e.Node) case NodeLeave: - // Automatic nodeLeave is intentionally not implemented. + c.logger.Printf("received node leave on %s: %s, uri: %v", c.Node, e.Node, e.Node.URI) + c.mu.Lock() + defer c.mu.Unlock() + if c.unprotectedIsCoordinator() { + // if removeNodeBasicSorted succeeds, that means that the node was + // 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) { + 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()) + } + } + c.logger.Printf("finished node leave on %s: %s, uri: %v", c.Node, e.Node, e.Node.URI) case NodeUpdate: + c.logger.Printf("received node update event: id: %v, string: %v, uri: %v", e.Node.ID, e.Node.String(), e.Node.URI) // NodeUpdate is intentionally not implemented. } - return nil + return err } +// nodeJoin should only be called by the coordinator. func (c *cluster) nodeJoin(node *Node) error { c.mu.Lock() defer c.mu.Unlock() @@ -1628,8 +1656,12 @@ func (c *cluster) nodeJoin(node *Node) error { // If the cluster already contains the node, just send it the cluster status. // This is useful in the case where a node is restarted or temporarily leaves // the cluster. - if node := c.unprotectedNodeByID(node.ID); node != nil { - return c.sendTo(node, c.unprotectedStatus()) + if cnode := c.unprotectedNodeByID(node.ID); cnode != nil { + if cnode.URI != node.URI { + c.logger.Printf("Node: %v changed URI from %s to %s", cnode.ID, cnode.URI, node.URI) + cnode.URI = node.URI + } + return c.unprotectedSetStateAndBroadcast(c.determineClusterState()) } // If the holder does not yet contain data, go ahead and add the node. diff --git a/server/server_test.go b/server/server_test.go index 2c222984d..6c937a116 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -24,6 +24,7 @@ import ( "strings" "testing" "testing/quick" + "time" "github.com/pelletier/go-toml" "github.com/pilosa/pilosa" @@ -374,3 +375,52 @@ type uint64Slice []uint64 func (p uint64Slice) Swap(i, j int) { p[i], p[j] = p[j], p[i] } func (p uint64Slice) Len() int { return len(p) } func (p uint64Slice) Less(i, j int) bool { return p[i] < p[j] } + +func TestClusteringNodes(t *testing.T) { + cluster := test.MustRunCluster(t, 3) + defer cluster.Close() + + var wait = true + for wait { + wait = false + for _, node := range cluster { + if node.API.State() != pilosa.ClusterStateNormal { + wait = true + } + } + time.Sleep(time.Millisecond * 1) + } + + if err := cluster[2].Command.Close(); err != nil { + t.Fatalf("closing third node: %v", err) + } + + // TODO: confirm that cluster stops accepting queries after one node closes + // TODO: implement and confirm that cluster keeps accepting queries if replication > 1 + // TODO: confirm that cluster stops accepting queries if 2 nodes fail and replication == 2 + // TODO: confirm that things keep working if a node is hard-closed (no nodeLeave event) and immediately restarted with a different address. + // TODO: confirm that a node can be removed using the remove endpoint after it has left teh cluster + // TODO: confirm that cluster still operates properly in state DEGRADED + + // Create new main with the same config. + config := cluster[2].Command.Config + // config.Bind = cluster[2].API.Node().URI.HostPort() + // config.Gossip.Port = strconv.Itoa(int(cluster[2].Command.GossipTransport().URI.Port)) + cluster[2].Command = server.NewCommand(cluster[2].Stdin, cluster[2].Stdout, cluster[2].Stderr) + cluster[2].Command.Config = config + + // Run new program. + if err := cluster[2].Start(); err != nil { + t.Fatalf("restarting node 2: %v", err) + } + + for wait { + wait = false + for _, node := range cluster { + if node.API.State() != pilosa.ClusterStateNormal { + wait = true + } + } + time.Sleep(time.Millisecond) + } +} diff --git a/test/pilosa.go b/test/pilosa.go index 91f8d9a05..743592569 100644 --- a/test/pilosa.go +++ b/test/pilosa.go @@ -21,6 +21,8 @@ import ( "io/ioutil" gohttp "net/http" "os" + "path" + "strconv" "strings" "testing" "time" @@ -204,6 +206,10 @@ func newCluster(size int, opts ...[]server.CommandOption) (Cluster, error) { commandOpts = opts[i%len(opts)] } m := NewCommandNode(i == 0, commandOpts...) + err := ioutil.WriteFile(path.Join(m.Config.DataDir, ".id"), []byte("node"+strconv.Itoa(i)), 0600) + if err != nil { + return nil, errors.Wrap(err, "writing node id") + } cluster[i] = m } From ba5b46898ff662b59f6af5bfd8fdbde59b5df19d Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Mon, 6 Aug 2018 14:55:38 -0500 Subject: [PATCH 2/5] fix determineClusterState bug and add more test cases --- cluster.go | 2 +- server/server_test.go | 106 +++++++++++++++++++++++++++++++++++++++--- 2 files changed, 100 insertions(+), 8 deletions(-) diff --git a/cluster.go b/cluster.go index 12513c14e..86fdf7712 100644 --- a/cluster.go +++ b/cluster.go @@ -474,7 +474,7 @@ func (c *cluster) determineClusterState() (clusterState string) { if c.haveTopologyAgreement() && c.allNodesReady() { return ClusterStateNormal } - if len(c.nodeIDs())-len(c.Topology.nodeIDs) >= c.ReplicaN && c.allNodesReady() { + if len(c.Topology.nodeIDs)-len(c.nodeIDs()) < c.ReplicaN && c.allNodesReady() { return ClusterStateDegraded } return ClusterStateStarting diff --git a/server/server_test.go b/server/server_test.go index 6c937a116..9fa2df34d 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -21,6 +21,7 @@ import ( "math/rand" "reflect" "sort" + "strconv" "strings" "testing" "testing/quick" @@ -376,7 +377,7 @@ func (p uint64Slice) Swap(i, j int) { p[i], p[j] = p[j], p[i] } func (p uint64Slice) Len() int { return len(p) } func (p uint64Slice) Less(i, j int) bool { return p[i] < p[j] } -func TestClusteringNodes(t *testing.T) { +func TestClusteringNodesReplica1(t *testing.T) { cluster := test.MustRunCluster(t, 3) defer cluster.Close() @@ -396,16 +397,17 @@ func TestClusteringNodes(t *testing.T) { } // TODO: confirm that cluster stops accepting queries after one node closes - // TODO: implement and confirm that cluster keeps accepting queries if replication > 1 - // TODO: confirm that cluster stops accepting queries if 2 nodes fail and replication == 2 - // TODO: confirm that things keep working if a node is hard-closed (no nodeLeave event) and immediately restarted with a different address. - // TODO: confirm that a node can be removed using the remove endpoint after it has left teh cluster - // TODO: confirm that cluster still operates properly in state DEGRADED + if _, err := cluster[0].API.Query(context.Background(), &pilosa.QueryRequest{}); !strings.Contains(err.Error(), "not allowed in state STARTING") { + t.Fatalf("got unexpected error querying an incomplete cluster: %v", err) + } // Create new main with the same config. config := cluster[2].Command.Config // config.Bind = cluster[2].API.Node().URI.HostPort() - // config.Gossip.Port = strconv.Itoa(int(cluster[2].Command.GossipTransport().URI.Port)) + + // this isn't necessary, but makes the test run way faster + config.Gossip.Port = strconv.Itoa(int(cluster[2].Command.GossipTransport().URI.Port)) + cluster[2].Command = server.NewCommand(cluster[2].Stdin, cluster[2].Stdout, cluster[2].Stderr) cluster[2].Command.Config = config @@ -424,3 +426,93 @@ func TestClusteringNodes(t *testing.T) { time.Sleep(time.Millisecond) } } + +func TestClusteringNodesReplica2(t *testing.T) { + cluster := test.MustNewCluster(t, 3) + for _, c := range cluster { + c.Config.Cluster.ReplicaN = 2 + } + err := cluster.Start() + if err != nil { + t.Fatalf("starting cluster: %v", err) + } + + var wait = true + for wait { + wait = false + for _, node := range cluster { + if node.API.State() != pilosa.ClusterStateNormal { + wait = true + } + } + time.Sleep(time.Millisecond * 1) + } + + if err := cluster[2].Command.Close(); err != nil { + t.Fatalf("closing third node: %v", err) + } + + if cluster[0].API.State() != pilosa.ClusterStateDegraded { + t.Fatalf("expected state to be DEGRADED, but got %s", cluster[0].API.State()) + } + + // TODO: implement and confirm that cluster keeps accepting queries if replication > 1 + if _, err := cluster[0].API.CreateIndex(context.Background(), "anewindex", pilosa.IndexOptions{}); err != nil { + t.Fatalf("got unexpected error creating index: %v", err) + } + + // TODO: confirm that cluster stops accepting queries if 2 nodes fail and replication == 2 + if err := cluster[1].Command.Close(); err != nil { + t.Fatalf("closing 2nd node: %v", err) + } + + if cluster[0].API.State() != pilosa.ClusterStateStarting { + t.Fatalf("expected state to be Starting, but got %s", cluster[0].API.State()) + } + + if _, err := cluster[0].API.Query(context.Background(), &pilosa.QueryRequest{}); !strings.Contains(err.Error(), "not allowed in state STARTING") { + t.Fatalf("got unexpected error querying an incomplete cluster: %v", err) + } + + // Create new main with the same config. + config := cluster[2].Command.Config + // config.Bind = cluster[2].API.Node().URI.HostPort() + + // this isn't necessary, but makes the test run way faster + config.Gossip.Port = strconv.Itoa(int(cluster[2].Command.GossipTransport().URI.Port)) + + cluster[2].Command = server.NewCommand(cluster[2].Stdin, cluster[2].Stdout, cluster[2].Stderr) + cluster[2].Command.Config = config + + // Run new program. + if err := cluster[2].Start(); err != nil { + t.Fatalf("restarting node 2: %v", err) + } + + // Create new main with the same config. + config = cluster[1].Command.Config + // config.Bind = cluster[1].API.Node().URI.HostPort() + + // this isn't necessary, but makes the test run way faster + config.Gossip.Port = strconv.Itoa(int(cluster[1].Command.GossipTransport().URI.Port)) + + cluster[1].Command = server.NewCommand(cluster[1].Stdin, cluster[1].Stdout, cluster[1].Stderr) + cluster[1].Command.Config = config + + // Run new program. + if err := cluster[1].Start(); err != nil { + t.Fatalf("restarting node 2: %v", err) + } + + defer cluster.Close() + + for wait { + wait = false + for _, node := range cluster { + if node.API.State() != pilosa.ClusterStateNormal { + wait = true + } + } + time.Sleep(time.Millisecond) + } +} From bb32706cb6e8089b05d415c6238dcb5305c32f26 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Mon, 6 Aug 2018 16:55:23 -0500 Subject: [PATCH 3/5] support DEGRADED in removeNode now, nodes which have failed and been removed from the running cluster state can still be manually removed to trigger a cluster resize event. This is important because otherwise there is no way to cause the cluster to resize itself if a node fails and you don't want to add a node to replace it. --- api.go | 11 +++++-- cluster.go | 67 ++++++++++++++++++++++++------------------- server/server_test.go | 51 ++++++++++++++++++++++++++++++-- 3 files changed, 93 insertions(+), 36 deletions(-) diff --git a/api.go b/api.go index c9242a9bd..847cc7ebd 100644 --- a/api.go +++ b/api.go @@ -734,13 +734,18 @@ func (api *API) RemoveNode(id string) (*Node, error) { return nil, errors.Wrap(err, "validating api method") } - removeNode := api.cluster.unprotectedNodeByID(id) + removeNode := api.cluster.nodeByID(id) if removeNode == nil { - return nil, errors.Wrap(ErrNodeIDNotExists, "finding node to remove") + if !api.cluster.topologyContainsNode(id) { + return nil, errors.Wrap(ErrNodeIDNotExists, "finding node to remove") + } + removeNode = &Node{ + ID: id, + } } // Start the resize process (similar to NodeJoin) - err := api.cluster.nodeLeave(removeNode) + err := api.cluster.nodeLeave(id) if err != nil { return removeNode, errors.Wrap(err, "calling node leave") } diff --git a/cluster.go b/cluster.go index 86fdf7712..309846e0b 100644 --- a/cluster.go +++ b/cluster.go @@ -341,17 +341,15 @@ func (c *cluster) addNode(node *Node) error { // removeNode removes a node from the Cluster and updates and saves the // new topology. unprotected. -func (c *cluster) removeNode(node *Node) error { +func (c *cluster) removeNode(nodeID string) error { // remove from cluster - if !c.removeNodeBasicSorted(node) { - return nil - } + c.removeNodeBasicSorted(nodeID) // remove from topology if c.Topology == nil { return fmt.Errorf("Cluster.Topology is nil") } - if !c.Topology.removeID(node.ID) { + if !c.Topology.removeID(nodeID) { return nil } @@ -511,6 +509,17 @@ func (c *cluster) unprotectedNodeByID(id string) *Node { return nil } +func (c *cluster) topologyContainsNode(id string) bool { + c.Topology.mu.RLock() + defer c.Topology.mu.RUnlock() + for _, nid := range c.Topology.nodeIDs { + if id == nid { + return true + } + } + return false +} + // nodePositionByID returns the position of the node in slice c.Nodes. func (c *cluster) nodePositionByID(nodeID string) int { for i, n := range c.Nodes { @@ -539,8 +548,8 @@ func (c *cluster) addNodeBasicSorted(node *Node) bool { // removeNodeBasicSorted removes a node from the cluster, maintaining the sort // order. Returns true if the node was removed. unprotected. -func (c *cluster) removeNodeBasicSorted(node *Node) bool { - i := c.nodePositionByID(node.ID) +func (c *cluster) removeNodeBasicSorted(nodeID string) bool { + i := c.nodePositionByID(nodeID) if i < 0 { return false } @@ -970,7 +979,7 @@ func (c *cluster) handleNodeAction(nodeAction nodeAction) error { if j.action == resizeJobActionRemove { c.mu.Lock() defer c.mu.Unlock() - return c.removeNode(nodeAction.node) + return c.removeNode(nodeAction.node.ID) } else if j.action == resizeJobActionAdd { c.mu.Lock() defer c.mu.Unlock() @@ -1100,7 +1109,7 @@ func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (* toCluster.partitionN = c.partitionN toCluster.ReplicaN = c.ReplicaN if nodeAction.action == resizeJobActionRemove { - toCluster.removeNodeBasicSorted(nodeAction.node) + toCluster.removeNodeBasicSorted(nodeAction.node.ID) } else if nodeAction.action == resizeJobActionAdd { toCluster.addNodeBasicSorted(nodeAction.node) } @@ -1598,7 +1607,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 c.removeNodeBasicSorted(e.Node) { + 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 @@ -1685,47 +1694,45 @@ func (c *cluster) nodeJoin(node *Node) error { } // nodeLeave initiates the removal of a node from the cluster. -func (c *cluster) nodeLeave(node *Node) error { +func (c *cluster) nodeLeave(nodeID string) error { c.mu.Lock() defer c.mu.Unlock() // Refuse the request if this is not the coordinator. if !c.unprotectedIsCoordinator() { - return fmt.Errorf("node removal requests are only valid on the coordinator node: %s", c.unprotectedCoordinatorNode().ID) + return fmt.Errorf("node removal requests are only valid on the coordinator node: %s", + c.unprotectedCoordinatorNode().ID) } - if c.state != ClusterStateNormal { - return fmt.Errorf("Cluster must be in state %s to remove a node. Current state: %s", ClusterStateNormal, c.state) + if c.state != ClusterStateNormal && c.state != ClusterStateDegraded { + return fmt.Errorf("Cluster must be in state %s to remove a node. Current state: %s", + ClusterStateNormal, c.state) } // Ensure that node is in the cluster. - if c.unprotectedNodeByID(node.ID) == nil { - return fmt.Errorf("Node is not a member of the cluster: %s", node.ID) + if !c.topologyContainsNode(nodeID) { + return fmt.Errorf("Node is not a member of the cluster: %s", nodeID) } // Prevent removing the coordinator node (this node). - if node.ID == c.Node.ID { + if nodeID == c.Node.ID { return fmt.Errorf("coordinator cannot be removed; first, make a different node the new coordinator.") } // See if resize job can be generated - if _, err := c.unprotectedGenerateResizeJobByAction(nodeAction{c.unprotectedNodeByID(node.ID), resizeJobActionRemove}); err != nil { + if _, err := c.unprotectedGenerateResizeJobByAction( + nodeAction{ + node: &Node{ID: nodeID}, + action: resizeJobActionRemove}, + ); err != nil { return errors.Wrap(err, "generating job") } - // Get the actual node in the local cluster. - n := c.unprotectedNodeByID(node.ID) - - // Don't do anything else if the cluster doesn't contain the node. - if n == nil { - return nil - } - // If the holder does not yet contain data, go ahead and remove the node. if ok, err := c.holder.HasData(); !ok && err == nil { - if err := c.removeNode(n); err != nil { + if err := c.removeNode(nodeID); err != nil { return errors.Wrap(err, "removing node") } - return c.unprotectedSetStateAndBroadcast(ClusterStateNormal) + return c.unprotectedSetStateAndBroadcast(c.determineClusterState()) } else if err != nil { return errors.Wrap(err, "checking if holder has data") } @@ -1735,7 +1742,7 @@ func (c *cluster) nodeLeave(node *Node) error { if err := c.unprotectedSetStateAndBroadcast(ClusterStateResizing); err != nil { return errors.Wrap(err, "broadcasting state") } - c.joiningLeavingNodes <- nodeAction{n, resizeJobActionRemove} + c.joiningLeavingNodes <- nodeAction{node: &Node{ID: nodeID}, action: resizeJobActionRemove} return nil } @@ -1777,7 +1784,7 @@ func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error { } for _, nodeID := range nodeIDsToRemove { - if err := c.removeNode(c.unprotectedNodeByID(nodeID)); err != nil { + if err := c.removeNode(nodeID); err != nil { return errors.Wrap(err, "removing node") } } diff --git a/server/server_test.go b/server/server_test.go index 9fa2df34d..1f00ebeb5 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -396,7 +396,7 @@ func TestClusteringNodesReplica1(t *testing.T) { t.Fatalf("closing third node: %v", err) } - // TODO: confirm that cluster stops accepting queries after one node closes + // confirm that cluster stops accepting queries after one node closes if _, err := cluster[0].API.Query(context.Background(), &pilosa.QueryRequest{}); !strings.Contains(err.Error(), "not allowed in state STARTING") { t.Fatalf("got unexpected error querying an incomplete cluster: %v", err) } @@ -456,12 +456,12 @@ func TestClusteringNodesReplica2(t *testing.T) { t.Fatalf("expected state to be DEGRADED, but got %s", cluster[0].API.State()) } - // TODO: implement and confirm that cluster keeps accepting queries if replication > 1 + // confirm that cluster keeps accepting queries if replication > 1 if _, err := cluster[0].API.CreateIndex(context.Background(), "anewindex", pilosa.IndexOptions{}); err != nil { t.Fatalf("got unexpected error creating index: %v", err) } - // TODO: confirm that cluster stops accepting queries if 2 nodes fail and replication == 2 + // confirm that cluster stops accepting queries if 2 nodes fail and replication == 2 if err := cluster[1].Command.Close(); err != nil { t.Fatalf("closing 2nd node: %v", err) } @@ -516,3 +516,48 @@ func TestClusteringNodesReplica2(t *testing.T) { time.Sleep(time.Millisecond) } } + +func TestRemoveNodeAfterItDies(t *testing.T) { + cluster := test.MustNewCluster(t, 3) + for _, c := range cluster { + c.Config.Cluster.ReplicaN = 2 + } + err := cluster.Start() + if err != nil { + t.Fatalf("starting cluster: %v", err) + } + + var wait = true + for wait { + wait = false + for _, node := range cluster { + if node.API.State() != pilosa.ClusterStateNormal { + wait = true + } + } + time.Sleep(time.Millisecond * 1) + } + + if err := cluster[2].Command.Close(); err != nil { + t.Fatalf("closing third node: %v", err) + } + + if cluster[0].API.State() != pilosa.ClusterStateDegraded { + t.Fatalf("expected state to be DEGRADED, but got %s", cluster[0].API.State()) + } + + if _, err := cluster[0].API.RemoveNode(cluster[2].API.Node().ID); err != nil { + t.Fatalf("removing failed node: %v", err) + } + + if cluster[0].API.State() != pilosa.ClusterStateNormal { + t.Fatalf("expected state to be DEGRADED, but got %s", cluster[0].API.State()) + } + + hosts := cluster[0].API.Hosts(context.Background()) + if len(hosts) != 2 { + t.Fatalf("unexpected hosts: %v", hosts) + } +} + +// TODO: confirm that things keep working if a node is hard-closed (no nodeLeave event) and immediately restarted with a different address. From 6e99a8757da56bf3e61621c1e2bfd29f9ee1acb8 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Wed, 8 Aug 2018 19:25:54 -0500 Subject: [PATCH 4/5] remove unnecessary memberlist constraint --- Gopkg.lock | 6 +++--- Gopkg.toml | 5 ----- 2 files changed, 3 insertions(+), 8 deletions(-) diff --git a/Gopkg.lock b/Gopkg.lock index dc3ddc49a..4a6068840 100644 --- a/Gopkg.lock +++ b/Gopkg.lock @@ -154,10 +154,10 @@ revision = "ef8a98b0bbce4a65b5aa4c368430a80ddc533168" [[projects]] - branch = "master" name = "github.com/hashicorp/memberlist" packages = ["."] - revision = "2288bf30e9c8d7b5f6549bf62e07120d72fd4b6c" + revision = "ce8abaa0c60c2d6bee7219f5ddf500e0a1457b28" + version = "v0.1.0" [[projects]] name = "github.com/inconshreveable/mousetrap" @@ -324,6 +324,6 @@ [solve-meta] analyzer-name = "dep" analyzer-version = 1 - inputs-digest = "6ae38dc521f55a89507ec21c575e0b65505de9514d1db6255a1b9b99fcf19ab3" + inputs-digest = "8290156ce8b4066c46ab83d743f4c81df0a17e148415bb1ee8409a51ac4c3ba4" solver-name = "gps-cdcl" solver-version = 1 diff --git a/Gopkg.toml b/Gopkg.toml index f5f81ad68..459775e54 100644 --- a/Gopkg.toml +++ b/Gopkg.toml @@ -17,11 +17,6 @@ # Only one of "branch", "version" or "revision" can be specified. branch = "master" -[[constraint]] - name = "github.com/hashicorp/memberlist" - branch = "master" - - [[constraint]] name = "github.com/gorilla/handlers" version = "=1.3.0" From 1901ffada651b92e4a4859a5899c86130c0020f7 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Wed, 8 Aug 2018 19:42:30 -0500 Subject: [PATCH 5/5] make sure cluster gets into DEGRADED state when adding nodes previously, losing a node could cause the cluster to go from NORMAL->DEGRADED, but adding a node would not cause it to go from STARTING->DEGRADED. Cody brought this up in code review. --- cluster.go | 12 +----------- server/server_test.go | 4 ++++ 2 files changed, 5 insertions(+), 11 deletions(-) diff --git a/cluster.go b/cluster.go index 309846e0b..715176b1c 100644 --- a/cluster.go +++ b/cluster.go @@ -456,12 +456,7 @@ func (c *cluster) receiveNodeState(nodeID string, state string) error { c.Topology.mu.Unlock() c.logger.Printf("received state %s (%s)", state, nodeID) - // Set cluster state to NORMAL. - if c.haveTopologyAgreement() && c.allNodesReady() { - return c.unprotectedSetStateAndBroadcast(ClusterStateNormal) - } - - return nil + return c.unprotectedSetStateAndBroadcast(c.determineClusterState()) } // determineClusterState is unprotected. @@ -1573,11 +1568,6 @@ func (c *cluster) considerTopology() error { return fmt.Errorf("coordinator %s is not in topology: %v", c.Node.ID, c.Topology.nodeIDs) } - // If local node is the only thing in .topology, continue. - //if len(c.Topology.NodeIDs) == 1 { - // return nil - //} - // Keep the cluster in state "STARTING" until hearing from all nodes. // Topology contains 2+ hosts. return nil diff --git a/server/server_test.go b/server/server_test.go index 1f00ebeb5..ab860dc1b 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -489,6 +489,10 @@ func TestClusteringNodesReplica2(t *testing.T) { t.Fatalf("restarting node 2: %v", err) } + if cluster[0].API.State() != pilosa.ClusterStateDegraded { + t.Fatalf("expected state to be DEGRADED, but got %s", cluster[0].API.State()) + } + // Create new main with the same config. config = cluster[1].Command.Config // config.Bind = cluster[1].API.Node().URI.HostPort()