From 7b0d4d75b49ff489b72e6f369e2bbc2850834aaf Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Thu, 13 Sep 2018 13:14:35 -0500 Subject: [PATCH] remove validators on import-roaring and test handler validators are for query args, not url vars. Also some misc cleanup and error handling in the handler. --- api.go | 4 +--- fragment.go | 1 - http/handler.go | 40 ++++++++++++++++++---------------------- server/handler_test.go | 18 ++++++++++++++++++ 4 files changed, 37 insertions(+), 26 deletions(-) diff --git a/api.go b/api.go index 137fdfe1c..a1f1271ea 100644 --- a/api.go +++ b/api.go @@ -332,14 +332,12 @@ func (api *API) ImportRoaring(ctx context.Context, indexName, fieldName string, }(node) } else { wg.Add(1) - //forward it on + // forward it on go func(node *Node) { - //execute on node err = api.server.defaultClient.ImportRoaring(ctx, node, indexName, fieldName, shard, data) wg.Done() }(node) } - } wg.Wait() return err diff --git a/fragment.go b/fragment.go index 19b0fdf5a..e551a2834 100644 --- a/fragment.go +++ b/fragment.go @@ -1371,7 +1371,6 @@ func (f *fragment) bulkImport(rowIDs, columnIDs []uint64) error { f.mu.Lock() defer f.mu.Unlock() - //f.storage.Unmmap() // Merge localBitmap into fragment's existing data. var results *roaring.Bitmap if f.storage.Count() > 0 { diff --git a/http/handler.go b/http/handler.go index f9a8c51c7..ac325de5c 100644 --- a/http/handler.go +++ b/http/handler.go @@ -177,12 +177,12 @@ func (h *Handler) populateValidators() { h.validators["GetFragmentData"] = queryValidationSpecRequired("index", "field", "shard") h.validators["PostFragmentData"] = queryValidationSpecRequired("index", "field", "shard") h.validators["GetFragmentBlocks"] = queryValidationSpecRequired("index", "field", "view", "shard") - h.validators["ImportRoaringBitmap"] = queryValidationSpecRequired("index", "field", "shard") } func (h *Handler) queryArgValidator(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { key := mux.CurrentRoute(r).GetName() + if validator, ok := h.validators[key]; ok { if err := validator.validate(r.URL.Query()); err != nil { // TODO: Return the response depending on the Accept header @@ -218,7 +218,7 @@ func newRouter(handler *Handler) *mux.Router { router.HandleFunc("/index/{index}/field/{field}", handler.handlePostField).Methods("POST") router.HandleFunc("/index/{index}/field/{field}", handler.handleDeleteField).Methods("DELETE") router.HandleFunc("/index/{index}/field/{field}/import", handler.handlePostImport).Methods("POST") - router.HandleFunc("/index/{index}/field/{field}/import-roaring/{shard}", handler.handlePostRoaringImport).Methods("POST") + router.HandleFunc("/index/{index}/field/{field}/import-roaring/{shard}", handler.handlePostImportRoaring).Methods("POST") router.HandleFunc("/index/{index}/query", handler.handlePostQuery).Methods("POST").Name("PostQuery") router.HandleFunc("/info", handler.handleGetInfo).Methods("GET") router.HandleFunc("/recalculate-caches", handler.handleRecalculateCaches).Methods("POST") @@ -1431,19 +1431,11 @@ func GetHTTPClient(t *tls.Config) *http.Client { } // handlPostRoaringImport -func (h *Handler) handlePostRoaringImport(w http.ResponseWriter, r *http.Request) { - // Verify that request is only communicating over protobufs. +func (h *Handler) handlePostImportRoaring(w http.ResponseWriter, r *http.Request) { if r.Header.Get("Content-Type") != "application/x-binary" { http.Error(w, "Unsupported media type", http.StatusUnsupportedMediaType) return - } /*else if validHeaderAcceptJSON(r.Header) { - http.Error(w, "Not acceptable", http.StatusNotAcceptable) - return - }*/ - - indexName := mux.Vars(r)["index"] - fieldName := mux.Vars(r)["field"] - shardName := mux.Vars(r)["shard"] + } // Read entire body. body, err := ioutil.ReadAll(r.Body) @@ -1452,25 +1444,29 @@ func (h *Handler) handlePostRoaringImport(w http.ResponseWriter, r *http.Request return } - shard, err := strconv.ParseUint(shardName, 10, 64) + urlVars := mux.Vars(r) + shard, err := strconv.ParseUint(urlVars["shard"], 10, 64) if err != nil { http.Error(w, "shard should be an unsigned integer", http.StatusBadRequest) return } - //TODO give meaningful stats for import - err = h.api.ImportRoaring(r.Context(), indexName, fieldName, shard, body) - // Marshal response object. - msg := string("") + // TODO give meaningful stats for import + err = h.api.ImportRoaring(r.Context(), urlVars["index"], urlVars["field"], shard, body) + resp := &pilosa.ImportResponse{} if err != nil { - msg = err.Error() + resp.Err = err.Error() } - buf, e := h.api.Serializer.Marshal(&pilosa.ImportResponse{Err: msg}) - if e != nil { - http.Error(w, fmt.Sprintf("marshal import response"), http.StatusInternalServerError) + // Marshal response object. + buf, err := h.api.Serializer.Marshal(resp) + if err != nil { + http.Error(w, fmt.Sprintf("marshal import response: %v", err), http.StatusInternalServerError) return } // Write response. - w.Write(buf) + _, err = w.Write(buf) + if err != nil { + h.logger.Printf("writing import-roaring response: %v", err) + } } diff --git a/server/handler_test.go b/server/handler_test.go index 0388dcec6..2c1aa9868 100644 --- a/server/handler_test.go +++ b/server/handler_test.go @@ -16,6 +16,8 @@ package server_test import ( "bytes" + "context" + "encoding/hex" "encoding/json" "fmt" "io" @@ -86,6 +88,22 @@ func TestHandler_Endpoints(t *testing.T) { } }) + t.Run("ImportRoaring", func(t *testing.T) { + w := httptest.NewRecorder() + roaringData, _ := hex.DecodeString("3B3001000100000900010000000100010009000100") + req := test.MustNewHTTPRequest("POST", "/index/i0/field/f1/import-roaring/0", bytes.NewBuffer(roaringData)) + req.Header.Set("Content-Type", "application/x-binary") + h.ServeHTTP(w, req) + resp, err := cmd.API.Query(context.Background(), &pilosa.QueryRequest{Index: "i0", Query: "TopN(f1)"}) + if err != nil { + t.Fatalf("querying: %v", err) + } + if !reflect.DeepEqual(resp.Results[0], []pilosa.Pair{{Count: 12, ID: 0}}) { + t.Fatalf("Unexpected result %v", resp.Results[0]) + } + + }) + t.Run("Status", func(t *testing.T) { w := httptest.NewRecorder() h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/status", nil))