Merge branch 'cluster-resize' into redundant-gossip-seeds

cluster.go

@@ -1690,6 +1690,16 @@ func (c *Cluster) NodeLeave(node *Node) error {
 		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.nodeByID(node.ID) == nil {
+		return fmt.Errorf("Node is not a member of the cluster: %s", node.ID)
+	}
+
+	// Prevent removing the coordinator node (this node).
+	if node.ID == c.Node.ID {
+		return fmt.Errorf("The coordinator node cannot be removed. First, make a different node the new coordinator.")
+	}
+
 	return c.nodeLeave(node)
 }
 

frame.go

@@ -22,7 +22,6 @@ import (
 	"os"
 	"path/filepath"
 	"sort"
-	"strings"
 	"sync"
 	"time"
 
@@ -574,53 +573,51 @@ func (f *Frame) RecalculateCaches() {
 // Additionally, a CreateViewMessage is sent to the cluster.
 func (f *Frame) CreateViewIfNotExists(name string) (*View, error) {
 
-	view, err := f.CreateViewIfNotExistsBase(name)
+	view, created, err := f.createViewIfNotExistsBase(name)
 	if err != nil {
 		return nil, err
 	}
 
-	// Broadcast view creation to the cluster.
-	err = f.broadcaster.SendSync(
-		&internal.CreateViewMessage{
-			Index: f.index,
-			Frame: f.name,
-			View:  name,
-		})
-	if err != nil {
-		return nil, err
+	if created {
+		// Broadcast view creation to the cluster.
+		err = f.broadcaster.SendSync(
+			&internal.CreateViewMessage{
+				Index: f.index,
+				Frame: f.name,
+				View:  name,
+			})
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	return view, nil
 }
 
-// CreateViewIfNotExistsBase returns the named view, creating it if necessary.
-func (f *Frame) CreateViewIfNotExistsBase(name string) (*View, error) {
+// createViewIfNotExistsBase returns the named view, creating it if necessary.
+// The returned bool indicates whether the view was created or not.
+func (f *Frame) createViewIfNotExistsBase(name string) (*View, bool, error) {
 	// Don't create inverse views if they are not enabled.
 	if !f.InverseEnabled() && IsInverseView(name) {
-		return nil, ErrFrameInverseDisabled
+		return nil, false, ErrFrameInverseDisabled
 	}
 
 	f.mu.Lock()
 	defer f.mu.Unlock()
 
 	if view := f.views[name]; view != nil {
-		return view, nil
+		return view, false, nil
 	}
 
 	view := f.newView(f.ViewPath(name), name)
 
-	// Never keep a cache for field views.
-	if strings.HasPrefix(name, ViewFieldPrefix) {
-		view.cacheType = CacheTypeNone
-	}
-
 	if err := view.Open(); err != nil {
-		return nil, err
+		return nil, false, err
 	}
 	view.RowAttrStore = f.rowAttrStore
 	f.views[view.Name()] = view
 
-	return view, nil
+	return view, true, nil
 }
 
 func (f *Frame) newView(path, name string) *View {

handler.go

@@ -2042,20 +2042,13 @@ func (h *Handler) handlePostClusterResizeRemoveNode(w http.ResponseWriter, r *ht
 
 	removeNode := h.Cluster.nodeByID(req.ID)
 	if removeNode == nil {
-		http.Error(w, err.Error(), http.StatusBadRequest)
+		http.Error(w, fmt.Sprintf("Node is not a member of the cluster: %s", req.ID), http.StatusBadRequest)
 		return
 	}
-	if err := func() error {
-		// TODO: prevent removing the coordinator node
 
-		// Start the resize process (similar to NodeJoin)
-		err := h.Cluster.NodeLeave(removeNode)
-		if err != nil {
-			return err
-		}
-
-		return nil
-	}(); err != nil {
+	// Start the resize process (similar to NodeJoin)
+	err = h.Cluster.NodeLeave(removeNode)
+	if err != nil {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
@@ -2208,7 +2201,6 @@ func GetTimeStamp(data map[string]interface{}, timeField string) (int64, error)
 func (h *Handler) handlePostClusterMessage(w http.ResponseWriter, r *http.Request) {
 	// Verify that request is only communicating over protobufs.
 	if r.Header.Get("Content-Type") != "application/x-protobuf" {
-		fmt.Println("**unsupported media type**")
 		http.Error(w, "Unsupported media type", http.StatusUnsupportedMediaType)
 		return
 	}

handler_test.go

@@ -1075,9 +1075,6 @@ func TestHandler_Frame_GetFields(t *testing.T) {
 			t.Fatal(err)
 		}
 		resp, err := http.Get(s.URL + "/index/i/frame/f/fields")
-		if err != nil {
-			t.Fatal(err)
-		}
 		if err != nil {
 			t.Fatal(err)
 		} else if resp.StatusCode != http.StatusOK {

server.go

@@ -424,7 +424,7 @@ func (s *Server) ReceiveMessage(pb proto.Message) error {
 		if f == nil {
 			return fmt.Errorf("Local Frame not found: %s", obj.Frame)
 		}
-		_, err := f.CreateViewIfNotExistsBase(obj.View)
+		_, _, err := f.createViewIfNotExistsBase(obj.View)
 		if err != nil {
 			return err
 		}

server/cluster_test.go

@@ -16,7 +16,10 @@ package server_test
 
 import (
 	"context"
+	"fmt"
+	"net/http"
 	"reflect"
+	"strings"
 	"testing"
 	"time"
 
@@ -496,3 +499,50 @@ func TestCluster_GossipMembership(t *testing.T) {
 		}
 	})
 }
+
+func TestClusterResize_RemoveNode(t *testing.T) {
+	cluster := test.MustRunMainWithCluster(t, 3)
+	m0 := cluster[0]
+	m1 := cluster[1]
+
+	t.Run("ErrorRemoveInvalidNode", func(t *testing.T) {
+		resp := test.MustDo("POST", m0.URL()+fmt.Sprintf("/cluster/resize/remove-node"), `{"id": "invalid-node-id"}`)
+		expBody := "Node is not a member of the cluster: invalid-node-id"
+		if resp.StatusCode != http.StatusBadRequest {
+			t.Fatalf("expected StatusCode %d but got %d", http.StatusBadRequest, resp.StatusCode)
+		} else if strings.TrimSpace(resp.Body) != expBody {
+			t.Fatalf("expected Body '%s' but got '%s'", expBody, strings.TrimSpace(resp.Body))
+		}
+	})
+
+	t.Run("ErrorRemoveCoordinator", func(t *testing.T) {
+		resp := test.MustDo("GET", m0.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 := "The coordinator node cannot be removed. First, make a different node the new coordinator."
+		if resp.StatusCode != http.StatusInternalServerError {
+			t.Fatalf("expected StatusCode %d but got %d", http.StatusInternalServerError, resp.StatusCode)
+		} else if strings.TrimSpace(resp.Body) != expBody {
+			t.Fatalf("expected Body '%s' but got '%s'", expBody, strings.TrimSpace(resp.Body))
+		}
+	})
+
+	t.Run("ErrorRemoveOnNonCoordinator", func(t *testing.T) {
+		resp := test.MustDo("GET", m0.URL()+fmt.Sprintf("/id"), "")
+		coordinatorNodeID := resp.Body
+
+		resp = test.MustDo("GET", m1.URL()+fmt.Sprintf("/id"), "")
+		nodeID := resp.Body
+
+		resp = test.MustDo("POST", m1.URL()+fmt.Sprintf("/cluster/resize/remove-node"), fmt.Sprintf(`{"id": "%s"}`, nodeID))
+
+		expBody := fmt.Sprintf("Node removal requests are only valid on the Coordinator node: %s", coordinatorNodeID)
+		if resp.StatusCode != http.StatusInternalServerError {
+			t.Fatalf("expected StatusCode %d but got %d", http.StatusInternalServerError, resp.StatusCode)
+		} else if strings.TrimSpace(resp.Body) != expBody {
+			t.Fatalf("expected Body '%s' but got '%s'", expBody, strings.TrimSpace(resp.Body))
+		}
+	})
+}

view.go

@@ -99,6 +99,12 @@ func (v *View) Path() string { return v.path }
 
 // Open opens and initializes the view.
 func (v *View) Open() error {
+
+	// Never keep a cache for field views.
+	if strings.HasPrefix(v.name, ViewFieldPrefix) {
+		v.cacheType = CacheTypeNone
+	}
+
 	if err := func() error {
 		// Ensure the view's path exists.
 		if err := os.MkdirAll(v.path, 0777); err != nil {