From d28a30ebd411acaee94ce218a2a2d4cd5a4e726c Mon Sep 17 00:00:00 2001 From: Cody Soyland Date: Tue, 6 Mar 2018 15:11:46 -0600 Subject: [PATCH 1/3] Proper error handling when attempting to remove node when there aren't enough replicas --- cluster.go | 15 +++++++++++++-- server/cluster_test.go | 31 +++++++++++++++++++++++++++++++ 2 files changed, 44 insertions(+), 2 deletions(-) diff --git a/cluster.go b/cluster.go index ed5318917..12d6fba50 100644 --- a/cluster.go +++ b/cluster.go @@ -737,7 +737,7 @@ func (c *Cluster) fragSources(to *Cluster, idx *Index) (map[string][]*internal.R // the fragment. srcNodeID, ok := srcNodesByFrag[frag] if !ok { - return nil, errors.New("not enough data to perform resize") + return nil, errors.New("not enough data to perform resize (replica factor may need to be increased)") } src := &internal.ResizeSource{ @@ -937,7 +937,11 @@ func (c *Cluster) allNodesReady() bool { func (c *Cluster) handleNodeAction(nodeAction nodeAction) error { j, err := c.generateResizeJob(nodeAction) if err != nil { - return err + c.logger().Printf("generateResizeJob error: err=%s", err) + if err := c.setStateAndBroadcast(ClusterStateNormal); err != nil { + c.logger().Printf("setStateAndBroadcast error: err=%s", err) + } + return c.setStateAndBroadcast(ClusterStateNormal) } // j.Run() runs in a goroutine because in the case where the @@ -1702,6 +1706,13 @@ func (c *Cluster) NodeLeave(node *Node) error { return fmt.Errorf("The coordinator node cannot be removed. First, make a different node the new coordinator.") } + // See if resize job can be generated + _, err := c.generateResizeJobByAction(nodeAction{c.nodeByID(node.ID), ResizeJobActionRemove}) + + if err != nil { + return err + } + return c.nodeLeave(node) } diff --git a/server/cluster_test.go b/server/cluster_test.go index 5ec805f81..9c518aecb 100644 --- a/server/cluster_test.go +++ b/server/cluster_test.go @@ -545,4 +545,35 @@ func TestClusterResize_RemoveNode(t *testing.T) { t.Fatalf("expected Body '%s' but got '%s'", expBody, strings.TrimSpace(resp.Body)) } }) + + t.Run("ErrorRemoveWithoutReplicas", func(t *testing.T) { + client0 := m0.Client() + + // Create indexes and frames on one node. + if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists { + t.Fatal(err) + } else if err := client0.CreateFrame(context.Background(), "i", "f", pilosa.FrameOptions{}); err != nil { + t.Fatal(err) + } + + setBits := "" + for i := 0; i < 20; i++ { + setBits += fmt.Sprintf("SetBit(rowID=1, frame=\"f\", columnID=%d) ", i*pilosa.SliceWidth) + } + + if _, err := m0.Query("i", "", setBits); err != nil { + t.Fatal(err) + } + + resp := test.MustDo("GET", m1.URL()+fmt.Sprintf("/id"), "") + nodeID := resp.Body + + resp = test.MustDo("POST", m0.URL()+fmt.Sprintf("/cluster/resize/remove-node"), fmt.Sprintf(`{"id": "%s"}`, nodeID)) + expBody := "not enough data to perform resize" + if resp.StatusCode != http.StatusInternalServerError { + t.Fatalf("expected StatusCode %d but got %d", http.StatusInternalServerError, resp.StatusCode) + } else if !strings.Contains(resp.Body, expBody) { + t.Fatalf("expected to contain '%s' but got '%s'", expBody, strings.TrimSpace(resp.Body)) + } + }) } From b7b92913d9c0013e21f7e582fcd9c0cb47565212 Mon Sep 17 00:00:00 2001 From: Cody Soyland Date: Tue, 6 Mar 2018 15:13:24 -0600 Subject: [PATCH 2/3] Add comment to listenForJoins --- cluster.go | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/cluster.go b/cluster.go index 12d6fba50..dcb9e0ecc 100644 --- a/cluster.go +++ b/cluster.go @@ -1002,7 +1002,13 @@ func (c *Cluster) ListenForJoins() { } func (c *Cluster) listenForJoins() { - var uriJoined bool + // When a cluster starts, the state is STARTING. + // We first want to wait for at least one node to join. + // Then we want to clear out the joiningLeavingNodes queue (buffered channel). + // Then we want to set the cluster state to NORMAL and resume processing of joiningLeavingNodes events. + // We use a bool `setNormal` to indicate when at least one node has joined. + + var setNormal bool for { @@ -1014,13 +1020,13 @@ func (c *Cluster) listenForJoins() { c.logger().Printf("handleNodeAction error: err=%s", err) continue } - uriJoined = true + setNormal = true continue default: } // Only change state to NORMAL if we have successfully added at least one host. - if uriJoined { + if setNormal { // Put the cluster back to state NORMAL and broadcast. if err := c.setStateAndBroadcast(ClusterStateNormal); err != nil { c.logger().Printf("setStateAndBroadcast error: err=%s", err) @@ -1037,7 +1043,7 @@ func (c *Cluster) listenForJoins() { c.logger().Printf("handleNodeAction error: err=%s", err) continue } - uriJoined = true + setNormal = true continue } } From 42682e12a828639aa953baf06e526cdb500ed5b4 Mon Sep 17 00:00:00 2001 From: Cody Soyland Date: Wed, 7 Mar 2018 09:45:27 -0600 Subject: [PATCH 3/3] Address code review: Fix error handling and add comment --- cluster.go | 2 +- server/cluster_test.go | 2 ++ 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/cluster.go b/cluster.go index dcb9e0ecc..8b56edaaf 100644 --- a/cluster.go +++ b/cluster.go @@ -941,7 +941,7 @@ func (c *Cluster) handleNodeAction(nodeAction nodeAction) error { if err := c.setStateAndBroadcast(ClusterStateNormal); err != nil { c.logger().Printf("setStateAndBroadcast error: err=%s", err) } - return c.setStateAndBroadcast(ClusterStateNormal) + return err } // j.Run() runs in a goroutine because in the case where the diff --git a/server/cluster_test.go b/server/cluster_test.go index 9c518aecb..299415bc3 100644 --- a/server/cluster_test.go +++ b/server/cluster_test.go @@ -556,6 +556,8 @@ func TestClusterResize_RemoveNode(t *testing.T) { t.Fatal(err) } + // This is an attempt to ensure there is data on both nodes, but is not guaranteed. + // TODO: Deterministic node IDs would ensure consistent results setBits := "" for i := 0; i < 20; i++ { setBits += fmt.Sprintf("SetBit(rowID=1, frame=\"f\", columnID=%d) ", i*pilosa.SliceWidth)