Merge pull request #1785 from travisturner/cluster-resize-fix

Cluster resize fix

Author: travisturner

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.FragmentData")
+	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

@@ -1218,7 +1218,7 @@ func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (*
 			Node:          toCluster.unprotectedNodeByID(id),
 			Coordinator:   c.unprotectedCoordinatorNode(),
 			Sources:       sources,
-			Schema:        &Schema{Indexes: c.holder.Schema()}, // Include the schema to ensure it's in sync on the receiving node.
+			NodeStatus:    c.nodeStatus(), // Include the NodeStatus in order to ensure that schema and availableShards are in sync on the receiving node.
 			ClusterStatus: c.unprotectedStatus(),
 		}
 		j.Instructions = append(j.Instructions, instr)
@@ -1277,12 +1277,30 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error {
 			span, ctx := tracing.StartSpanFromContext(context.Background(), "Cluster.followResizeInstruction")
 			defer span.Finish()
 
-			// Sync the schema received in the resize instruction.
+			// Sync the NodeStatus received in the resize instruction.
+			// Sync schema.
 			c.logger.Debugf("holder applySchema")
-			if err := c.holder.applySchema(instr.Schema); err != nil {
+			if err := c.holder.applySchema(instr.NodeStatus.Schema); err != nil {
 				return errors.Wrap(err, "applying schema")
 			}
 
+			// Sync available shards.
+			for _, is := range instr.NodeStatus.Indexes {
+				for _, fs := range is.Fields {
+					f := c.holder.Field(is.Name, fs.Name)
+
+					// if we don't know about a field locally, log an error because
+					// fields should be created and synced prior to shard creation
+					if f == nil {
+						c.logger.Printf("local field not found: %s/%s", is.Name, fs.Name)
+						continue
+					}
+					if err := f.AddRemoteAvailableShards(fs.AvailableShards); err != nil {
+						return errors.Wrap(err, "adding remote available shards")
+					}
+				}
+			}
+
 			// Request each source file in ResizeSources.
 			for _, src := range instr.Sources {
 				c.logger.Printf("get shard %d for index %s from host %s", src.Shard, src.Index, src.Node.URI)
@@ -1309,7 +1327,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 +1336,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 {
@@ -1817,6 +1835,30 @@ func (c *cluster) nodeLeave(nodeID string) error {
 	return nil
 }
 
+func (c *cluster) nodeStatus() *NodeStatus {
+	ns := &NodeStatus{
+		Node:   c.Node,
+		Schema: &Schema{Indexes: c.holder.Schema()},
+	}
+	var availableShards *roaring.Bitmap
+	for _, idx := range ns.Schema.Indexes {
+		is := &IndexStatus{Name: idx.Name}
+		for _, f := range idx.Fields {
+			if field := c.holder.Field(idx.Name, f.Name); field != nil {
+				availableShards = field.AvailableShards()
+			} else {
+				availableShards = roaring.NewBitmap()
+			}
+			is.Fields = append(is.Fields, &FieldStatus{
+				Name:            f.Name,
+				AvailableShards: availableShards,
+			})
+		}
+		ns.Indexes = append(ns.Indexes, is)
+	}
+	return ns
+}
+
 func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
@@ -1918,7 +1960,7 @@ type ResizeInstruction struct {
 	Node          *Node
 	Coordinator   *Node
 	Sources       []*ResizeSource
-	Schema        *Schema
+	NodeStatus    *NodeStatus
 	ClusterStatus *ClusterStatus
 }
 

encoding/proto/proto.go

@@ -457,7 +457,7 @@ func encodeResizeInstruction(m *pilosa.ResizeInstruction) *internal.ResizeInstru
 		Node:          encodeNode(m.Node),
 		Coordinator:   encodeNode(m.Coordinator),
 		Sources:       encodeResizeSources(m.Sources),
-		Schema:        encodeSchema(m.Schema),
+		NodeStatus:    encodeNodeStatus(m.NodeStatus),
 		ClusterStatus: encodeClusterStatus(m.ClusterStatus),
 	}
 }
@@ -737,8 +737,8 @@ func decodeResizeInstruction(ri *internal.ResizeInstruction, m *pilosa.ResizeIns
 	decodeNode(ri.Coordinator, m.Coordinator)
 	m.Sources = make([]*pilosa.ResizeSource, len(ri.Sources))
 	decodeResizeSources(ri.Sources, m.Sources)
-	m.Schema = &pilosa.Schema{}
-	decodeSchema(ri.Schema, m.Schema)
+	m.NodeStatus = &pilosa.NodeStatus{}
+	decodeNodeStatus(ri.NodeStatus, m.NodeStatus)
 	m.ClusterStatus = &pilosa.ClusterStatus{}
 	decodeClusterStatus(ri.ClusterStatus, m.ClusterStatus)
 }

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) {