Replace the /fragment/data endpoint to support cluster resizing

api.go

@@ -559,6 +559,23 @@ func (api *API) FragmentBlocks(ctx context.Context, indexName, fieldName, viewNa
 	return blocks, nil
 }
 
+// FragmentData returns all data in the specified fragment.
+func (api *API) FragmentData(ctx context.Context, indexName, fieldName, viewName string, shard uint64) (io.WriterTo, error) {
+	span, _ := tracing.StartSpanFromContext(ctx, "API.FragmentBlocks")
+	defer span.Finish()
+
+	if err := api.validate(apiFragmentData); err != nil {
+		return nil, errors.Wrap(err, "validating api method")
+	}
+
+	// Retrieve fragment from holder.
+	f := api.holder.fragment(indexName, fieldName, viewName, shard)
+	if f == nil {
+		return nil, ErrFragmentNotFound
+	}
+	return f, nil
+}
+
 // Hosts returns a list of the hosts in the cluster including their ID,
 // URL, and which is the coordinator.
 func (api *API) Hosts(ctx context.Context) []*Node {
@@ -1203,6 +1220,7 @@ const (
 	apiExportCSV
 	apiFragmentBlockData
 	apiFragmentBlocks
+	apiFragmentData
 	apiField
 	apiFieldAttrDiff
 	//apiHosts // not implemented
@@ -1232,7 +1250,8 @@ var methodsCommon = map[apiMethod]struct{}{
 }
 
 var methodsResizing = map[apiMethod]struct{}{
-	apiResizeAbort: {},
+	apiFragmentData: {},
+	apiResizeAbort:  {},
 }
 
 var methodsNormal = map[apiMethod]struct{}{

client.go

@@ -52,7 +52,7 @@ type InternalClient interface {
 	ColumnAttrDiff(ctx context.Context, uri *URI, index string, blks []AttrBlock) (map[uint64]map[string]interface{}, error)
 	RowAttrDiff(ctx context.Context, uri *URI, index, field string, blks []AttrBlock) (map[uint64]map[string]interface{}, error)
 	SendMessage(ctx context.Context, uri *URI, msg []byte) error
-	RetrieveShardFromURI(ctx context.Context, index, field string, shard uint64, uri URI) (io.ReadCloser, error)
+	RetrieveShardFromURI(ctx context.Context, index, field, view string, shard uint64, uri URI) (io.ReadCloser, error)
 	ImportRoaring(ctx context.Context, uri *URI, index, field string, shard uint64, remote bool, req *ImportRoaringRequest) error
 }
 
@@ -149,6 +149,6 @@ func (n nopInternalClient) RowAttrDiff(ctx context.Context, uri *URI, index, fie
 func (n nopInternalClient) SendMessage(ctx context.Context, uri *URI, msg []byte) error {
 	return nil
 }
-func (n nopInternalClient) RetrieveShardFromURI(ctx context.Context, index, field string, shard uint64, uri URI) (io.ReadCloser, error) {
+func (n nopInternalClient) RetrieveShardFromURI(ctx context.Context, index, field, view string, shard uint64, uri URI) (io.ReadCloser, error) {
 	return nil, nil
 }

cluster.go

@@ -1309,7 +1309,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error {
 
 				// Stream shard from remote node.
 				c.logger.Printf("retrieve shard %d for index %s from host %s", src.Shard, src.Index, src.Node.URI)
-				rd, err := c.InternalClient.RetrieveShardFromURI(ctx, src.Index, src.Field, src.Shard, srcURI)
+				rd, err := c.InternalClient.RetrieveShardFromURI(ctx, src.Index, src.Field, src.View, src.Shard, srcURI)
 				if err != nil {
 					// For now it is an acceptable error if the fragment is not found
 					// on the remote node. This occurs when a shard has been skipped and
@@ -1318,7 +1318,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error {
 					// TODO: figure out a way to distinguish from "fragment not found" errors
 					// which are true errors and which simply mean the fragment doesn't have data.
 					if err == ErrFragmentNotFound {
-						return nil
+						continue
 					}
 					return errors.Wrap(err, "retrieving shard")
 				} else if rd == nil {

http/client.go

@@ -705,24 +705,19 @@ func (c *InternalClient) exportNodeCSV(ctx context.Context, node *pilosa.Node, i
 	return nil
 }
 
-func (c *InternalClient) RetrieveShardFromURI(ctx context.Context, index, field string, shard uint64, uri pilosa.URI) (io.ReadCloser, error) {
+func (c *InternalClient) RetrieveShardFromURI(ctx context.Context, index, field, view string, shard uint64, uri pilosa.URI) (io.ReadCloser, error) {
 	span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.RetrieveShardFromURI")
 	defer span.Finish()
 
 	node := &pilosa.Node{
 		URI: uri,
 	}
-	return c.backupShardNode(ctx, index, field, shard, node)
-}
 
-func (c *InternalClient) backupShardNode(ctx context.Context, index, field string, shard uint64, node *pilosa.Node) (io.ReadCloser, error) {
-	span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.backupShardNode")
-	defer span.Finish()
-
-	u := nodePathToURL(node, "/fragment/data")
+	u := nodePathToURL(node, "/internal/fragment/data")
 	u.RawQuery = url.Values{
 		"index": {index},
 		"field": {field},
+		"view":  {view},
 		"shard": {strconv.FormatUint(shard, 10)},
 	}.Encode()
 

http/handler.go

@@ -192,6 +192,7 @@ func (h *Handler) populateValidators() {
 	h.validators["PostClusterMessage"] = queryValidationSpecRequired()
 	h.validators["GetFragmentBlockData"] = queryValidationSpecRequired()
 	h.validators["GetFragmentBlocks"] = queryValidationSpecRequired("index", "field", "view", "shard")
+	h.validators["GetFragmentData"] = queryValidationSpecRequired("index", "field", "view", "shard")
 	h.validators["GetFragmentNodes"] = queryValidationSpecRequired("shard", "index")
 	h.validators["PostIndexAttrDiff"] = queryValidationSpecRequired()
 	h.validators["PostFieldAttrDiff"] = queryValidationSpecRequired()
@@ -262,6 +263,7 @@ func newRouter(handler *Handler) *mux.Router {
 	router.HandleFunc("/internal/cluster/message", handler.handlePostClusterMessage).Methods("POST").Name("PostClusterMessage")
 	router.HandleFunc("/internal/fragment/block/data", handler.handleGetFragmentBlockData).Methods("GET").Name("GetFragmentBlockData")
 	router.HandleFunc("/internal/fragment/blocks", handler.handleGetFragmentBlocks).Methods("GET").Name("GetFragmentBlocks")
+	router.HandleFunc("/internal/fragment/data", handler.handleGetFragmentData).Methods("GET").Name("GetFragmentData")
 	router.HandleFunc("/internal/fragment/nodes", handler.handleGetFragmentNodes).Methods("GET").Name("GetFragmentNodes")
 	router.HandleFunc("/internal/index/{index}/attr/diff", handler.handlePostIndexAttrDiff).Methods("POST").Name("PostIndexAttrDiff")
 	router.HandleFunc("/internal/index/{index}/field/{field}/attr/diff", handler.handlePostFieldAttrDiff).Methods("POST").Name("PostFieldAttrDiff")
@@ -1214,6 +1216,27 @@ type getFragmentBlocksResponse struct {
 	Blocks []pilosa.FragmentBlock `json:"blocks"`
 }
 
+// handleGetFragmentData handles GET /internal/fragment/data requests.
+func (h *Handler) handleGetFragmentData(w http.ResponseWriter, r *http.Request) {
+	// Read shard parameter.
+	q := r.URL.Query()
+	shard, err := strconv.ParseUint(q.Get("shard"), 10, 64)
+	if err != nil {
+		http.Error(w, "shard required", http.StatusBadRequest)
+		return
+	}
+	// Retrieve fragment data from holder.
+	f, err := h.api.FragmentData(r.Context(), q.Get("index"), q.Get("field"), q.Get("view"), shard)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusNotFound)
+		return
+	}
+	// Stream fragment to response body.
+	if _, err := f.WriteTo(w); err != nil {
+		h.logger.Printf("error streaming fragment data: %s", err)
+	}
+}
+
 // handleGetVersion handles /version requests.
 func (h *Handler) handleGetVersion(w http.ResponseWriter, r *http.Request) {
 	if !validHeaderAcceptJSON(r.Header) {