From 84fddbc67f245d4dc95a3ad823f65c293b87684d Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Wed, 12 Dec 2018 12:03:57 -0600 Subject: [PATCH] Replace the /fragment/data endpoint to support cluster resizing --- api.go | 21 ++++++++++++++++++++- client.go | 4 ++-- cluster.go | 4 ++-- http/client.go | 11 +++-------- http/handler.go | 23 +++++++++++++++++++++++ 5 files changed, 50 insertions(+), 13 deletions(-) diff --git a/api.go b/api.go index ce4fbd0ca..c6a1f3725 100644 --- a/api.go +++ b/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{}{ diff --git a/client.go b/client.go index b16222b7a..3762a27b9 100644 --- a/client.go +++ b/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 } diff --git a/cluster.go b/cluster.go index ad6cdbe6a..4c1c0853a 100644 --- a/cluster.go +++ b/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 { diff --git a/http/client.go b/http/client.go index b18757a46..46800fd70 100644 --- a/http/client.go +++ b/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() diff --git a/http/handler.go b/http/handler.go index 68719fdfe..c370656fa 100644 --- a/http/handler.go +++ b/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) {