scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 02:44:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

consolidate import and import-value endpoints

Browse source
This commit is contained in:
Travis Turner 2018-07-02 10:20:55 -05:00
parent 7bd55f47e7
commit 2898a422bf
No known key found for this signature in database
GPG key ID: 7F08008DFD9314C9
4 changed files with 70 additions and 75 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
api.go
View file

@ -277,6 +277,19 @@ func (api *API) CreateField(ctx context.Context, indexName string, fieldName str
return field, nil
}
// Field retrieves the named field.
func (api *API) Field(ctx context.Context, indexName, fieldName string) (*Field, error) {
if err := api.validate(apiField); err != nil {
return nil, errors.Wrap(err, "validating api method")
}
field := api.holder.Field(indexName, fieldName)
if field == nil {
return nil, NewNotFoundError(ErrFieldNotFound)
}
return field, nil
}
// DeleteField removes the named field from the named index. If the index is not
// found, an error is returned. If the field is not found, it is ignored and no
// action is taken.
@ -866,6 +879,7 @@ const (
apiExportCSV
apiFragmentBlockData
apiFragmentBlocks
apiField
apiFieldAttrDiff
//apiHosts // not implemented
apiImport
@ -909,6 +923,7 @@ var methodsNormal = map[apiMethod]struct{}{
apiExportCSV: struct{}{},
apiFragmentBlockData: struct{}{},
apiFragmentBlocks: struct{}{},
apiField: struct{}{},
apiFieldAttrDiff: struct{}{},
apiImport: struct{}{},
apiImportValue: struct{}{},

4
apimethod_string.go
View file

@ -4,9 +4,9 @@ package pilosa
import "strconv"
const _apiMethod_name = "apiClusterMessageapiCreateFieldapiCreateIndexapiDeleteFieldapiDeleteIndexapiDeleteViewapiExportCSVapiFragmentBlockDataapiFragmentBlocksapiFieldAttrDiffapiImportapiImportValueapiIndexapiIndexAttrDiffapiMarshalFragmentapiQueryapiRecalculateCachesapiRemoveNodeapiResizeAbortapiSetCoordinatorapiShardNodesapiUnmarshalFragmentapiViews"
const _apiMethod_name = "apiClusterMessageapiCreateFieldapiCreateIndexapiDeleteFieldapiDeleteIndexapiDeleteViewapiExportCSVapiFragmentBlockDataapiFragmentBlocksapiFieldapiFieldAttrDiffapiImportapiImportValueapiIndexapiIndexAttrDiffapiMarshalFragmentapiQueryapiRecalculateCachesapiRemoveNodeapiResizeAbortapiSetCoordinatorapiShardNodesapiUnmarshalFragmentapiViews"
var _apiMethod_index = [...]uint16{0, 17, 31, 45, 59, 73, 86, 98, 118, 135, 151, 160, 174, 182, 198, 216, 224, 244, 257, 271, 288, 301, 321, 329}
var _apiMethod_index = [...]uint16{0, 17, 31, 45, 59, 73, 86, 98, 118, 135, 143, 159, 168, 182, 190, 206, 224, 232, 252, 265, 279, 296, 309, 329, 337}
func (i apiMethod) String() string {
if i < 0 || i >= apiMethod(len(_apiMethod_index)-1) {

16
http/client.go
View file

@ -293,7 +293,7 @@ func (c *InternalClient) Import(ctx context.Context, index, field string, shard
// Import to each node.
for _, node := range nodes {
if err := c.importNode(ctx, node, buf); err != nil {
if err := c.importNode(ctx, node, index, field, buf); err != nil {
return fmt.Errorf("import node: host=%s, err=%s", node.URI, err)
}
}
@ -319,7 +319,7 @@ func (c *InternalClient) ImportK(ctx context.Context, index, field string, colum
}
// Import to node.
if err := c.importNode(ctx, node, buf); err != nil {
if err := c.importNode(ctx, node, index, field, buf); err != nil {
return fmt.Errorf("import node: host=%s, err=%s", node.URI, err)
}
@ -386,9 +386,10 @@ func marshalImportPayloadK(index, field string, bits []pilosa.Bit) ([]byte, erro
}
// importNode sends a pre-marshaled import request to a node.
func (c *InternalClient) importNode(ctx context.Context, node *pilosa.Node, buf []byte) error {
func (c *InternalClient) importNode(ctx context.Context, node *pilosa.Node, index, field string, buf []byte) error {
// Create URL & HTTP request.
u := nodePathToURL(node, "/import")
path := fmt.Sprintf("/index/%s/field/%s/import", index, field)
u := nodePathToURL(node, path)
req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf))
if err != nil {
return errors.Wrap(err, "creating request")
@ -444,7 +445,7 @@ func (c *InternalClient) ImportValue(ctx context.Context, index, field string, s
// Import to each node.
for _, node := range nodes {
if err := c.importValueNode(ctx, node, buf); err != nil {
if err := c.importValueNode(ctx, node, index, field, buf); err != nil {
return fmt.Errorf("import node: host=%s, err=%s", node.URI, err)
}
}
@ -473,9 +474,10 @@ func marshalImportValuePayload(index, field string, shard uint64, vals []pilosa.
}
// importValueNode sends a pre-marshaled import request to a node.
func (c *InternalClient) importValueNode(ctx context.Context, node *pilosa.Node, buf []byte) error {
func (c *InternalClient) importValueNode(ctx context.Context, node *pilosa.Node, index, field string, buf []byte) error {
// Create URL & HTTP request.
u := nodePathToURL(node, "/import-value")
path := fmt.Sprintf("/index/%s/field/%s/import", index, field)
u := nodePathToURL(node, path)
req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf))
if err != nil {
return errors.Wrap(err, "creating request")

110
http/handler.go
View file

@ -195,8 +195,6 @@ func NewRouter(handler *Handler) *mux.Router {
router.PathPrefix("/debug/pprof/").Handler(http.DefaultServeMux).Methods("GET")
router.Handle("/debug/vars", expvar.Handler()).Methods("GET")
router.HandleFunc("/export", handler.handleGetExport).Methods("GET").Name("GetExport")
router.HandleFunc("/import", handler.handlePostImport).Methods("POST")
router.HandleFunc("/import-value", handler.handlePostImportValue).Methods("POST")
router.HandleFunc("/index", handler.handleGetIndexes).Methods("GET")
router.HandleFunc("/index/{index}", handler.handleGetIndex).Methods("GET")
router.HandleFunc("/index/{index}", handler.handlePostIndex).Methods("POST")
@ -204,6 +202,7 @@ func NewRouter(handler *Handler) *mux.Router {
//router.HandleFunc("/index/{index}/field", handler.handleGetFields).Methods("GET") // Not implemented.
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}/query", handler.handlePostQuery).Methods("POST").Name("PostQuery")
router.HandleFunc("/info", handler.handleGetInfo).Methods("GET")
router.HandleFunc("/recalculate-caches", handler.handleRecalculateCaches).Methods("POST")
@ -886,60 +885,24 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) {
http.Error(w, "Not acceptable", http.StatusNotAcceptable)
return
}
indexName := mux.Vars(r)["index"]
fieldName := mux.Vars(r)["field"]
// Read entire body.
body, err := ioutil.ReadAll(r.Body)
// Get index and field type to determine how to handle the
// import data.
field, err := h.API.Field(r.Context(), indexName, fieldName)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
// Marshal into request object.
var req internal.ImportRequest
if err := proto.Unmarshal(body, &req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
if err := h.API.Import(r.Context(), req); err != nil {
switch errors.Cause(err) {
case pilosa.ErrIndexNotFound:
fallthrough
case pilosa.ErrFieldNotFound:
http.Error(w, err.Error(), http.StatusNotFound)
case pilosa.ErrClusterDoesNotOwnShard:
http.Error(w, err.Error(), http.StatusPreconditionFailed)
default:
http.Error(w, err.Error(), http.StatusInternalServerError)
}
return
}
// Marshal response object.
buf, e := proto.Marshal(&internal.ImportResponse{Err: errorString(err)})
if e != nil {
http.Error(w, fmt.Sprintf("marshal import response: %s", err), http.StatusInternalServerError)
return
}
// Write response.
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
}
w.Write(buf)
}
// handlePostImportValue handles /import-value requests.
func (h *Handler) handlePostImportValue(w http.ResponseWriter, r *http.Request) {
// Verify that request is only communicating over protobufs.
if r.Header.Get("Content-Type") != "application/x-protobuf" {
http.Error(w, "Unsupported media type", http.StatusUnsupportedMediaType)
return
} else if r.Header.Get("Accept") != "application/x-protobuf" {
http.Error(w, "Not acceptable", http.StatusNotAcceptable)
return
}
// Read entire body.
body, err := ioutil.ReadAll(r.Body)
if err != nil {
@ -947,38 +910,53 @@ func (h *Handler) handlePostImportValue(w http.ResponseWriter, r *http.Request)
return
}
// Marshal into request object.
var req internal.ImportValueRequest
if err := proto.Unmarshal(body, &req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
if err = h.API.ImportValue(r.Context(), req); err != nil {
switch errors.Cause(err) {
case pilosa.ErrIndexNotFound:
fallthrough
case pilosa.ErrFieldNotFound:
http.Error(w, err.Error(), http.StatusNotFound)
case pilosa.ErrClusterDoesNotOwnShard:
http.Error(w, err.Error(), http.StatusPreconditionFailed)
default:
http.Error(w, err.Error(), http.StatusInternalServerError)
// Unmarshal request based on field type.
if field.Type() == pilosa.FieldTypeInt {
// Field type: Int
// Marshal into request object.
var req internal.ImportValueRequest
if err := proto.Unmarshal(body, &req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
if err := h.API.ImportValue(r.Context(), req); err != nil {
switch errors.Cause(err) {
case pilosa.ErrClusterDoesNotOwnShard:
http.Error(w, err.Error(), http.StatusPreconditionFailed)
default:
http.Error(w, err.Error(), http.StatusInternalServerError)
}
return
}
} else {
// Field type: Set, Time
// Marshal into request object.
var req internal.ImportRequest
if err := proto.Unmarshal(body, &req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
if err := h.API.Import(r.Context(), req); err != nil {
switch errors.Cause(err) {
case pilosa.ErrClusterDoesNotOwnShard:
http.Error(w, err.Error(), http.StatusPreconditionFailed)
default:
http.Error(w, err.Error(), http.StatusInternalServerError)
}
return
}
return
}
// Marshal response object.
buf, e := proto.Marshal(&internal.ImportResponse{Err: errorString(err)})
buf, e := proto.Marshal(&internal.ImportResponse{Err: ""})
if e != nil {
http.Error(w, fmt.Sprintf("marshal import response: %s", err), http.StatusInternalServerError)
http.Error(w, fmt.Sprintf("marshal import response"), http.StatusInternalServerError)
return
}
// Write response.
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
}
w.Write(buf)
}