diff --git a/api.go b/api.go index 27827fcec..f57d159f2 100644 --- a/api.go +++ b/api.go @@ -19,6 +19,7 @@ import ( "encoding/csv" "fmt" "io" + "io/ioutil" "net/http" "reflect" "strconv" @@ -143,6 +144,7 @@ func (api *API) CreateIndex(ctx context.Context, indexName string, options Index return index, nil } +// Index retrieves the named index. func (api *API) Index(ctx context.Context, indexName string) (*Index, error) { index := api.Holder.Index(indexName) if index == nil { @@ -230,9 +232,11 @@ func (api *API) DeleteFrame(ctx context.Context, indexName string, frameName str return nil } +// ExportCSV encodes the fragment designated by the index,frame,view,slice as +// CSV of the form , func (api *API) ExportCSV(ctx context.Context, indexName string, frameName string, viewName string, slice uint64, w io.Writer) error { // Validate that this handler owns the slice. - if !api.Cluster.OwnsFragment(api.LocalID(), indexName, slice) { + if !api.Cluster.OwnsSlice(api.LocalID(), indexName, slice) { api.Logger.Printf("host does not own slice %s-%s slice:%d", api.URI, indexName, slice) return ErrClusterDoesNotOwnSlice } @@ -262,11 +266,20 @@ func (api *API) ExportCSV(ctx context.Context, indexName string, frameName strin return nil } -func (api *API) FragmentNodes(ctx context.Context, indexName string, slice uint64) []*Node { - return api.Cluster.FragmentNodes(indexName, slice) +// SliceNodes returns the node and all replicas which should contain a slice's data. +func (api *API) SliceNodes(ctx context.Context, indexName string, slice uint64) []*Node { + return api.Cluster.SliceNodes(indexName, slice) } -func (api *API) FragmentData(ctx context.Context, indexName string, frameName string, viewName string, slice uint64) (*Fragment, error) { +// WriterTo is an interface for any Object which knows how to serialize itself to an io.Writer +type WriterTo interface { + WriteTo(w io.Writer) (n int64, err error) +} + +// MarshalFragment returns an object which can write the specified fragment's data +// to an io.Writer. The serialized data can be read back into a fragment with +// the UnmarshalFragment API call. +func (api *API) MarshalFragment(ctx context.Context, indexName string, frameName string, viewName string, slice uint64) (WriterTo, error) { // Retrieve fragment from holder. f := api.Holder.Fragment(indexName, frameName, viewName, slice) if f == nil { @@ -275,7 +288,10 @@ func (api *API) FragmentData(ctx context.Context, indexName string, frameName st return f, nil } -func (api *API) WriteFragmentData(ctx context.Context, indexName string, frameName string, viewName string, slice uint64, reader io.ReadCloser) error { +// UnmarshalFragment creates a new fragment (if necessary) and reads data from a +// Reader which was previously written by MarshalFragment to populate the +// fragment's data. +func (api *API) UnmarshalFragment(ctx context.Context, indexName string, frameName string, viewName string, slice uint64, reader io.ReadCloser) error { // Retrieve frame. f := api.Holder.Frame(indexName, frameName) if f == nil { @@ -301,19 +317,38 @@ func (api *API) WriteFragmentData(ctx context.Context, indexName string, frameNa return nil } -func (api *API) FragmentBlockData(ctx context.Context, req internal.BlockDataRequest) (internal.BlockDataResponse, error) { +// FragmentBlockData is an endpoint for internal usage. It is not guaranteed to +// return anything useful. Currently it returns protobuf encoded row and column +// ids from a "block" which is a subdivision of a fragment. +func (api *API) FragmentBlockData(ctx context.Context, body io.Reader) ([]byte, error) { + reqBytes, err := ioutil.ReadAll(body) + if err != nil { + return nil, BadRequestError{errors.Wrap(err, "read body error")} + } + var req internal.BlockDataRequest + if err := proto.Unmarshal(reqBytes, &req); err != nil { + return nil, BadRequestError{errors.Wrap(err, "unmarshal body error")} + } + // Retrieve fragment from holder. f := api.Holder.Fragment(req.Index, req.Frame, req.View, req.Slice) if f == nil { - return internal.BlockDataResponse{}, ErrFragmentNotFound + return nil, ErrFragmentNotFound } - // Read data - var resp internal.BlockDataResponse + var resp = internal.BlockDataResponse{} resp.RowIDs, resp.ColumnIDs = f.BlockData(int(req.Block)) - return resp, nil + + // Encode response. + buf, err := proto.Marshal(&resp) + if err != nil { + return nil, errors.Wrap(err, "merge block response encoding error: %s") + } + + return buf, nil } +// FragmentBlocks returns the checksums and block ids for all blocks in the specified fragment. func (api *API) FragmentBlocks(ctx context.Context, indexName string, frameName string, viewName string, slice uint64) ([]FragmentBlock, error) { // Retrieve fragment from holder. f := api.Holder.Fragment(indexName, frameName, viewName, slice) @@ -326,6 +361,8 @@ func (api *API) FragmentBlocks(ctx context.Context, indexName string, frameName return blocks, nil } +// RestoreFrame reads all the data that this host should have for a given frame +// from replicas in the cluster and restores that data to it. func (api *API) RestoreFrame(ctx context.Context, indexName string, frameName string, host *URI) error { // Create a client for the remote cluster. client := NewInternalHTTPClientFromURI(host, api.RemoteClient) @@ -351,7 +388,7 @@ func (api *API) RestoreFrame(ctx context.Context, indexName string, frameName st // Loop over each slice and import it if this node owns it. for slice := uint64(0); slice <= maxSlices[indexName]; slice++ { // Ignore this slice if we don't own it. - if !api.Cluster.OwnsFragment(api.LocalID(), indexName, slice) { + if !api.Cluster.OwnsSlice(api.LocalID(), indexName, slice) { continue } @@ -399,7 +436,10 @@ func (api *API) Hosts(ctx context.Context) []*Node { return api.Cluster.Nodes } +// CreateInputDefinition is deprecated and will be removed. Do not use it. func (api *API) CreateInputDefinition(ctx context.Context, indexName string, inputDefName string, inputDef InputDefinitionInfo) error { + api.Logger.Printf(`CreateInputDefinition is deprecated and will be removed. +Please open an issue if you need to continue using it.`) // Find index. index := api.Holder.Index(indexName) if index == nil { @@ -430,7 +470,9 @@ func (api *API) CreateInputDefinition(ctx context.Context, indexName string, inp return nil } +// InputDefinition is deprecated and will be removed. func (api *API) InputDefinition(ctx context.Context, indexName string, inputDefName string) (*InputDefinition, error) { + api.Logger.Printf(`InputDefinition is deprecated and will be removed.`) // Find index. index := api.Holder.Index(indexName) if index == nil { @@ -444,7 +486,9 @@ func (api *API) InputDefinition(ctx context.Context, indexName string, inputDefN return inputDef, nil } +// DeleteInputDefinition is deprecated and will be removed. func (api *API) DeleteInputDefinition(ctx context.Context, indexName string, inputDefName string) error { + api.Logger.Printf("DeleteInputDefinition is deprecated and will be removed.") // Find index. index := api.Holder.Index(indexName) if index == nil { @@ -467,7 +511,9 @@ func (api *API) DeleteInputDefinition(ctx context.Context, indexName string, inp return nil } +// WriteInput is deprecated and will be removed. func (api *API) WriteInput(ctx context.Context, indexName string, inputDefName string, reqs []interface{}) error { + api.Logger.Printf("WriteInput is deprecated and will be removed.") // Find index. index := api.Holder.Index(indexName) if index == nil { @@ -499,10 +545,24 @@ func (api *API) RecalculateCaches(ctx context.Context) error { return nil } -func (api *API) PostClusterMessage(ctx context.Context, pb proto.Message) error { +// PostClusterMessage is for internal use. It decodes a protobuf message out of +// the body and forwards it to the BroadcastHandler. +func (api *API) PostClusterMessage(ctx context.Context, reqBody io.Reader) error { + // Read entire body. + body, err := ioutil.ReadAll(reqBody) + if err != nil { + return errors.Wrap(err, "reading body") + } + + // Marshal into request object. + pb, err := UnmarshalMessage(body) + if err != nil { + return errors.Wrap(err, "unmarshaling message") + } + // Forward the error message. if err := api.BroadcastHandler.ReceiveMessage(pb); err != nil { - return err + return errors.Wrap(err, "receiving message") } return nil } @@ -518,6 +578,7 @@ func (api *API) Schema(ctx context.Context) []*IndexInfo { return api.Holder.Schema() } +// CreateField creates a new BSI field in the given index and frame. func (api *API) CreateField(ctx context.Context, indexName string, frameName string, field *Field) error { // Retrieve frame by name. f := api.Holder.Frame(indexName, frameName) @@ -543,6 +604,7 @@ func (api *API) CreateField(ctx context.Context, indexName string, frameName str return err } +// DeleteField deletes the given field. func (api *API) DeleteField(ctx context.Context, indexName string, frameName string, fieldName string) error { // Retrieve frame by name. f := api.Holder.Frame(indexName, frameName) @@ -568,6 +630,7 @@ func (api *API) DeleteField(ctx context.Context, indexName string, frameName str return err } +// Fields returns the fields in the given frame. func (api *API) Fields(ctx context.Context, indexName string, frameName string) ([]*Field, error) { index := api.Holder.index(indexName) if index == nil { @@ -582,6 +645,7 @@ func (api *API) Fields(ctx context.Context, indexName string, frameName string) return frame.GetFields() } +// Views returns the views in the given frame. func (api *API) Views(ctx context.Context, indexName string, frameName string) ([]*View, error) { // Retrieve views. f := api.Holder.Frame(indexName, frameName) @@ -594,6 +658,7 @@ func (api *API) Views(ctx context.Context, indexName string, frameName string) ( return views, nil } +// DeleteView removes the given view. func (api *API) DeleteView(ctx context.Context, indexName string, frameName string, viewName string) error { // Retrieve frame. f := api.Holder.Frame(indexName, frameName) @@ -623,6 +688,7 @@ func (api *API) DeleteView(ctx context.Context, indexName string, frameName stri return err } +// IndexAttrDiff func (api *API) IndexAttrDiff(ctx context.Context, indexName string, blocks []AttrBlock) (map[uint64]map[string]interface{}, error) { // Retrieve index from holder. index := api.Holder.Index(indexName) @@ -723,6 +789,7 @@ func (api *API) ImportValue(ctx context.Context, req internal.ImportValueRequest return err } +// ModifyIndexTimeQuantum changes the default time quantum on the given index. func (api *API) ModifyIndexTimeQuantum(ctx context.Context, indexName string, timeQuantum TimeQuantum) error { // Retrieve index by name. index := api.Holder.Index(indexName) @@ -734,6 +801,8 @@ func (api *API) ModifyIndexTimeQuantum(ctx context.Context, indexName string, ti return index.SetTimeQuantum(timeQuantum) } +// ModifyFrameTimeQuantum changes the time quantum on the given frame. TODO: +// what happens if there is already data in the frame? func (api *API) ModifyFrameTimeQuantum(ctx context.Context, indexName string, frameName string, timeQuantum TimeQuantum) error { // Retrieve index by name. frame := api.Holder.Frame(indexName, frameName) @@ -745,14 +814,19 @@ func (api *API) ModifyFrameTimeQuantum(ctx context.Context, indexName string, fr return frame.SetTimeQuantum(timeQuantum) } +// MaxSlices returns the maximum slice number for each index in a map. func (api *API) MaxSlices(ctx context.Context) map[string]uint64 { return api.Holder.MaxSlices() } +// MaxInverseSlices returns the maximum inverse slice number for each index in a +// map. func (api *API) MaxInverseSlices(ctx context.Context) map[string]uint64 { return api.Holder.MaxInverseSlices() } +// StatsWithTags returns an instance of whatever implementation of StatsClient +// pilosa is using with the given tags. func (api *API) StatsWithTags(tags []string) StatsClient { if api.Holder == nil || api.Cluster == nil { return nil @@ -771,7 +845,7 @@ func (api *API) LongQueryTime() time.Duration { func (api *API) indexFrame(indexName string, frameName string, slice uint64) (*Index, *Frame, error) { // Validate that this handler owns the slice. - if !api.Cluster.OwnsFragment(api.LocalID(), indexName, slice) { + if !api.Cluster.OwnsSlice(api.LocalID(), indexName, slice) { api.Logger.Printf("host does not own slice %s-%s slice:%d", api.URI, indexName, slice) return nil, nil, ErrClusterDoesNotOwnSlice } @@ -793,7 +867,7 @@ func (api *API) indexFrame(indexName string, frameName string, slice uint64) (*I return index, frame, nil } -// inputJSONDataParser validates input json file and executes SetBit. +// inputJSONDataParser validates input json file and executes SetBit. Deprecated - remove with input definition stuff. func (api *API) inputJSONDataParser(req map[string]interface{}, index *Index, name string) (map[string][]*Bit, error) { inputDef, err := index.InputDefinition(name) if err != nil { @@ -903,6 +977,7 @@ func (api *API) RemoveNode(id string) (*Node, error) { return removeNode, nil } +// ResizeAbort stops the current resize job. func (api *API) ResizeAbort() error { if !api.Cluster.IsCoordinator() { return ErrNodeNotCoordinator diff --git a/cluster.go b/cluster.go index c490c58cb..abf753892 100644 --- a/cluster.go +++ b/cluster.go @@ -654,7 +654,7 @@ func (c *Cluster) fragsByHost(idx *Index) fragsByHost { func (c *Cluster) fragCombos(idx string, maxSlice uint64, frameViews viewsByFrame) fragsByHost { t := make(fragsByHost) for i := uint64(0); i <= maxSlice; i++ { - nodes := c.FragmentNodes(idx, i) + nodes := c.SliceNodes(idx, i) for _, n := range nodes { // for each frame/view combination: for frame, views := range frameViews { @@ -807,14 +807,14 @@ func (c *Cluster) Partition(index string, slice uint64) int { return int(h.Sum64() % uint64(c.PartitionN)) } -// FragmentNodes returns a list of nodes that own a fragment. -func (c *Cluster) FragmentNodes(index string, slice uint64) []*Node { +// SliceNodes returns a list of nodes that own a fragment. +func (c *Cluster) SliceNodes(index string, slice uint64) []*Node { return c.PartitionNodes(c.Partition(index, slice)) } -// OwnsFragment returns true if a host owns a fragment. -func (c *Cluster) OwnsFragment(nodeID string, index string, slice uint64) bool { - return Nodes(c.FragmentNodes(index, slice)).ContainsID(nodeID) +// OwnsSlice returns true if a host owns a fragment. +func (c *Cluster) OwnsSlice(nodeID string, index string, slice uint64) bool { + return Nodes(c.SliceNodes(index, slice)).ContainsID(nodeID) } // PartitionNodes returns a list of nodes that own a partition. diff --git a/executor.go b/executor.go index 2154b60ff..c2cd860c4 100644 --- a/executor.go +++ b/executor.go @@ -941,7 +941,7 @@ func (e *Executor) executeClearBit(ctx context.Context, index string, c *pql.Cal func (e *Executor) executeClearBitView(ctx context.Context, index string, c *pql.Call, f *Frame, view string, colID, rowID uint64, opt *ExecOptions) (bool, error) { slice := colID / SliceWidth ret := false - for _, node := range e.Cluster.FragmentNodes(index, slice) { + for _, node := range e.Cluster.SliceNodes(index, slice) { // Update locally if host matches. if node.ID == e.Node.ID { val, err := f.ClearBit(view, rowID, colID, nil) @@ -1042,7 +1042,7 @@ func (e *Executor) executeSetBitView(ctx context.Context, index string, c *pql.C slice := colID / SliceWidth ret := false - for _, node := range e.Cluster.FragmentNodes(index, slice) { + for _, node := range e.Cluster.SliceNodes(index, slice) { // Update locally if host matches. if node.ID == e.Node.ID { val, err := f.SetBit(view, rowID, colID, timestamp) @@ -1385,7 +1385,7 @@ func (e *Executor) slicesByNode(nodes []*Node, index string, slices []uint64) (m loop: for _, slice := range slices { - for _, node := range e.Cluster.FragmentNodes(index, slice) { + for _, node := range e.Cluster.SliceNodes(index, slice) { if Nodes(nodes).Contains(node) { m[node] = append(m[node], slice) continue loop diff --git a/fragment.go b/fragment.go index 438a80ae7..c44a07380 100644 --- a/fragment.go +++ b/fragment.go @@ -1702,7 +1702,7 @@ func (s *FragmentSyncer) isClosing() bool { // then merges any blocks which have differences. func (s *FragmentSyncer) SyncFragment() error { // Determine replica set. - nodes := s.Cluster.FragmentNodes(s.Fragment.Index(), s.Fragment.Slice()) + nodes := s.Cluster.SliceNodes(s.Fragment.Index(), s.Fragment.Slice()) if len(nodes) == 1 { return nil } @@ -1784,7 +1784,7 @@ func (s *FragmentSyncer) syncBlock(id int) error { // Read pairs from each remote block. var pairSets []PairSet var clients []InternalClient - for _, node := range s.Cluster.FragmentNodes(f.Index(), f.Slice()) { + for _, node := range s.Cluster.SliceNodes(f.Index(), f.Slice()) { if s.Node.ID == node.ID { continue } diff --git a/handler.go b/handler.go index 4ce8560b5..33d2e5846 100644 --- a/handler.go +++ b/handler.go @@ -1169,7 +1169,7 @@ func (h *Handler) handleGetFragmentNodes(w http.ResponseWriter, r *http.Request) } // Retrieve fragment owner nodes. - nodes := h.API.FragmentNodes(r.Context(), index, slice) + nodes := h.API.SliceNodes(r.Context(), index, slice) // Write to response. if err := json.NewEncoder(w).Encode(nodes); err != nil { @@ -1188,7 +1188,7 @@ func (h *Handler) handleGetFragmentData(w http.ResponseWriter, r *http.Request) } // Retrieve fragment from holder. - f, err := h.API.FragmentData(r.Context(), q.Get("index"), q.Get("frame"), q.Get("view"), slice) + f, err := h.API.MarshalFragment(r.Context(), q.Get("index"), q.Get("frame"), q.Get("view"), slice) if err != nil { http.Error(w, err.Error(), http.StatusNotFound) return @@ -1210,7 +1210,7 @@ func (h *Handler) handlePostFragmentData(w http.ResponseWriter, r *http.Request) return } - if err = h.API.WriteFragmentData(r.Context(), q.Get("index"), q.Get("frame"), q.Get("view"), slice, r.Body); err != nil { + if err = h.API.UnmarshalFragment(r.Context(), q.Get("index"), q.Get("frame"), q.Get("view"), slice, r.Body); err != nil { if err == ErrFrameNotFound { http.Error(w, ErrFrameNotFound.Error(), http.StatusNotFound) } else { @@ -1221,19 +1221,11 @@ func (h *Handler) handlePostFragmentData(w http.ResponseWriter, r *http.Request) // handleGetFragmentBlockData handles GET /fragment/block/data requests. func (h *Handler) handleGetFragmentBlockData(w http.ResponseWriter, r *http.Request) { - // Read request object. - var req internal.BlockDataRequest - if body, err := ioutil.ReadAll(r.Body); err != nil { - http.Error(w, "ready body error", http.StatusBadRequest) - return - } else if err := proto.Unmarshal(body, &req); err != nil { - http.Error(w, "unmarshal body error", http.StatusBadRequest) - return - } - - resp, err := h.API.FragmentBlockData(r.Context(), req) + buf, err := h.API.FragmentBlockData(r.Context(), r.Body) if err != nil { - if err == ErrFragmentNotFound { + if _, ok := err.(BadRequestError); ok { + http.Error(w, err.Error(), http.StatusBadRequest) + } else if err == ErrFragmentNotFound { http.Error(w, err.Error(), http.StatusNotFound) } else { http.Error(w, err.Error(), http.StatusInternalServerError) @@ -1241,13 +1233,6 @@ func (h *Handler) handleGetFragmentBlockData(w http.ResponseWriter, r *http.Requ return } - // Encode response. - buf, err := proto.Marshal(&resp) - if err != nil { - h.Logger.Printf("merge block response encoding error: %s", err) - return - } - // Write response. w.Header().Set("Content-Type", "application/protobuf") w.Header().Set("Content-Length", strconv.Itoa(len(buf))) @@ -1742,23 +1727,10 @@ func (h *Handler) handlePostClusterMessage(w http.ResponseWriter, r *http.Reques return } - // Read entire body. - body, err := ioutil.ReadAll(r.Body) + err := h.API.PostClusterMessage(r.Context(), r.Body) if err != nil { + // TODO this was the previous behavior, but perhaps not everything is a bad request http.Error(w, err.Error(), http.StatusBadRequest) - return - } - - // Marshal into request object. - pb, err := UnmarshalMessage(body) - if err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return - } - - if err := h.API.PostClusterMessage(r.Context(), pb); err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return } if err := json.NewEncoder(w).Encode(defaultClusterMessageResponse{}); err != nil { diff --git a/holder.go b/holder.go index 22c53c7a4..570433ce4 100644 --- a/holder.go +++ b/holder.go @@ -614,7 +614,7 @@ func (s *HolderSyncer) SyncHolder() error { for slice := uint64(0); slice <= s.Holder.Index(di.Name).MaxSlice(); slice++ { // Ignore slices that this host doesn't own. - if !s.Cluster.OwnsFragment(s.Node.ID, di.Name, slice) { + if !s.Cluster.OwnsSlice(s.Node.ID, di.Name, slice) { continue } diff --git a/pilosa.go b/pilosa.go index 4b28daccb..ffb836ae8 100644 --- a/pilosa.go +++ b/pilosa.go @@ -82,6 +82,13 @@ var ( ErrResizeNotRunning = errors.New("no resize job currently running") ) +// BadRequestError wraps an error value to signify that a request could not be +// read, decoded, or parsed such that in an HTTP scenario, http.StatusBadRequest +// would be returned. +type BadRequestError struct { + error +} + // Regular expression to validate index and frame names. var nameRegexp = regexp.MustCompile(`^[a-z][a-z0-9_-]{0,63}$`) diff --git a/test/cluster.go b/test/cluster.go index 8362ca964..308a7e69c 100644 --- a/test/cluster.go +++ b/test/cluster.go @@ -129,7 +129,7 @@ func (t *TestCluster) SetBit(index, frame, view string, rowID, colID uint64, x * // Determine which node should receive the SetBit. c0 := t.Clusters[0] // use the first node's cluster to determine slice location. slice := colID / pilosa.SliceWidth - nodes := c0.FragmentNodes(index, slice) + nodes := c0.SliceNodes(index, slice) for _, node := range nodes { c := t.clusterByID(node.ID) @@ -153,7 +153,7 @@ func (t *TestCluster) SetFieldValue(index, frame string, columnID uint64, name s // Determine which node should receive the SetFieldValue. c0 := t.Clusters[0] // use the first node's cluster to determine slice location. slice := columnID / pilosa.SliceWidth - nodes := c0.FragmentNodes(index, slice) + nodes := c0.SliceNodes(index, slice) for _, node := range nodes { c := t.clusterByID(node.ID)