scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1584 from jaffee/1249-node-failure

Browse source 
Add DEGRADED cluster state and handle gossip NodeLeave events correctly
This commit is contained in:
Matthew Jaffee 2018-08-09 09:39:23 -05:00 • committed by GitHub
commit 6070db077c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 304 additions and 68 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
Gopkg.lock generated
View file

@ -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"
@ -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 = "8290156ce8b4066c46ab83d743f4c81df0a17e148415bb1ee8409a51ac4c3ba4"
solver-name = "gps-cdcl"
solver-version = 1

4
Gopkg.toml
View file

@ -16,3 +16,7 @@
# 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/gorilla/handlers"
version = "=1.3.0"

12
api.go
View file

@ -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),
}
@ -733,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")
}

135
cluster.go
View file

@ -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"
@ -339,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
}
@ -396,7 +396,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,22 +451,26 @@ 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()
c.logger.Printf("received state %s (%s)", state, nodeID)
// Set cluster state to NORMAL.
if c.haveTopologyAgreement() && c.allNodesReady() {
return c.unprotectedSetStateAndBroadcast(ClusterStateNormal)
}
return c.unprotectedSetStateAndBroadcast(c.determineClusterState())
}
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.Topology.nodeIDs)-len(c.nodeIDs()) < c.ReplicaN && c.allNodesReady() {
return ClusterStateDegraded
}
return ClusterStateStarting
}
func (c *cluster) status() *ClusterStatus {
@ -500,6 +504,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 {
@ -528,8 +543,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
}
@ -907,12 +922,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
}
}
@ -959,7 +974,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()
@ -1089,7 +1104,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)
}
@ -1553,18 +1568,13 @@ 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
}
// 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 +1589,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.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())
}
}
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 +1655,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.
@ -1653,47 +1684,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")
}
@ -1703,7 +1732,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
}
@ -1745,7 +1774,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")
}
}

191
server/server_test.go
View file

@ -21,9 +21,11 @@ import (
"math/rand"
"reflect"
"sort"
"strconv"
"strings"
"testing"
"testing/quick"
"time"
"github.com/pelletier/go-toml"
"github.com/pilosa/pilosa"
@ -374,3 +376,192 @@ 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 TestClusteringNodesReplica1(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)
}
// 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)
}
// 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)
}
for wait {
wait = false
for _, node := range cluster {
if node.API.State() != pilosa.ClusterStateNormal {
wait = true
}
}
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())
}
// 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)
}
// 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)
}
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()
// 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)
}
}
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.

6
test/pilosa.go
View file

@ -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
}