diff --git a/cluster.go b/cluster.go index d9575b58e..c7f5c4193 100644 --- a/cluster.go +++ b/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) } diff --git a/frame.go b/frame.go index b5807dc71..069e0bc89 100644 --- a/frame.go +++ b/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 { diff --git a/handler.go b/handler.go index 5d7af3987..fc1771128 100644 --- a/handler.go +++ b/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 } diff --git a/handler_test.go b/handler_test.go index c8030a9ea..419cbcecd 100644 --- a/handler_test.go +++ b/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 { diff --git a/server.go b/server.go index 6a92a7768..3751e28ad 100644 --- a/server.go +++ b/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 } diff --git a/server/cluster_test.go b/server/cluster_test.go index bfc3640b9..65d8844d3 100644 --- a/server/cluster_test.go +++ b/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)) + } + }) +} diff --git a/view.go b/view.go index d77d9257a..db2eb97f1 100644 --- a/view.go +++ b/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 {