Merge pull request #913 from travisturner/abort-resize-endpoint

endpoint to abort a cluster resize that is in progress

cluster.go

@@ -569,11 +569,15 @@ func (c *Cluster) handleJoiningHost(uri URI) error {
 	jobResult := <-j.result
 	switch jobResult {
 	case ResizeJobStateDone:
-		c.CompleteCurrentJob(ResizeJobStateDone)
+		if err := c.CompleteCurrentJob(ResizeJobStateDone); err != nil {
+			return err
+		}
 		// Add uri to the cluster.
 		return c.AddNode(uri)
 	case ResizeJobStateAborted:
-		c.CompleteCurrentJob(ResizeJobStateAborted)
+		if err := c.CompleteCurrentJob(ResizeJobStateAborted); err != nil {
+			return err
+		}
 	}
 	return nil
 }
@@ -691,14 +695,15 @@ func (c *Cluster) generateResizeJob(addURI URI) *ResizeJob {
 
 // CompleteCurrentJob sets the state of the current ResizeJob
 // then removes the pointer to currentJob.
-func (c *Cluster) CompleteCurrentJob(state string) {
+func (c *Cluster) CompleteCurrentJob(state string) error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	if c.currentJob == nil {
-		return
+		return fmt.Errorf("no resize job currently running")
 	}
 	c.currentJob.SetState(state)
 	c.currentJob = nil
+	return nil
 }
 
 // followResizeInstruction is run by any node that receives a ResizeInstruction.

handler.go

@@ -98,6 +98,7 @@ func NewRouter(handler *Handler) *mux.Router {
 	router := mux.NewRouter()
 	router.HandleFunc("/", handler.handleWebUI).Methods("GET")
 	router.HandleFunc("/assets/{file}", handler.handleWebUI).Methods("GET")
+	router.HandleFunc("/cluster/resize/abort", handler.handlePostClusterResizeAbort).Methods("POST")
 	router.PathPrefix("/debug/pprof/").Handler(http.DefaultServeMux).Methods("GET")
 	router.HandleFunc("/debug/vars", handler.handleExpvar).Methods("GET")
 	router.HandleFunc("/export", handler.handleGetExport).Methods("GET")
@@ -1891,6 +1892,35 @@ func (h *Handler) handlePostInput(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+//handlePostClusterResizeAbort handles POST /cluster/resize/abort request.
+func (h *Handler) handlePostClusterResizeAbort(w http.ResponseWriter, r *http.Request) {
+	var msg string
+
+	if err := func() error {
+		if !h.Cluster.IsCoordinator() {
+			return fmt.Errorf("abort requests must be made on the coordinator node")
+		}
+		err := h.Cluster.CompleteCurrentJob(ResizeJobStateAborted)
+		if err != nil {
+			return err
+		}
+		return nil
+	}(); err != nil {
+		msg = err.Error()
+	}
+
+	// Encode response.
+	if err := json.NewEncoder(w).Encode(clusterResizeAbortResponse{
+		Info: msg,
+	}); err != nil {
+		h.logger().Printf("response encoding error: %s", err)
+	}
+}
+
+type clusterResizeAbortResponse struct {
+	Info string `json:"info"`
+}
+
 // InputJSONDataParser validates input json file and executes SetBit.
 func (h *Handler) InputJSONDataParser(req map[string]interface{}, index *Index, name string) (map[string][]*Bit, error) {
 	inputDef, err := index.InputDefinition(name)

handler_test.go

@@ -152,6 +152,24 @@ func TestHandler_Status(t *testing.T) {
 	}
 }
 
+// Ensure the handler can abort a cluster resize.
+func TestHandler_ClusterResizeAbort(t *testing.T) {
+
+	t.Run("No resize job", func(t *testing.T) {
+		h := test.NewHandler()
+		h.Cluster = test.NewCluster(1)
+
+		w := httptest.NewRecorder()
+		h.ServeHTTP(w, test.MustNewHTTPRequest("POST", "/cluster/resize/abort", nil))
+		if w.Code != http.StatusOK {
+			t.Fatalf("unexpected status code: %d", w.Code)
+		} else if body := w.Body.String(); body != `{"info":"no resize job currently running"}`+"\n" {
+			t.Fatalf("unexpected body: %s", body)
+		}
+	})
+
+}
+
 // Ensure the handler can return the maxslice map.
 func TestHandler_MaxSlices(t *testing.T) {
 	hldr := test.MustOpenHolder()