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

Merge branch 'master' into wrap-errors-index

Browse source
This commit is contained in:
alanbernstein 2018-05-10 18:57:44 -05:00 • committed by GitHub
commit 64ebae3a8a
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
40 changed files with 568 additions and 461 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

150
api.go
View file

@ -46,7 +46,6 @@ type API struct {
BroadcastHandler BroadcastHandler
StatusHandler StatusHandler
Cluster *Cluster
URI URI
RemoteClient *http.Client
Logger Logger
}
@ -91,14 +90,14 @@ func (api *API) validate(f apiMethod) error {
// Query parses a PQL query out of the request and executes it.
func (api *API) Query(ctx context.Context, req *QueryRequest) (QueryResponse, error) {
if err := api.validate(apiQuery); err != nil {
return QueryResponse{}, errors.Wrap(err, "validate api method")
return QueryResponse{}, errors.Wrap(err, "validating api method")
}
resp := QueryResponse{}
q, err := pql.NewParser(strings.NewReader(req.Query)).Parse()
if err != nil {
return resp, err
return resp, errors.Wrap(err, "parsing")
}
execOpts := &ExecOptions{
Remote: req.Remote,
@ -107,7 +106,7 @@ func (api *API) Query(ctx context.Context, req *QueryRequest) (QueryResponse, er
}
results, err := api.Executor.Execute(ctx, req.Index, q, req.Slices, execOpts)
if err != nil {
return resp, err
return resp, errors.Wrap(err, "executing")
}
resp.Results = results
@ -126,7 +125,7 @@ func (api *API) Query(ctx context.Context, req *QueryRequest) (QueryResponse, er
// Retrieve column attributes across all calls.
columnAttrSets, err := api.readColumnAttrSets(api.Holder.Index(req.Index), columnIDs)
if err != nil {
return resp, err
return resp, errors.Wrap(err, "reading column attrs")
}
resp.ColumnAttrSets = columnAttrSets
}
@ -144,7 +143,7 @@ func (api *API) readColumnAttrSets(index *Index, ids []uint64) ([]*ColumnAttrSet
// Read attributes for column. Skip column if empty.
attrs, err := index.ColumnAttrStore().Attrs(id)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "getting attrs")
} else if len(attrs) == 0 {
continue
}
@ -159,13 +158,13 @@ func (api *API) readColumnAttrSets(index *Index, ids []uint64) ([]*ColumnAttrSet
// CreateIndex makes a new Pilosa index.
func (api *API) CreateIndex(ctx context.Context, indexName string, options IndexOptions) (*Index, error) {
if err := api.validate(apiCreateIndex); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
// Create index.
index, err := api.Holder.CreateIndex(indexName, options)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "creating index")
}
// Send the create index message to all nodes.
err = api.Broadcaster.SendSync(
@ -175,7 +174,7 @@ func (api *API) CreateIndex(ctx context.Context, indexName string, options Index
})
if err != nil {
api.Logger.Printf("problem sending CreateIndex message: %s", err)
return nil, err
return nil, errors.Wrap(err, "sending CreateIndex message")
}
api.Holder.Stats.Count("createIndex", 1, 1.0)
return index, nil
@ -184,7 +183,7 @@ func (api *API) CreateIndex(ctx context.Context, indexName string, options Index
// Index retrieves the named index.
func (api *API) Index(ctx context.Context, indexName string) (*Index, error) {
if err := api.validate(apiIndex); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
index := api.Holder.Index(indexName)
@ -198,13 +197,13 @@ func (api *API) Index(ctx context.Context, indexName string) (*Index, error) {
// nothing and returns no error.
func (api *API) DeleteIndex(ctx context.Context, indexName string) error {
if err := api.validate(apiDeleteIndex); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
// Delete index from the holder.
err := api.Holder.DeleteIndex(indexName)
if err != nil {
return err
return errors.Wrap(err, "deleting index")
}
// Send the delete index message to all nodes.
err = api.Broadcaster.SendSync(
@ -213,7 +212,7 @@ func (api *API) DeleteIndex(ctx context.Context, indexName string) error {
})
if err != nil {
api.Logger.Printf("problem sending DeleteIndex message: %s", err)
return err
return errors.Wrap(err, "sending DeleteIndex message")
}
api.Holder.Stats.Count("deleteIndex", 1, 1.0)
return nil
@ -222,7 +221,7 @@ func (api *API) DeleteIndex(ctx context.Context, indexName string) error {
// CreateFrame makes the named frame in the named index with the given options.
func (api *API) CreateFrame(ctx context.Context, indexName string, frameName string, options FrameOptions) (*Frame, error) {
if err := api.validate(apiCreateFrame); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
// Find index.
@ -234,7 +233,7 @@ func (api *API) CreateFrame(ctx context.Context, indexName string, frameName str
// Create frame.
frame, err := index.CreateFrame(frameName, options)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "creating frame")
}
// Send the create frame message to all nodes.
@ -246,7 +245,7 @@ func (api *API) CreateFrame(ctx context.Context, indexName string, frameName str
})
if err != nil {
api.Logger.Printf("problem sending CreateFrame message: %s", err)
return nil, err
return nil, errors.Wrap(err, "sending CreateFrame message")
}
api.Holder.Stats.CountWithCustomTags("createFrame", 1, 1.0, []string{fmt.Sprintf("index:%s", indexName)})
return frame, nil
@ -257,7 +256,7 @@ func (api *API) CreateFrame(ctx context.Context, indexName string, frameName str
// action is taken.
func (api *API) DeleteFrame(ctx context.Context, indexName string, frameName string) error {
if err := api.validate(apiDeleteFrame); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
// Find index.
@ -268,7 +267,7 @@ func (api *API) DeleteFrame(ctx context.Context, indexName string, frameName str
// Delete frame from the index.
if err := index.DeleteFrame(frameName); err != nil {
return err
return errors.Wrap(err, "deleting frame")
}
// Send the delete frame message to all nodes.
@ -279,7 +278,7 @@ func (api *API) DeleteFrame(ctx context.Context, indexName string, frameName str
})
if err != nil {
api.Logger.Printf("problem sending DeleteFrame message: %s", err)
return err
return errors.Wrap(err, "sending DeleteFrame message")
}
api.Holder.Stats.CountWithCustomTags("deleteFrame", 1, 1.0, []string{fmt.Sprintf("index:%s", indexName)})
return nil
@ -289,12 +288,12 @@ func (api *API) DeleteFrame(ctx context.Context, indexName string, frameName str
// CSV of the form <row>,<col>
func (api *API) ExportCSV(ctx context.Context, indexName string, frameName string, viewName string, slice uint64, w io.Writer) error {
if err := api.validate(apiExportCSV); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
// Validate that this handler owns the 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)
api.Logger.Printf("node %s does not own slice %d of index %s", api.LocalID(), slice, indexName)
return ErrClusterDoesNotOwnSlice
}
@ -314,7 +313,7 @@ func (api *API) ExportCSV(ctx context.Context, indexName string, frameName strin
strconv.FormatUint(columnID, 10),
})
}); err != nil {
return err
return errors.Wrap(err, "writing CSV")
}
// Ensure data is flushed.
@ -326,7 +325,7 @@ func (api *API) ExportCSV(ctx context.Context, indexName string, frameName strin
// 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, error) {
if err := api.validate(apiSliceNodes); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
return api.Cluster.SliceNodes(indexName, slice), nil
@ -337,7 +336,7 @@ func (api *API) SliceNodes(ctx context.Context, indexName string, slice uint64)
// the UnmarshalFragment API call.
func (api *API) MarshalFragment(ctx context.Context, indexName string, frameName string, viewName string, slice uint64) (io.WriterTo, error) {
if err := api.validate(apiMarshalFragment); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
// Retrieve fragment from holder.
@ -353,7 +352,7 @@ func (api *API) MarshalFragment(ctx context.Context, indexName string, frameName
// fragment's data.
func (api *API) UnmarshalFragment(ctx context.Context, indexName string, frameName string, viewName string, slice uint64, reader io.ReadCloser) error {
if err := api.validate(apiUnmarshalFragment); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
// Retrieve frame.
@ -365,18 +364,18 @@ func (api *API) UnmarshalFragment(ctx context.Context, indexName string, frameNa
// Retrieve view.
view, err := f.CreateViewIfNotExists(viewName)
if err != nil {
return err
return errors.Wrap(err, "creating view")
}
// Retrieve fragment from frame.
frag, err := view.CreateFragmentIfNotExists(slice)
if err != nil {
return err
return errors.Wrap(err, "creating fragment")
}
// Read fragment in from request body.
if _, err := frag.ReadFrom(reader); err != nil {
return err
return errors.Wrap(err, "reading fragment")
}
return nil
}
@ -386,7 +385,7 @@ func (api *API) UnmarshalFragment(ctx context.Context, indexName string, frameNa
// ids from a "block" which is a subdivision of a fragment.
func (api *API) FragmentBlockData(ctx context.Context, body io.Reader) ([]byte, error) {
if err := api.validate(apiFragmentBlockData); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
reqBytes, err := ioutil.ReadAll(body)
@ -419,7 +418,7 @@ func (api *API) FragmentBlockData(ctx context.Context, body io.Reader) ([]byte,
// 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) {
if err := api.validate(apiFragmentBlocks); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
// Retrieve fragment from holder.
@ -437,7 +436,7 @@ func (api *API) FragmentBlocks(ctx context.Context, indexName string, frameName
// 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 {
if err := api.validate(apiRestoreFrame); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
// Create a client for the remote cluster.
@ -446,7 +445,7 @@ func (api *API) RestoreFrame(ctx context.Context, indexName string, frameName st
// Determine the maximum number of slices.
maxSlices, err := client.MaxSliceByIndex(ctx)
if err != nil {
return err
return errors.Wrap(err, "getting max slice")
}
// Retrieve frame.
@ -458,7 +457,7 @@ func (api *API) RestoreFrame(ctx context.Context, indexName string, frameName st
// Retrieve list of all views.
views, err := client.FrameViews(ctx, indexName, frameName)
if err != nil {
return err
return errors.Wrap(err, "getting views")
}
// Loop over each slice and import it if this node owns it.
@ -473,19 +472,19 @@ func (api *API) RestoreFrame(ctx context.Context, indexName string, frameName st
// Create view.
v, err := f.CreateViewIfNotExists(view)
if err != nil {
return err
return errors.Wrap(err, "creating view")
}
// Otherwise retrieve the local fragment.
frag, err := v.CreateFragmentIfNotExists(slice)
if err != nil {
return err
return errors.Wrap(err, "creating fragment")
}
// Stream backup from remote node.
rd, err := client.BackupSlice(ctx, indexName, frameName, view, slice)
if err != nil {
return err
return errors.Wrap(err, "getting backup")
} else if rd == nil {
continue // slice doesn't exist
}
@ -494,7 +493,7 @@ func (api *API) RestoreFrame(ctx context.Context, indexName string, frameName st
if err := func() error {
defer rd.Close()
if _, err := frag.ReadFrom(rd); err != nil {
return err
return errors.Wrap(err, "reading fragment")
}
return nil
}(); err != nil {
@ -515,7 +514,7 @@ func (api *API) Hosts(ctx context.Context) []*Node {
// 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 {
if err := api.validate(apiCreateInputDefinition); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
api.Logger.Printf(`CreateInputDefinition is deprecated and will be removed.
@ -553,7 +552,7 @@ Please open an issue if you need to continue using it.`)
// InputDefinition is deprecated and will be removed.
func (api *API) InputDefinition(ctx context.Context, indexName string, inputDefName string) (*InputDefinition, error) {
if err := api.validate(apiInputDefinition); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
api.Logger.Printf(`InputDefinition is deprecated and will be removed.`)
@ -573,7 +572,7 @@ func (api *API) InputDefinition(ctx context.Context, indexName string, inputDefN
// DeleteInputDefinition is deprecated and will be removed.
func (api *API) DeleteInputDefinition(ctx context.Context, indexName string, inputDefName string) error {
if err := api.validate(apiDeleteInputDefinition); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
api.Logger.Printf("DeleteInputDefinition is deprecated and will be removed.")
@ -602,7 +601,7 @@ func (api *API) DeleteInputDefinition(ctx context.Context, indexName string, inp
// WriteInput is deprecated and will be removed.
func (api *API) WriteInput(ctx context.Context, indexName string, inputDefName string, reqs []interface{}) error {
if err := api.validate(apiWriteInput); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
api.Logger.Printf("WriteInput is deprecated and will be removed.")
@ -630,7 +629,7 @@ func (api *API) WriteInput(ctx context.Context, indexName string, inputDefName s
// RecalculateCaches forces all TopN caches to be updated. Used mainly for integration tests.
func (api *API) RecalculateCaches(ctx context.Context) error {
if err := api.validate(apiRecalculateCaches); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
err := api.Broadcaster.SendSync(&internal.RecalculateCaches{})
@ -645,7 +644,7 @@ func (api *API) RecalculateCaches(ctx context.Context) error {
// the body and forwards it to the BroadcastHandler.
func (api *API) ClusterMessage(ctx context.Context, reqBody io.Reader) error {
if err := api.validate(apiClusterMessage); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
// Read entire body.
@ -681,7 +680,7 @@ func (api *API) Schema(ctx context.Context) []*IndexInfo {
// 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 {
if err := api.validate(apiCreateField); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
// Retrieve frame by name.
@ -692,7 +691,7 @@ func (api *API) CreateField(ctx context.Context, indexName string, frameName str
// Create new field.
if err := f.CreateField(field); err != nil {
return err
return errors.Wrap(err, "creating field")
}
// Send the create field message to all nodes.
@ -705,13 +704,13 @@ func (api *API) CreateField(ctx context.Context, indexName string, frameName str
if err != nil {
api.Logger.Printf("problem sending CreateField message: %s", err)
}
return err
return errors.Wrap(err, "sending CreateField message")
}
// DeleteField deletes the given field.
func (api *API) DeleteField(ctx context.Context, indexName string, frameName string, fieldName string) error {
if err := api.validate(apiDeleteField); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
// Retrieve frame by name.
@ -722,7 +721,7 @@ func (api *API) DeleteField(ctx context.Context, indexName string, frameName str
// Delete field.
if err := f.DeleteField(fieldName); err != nil {
return err
return errors.Wrap(err, "deleting field")
}
// Send the delete field message to all nodes.
@ -735,13 +734,13 @@ func (api *API) DeleteField(ctx context.Context, indexName string, frameName str
if err != nil {
api.Logger.Printf("problem sending DeleteField message: %s", err)
}
return err
return errors.Wrap(err, "sending DeleteField message")
}
// Fields returns the fields in the given frame.
func (api *API) Fields(ctx context.Context, indexName string, frameName string) ([]*Field, error) {
if err := api.validate(apiFields); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
index := api.Holder.index(indexName)
@ -760,7 +759,7 @@ func (api *API) Fields(ctx context.Context, indexName string, frameName string)
// Views returns the views in the given frame.
func (api *API) Views(ctx context.Context, indexName string, frameName string) ([]*View, error) {
if err := api.validate(apiViews); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
// Retrieve views.
@ -777,7 +776,7 @@ func (api *API) Views(ctx context.Context, indexName string, frameName string) (
// DeleteView removes the given view.
func (api *API) DeleteView(ctx context.Context, indexName string, frameName string, viewName string) error {
if err := api.validate(apiDeleteView); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
// Retrieve frame.
@ -790,7 +789,7 @@ func (api *API) DeleteView(ctx context.Context, indexName string, frameName stri
if err := f.DeleteView(viewName); err != nil {
// Ignore this error because views do not exist on all nodes due to slice distribution.
if err != ErrInvalidView {
return err
return errors.Wrap(err, "deleting view")
}
}
@ -805,13 +804,13 @@ func (api *API) DeleteView(ctx context.Context, indexName string, frameName stri
api.Logger.Printf("problem sending DeleteView message: %s", err)
}
return err
return errors.Wrap(err, "sending DeleteView message")
}
// IndexAttrDiff
func (api *API) IndexAttrDiff(ctx context.Context, indexName string, blocks []AttrBlock) (map[uint64]map[string]interface{}, error) {
if err := api.validate(apiIndexAttrDiff); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
// Retrieve index from holder.
@ -823,7 +822,7 @@ func (api *API) IndexAttrDiff(ctx context.Context, indexName string, blocks []At
// Retrieve local blocks.
localBlocks, err := index.ColumnAttrStore().Blocks()
if err != nil {
return nil, err
return nil, errors.Wrap(err, "getting blocks")
}
// Read all attributes from all mismatched blocks.
@ -832,7 +831,7 @@ func (api *API) IndexAttrDiff(ctx context.Context, indexName string, blocks []At
// Retrieve block data.
m, err := index.ColumnAttrStore().BlockData(blockID)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "getting block")
}
// Copy to index-wide struct.
@ -845,7 +844,7 @@ func (api *API) IndexAttrDiff(ctx context.Context, indexName string, blocks []At
func (api *API) FrameAttrDiff(ctx context.Context, indexName string, frameName string, blocks []AttrBlock) (map[uint64]map[string]interface{}, error) {
if err := api.validate(apiFrameAttrDiff); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
// Retrieve index from holder.
@ -857,7 +856,7 @@ func (api *API) FrameAttrDiff(ctx context.Context, indexName string, frameName s
// Retrieve local blocks.
localBlocks, err := f.RowAttrStore().Blocks()
if err != nil {
return nil, err
return nil, errors.Wrap(err, "getting blocks")
}
// Read all attributes from all mismatched blocks.
@ -866,7 +865,7 @@ func (api *API) FrameAttrDiff(ctx context.Context, indexName string, frameName s
// Retrieve block data.
m, err := f.RowAttrStore().BlockData(blockID)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "getting block")
}
// Copy to index-wide struct.
@ -880,12 +879,12 @@ func (api *API) FrameAttrDiff(ctx context.Context, indexName string, frameName s
// Import bulk imports data into a particular index,frame,slice.
func (api *API) Import(ctx context.Context, req internal.ImportRequest) error {
if err := api.validate(apiImport); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
_, frame, err := api.indexFrame(req.Index, req.Frame, req.Slice)
if err != nil {
return err
return errors.Wrap(err, "getting frame")
}
// Convert timestamps to time.Time.
@ -903,18 +902,18 @@ func (api *API) Import(ctx context.Context, req internal.ImportRequest) error {
if err != nil {
api.Logger.Printf("import error: index=%s, frame=%s, slice=%d, bits=%d, err=%s", req.Index, req.Frame, req.Slice, len(req.ColumnIDs), err)
}
return err
return errors.Wrap(err, "importing")
}
// ImportValue bulk imports values into a particular field.
func (api *API) ImportValue(ctx context.Context, req internal.ImportValueRequest) error {
if err := api.validate(apiImportValue); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
_, frame, err := api.indexFrame(req.Index, req.Frame, req.Slice)
if err != nil {
return err
return errors.Wrap(err, "getting frame")
}
// Import into fragment.
@ -922,7 +921,7 @@ func (api *API) ImportValue(ctx context.Context, req internal.ImportValueRequest
if err != nil {
api.Logger.Printf("import error: index=%s, frame=%s, slice=%d, field=%s, bits=%d, err=%s", req.Index, req.Frame, req.Slice, req.Field, len(req.ColumnIDs), err)
}
return err
return errors.Wrap(err, "importing")
}
// MaxSlices returns the maximum slice number for each index in a map.
@ -957,7 +956,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.OwnsSlice(api.LocalID(), indexName, slice) {
api.Logger.Printf("host does not own slice %s-%s slice:%d", api.URI, indexName, slice)
api.Logger.Printf("node %s does not own slice %d of index %s", api.LocalID(), slice, indexName)
return nil, nil, ErrClusterDoesNotOwnSlice
}
@ -1050,7 +1049,7 @@ func (api *API) inputJSONDataParser(req map[string]interface{}, index *Index, na
// SetCoordinator makes a new Node the cluster coordinator.
func (api *API) SetCoordinator(ctx context.Context, id string) (oldNode, newNode *Node, err error) {
if err := api.validate(apiSetCoordinator); err != nil {
return nil, nil, errors.Wrap(err, "validate api method")
return nil, nil, errors.Wrap(err, "validating api method")
}
oldNode = api.Cluster.NodeByID(api.Cluster.Coordinator)
@ -1080,7 +1079,7 @@ func (api *API) SetCoordinator(ctx context.Context, id string) (oldNode, newNode
// removing the given node.
func (api *API) RemoveNode(id string) (*Node, error) {
if err := api.validate(apiRemoveNode); err != nil {
return nil, errors.Wrap(err, "validate api method")
return nil, errors.Wrap(err, "validating api method")
}
removeNode := api.Cluster.nodeByID(id)
@ -1099,7 +1098,7 @@ func (api *API) RemoveNode(id string) (*Node, error) {
// ResizeAbort stops the current resize job.
func (api *API) ResizeAbort() error {
if err := api.validate(apiResizeAbort); err != nil {
return errors.Wrap(err, "validate api method")
return errors.Wrap(err, "validating api method")
}
err := api.Cluster.CompleteCurrentJob(ResizeJobStateAborted)
@ -1118,6 +1117,17 @@ func (api *API) Version() string {
return strings.TrimPrefix(Version, "v")
}
// Info returns information about this server instance
func (api *API) Info() ServerInfo {
return ServerInfo{
SliceWidth: SliceWidth,
}
}
type ServerInfo struct {
SliceWidth uint64 `json:"sliceWidth"`
}
type apiMethod int
// API validation constants.

21
boltdb/attrstore.go
View file

@ -27,6 +27,7 @@ import (
"github.com/boltdb/bolt"
"github.com/pilosa/pilosa"
"github.com/pkg/errors"
)
// AttrBlockSize is the size of attribute blocks for anti-entropy.
@ -93,7 +94,7 @@ func (s *AttrStore) Open() error {
// Open storage.
db, err := bolt.Open(s.path, 0666, &bolt.Options{Timeout: 1 * time.Second})
if err != nil {
return err
return errors.Wrap(err, "opening storage")
}
s.db = db
@ -104,7 +105,7 @@ func (s *AttrStore) Open() error {
}
return nil
}); err != nil {
return err
return errors.Wrap(err, "initializing")
}
return nil
@ -136,7 +137,7 @@ func (s *AttrStore) Attrs(id uint64) (m map[string]interface{}, err error) {
}
return nil
}); err != nil {
return nil, err
return nil, errors.Wrap(err, "finding attributes")
}
// Add to cache.
@ -154,7 +155,7 @@ func (s *AttrStore) SetAttrs(id uint64, m map[string]interface{}) error {
// Check if the attributes already exist under a read-only lock.
if attr, err := s.Attrs(id); err != nil {
return err
return errors.Wrap(err, "checking attrs")
} else if attr != nil && mapContains(attr, m) {
return nil
}
@ -173,7 +174,7 @@ func (s *AttrStore) SetAttrs(id uint64, m map[string]interface{}) error {
return nil
}); err != nil {
return err
return errors.Wrap(err, "updating store")
}
// Swap attributes map in cache.
@ -222,7 +223,7 @@ func (s *AttrStore) SetBulkAttrs(m map[uint64]map[string]interface{}) error {
func (s *AttrStore) Blocks() ([]pilosa.AttrBlock, error) {
tx, err := s.db.Begin(false)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "starting transaction")
}
defer tx.Rollback()
@ -256,7 +257,7 @@ func (s *AttrStore) BlockData(i uint64) (map[uint64]map[string]interface{}, erro
// Start read-only transaction.
tx, err := s.db.Begin(false)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "starting transaction")
}
defer tx.Rollback()
@ -273,7 +274,7 @@ func (s *AttrStore) BlockData(i uint64) (map[uint64]map[string]interface{}, erro
// Decode attribute map and associate with id.
attrs, err := pilosa.DecodeAttrs(v)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "decoding attrs")
}
m[btou64(k)] = attrs
@ -329,10 +330,10 @@ func txUpdateAttrs(tx *bolt.Tx, id uint64, m map[string]interface{}) (map[string
// Marshal and save new values.
buf, err := pilosa.EncodeAttrs(attr)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "encoding attrs")
}
if err := tx.Bucket([]byte("attrs")).Put(u64tob(id), buf); err != nil {
return nil, err
return nil, errors.Wrap(err, "saving attrs")
}
return attr, nil
}

5
broadcast.go
View file

@ -20,6 +20,7 @@ import (
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/internal"
"github.com/pkg/errors"
)
// MemberSet represents an interface for Node membership and inter-node communication.
@ -187,7 +188,7 @@ func MarshalMessage(m proto.Message) ([]byte, error) {
}
buf, err := proto.Marshal(m)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "marshalling")
}
return append([]byte{typ}, buf...), nil
}
@ -241,7 +242,7 @@ func UnmarshalMessage(buf []byte) (proto.Message, error) {
}
if err := proto.Unmarshal(buf, m); err != nil {
return nil, err
return nil, errors.Wrap(err, "unmarshalling")
}
return m, nil
}

7
broadcast_test.go
View file

@ -56,9 +56,14 @@ func testMessageMarshal(t *testing.T, m proto.Message) {
// Ensure that BroadcastReceiver can register a BroadcastHandler.
func TestBroadcast_BroadcastReceiver(t *testing.T) {
path, err := ioutil.TempDir("", "pilosa-")
if err != nil {
panic(err)
}
com := server.NewCommand(bytes.NewBuffer([]byte{}), ioutil.Discard, ioutil.Discard)
com.Config.Bind = "localhost:0"
err := com.SetupServer() // this test shouldn't need to import pilosa/server just to set up the Server, but it really shouldn't need to setup the Server at all. The Server should not be the implementation of Broadcast* TODO
com.Config.DataDir = path
err = com.SetupServer() // this test shouldn't need to import pilosa/server just to set up the Server, but it really shouldn't need to setup the Server at all. The Server should not be the implementation of Broadcast* TODO
if err != nil {
t.Fatalf("setting up server: %v", err)
}

126
client.go
View file

@ -19,7 +19,6 @@ import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"io/ioutil"
@ -35,6 +34,7 @@ import (
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/internal"
"github.com/pkg/errors"
)
// ClientOptions represents the configuration for a InternalHTTPClient
@ -58,7 +58,7 @@ func NewInternalHTTPClient(host string, remoteClient *http.Client) (*InternalHTT
uri, err := NewURIFromAddress(host)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "getting URI")
}
client := NewInternalHTTPClientFromURI(uri, remoteClient)
@ -93,7 +93,7 @@ func (c *InternalHTTPClient) maxSliceByIndex(ctx context.Context, inverse bool)
// Build request.
req, err := http.NewRequest("GET", u.String(), nil)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("User-Agent", "pilosa/"+Version)
@ -101,7 +101,7 @@ func (c *InternalHTTPClient) maxSliceByIndex(ctx context.Context, inverse bool)
// Execute request.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
if err != nil {
return nil, err
return nil, errors.Wrap(err, "executing request")
}
defer resp.Body.Close()
@ -126,7 +126,7 @@ func (c *InternalHTTPClient) Schema(ctx context.Context) ([]*IndexInfo, error) {
// Build request.
req, err := http.NewRequest("GET", u, nil)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("User-Agent", "pilosa/"+Version)
@ -134,7 +134,7 @@ func (c *InternalHTTPClient) Schema(ctx context.Context) ([]*IndexInfo, error) {
// Execute request.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
if err != nil {
return nil, err
return nil, errors.Wrap(err, "executing request")
}
defer resp.Body.Close()
@ -154,14 +154,14 @@ func (c *InternalHTTPClient) CreateIndex(ctx context.Context, index string, opt
Options: opt,
})
if err != nil {
return err
return errors.Wrap(err, "encoding request")
}
// Create URL & HTTP request.
u := uriPathToURL(c.defaultURI, fmt.Sprintf("/index/%s", index))
req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf))
if err != nil {
return err
return errors.Wrap(err, "creating request")
}
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
req.Header.Set("Content-Type", "application/json")
@ -171,14 +171,14 @@ func (c *InternalHTTPClient) CreateIndex(ctx context.Context, index string, opt
// Execute request against the host.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
if err != nil {
return err
return errors.Wrap(err, "executing request")
}
defer resp.Body.Close()
// Read body.
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return err
return errors.Wrap(err, "reading")
}
// Handle response based on status code.
@ -201,7 +201,7 @@ func (c *InternalHTTPClient) FragmentNodes(ctx context.Context, index string, sl
// Build request.
req, err := http.NewRequest("GET", u.String(), nil)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("User-Agent", "pilosa/"+Version)
@ -209,7 +209,7 @@ func (c *InternalHTTPClient) FragmentNodes(ctx context.Context, index string, sl
// Execute request.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
if err != nil {
return nil, err
return nil, errors.Wrap(err, "executing request")
}
defer resp.Body.Close()
@ -239,14 +239,14 @@ func (c *InternalHTTPClient) QueryNode(ctx context.Context, uri *URI, index stri
// Encode request object.
buf, err := proto.Marshal(queryRequest)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "marshaling")
}
// Create HTTP request.
u := uri.Path(fmt.Sprintf("/index/%s/query", index))
req, err := http.NewRequest("POST", u, bytes.NewReader(buf))
if err != nil {
return nil, err
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
@ -257,14 +257,14 @@ func (c *InternalHTTPClient) QueryNode(ctx context.Context, uri *URI, index stri
// Execute request against the host.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
if err != nil {
return nil, err
return nil, errors.Wrap(err, "executing request")
}
defer resp.Body.Close()
// Read body and unmarshal response.
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "reading")
} else if resp.StatusCode != http.StatusOK {
return nil, errors.New(string(body))
}
@ -398,7 +398,7 @@ func (c *InternalHTTPClient) importNode(ctx context.Context, node *Node, buf []b
u := nodePathToURL(node, "/import")
req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf))
if err != nil {
return err
return errors.Wrap(err, "creating request")
}
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
req.Header.Set("Content-Type", "application/x-protobuf")
@ -408,14 +408,14 @@ func (c *InternalHTTPClient) importNode(ctx context.Context, node *Node, buf []b
// Execute request against the host.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
if err != nil {
return err
return errors.Wrap(err, "executing request")
}
defer resp.Body.Close()
// Read body and unmarshal response.
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return err
return errors.Wrap(err, "reading")
} else if resp.StatusCode != http.StatusOK {
return errors.New(string(body))
}
@ -486,7 +486,7 @@ func (c *InternalHTTPClient) importValueNode(ctx context.Context, node *Node, bu
u := nodePathToURL(node, "/import-value")
req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf))
if err != nil {
return err
return errors.Wrap(err, "creating request")
}
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
req.Header.Set("Content-Type", "application/x-protobuf")
@ -496,14 +496,14 @@ func (c *InternalHTTPClient) importValueNode(ctx context.Context, node *Node, bu
// Execute request against the host.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
if err != nil {
return err
return errors.Wrap(err, "executing request")
}
defer resp.Body.Close()
// Read body and unmarshal response.
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return err
return errors.Wrap(err, "reading")
} else if resp.StatusCode != http.StatusOK {
return errors.New(string(body))
}
@ -564,7 +564,7 @@ func (c *InternalHTTPClient) exportNodeCSV(ctx context.Context, node *Node, inde
// Generate HTTP request.
req, err := http.NewRequest("GET", u.String(), nil)
if err != nil {
return err
return errors.Wrap(err, "creating request")
}
req.Header.Set("Accept", "text/csv")
req.Header.Set("User-Agent", "pilosa/"+Version)
@ -572,7 +572,7 @@ func (c *InternalHTTPClient) exportNodeCSV(ctx context.Context, node *Node, inde
// Execute request against the host.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
if err != nil {
return err
return errors.Wrap(err, "executing request")
}
defer resp.Body.Close()
@ -583,7 +583,7 @@ func (c *InternalHTTPClient) exportNodeCSV(ctx context.Context, node *Node, inde
// Copy body to writer.
if _, err := io.Copy(w, resp.Body); err != nil {
return err
return errors.Wrap(err, "copying")
}
return nil
@ -618,13 +618,13 @@ func (c *InternalHTTPClient) BackupTo(ctx context.Context, w io.Writer, index, f
// Backup every slice to the tar file.
for i := uint64(0); i <= maxSlices[index]; i++ {
if err := c.backupSliceTo(ctx, tw, index, frame, view, i); err != nil {
return err
return errors.Wrap(err, "backing up slice")
}
}
// Close tar file.
if err := tw.Close(); err != nil {
return err
return errors.Wrap(err, "closing")
}
return nil
@ -644,9 +644,9 @@ func (c *InternalHTTPClient) backupSliceTo(ctx context.Context, tw *tar.Writer,
// Read entire buffer to determine file size.
data, err := ioutil.ReadAll(r)
if err != nil {
return err
return errors.Wrap(err, "reading")
} else if err := r.Close(); err != nil {
return err
return errors.Wrap(err, "closing")
}
// Write slice file header.
@ -656,12 +656,12 @@ func (c *InternalHTTPClient) backupSliceTo(ctx context.Context, tw *tar.Writer,
Size: int64(len(data)),
ModTime: time.Now(),
}); err != nil {
return err
return errors.Wrap(err, "writing header")
}
// Write buffer to file.
if _, err := tw.Write(data); err != nil {
return fmt.Errorf("write buffer: %s", err)
return errors.Wrap(err, "writing buffer")
}
return nil
@ -711,7 +711,7 @@ func (c *InternalHTTPClient) backupSliceNode(ctx context.Context, index, frame,
// Build request.
req, err := http.NewRequest("GET", u.String(), nil)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("User-Agent", "pilosa/"+Version)
@ -719,7 +719,7 @@ func (c *InternalHTTPClient) backupSliceNode(ctx context.Context, index, frame,
// Execute request.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
if err != nil {
return nil, err
return nil, errors.Wrap(err, "executing request")
}
// Return error if status is not OK.
@ -751,7 +751,7 @@ func (c *InternalHTTPClient) RestoreFrom(ctx context.Context, r io.Reader, index
if err == io.EOF {
return nil
} else if err != nil {
return err
return errors.Wrap(err, "opening")
}
// Parse slice from entry name.
@ -763,12 +763,12 @@ func (c *InternalHTTPClient) RestoreFrom(ctx context.Context, r io.Reader, index
// Read file into buffer.
var buf bytes.Buffer
if _, err := io.CopyN(&buf, tr, hdr.Size); err != nil {
return err
return errors.Wrap(err, "copying")
}
// Restore file to all nodes that own it.
if err := c.restoreSliceFrom(ctx, buf.Bytes(), index, frame, view, slice); err != nil {
return err
return errors.Wrap(err, "restoring")
}
}
}
@ -794,14 +794,14 @@ func (c *InternalHTTPClient) restoreSliceFrom(ctx context.Context, buf []byte, i
// Build request.
req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf))
if err != nil {
return err
return errors.Wrap(err, "creating request")
}
req.Header.Set("Content-Type", "application/octet-stream")
req.Header.Set("User-Agent", "pilosa/"+Version)
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
if err != nil {
return err
return errors.Wrap(err, "executing request")
}
resp.Body.Close()
@ -825,14 +825,14 @@ func (c *InternalHTTPClient) CreateFrame(ctx context.Context, index, frame strin
Options: opt,
})
if err != nil {
return err
return errors.Wrap(err, "marshaling")
}
// Create URL & HTTP request.
u := uriPathToURL(c.defaultURI, fmt.Sprintf("/index/%s/frame/%s", index, frame))
req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf))
if err != nil {
return err
return errors.Wrap(err, "creating request")
}
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
req.Header.Set("Content-Type", "application/json")
@ -842,14 +842,14 @@ func (c *InternalHTTPClient) CreateFrame(ctx context.Context, index, frame strin
// Execute request against the host.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
if err != nil {
return err
return errors.Wrap(err, "executing request")
}
defer resp.Body.Close()
// Read body.
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return err
return errors.Wrap(err, "reading")
}
// Handle response based on status code.
@ -873,7 +873,7 @@ func (c *InternalHTTPClient) RestoreFrame(ctx context.Context, host, index, fram
// Build request.
req, err := http.NewRequest("POST", u.String(), nil)
if err != nil {
return err
return errors.Wrap(err, "creating request")
}
req.Header.Set("Content-Type", "application/octet-stream")
req.Header.Set("User-Agent", "pilosa/"+Version)
@ -881,7 +881,7 @@ func (c *InternalHTTPClient) RestoreFrame(ctx context.Context, host, index, fram
// Execute request.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
if err != nil {
return err
return errors.Wrap(err, "executing request")
}
resp.Body.Close()
@ -899,7 +899,7 @@ func (c *InternalHTTPClient) FrameViews(ctx context.Context, index, frame string
u := uriPathToURL(c.defaultURI, fmt.Sprintf("/index/%s/frame/%s/views", index, frame))
req, err := http.NewRequest("GET", u.String(), nil)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+Version)
@ -907,7 +907,7 @@ func (c *InternalHTTPClient) FrameViews(ctx context.Context, index, frame string
// Execute request against the host.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
if err != nil {
return nil, err
return nil, errors.Wrap(err, "executing request")
}
defer resp.Body.Close()
@ -924,7 +924,7 @@ func (c *InternalHTTPClient) FrameViews(ctx context.Context, index, frame string
// Decode response.
var rsp getFrameViewsResponse
if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil {
return nil, err
return nil, errors.Wrap(err, "decoding")
}
return rsp.Views, nil
}
@ -943,7 +943,7 @@ func (c *InternalHTTPClient) FragmentBlocks(ctx context.Context, index, frame, v
// Build request.
req, err := http.NewRequest("GET", u.String(), nil)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("User-Agent", "pilosa/"+Version)
@ -951,7 +951,7 @@ func (c *InternalHTTPClient) FragmentBlocks(ctx context.Context, index, frame, v
// Execute request.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
if err != nil {
return nil, err
return nil, errors.Wrap(err, "executing request")
}
defer resp.Body.Close()
@ -967,7 +967,7 @@ func (c *InternalHTTPClient) FragmentBlocks(ctx context.Context, index, frame, v
// Decode response object.
var rsp getFragmentBlocksResponse
if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil {
return nil, err
return nil, errors.Wrap(err, "decoding")
}
return rsp.Blocks, nil
}
@ -982,13 +982,13 @@ func (c *InternalHTTPClient) BlockData(ctx context.Context, index, frame, view s
Block: uint64(block),
})
if err != nil {
return nil, nil, err
return nil, nil, errors.Wrap(err, "marshaling")
}
u := uriPathToURL(c.defaultURI, "/fragment/block/data")
req, err := http.NewRequest("GET", u.String(), bytes.NewReader(buf))
if err != nil {
return nil, nil, err
return nil, nil, errors.Wrap(err, "creating request")
}
req.Header.Set("Content-Type", "application/protobuf")
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
@ -997,7 +997,7 @@ func (c *InternalHTTPClient) BlockData(ctx context.Context, index, frame, view s
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
if err != nil {
return nil, nil, err
return nil, nil, errors.Wrap(err, "executing request")
}
defer resp.Body.Close()
@ -1013,9 +1013,9 @@ func (c *InternalHTTPClient) BlockData(ctx context.Context, index, frame, view s
// Decode response object.
var rsp internal.BlockDataResponse
if body, err := ioutil.ReadAll(resp.Body); err != nil {
return nil, nil, err
return nil, nil, errors.Wrap(err, "reading")
} else if err := proto.Unmarshal(body, &rsp); err != nil {
return nil, nil, err
return nil, nil, errors.Wrap(err, "unmarshalling")
}
return rsp.RowIDs, rsp.ColumnIDs, nil
}
@ -1027,13 +1027,13 @@ func (c *InternalHTTPClient) ColumnAttrDiff(ctx context.Context, index string, b
// Encode request.
buf, err := json.Marshal(postIndexAttrDiffRequest{Blocks: blks})
if err != nil {
return nil, err
return nil, errors.Wrap(err, "marshaling")
}
// Build request.
req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf))
if err != nil {
return nil, err
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("User-Agent", "pilosa/"+Version)
@ -1041,7 +1041,7 @@ func (c *InternalHTTPClient) ColumnAttrDiff(ctx context.Context, index string, b
// Execute request.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
if err != nil {
return nil, err
return nil, errors.Wrap(err, "executing request")
}
defer resp.Body.Close()
@ -1055,7 +1055,7 @@ func (c *InternalHTTPClient) ColumnAttrDiff(ctx context.Context, index string, b
// Decode response object.
var rsp postIndexAttrDiffResponse
if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil {
return nil, err
return nil, errors.Wrap(err, "decoding")
}
return rsp.Attrs, nil
}
@ -1067,13 +1067,13 @@ func (c *InternalHTTPClient) RowAttrDiff(ctx context.Context, index, frame strin
// Encode request.
buf, err := json.Marshal(postFrameAttrDiffRequest{Blocks: blks})
if err != nil {
return nil, err
return nil, errors.Wrap(err, "marshaling")
}
// Build request.
req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf))
if err != nil {
return nil, err
return nil, errors.Wrap(err, "creating request")
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("User-Agent", "pilosa/"+Version)
@ -1081,7 +1081,7 @@ func (c *InternalHTTPClient) RowAttrDiff(ctx context.Context, index, frame strin
// Execute request.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
if err != nil {
return nil, err
return nil, errors.Wrap(err, "executing request")
}
defer resp.Body.Close()
@ -1097,7 +1097,7 @@ func (c *InternalHTTPClient) RowAttrDiff(ctx context.Context, index, frame strin
// Decode response object.
var rsp postFrameAttrDiffResponse
if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil {
return nil, err
return nil, errors.Wrap(err, "decoding")
}
return rsp.Attrs, nil
}

8
client_test.go
View file

@ -37,7 +37,6 @@ func createCluster(c *pilosa.Cluster) ([]*test.Server, []*test.Holder) {
for i := 0; i < numNodes; i++ {
hldr[i] = test.MustOpenHolder()
server[i] = test.NewServer()
server[i].Handler.API.URI = server[i].HostURI()
server[i].Handler.API.Cluster = c
server[i].Handler.API.Cluster.Nodes[i].URI = server[i].HostURI()
server[i].Handler.API.Holder = hldr[i].Holder
@ -218,7 +217,6 @@ func TestClient_Import(t *testing.T) {
s := test.NewServer()
defer s.Close()
s.Handler.API.URI = s.HostURI()
s.Handler.API.Cluster = test.NewCluster(1)
s.Handler.API.Cluster.Nodes[0].URI = s.HostURI()
s.Handler.API.Holder = hldr.Holder
@ -269,7 +267,6 @@ func TestClient_ImportInverseEnabled(t *testing.T) {
s := test.NewServer()
defer s.Close()
s.Handler.API.URI = s.HostURI()
s.Handler.API.Cluster = test.NewCluster(1)
s.Handler.API.Cluster.Nodes[0].URI = s.HostURI()
s.Handler.API.Holder = hldr.Holder
@ -318,7 +315,6 @@ func TestClient_ImportValue(t *testing.T) {
s := test.NewServer()
defer s.Close()
s.Handler.API.URI = s.HostURI()
s.Handler.API.Cluster = test.NewCluster(1)
s.Handler.API.Cluster.Nodes[0].URI = s.HostURI()
s.Handler.API.Holder = hldr.Holder
@ -386,7 +382,6 @@ func TestClient_BackupRestore(t *testing.T) {
s := test.NewServer()
defer s.Close()
s.Handler.API.URI = s.HostURI()
s.Handler.API.Cluster = test.NewCluster(1)
s.Handler.API.Cluster.Nodes[0].URI = s.HostURI()
s.Handler.API.Holder = hldr.Holder
@ -452,7 +447,6 @@ func TestClient_BackupInverseView(t *testing.T) {
s := test.NewServer()
defer s.Close()
s.Handler.API.URI = s.HostURI()
s.Handler.API.Cluster = test.NewCluster(1)
s.Handler.API.Cluster.Nodes[0].URI = s.HostURI()
s.Handler.API.Holder = hldr.Holder
@ -489,7 +483,6 @@ func TestClient_BackupInvalidView(t *testing.T) {
s := test.NewServer()
defer s.Close()
s.Handler.API.URI = s.HostURI()
s.Handler.API.Cluster = test.NewCluster(1)
s.Handler.API.Cluster.Nodes[0].URI = s.HostURI()
s.Handler.API.Holder = hldr.Holder
@ -518,7 +511,6 @@ func TestClient_FragmentBlocks(t *testing.T) {
s := test.NewServer()
defer s.Close()
s.Handler.API.URI = s.HostURI()
s.Handler.API.Cluster = test.NewCluster(1)
s.Handler.API.Cluster.Nodes[0].URI = s.HostURI()
s.Handler.API.Holder = hldr.Holder

66
cluster.go
View file

@ -307,16 +307,16 @@ func (c *Cluster) isCoordinator() bool {
// nodes with its version of Cluster.Status.
func (c *Cluster) SetCoordinator(n *Node) error {
c.mu.Lock()
defer c.mu.Unlock()
// Verify that the new Coordinator value matches
// this node.
if c.Node.ID != n.ID {
c.mu.Unlock()
return fmt.Errorf("coordinator node does not match this node")
}
// Update IsCoordinator on all nodes (locally).
_ = c.updateCoordinator(n)
c.mu.Unlock()
// Send the update coordinator message to all nodes.
err := c.Broadcaster.SendSync(
&internal.UpdateCoordinatorMessage{
@ -721,7 +721,7 @@ func (c *Cluster) fragSources(to *Cluster, idx *Index) (map[string][]*internal.R
// Determine if a node is being added or removed.
action, diffNodeID, err := c.diff(to)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "diffing")
}
// Initialize the map with all the nodes in `to`.
@ -905,7 +905,7 @@ func (c *Cluster) Open() error {
// Load topology file if it exists.
if err := c.loadTopology(); err != nil {
return fmt.Errorf("load topology: %v", err)
return errors.Wrap(err, "loading topology")
}
c.ID = c.Topology.ClusterID
@ -1006,7 +1006,7 @@ func (c *Cluster) handleNodeAction(nodeAction nodeAction) error {
if err := c.setStateAndBroadcast(ClusterStateNormal); err != nil {
c.Logger.Printf("setStateAndBroadcast error: err=%s", err)
}
return err
return errors.Wrap(err, "setting state")
}
// j.Run() runs in a goroutine because in the case where the
@ -1023,14 +1023,14 @@ func (c *Cluster) handleNodeAction(nodeAction nodeAction) error {
// Make sure j.Run() didn't return an error.
if eg.Wait() != nil {
return err
return errors.Wrap(err, "running job")
}
c.Logger.Printf("received jobResult: %s", jobResult)
switch jobResult {
case ResizeJobStateDone:
if err := c.CompleteCurrentJob(ResizeJobStateDone); err != nil {
return err
return errors.Wrap(err, "completing finished job")
}
// Add/remove uri to/from the cluster.
if j.action == ResizeJobActionRemove {
@ -1040,7 +1040,7 @@ func (c *Cluster) handleNodeAction(nodeAction nodeAction) error {
}
case ResizeJobStateAborted:
if err := c.CompleteCurrentJob(ResizeJobStateAborted); err != nil {
return err
return errors.Wrap(err, "completing aborted job")
}
}
return nil
@ -1055,7 +1055,7 @@ func (c *Cluster) setStateAndBroadcast(state string) error {
func (c *Cluster) sendTo(node *Node, msg proto.Message) error {
if err := c.Broadcaster.SendTo(node, msg); err != nil {
return err
return errors.Wrap(err, "sending")
}
return nil
}
@ -1124,7 +1124,7 @@ func (c *Cluster) generateResizeJob(nodeAction nodeAction) (*ResizeJob, error) {
j, err := c.generateResizeJobByAction(nodeAction)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "generating job")
}
c.Logger.Printf("generated ResizeJob: %d", j.ID)
@ -1171,7 +1171,7 @@ func (c *Cluster) generateResizeJobByAction(nodeAction nodeAction) (*ResizeJob,
for _, idx := range c.Holder.Indexes() {
fragSources, err := c.fragSources(toCluster, idx)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "getting sources")
}
for id, sources := range fragSources {
@ -1223,7 +1223,7 @@ func (c *Cluster) FollowResizeInstruction(instr *internal.ResizeInstruction) err
// Make sure the cluster status on this node agrees with the Coordinator
// before attempting a resize.
if err := c.MergeClusterStatus(instr.ClusterStatus); err != nil {
return err
return errors.Wrap(err, "merging cluster status")
}
c.Logger.Printf("MergeClusterStatus done, start goroutine")
@ -1248,7 +1248,7 @@ func (c *Cluster) FollowResizeInstruction(instr *internal.ResizeInstruction) err
// Sync the schema received in the resize instruction.
c.Logger.Printf("Holder ApplySchema")
if err := c.Holder.ApplySchema(instr.Schema); err != nil {
return err
return errors.Wrap(err, "applying schema")
}
// Create a client for calling remote nodes.
@ -1269,13 +1269,13 @@ func (c *Cluster) FollowResizeInstruction(instr *internal.ResizeInstruction) err
// Create view.
v, err := f.CreateViewIfNotExists(src.View)
if err != nil {
return err
return errors.Wrap(err, "creating view")
}
// Create the local fragment.
frag, err := v.CreateFragmentIfNotExists(src.Slice)
if err != nil {
return err
return errors.Wrap(err, "creating fragment")
}
// Stream slice from remote node.
@ -1291,7 +1291,7 @@ func (c *Cluster) FollowResizeInstruction(instr *internal.ResizeInstruction) err
if err == ErrFragmentNotFound {
return nil
}
return err
return errors.Wrap(err, "retrieving slice")
} else if rd == nil {
return fmt.Errorf("slice %v doesn't exist on host: %s", src.Slice, src.Node.URI)
}
@ -1304,7 +1304,7 @@ func (c *Cluster) FollowResizeInstruction(instr *internal.ResizeInstruction) err
}
return nil
}(); err != nil {
return err
return errors.Wrap(err, "copying remote slice")
}
}
return nil
@ -1438,7 +1438,7 @@ func (j *ResizeJob) Run() error {
err := j.distributeResizeInstructions()
if err != nil {
j.result <- ResizeJobStateAborted
return err
return errors.Wrap(err, "distributing instructions")
}
return nil
}
@ -1475,7 +1475,7 @@ func (j *ResizeJob) distributeResizeInstructions() error {
}
j.Logger.Printf("send resize instructions: %v", instr)
if err := j.Broadcaster.SendTo(node, instr); err != nil {
return err
return errors.Wrap(err, "sending instruction")
}
}
return nil
@ -1581,16 +1581,16 @@ func (c *Cluster) loadTopology() error {
c.Topology = NewTopology()
return nil
} else if err != nil {
return err
return errors.Wrap(err, "reading file")
}
var pb internal.Topology
if err := proto.Unmarshal(buf, &pb); err != nil {
return err
return errors.Wrap(err, "unmarshalling")
}
top, err := decodeTopology(&pb)
if err != nil {
return err
return errors.Wrap(err, "decoding")
}
c.Topology = top
@ -1601,13 +1601,13 @@ func (c *Cluster) loadTopology() error {
func (c *Cluster) saveTopology() error {
if err := os.MkdirAll(c.Path, 0777); err != nil {
return err
return errors.Wrap(err, "creating directory")
}
if buf, err := proto.Marshal(encodeTopology(c.Topology)); err != nil {
return err
return errors.Wrap(err, "marshalling")
} else if err := ioutil.WriteFile(filepath.Join(c.Path, ".topology"), buf, 0666); err != nil {
return err
return errors.Wrap(err, "writing file")
}
return nil
}
@ -1704,7 +1704,7 @@ func (c *Cluster) nodeJoin(node *Node) error {
}
if err := c.AddNode(node); err != nil {
return err
return errors.Wrap(err, "adding node for agreement")
}
// Only change to normal if there is no existing data. Otherwise,
@ -1742,7 +1742,7 @@ func (c *Cluster) nodeJoin(node *Node) error {
// If the holder does not yet contain data, go ahead and add the node.
if ok, err := c.Holder.HasData(); !ok && err == nil {
if err := c.AddNode(node); err != nil {
return err
return errors.Wrap(err, "adding node")
}
return c.setStateAndBroadcast(ClusterStateNormal)
} else if err != nil {
@ -1752,7 +1752,7 @@ func (c *Cluster) nodeJoin(node *Node) error {
// If the cluster has data, we need to change to RESIZING and
// kick off the resizing process.
if err := c.setStateAndBroadcast(ClusterStateResizing); err != nil {
return err
return errors.Wrap(err, "broadcasting state")
}
c.joiningLeavingNodes <- nodeAction{node, ResizeJobActionAdd}
@ -1784,7 +1784,7 @@ func (c *Cluster) NodeLeave(node *Node) error {
_, err := c.generateResizeJobByAction(nodeAction{c.nodeByID(node.ID), ResizeJobActionRemove})
if err != nil {
return err
return errors.Wrap(err, "generating job")
}
return c.nodeLeave(node)
@ -1802,7 +1802,7 @@ func (c *Cluster) nodeLeave(node *Node) error {
// If the holder does not yet contain data, go ahead and remove the node.
if ok, err := c.Holder.HasData(); !ok && err == nil {
if err := c.RemoveNode(n); err != nil {
return err
return errors.Wrap(err, "removing node")
}
return c.setStateAndBroadcast(ClusterStateNormal)
} else if err != nil {
@ -1812,7 +1812,7 @@ func (c *Cluster) nodeLeave(node *Node) error {
// If the cluster has data then change state to RESIZING and
// kick off the resizing process.
if err := c.setStateAndBroadcast(ClusterStateResizing); err != nil {
return err
return errors.Wrap(err, "broadcasting state")
}
c.joiningLeavingNodes <- nodeAction{n, ResizeJobActionRemove}
@ -1836,7 +1836,7 @@ func (c *Cluster) MergeClusterStatus(cs *internal.ClusterStatus) error {
// Add all nodes from the coordinator.
for _, node := range officialNodes {
if err := c.AddNode(node); err != nil {
return err
return errors.Wrap(err, "adding node")
}
}
@ -1857,7 +1857,7 @@ func (c *Cluster) MergeClusterStatus(cs *internal.ClusterStatus) error {
for _, nodeID := range nodeIDsToRemove {
if err := c.RemoveNode(c.nodeByID(nodeID)); err != nil {
return err
return errors.Wrap(err, "removing node")
}
}

5
cluster_internal_test.go
View file

@ -17,6 +17,7 @@ package pilosa
import (
"io/ioutil"
"reflect"
"strings"
"testing"
"github.com/pilosa/pilosa/internal"
@ -243,8 +244,8 @@ func TestFragSources(t *testing.T) {
actual, err := (test.from).fragSources(test.to, test.idx)
if test.err != "" {
if err.Error() != test.err {
t.Fatalf("expected error: %s", test.err)
if !strings.Contains(err.Error(), test.err) {
t.Fatalf("expected error: %s, got: %s", test.err, err.Error())
}
} else {
if err != nil {

12
ctl/backup.go
View file

@ -16,12 +16,12 @@ package ctl
import (
"context"
"errors"
"io"
"os"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/server"
"github.com/pkg/errors"
)
// BackupCommand represents a command for backing up a view.
@ -60,26 +60,26 @@ func (cmd *BackupCommand) Run(ctx context.Context) error {
// Create a client to the server.
client, err := CommandClient(cmd)
if err != nil {
return err
return errors.Wrap(err, "creating client")
}
// Open output file.
f, err := os.Create(cmd.Path)
if err != nil {
return err
return errors.Wrap(err, "creating file")
}
defer f.Close()
// Begin streaming backup.
if err := client.BackupTo(ctx, f, cmd.Index, cmd.Frame, cmd.View); err != nil {
return err
return errors.Wrap(err, "backing up")
}
// Sync & close file to ensure durability.
if err := f.Sync(); err != nil {
return err
return errors.Wrap(err, "syncing")
} else if err = f.Close(); err != nil {
return err
return errors.Wrap(err, "closing file")
}
return nil

6
ctl/backup_test.go
View file

@ -46,11 +46,7 @@ func TestBackupCommand_Run(t *testing.T) {
s := test.NewServer()
defer s.Close()
uri, err := pilosa.NewURIFromAddress(s.Host())
if err != nil {
t.Fatal(err)
}
s.Handler.API.URI = *uri
s.Handler.API.Cluster = test.NewCluster(1)
s.Handler.API.Cluster.Nodes[0].URI = s.HostURI()
s.Handler.API.Holder = hldr.Holder

4
ctl/bench.go
View file

@ -16,7 +16,6 @@ package ctl
import (
"context"
"errors"
"fmt"
"io"
"math/rand"
@ -25,6 +24,7 @@ import (
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/server"
"github.com/pkg/errors"
)
// BenchCommand represents a command for benchmarking index operations.
@ -58,7 +58,7 @@ func (cmd *BenchCommand) Run(ctx context.Context) error {
// Create a client to the server.
client, err := CommandClient(cmd)
if err != nil {
return err
return errors.Wrap(err, "creating client")
}
switch cmd.Op {

3
ctl/bench_test.go
View file

@ -23,6 +23,7 @@ import (
"testing"
"github.com/pilosa/pilosa"
"github.com/pkg/errors"
)
func TestBenchCommand_InvalidOption(t *testing.T) {
@ -31,7 +32,7 @@ func TestBenchCommand_InvalidOption(t *testing.T) {
cm := NewBenchCommand(stdin, stdout, stderr)
err := cm.Run(context.Background())
if err != pilosa.ErrHostRequired {
if errors.Cause(err) != pilosa.ErrHostRequired {
t.Fatalf("Expect err: %s, actual err: %s", pilosa.ErrHostRequired, err)
}

15
ctl/check.go
View file

@ -24,6 +24,7 @@ import (
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/roaring"
"github.com/pkg/errors"
)
// CheckCommand represents a command for performing consistency checks on data files.
@ -48,17 +49,17 @@ func (cmd *CheckCommand) Run(ctx context.Context) error {
switch filepath.Ext(path) {
case "":
if err := cmd.checkBitmapFile(path); err != nil {
return err
return errors.Wrap(err, "checking bitmap")
}
case ".cache":
if err := cmd.checkCacheFile(path); err != nil {
return err
return errors.Wrap(err, "checking cache")
}
case ".snapshotting":
if err := cmd.checkSnapshotFile(path); err != nil {
return err
return errors.Wrap(err, "checking snapshot")
}
}
}
@ -71,26 +72,26 @@ func (cmd *CheckCommand) checkBitmapFile(path string) error {
// Open file handle.
f, err := os.Open(path)
if err != nil {
return err
return errors.Wrap(err, "opening file")
}
defer f.Close()
fi, err := f.Stat()
if err != nil {
return err
return errors.Wrap(err, "statting file")
}
// Memory map the file.
data, err := syscall.Mmap(int(f.Fd()), 0, int(fi.Size()), syscall.PROT_READ, syscall.MAP_SHARED)
if err != nil {
return err
return errors.Wrap(err, "mmapping")
}
defer syscall.Munmap(data)
// Attach the mmap file to the bitmap.
bm := roaring.NewBitmap()
if err := bm.UnmarshalBinary(data); err != nil {
return err
return errors.Wrap(err, "unmarshalling")
}
// Perform consistency check.

2
ctl/check_test.go
View file

@ -85,7 +85,7 @@ func TestCheckCommand_Run(t *testing.T) {
var buf bytes.Buffer
io.Copy(&buf, r)
if !strings.HasPrefix(err.Error(), "invalid roaring file") {
if !strings.HasPrefix(err.Error(), "checking bitmap: unmarshalling: invalid roaring file") {
t.Fatalf("expect error: invalid roaring file, actual: '%s'", err)
}
// Todo: need correct roaring file for happy path

5
ctl/common.go
View file

@ -19,6 +19,7 @@ import (
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/server"
"github.com/pkg/errors"
"github.com/spf13/pflag"
)
@ -42,7 +43,7 @@ func CommandClient(cmd CommandWithTLSSupport) (*pilosa.InternalHTTPClient, error
if tlsConfig.CertificatePath != "" && tlsConfig.CertificateKeyPath != "" {
cert, err := tls.LoadX509KeyPair(tlsConfig.CertificatePath, tlsConfig.CertificateKeyPath)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "loading keypair")
}
TLSConfig = &tls.Config{
Certificates: []tls.Certificate{cert},
@ -51,7 +52,7 @@ func CommandClient(cmd CommandWithTLSSupport) (*pilosa.InternalHTTPClient, error
}
client, err := pilosa.NewInternalHTTPClient(cmd.TLSHost(), server.GetHTTPClient(TLSConfig))
if err != nil {
return nil, err
return nil, errors.Wrap(err, "getting internal client")
}
return client, err
}

11
ctl/export.go
View file

@ -22,6 +22,7 @@ import (
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/server"
"github.com/pkg/errors"
)
// ExportCommand represents a command for bulk exporting data from a server.
@ -68,7 +69,7 @@ func (cmd *ExportCommand) Run(ctx context.Context) error {
if cmd.Path != "" {
f, err := os.Create(cmd.Path)
if err != nil {
return err
return errors.Wrap(err, "creating file")
}
defer f.Close()
@ -78,7 +79,7 @@ func (cmd *ExportCommand) Run(ctx context.Context) error {
// Create a client to the server.
client, err := CommandClient(cmd)
if err != nil {
return err
return errors.Wrap(err, "creating client")
}
// Determine slice count.
@ -90,21 +91,21 @@ func (cmd *ExportCommand) Run(ctx context.Context) error {
}
if err != nil {
return err
return errors.Wrap(err, "getting slice count")
}
// Export each slice.
for slice := uint64(0); slice <= maxSlices[cmd.Index]; slice++ {
logger.Printf("exporting slice: %d", slice)
if err := client.ExportCSV(ctx, cmd.Index, cmd.Frame, cmd.View, slice, w); err != nil {
return err
return errors.Wrap(err, "exporting")
}
}
// Close writer, if applicable.
if w, ok := w.(io.Closer); ok {
if err := w.Close(); err != nil {
return err
return errors.Wrap(err, "closing")
}
}

9
ctl/export_test.go
View file

@ -59,11 +59,7 @@ func TestExportCommand_Run(t *testing.T) {
defer hldr.Close()
s := test.NewServer()
defer s.Close()
uri, err := pilosa.NewURIFromAddress(s.Host())
if err != nil {
t.Fatal(err)
}
s.Handler.API.URI = *uri
s.Handler.API.Cluster = test.NewCluster(1)
s.Handler.API.Cluster.Nodes[0].URI = s.HostURI()
s.Handler.API.Holder = hldr.Holder
@ -75,8 +71,7 @@ func TestExportCommand_Run(t *testing.T) {
cm.Index = "i"
cm.Frame = "f"
cm.View = pilosa.ViewStandard
err = cm.Run(context.Background())
if err != nil {
if err := cm.Run(context.Background()); err != nil {
t.Fatalf("Export Run doesn't work: %s", err)
}
}

24
ctl/import.go
View file

@ -17,7 +17,6 @@ package ctl
import (
"context"
"encoding/csv"
"errors"
"fmt"
"io"
"log"
@ -28,6 +27,7 @@ import (
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/server"
"github.com/pkg/errors"
)
// ImportCommand represents a command for bulk importing data.
@ -94,14 +94,14 @@ func (cmd *ImportCommand) Run(ctx context.Context) error {
// Create a client to the server.
client, err := CommandClient(cmd)
if err != nil {
return err
return errors.Wrap(err, "creating client")
}
cmd.Client = client
if cmd.CreateSchema {
err := cmd.ensureSchema(ctx)
if err != nil {
return err
return errors.Wrap(err, "ensuring schema")
}
}
@ -153,7 +153,7 @@ func (cmd *ImportCommand) bufferBits(ctx context.Context, path string) error {
// Open file for reading.
f, err := os.Open(path)
if err != nil {
return err
return errors.Wrap(err, "opening file")
}
defer f.Close()
@ -173,7 +173,7 @@ func (cmd *ImportCommand) bufferBits(ctx context.Context, path string) error {
if err == io.EOF {
break
} else if err != nil {
return err
return errors.Wrap(err, "reading")
}
// Ignore blank rows.
@ -243,7 +243,7 @@ func (cmd *ImportCommand) importBits(ctx context.Context, bits []pilosa.Bit) err
logger.Printf("importing slice: %d, n=%d", slice, len(bits))
if err := cmd.Client.Import(ctx, cmd.Index, cmd.Frame, slice, bits); err != nil {
return err
return errors.Wrap(err, "importing")
}
}
@ -260,7 +260,7 @@ func (cmd *ImportCommand) bufferBitsK(ctx context.Context, path string) error {
// Open file for reading.
f, err := os.Open(path)
if err != nil {
return err
return errors.Wrap(err, "opening file")
}
defer f.Close()
@ -280,7 +280,7 @@ func (cmd *ImportCommand) bufferBitsK(ctx context.Context, path string) error {
if err == io.EOF {
break
} else if err != nil {
return err
return errors.Wrap(err, "reading")
}
// Ignore blank rows.
@ -340,7 +340,7 @@ func (cmd *ImportCommand) importBitsK(ctx context.Context, bits []pilosa.Bit) er
logger.Printf("importing keys: n=%d", len(bits))
if err := cmd.Client.ImportK(ctx, cmd.Index, cmd.Frame, bits); err != nil {
return err
return errors.Wrap(err, "importing keys")
}
return nil
@ -356,7 +356,7 @@ func (cmd *ImportCommand) bufferFieldValues(ctx context.Context, path string) er
// Open file for reading.
f, err := os.Open(path)
if err != nil {
return err
return errors.Wrap(err, "opening file")
}
defer f.Close()
@ -376,7 +376,7 @@ func (cmd *ImportCommand) bufferFieldValues(ctx context.Context, path string) er
if err == io.EOF {
break
} else if err != nil {
return err
return errors.Wrap(err, "reading")
}
// Ignore blank rows.
@ -437,7 +437,7 @@ func (cmd *ImportCommand) importFieldValues(ctx context.Context, vals []pilosa.F
logger.Printf("importing slice: %d, n=%d", slice, len(vals))
if err := cmd.Client.ImportValue(ctx, cmd.Index, cmd.Frame, cmd.Field, slice, vals); err != nil {
return err
return errors.Wrap(err, "importing values")
}
}

11
ctl/import_test.go
View file

@ -65,11 +65,7 @@ func TestImportCommand_Run(t *testing.T) {
defer hldr.Close()
s := test.NewServer()
defer s.Close()
uri, err := pilosa.NewURIFromAddress(s.Host())
if err != nil {
t.Fatal(err)
}
s.Handler.API.URI = *uri
s.Handler.API.Cluster = test.NewCluster(1)
s.Handler.API.Cluster.Nodes[0].URI = s.HostURI()
s.Handler.API.Holder = hldr.Holder
@ -103,12 +99,7 @@ func TestImportCommand_RunValue(t *testing.T) {
defer hldr.Close()
s := test.NewServer()
defer s.Close()
uri, err := pilosa.NewURIFromAddress(s.Host())
if err != nil {
t.Fatal(err)
}
s.Handler.API.URI = *uri
s.Handler.API.Cluster = test.NewCluster(1)
s.Handler.API.Cluster.Nodes[0].URI = s.HostURI()
s.Handler.API.Holder = hldr.Holder

9
ctl/inspect.go
View file

@ -26,6 +26,7 @@ import (
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/roaring"
"github.com/pkg/errors"
)
// InspectCommand represents a command for inspecting fragment data files.
@ -49,19 +50,19 @@ func (cmd *InspectCommand) Run(ctx context.Context) error {
// Open file handle.
f, err := os.Open(cmd.Path)
if err != nil {
return err
return errors.Wrap(err, "opening file")
}
defer f.Close()
fi, err := f.Stat()
if err != nil {
return err
return errors.Wrap(err, "statting file")
}
// Memory map the file.
data, err := syscall.Mmap(int(f.Fd()), 0, int(fi.Size()), syscall.PROT_READ, syscall.MAP_SHARED)
if err != nil {
return err
return errors.Wrap(err, "mmapping")
}
defer syscall.Munmap(data)
@ -70,7 +71,7 @@ func (cmd *InspectCommand) Run(ctx context.Context) error {
fmt.Fprintf(cmd.Stderr, "unmarshaling bitmap...")
bm := roaring.NewBitmap()
if err := bm.UnmarshalBinary(data); err != nil {
return err
return errors.Wrap(err, "unmarshalling")
}
fmt.Fprintf(cmd.Stderr, " (%s)\n", time.Since(t))

8
ctl/restore.go
View file

@ -16,12 +16,12 @@ package ctl
import (
"context"
"errors"
"io"
"os"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/server"
"github.com/pkg/errors"
)
// RestoreCommand represents a command for restoring a frame from a backup.
@ -60,19 +60,19 @@ func (cmd *RestoreCommand) Run(ctx context.Context) error {
// Create a client to the server.
client, err := CommandClient(cmd)
if err != nil {
return err
return errors.Wrap(err, "creating client")
}
// Open backup file.
f, err := os.Open(cmd.Path)
if err != nil {
return err
return errors.Wrap(err, "opening file")
}
defer f.Close()
// Restore backup file to the cluster.
if err := client.RestoreFrom(ctx, f, cmd.Index, cmd.Frame, cmd.View); err != nil {
return err
return errors.Wrap(err, "restoring")
}
return nil

6
ctl/restore_test.go
View file

@ -48,11 +48,7 @@ func TestRestoreCommand_Run(t *testing.T) {
s := test.NewServer()
defer s.Close()
uri, err := pilosa.NewURIFromAddress(s.Host())
if err != nil {
t.Fatal(err)
}
s.Handler.API.URI = *uri
s.Handler.API.Cluster = test.NewCluster(1)
s.Handler.API.Cluster.Nodes[0].URI = s.HostURI()
s.Handler.API.Holder = hldr.Holder

8
diagnostics.go
View file

@ -23,6 +23,8 @@ import (
"strings"
"sync"
"time"
"github.com/pkg/errors"
)
// Default version check URL.
@ -80,13 +82,13 @@ func (d *DiagnosticsCollector) Flush() error {
d.metrics["Uptime"] = (time.Now().Unix() - d.startTime)
buf, err := d.encode()
if err != nil {
return err
return errors.Wrap(err, "encoding")
}
req, err := http.NewRequest("POST", d.host, bytes.NewReader(buf))
req.Header.Set("Content-Type", "application/json")
resp, err := d.client.Do(req)
if err != nil {
return err
return errors.Wrap(err, "posting")
}
// Intentionally ignoring response body, as user does not need to be notified of error.
defer resp.Body.Close()
@ -99,7 +101,7 @@ func (d *DiagnosticsCollector) CheckVersion() error {
req, err := http.NewRequest("GET", d.VersionURL, nil)
resp, err := d.client.Do(req)
if err != nil {
return err
return errors.Wrap(err, "getting version")
}
defer resp.Body.Close()

4
docs/query-language.md
View file

@ -160,8 +160,8 @@ Delete url value for repo 10.
**Spec:**
```
SetBit(<frame=STRING>, <row=UINT>, <col=UINT>,
[timestamp=TIMESTAMP])
ClearBit(<frame=STRING>, <row=UINT>, <col=UINT>,
[timestamp=TIMESTAMP])
```
**Description:**

22
executor.go
View file

@ -16,7 +16,6 @@ package pilosa
import (
"context"
"errors"
"fmt"
"net/http"
"sort"
@ -24,6 +23,7 @@ import (
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/pql"
"github.com/pkg/errors"
)
// DefaultFrame is the frame used if one is not specified.
@ -149,7 +149,7 @@ func (e *Executor) Execute(ctx context.Context, index string, q *pql.Query, slic
// executeCall executes a call.
func (e *Executor) executeCall(ctx context.Context, index string, c *pql.Call, slices []uint64, opt *ExecOptions) (interface{}, error) {
if err := e.validateCallArgs(c); err != nil {
return nil, err
return nil, errors.Wrap(err, "validating args")
}
indexTag := fmt.Sprintf("index:%s", index)
// Special handling for mutation and top-n calls.
@ -344,7 +344,7 @@ func (e *Executor) executeBitmapCall(ctx context.Context, index string, c *pql.C
if columnID, ok, err := c.UintArg(columnLabel); ok && err == nil {
attrs, err := idx.ColumnAttrStore().Attrs(columnID)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "getting column attrs")
}
bm.Attrs = attrs
} else if err != nil {
@ -354,11 +354,11 @@ func (e *Executor) executeBitmapCall(ctx context.Context, index string, c *pql.C
if fr := idx.Frame(frame); fr != nil {
rowID, _, err := c.UintArg(rowLabel)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "getting row")
}
attrs, err := fr.RowAttrStore().Attrs(rowID)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "getting row attrs")
}
bm.Attrs = attrs
}
@ -400,7 +400,7 @@ func (e *Executor) executeSumCountSlice(ctx context.Context, index string, c *pq
if len(c.Children) == 1 {
bm, err := e.executeBitmapCallSlice(ctx, index, c.Children[0], slice)
if err != nil {
return ValCount{}, err
return ValCount{}, errors.Wrap(err, "executing bitmap call")
}
filter = bm
}
@ -425,7 +425,7 @@ func (e *Executor) executeSumCountSlice(ctx context.Context, index string, c *pq
vsum, vcount, err := fragment.FieldSum(filter, field.BitDepth())
if err != nil {
return ValCount{}, err
return ValCount{}, errors.Wrap(err, "computing sum")
}
return ValCount{
Val: int64(vsum) + (int64(vcount) * field.Min),
@ -527,7 +527,7 @@ func (e *Executor) executeTopN(ctx context.Context, index string, c *pql.Call, s
// Execute original query.
pairs, err := e.executeTopNSlices(ctx, index, c, slices, opt)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "finding top results")
}
// If this call is against specific ids, or we didn't get results,
@ -544,7 +544,7 @@ func (e *Executor) executeTopN(ctx context.Context, index string, c *pql.Call, s
trimmedList, err := e.executeTopNSlices(ctx, index, other, slices, opt)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "retrieving full counts")
}
if n != 0 && int(n) < len(trimmedList) {
@ -883,7 +883,7 @@ func (e *Executor) executeFieldRangeSlice(ctx context.Context, index string, c *
predicates, err := cond.IntSliceValue()
if err != nil {
return nil, err
return nil, errors.Wrap(err, "getting condition value")
}
// Only support two integers for the between operation.
@ -1573,7 +1573,7 @@ func (e *Executor) mapReduce(ctx context.Context, index string, slices []uint64,
// Start mapping across all primary owners.
if err := e.mapper(ctx, ch, nodes, index, slices, c, opt, mapFn, reduceFn); err != nil {
return nil, err
return nil, errors.Wrap(err, "starting mapper")
}
// Iterate over all map responses and reduce.

110
fragment.go
View file

@ -21,7 +21,6 @@ import (
"container/heap"
"context"
"encoding/binary"
"errors"
"fmt"
"hash"
"io"
@ -42,6 +41,7 @@ import (
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/pql"
"github.com/pilosa/pilosa/roaring"
"github.com/pkg/errors"
)
const (
@ -160,12 +160,12 @@ func (f *Fragment) Open() error {
if err := func() error {
// Initialize storage in a function so we can close if anything goes wrong.
if err := f.openStorage(); err != nil {
return err
return errors.Wrap(err, "opening storage")
}
// Fill cache with rows persisted to disk.
if err := f.openCache(); err != nil {
return err
return errors.Wrap(err, "opening cache")
}
// Clear checksums.
@ -206,7 +206,7 @@ func (f *Fragment) openStorage() error {
// If the file is empty then initialize it with an empty bitmap.
fi, err := f.file.Stat()
if err != nil {
return err
return errors.Wrap(err, "statting file before")
} else if fi.Size() == 0 {
bi := bufio.NewWriter(f.file)
if _, err := f.storage.WriteTo(bi); err != nil {
@ -215,7 +215,7 @@ func (f *Fragment) openStorage() error {
bi.Flush()
fi, err = f.file.Stat()
if err != nil {
return err
return errors.Wrap(err, "statting file after")
}
}
@ -298,13 +298,13 @@ func (f *Fragment) close() error {
// Flush cache if closing gracefully.
if err := f.flushCache(); err != nil {
f.Logger.Printf("fragment: error flushing cache on close: err=%s, path=%s", err, f.path)
return err
return errors.Wrap(err, "flushing cache")
}
// Close underlying storage.
if err := f.closeStorage(); err != nil {
f.Logger.Printf("fragment: error closing storage: err=%s, path=%s", err, f.path)
return err
return errors.Wrap(err, "closing storage")
}
// Remove checksums.
@ -393,12 +393,12 @@ func (f *Fragment) setBit(rowID, columnID uint64) (changed bool, err error) {
// Determine the position of the bit in the storage.
pos, err := f.pos(rowID, columnID)
if err != nil {
return false, err
return false, errors.Wrap(err, "getting bit ops")
}
// Write to storage.
if changed, err = f.storage.Add(pos); err != nil {
return false, err
return false, errors.Wrap(err, "writing")
}
// Don't update the cache if nothing changed.
@ -411,7 +411,7 @@ func (f *Fragment) setBit(rowID, columnID uint64) (changed bool, err error) {
// Increment number of operations until snapshot is required.
if err := f.incrementOpN(); err != nil {
return false, err
return false, errors.Wrap(err, "incrementing")
}
// Get the row from row cache or fragment.storage.
@ -445,12 +445,12 @@ func (f *Fragment) clearBit(rowID, columnID uint64) (changed bool, err error) {
// Determine the position of the bit in the storage.
pos, err := f.pos(rowID, columnID)
if err != nil {
return false, err
return false, errors.Wrap(err, "getting bit pos")
}
// Write to storage.
if changed, err = f.storage.Remove(pos); err != nil {
return false, err
return false, errors.Wrap(err, "writing")
}
// Don't update the cache if nothing changed.
@ -463,7 +463,7 @@ func (f *Fragment) clearBit(rowID, columnID uint64) (changed bool, err error) {
// Increment number of operations until snapshot is required.
if err := f.incrementOpN(); err != nil {
return false, err
return false, errors.Wrap(err, "incrementing")
}
// Get the row from cache or fragment.storage.
@ -493,7 +493,7 @@ func (f *Fragment) FieldValue(columnID uint64, bitDepth uint) (value uint64, exi
// If existence bit is unset then ignore remaining bits.
if v, err := f.bit(uint64(bitDepth), columnID); err != nil {
return 0, false, err
return 0, false, errors.Wrap(err, "getting existence bit")
} else if !v {
return 0, false, nil
}
@ -501,7 +501,7 @@ func (f *Fragment) FieldValue(columnID uint64, bitDepth uint) (value uint64, exi
// Compute other bits into a value.
for i := uint(0); i < bitDepth; i++ {
if v, err := f.bit(uint64(i), columnID); err != nil {
return 0, false, err
return 0, false, errors.Wrapf(err, "getting value bit %d", i)
} else if v {
value |= (1 << i)
}
@ -533,7 +533,7 @@ func (f *Fragment) SetFieldValue(columnID uint64, bitDepth uint, value uint64) (
// Mark value as set.
if c, err := f.setBit(uint64(bitDepth), columnID); err != nil {
return changed, err
return changed, errors.Wrap(err, "marking not-null")
} else if c {
changed = true
}
@ -548,20 +548,20 @@ func (f *Fragment) importSetFieldValue(columnID uint64, bitDepth uint, value uin
if value&(1<<i) != 0 {
bit, err := f.pos(uint64(i), columnID)
if err != nil {
return changed, err
return changed, errors.Wrap(err, "getting set pos")
}
if c, err := f.storage.Add(bit); err != nil {
return changed, err
return changed, errors.Wrap(err, "adding")
} else if c {
changed = true
}
} else {
bit, err := f.pos(uint64(i), columnID)
if err != nil {
return changed, err
return changed, errors.Wrap(err, "getting clear pos")
}
if c, err := f.storage.Remove(bit); err != nil {
return changed, err
return changed, errors.Wrap(err, "removing")
} else if c {
changed = true
}
@ -571,10 +571,10 @@ func (f *Fragment) importSetFieldValue(columnID uint64, bitDepth uint, value uin
// Mark value as set.
p, err := f.pos(uint64(bitDepth), columnID)
if err != nil {
return changed, err
return changed, errors.Wrap(err, "marking not-null")
}
if c, err := f.storage.Add(p); err != nil {
return changed, err
return changed, errors.Wrap(err, "adding to storage")
} else if c {
changed = true
}
@ -945,7 +945,7 @@ func (f *Fragment) Top(opt TopOptions) ([]Pair, error) {
if filters != nil {
attr, err := f.RowAttrStore.Attrs(rowID)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "getting attrs")
} else if attr == nil {
continue
} else if attrValue := attr[opt.FilterField]; attrValue == nil {
@ -1308,14 +1308,14 @@ func (f *Fragment) MergeBlock(id int, data []PairSet) (sets, clears []PairSet, e
// Set local bits.
for i := range sets[0].ColumnIDs {
if _, err := f.setBit(sets[0].RowIDs[i], (f.Slice()*SliceWidth)+sets[0].ColumnIDs[i]); err != nil {
return nil, nil, err
return nil, nil, errors.Wrap(err, "setting")
}
}
// Clear local bits.
for i := range clears[0].ColumnIDs {
if _, err := f.clearBit(clears[0].RowIDs[i], (f.Slice()*SliceWidth)+clears[0].ColumnIDs[i]); err != nil {
return nil, nil, err
return nil, nil, errors.Wrap(err, "clearing")
}
}
@ -1346,13 +1346,13 @@ func (f *Fragment) Import(rowIDs, columnIDs []uint64) error {
// Determine the position of the bit in the storage.
pos, err := f.pos(rowID, columnID)
if err != nil {
return err
return errors.Wrap(err, "getting bit pos")
}
// Write to storage.
_, err = f.storage.Add(pos)
if err != nil {
return err
return errors.Wrap(err, "writing")
}
// Reduce the StatsD rate for high volume stats
f.stats.Count("ImportBit", 1, 0.0001)
@ -1386,7 +1386,7 @@ func (f *Fragment) Import(rowIDs, columnIDs []uint64) error {
// Write the storage to disk and reload.
if err := f.snapshot(); err != nil {
return err
return errors.Wrap(err, "snapshotting")
}
return nil
@ -1410,7 +1410,7 @@ func (f *Fragment) ImportValue(columnIDs, values []uint64, bitDepth uint) error
_, err := f.importSetFieldValue(columnID, bitDepth, value)
if err != nil {
return err
return errors.Wrap(err, "setting")
}
}
return nil
@ -1420,7 +1420,7 @@ func (f *Fragment) ImportValue(columnIDs, values []uint64, bitDepth uint) error
return err
}
if err := f.snapshot(); err != nil {
return err
return errors.Wrap(err, "snapshotting")
}
return nil
}
@ -1525,12 +1525,12 @@ func (f *Fragment) flushCache() error {
// Marshal cache data to bytes.
buf, err := proto.Marshal(&internal.Cache{IDs: ids})
if err != nil {
return err
return errors.Wrap(err, "marshalling")
}
// Write to disk.
if err := ioutil.WriteFile(f.CachePath(), buf, 0666); err != nil {
return err
return errors.Wrap(err, "writing")
}
return nil
@ -1540,7 +1540,7 @@ func (f *Fragment) flushCache() error {
func (f *Fragment) WriteTo(w io.Writer) (n int64, err error) {
// Force cache flush.
if err := f.FlushCache(); err != nil {
return 0, err
return 0, errors.Wrap(err, "flushing cache")
}
// Write out data and cache to a tar archive.
@ -1558,7 +1558,7 @@ func (f *Fragment) writeStorageToArchive(tw *tar.Writer) error {
// Open separate file descriptor to read from.
file, err := os.Open(f.path)
if err != nil {
return err
return errors.Wrap(err, "opening file")
}
defer file.Close()
@ -1571,7 +1571,7 @@ func (f *Fragment) writeStorageToArchive(tw *tar.Writer) error {
fi, err := file.Stat()
if err != nil {
return err
return errors.Wrap(err, "statting")
}
sz = fi.Size()
@ -1587,13 +1587,13 @@ func (f *Fragment) writeStorageToArchive(tw *tar.Writer) error {
Size: sz,
ModTime: time.Now(),
}); err != nil {
return err
return errors.Wrap(err, "writing header")
}
// Copy the file up to the last known size.
// This is done outside the lock because the storage format is append-only.
if _, err := io.CopyN(tw, file, sz); err != nil {
return err
return errors.Wrap(err, "copying")
}
return nil
}
@ -1607,7 +1607,7 @@ func (f *Fragment) writeCacheToArchive(tw *tar.Writer) error {
if os.IsNotExist(err) {
return nil
} else if err != nil {
return err
return errors.Wrap(err, "reading cache")
}
// Write archive header.
@ -1617,12 +1617,12 @@ func (f *Fragment) writeCacheToArchive(tw *tar.Writer) error {
Size: int64(len(buf)),
ModTime: time.Now(),
}); err != nil {
return err
return errors.Wrap(err, "writing header")
}
// Write data to archive.
if _, err := tw.Write(buf); err != nil {
return err
return errors.Wrap(err, "writing")
}
return nil
}
@ -1639,18 +1639,18 @@ func (f *Fragment) ReadFrom(r io.Reader) (n int64, err error) {
if err == io.EOF {
break
} else if err != nil {
return 0, err
return 0, errors.Wrap(err, "opening")
}
// Process file based on file name.
switch hdr.Name {
case "data":
if err := f.readStorageFromArchive(tr); err != nil {
return 0, err
return 0, errors.Wrap(err, "reading storage")
}
case "cache":
if err := f.readCacheFromArchive(tr); err != nil {
return 0, err
return 0, errors.Wrap(err, "reading cache")
}
default:
return 0, fmt.Errorf("invalid fragment archive file: %s", hdr.Name)
@ -1665,28 +1665,28 @@ func (f *Fragment) readStorageFromArchive(r io.Reader) error {
path := f.path + CopyExt
file, err := os.Create(path)
if err != nil {
return err
return errors.Wrap(err, "creating directory")
}
defer file.Close()
// Copy reader into temporary path.
if _, err = io.Copy(file, r); err != nil {
return err
return errors.Wrap(err, "copying")
}
// Close current storage.
if err := f.closeStorage(); err != nil {
return err
return errors.Wrap(err, "closing")
}
// Move snapshot to data file location.
if err := os.Rename(path, f.path); err != nil {
return err
return errors.Wrap(err, "renaming")
}
// Reopen storage.
if err := f.openStorage(); err != nil {
return err
return errors.Wrap(err, "opening")
}
return nil
@ -1696,14 +1696,14 @@ func (f *Fragment) readCacheFromArchive(r io.Reader) error {
// Slurp data from reader and write to disk.
buf, err := ioutil.ReadAll(r)
if err != nil {
return err
return errors.Wrap(err, "reading")
} else if err := ioutil.WriteFile(f.CachePath(), buf, 0666); err != nil {
return err
return errors.Wrap(err, "writing")
}
// Re-open cache.
if err := f.openCache(); err != nil {
return err
return errors.Wrap(err, "opening")
}
return nil
@ -1785,7 +1785,7 @@ func (s *FragmentSyncer) SyncFragment() error {
client := NewInternalHTTPClientFromURI(&node.URI, s.RemoteClient)
blocks, err := client.FragmentBlocks(context.Background(), s.Fragment.Index(), s.Fragment.Frame(), s.Fragment.View(), s.Fragment.Slice())
if err != nil && err != ErrFragmentNotFound {
return err
return errors.Wrap(err, "getting blocks")
}
blockSets = append(blockSets, blocks)
@ -1864,7 +1864,7 @@ func (s *FragmentSyncer) syncBlock(id int) error {
// Only sync the standard block.
rowIDs, columnIDs, err := client.BlockData(context.Background(), f.Index(), f.Frame(), ViewStandard, f.Slice(), id)
if err != nil {
return err
return errors.Wrap(err, "getting block")
}
pairSets = append(pairSets, PairSet{
@ -1881,7 +1881,7 @@ func (s *FragmentSyncer) syncBlock(id int) error {
// Merge blocks together.
sets, clears, err := f.MergeBlock(id, pairSets)
if err != nil {
return err
return errors.Wrap(err, "merging")
}
// Write updates to remote blocks.
@ -1926,7 +1926,7 @@ func (s *FragmentSyncer) syncBlock(id int) error {
}
_, err := clients[i].Query(context.Background(), f.Index(), queryRequest)
if err != nil {
return err
return errors.Wrap(err, "executing")
}
}
}

65
fragment_test.go
View file

@ -1171,3 +1171,68 @@ func BenchmarkFragment_Snapshot(b *testing.B) {
}
}
}
func BenchmarkFragment_FullSnapshot(b *testing.B) {
f := test.MustOpenFragment("i", "f", pilosa.ViewStandard, 0, "")
defer f.Close()
// Generate some intersecting data.
maxX := 1048576 / 2
sz := maxX
rows := make([]uint64, sz, sz)
cols := make([]uint64, sz, sz)
max := 0
for row := 0; row < 100; row++ {
val := 1
i := 0
for col := 0; col < SliceWidth/2; col++ {
rows[i] = uint64(row)
cols[i] = uint64(val)
val += 2
i++
}
if err := f.Import(rows, cols); err != nil {
b.Fatalf("Error Building Sample: %s", err)
}
if row > max {
max = row
}
}
b.ResetTimer()
b.ReportAllocs()
for i := 0; i < b.N; i++ {
if err := f.Snapshot(); err != nil {
b.Fatal(err)
}
}
}
func BenchmarkFragment_Import(b *testing.B) {
f := test.MustOpenFragment("i", "f", pilosa.ViewStandard, 0, "")
defer f.Close()
maxX := 1048576 * 5 * 2
sz := maxX
rows := make([]uint64, sz, sz)
cols := make([]uint64, sz, sz)
i := 0
for row := 0; row < 100; row++ {
val := 1
for col := 0; col < SliceWidth/2; col++ {
rows[i] = uint64(row)
cols[i] = uint64(val)
val += 2
i++
}
if i == maxX {
break
}
}
b.ResetTimer()
b.ReportAllocs()
for i := 0; i < b.N; i++ {
if err := f.Import(rows, cols); err != nil {
b.Fatalf("Error Building Sample: %s", err)
}
}
}

74
frame.go
View file

@ -15,7 +15,6 @@
package pilosa
import (
"errors"
"fmt"
"io/ioutil"
"os"
@ -27,6 +26,7 @@ import (
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/pql"
"github.com/pkg/errors"
)
// Default frame settings.
@ -156,7 +156,7 @@ func (f *Frame) SetCacheSize(v uint32) error {
// Persist meta data to disk on change.
f.cacheSize = v
if err := f.saveMeta(); err != nil {
return err
return errors.Wrap(err, "saving")
}
return nil
@ -192,19 +192,19 @@ func (f *Frame) Open() error {
if err := func() error {
// Ensure the frame's path exists.
if err := os.MkdirAll(f.path, 0777); err != nil {
return err
return errors.Wrap(err, "creating frame dir")
}
if err := f.loadMeta(); err != nil {
return err
return errors.Wrap(err, "loading meta")
}
if err := f.openViews(); err != nil {
return err
return errors.Wrap(err, "opening views")
}
if err := f.rowAttrStore.Open(); err != nil {
return err
return errors.Wrap(err, "opening attrstore")
}
return nil
@ -222,13 +222,13 @@ func (f *Frame) openViews() error {
if os.IsNotExist(err) {
return nil
} else if err != nil {
return err
return errors.Wrap(err, "opening view directory")
}
defer file.Close()
fis, err := file.Readdir(0)
if err != nil {
return err
return errors.Wrap(err, "reading directory")
}
for _, fi := range fis {
@ -262,10 +262,10 @@ func (f *Frame) loadMeta() error {
//f.fields
return nil
} else if err != nil {
return err
return errors.Wrap(err, "reading meta")
} else {
if err := proto.Unmarshal(buf, &pb); err != nil {
return err
return errors.Wrap(err, "unmarshaling")
}
}
@ -288,12 +288,12 @@ func (f *Frame) saveMeta() error {
fo := f.options()
buf, err := proto.Marshal(fo.Encode())
if err != nil {
return err
return errors.Wrap(err, "marshaling")
}
// Write to meta file.
if err := ioutil.WriteFile(filepath.Join(f.path, ".meta"), buf, 0666); err != nil {
return err
return errors.Wrap(err, "writing meta")
}
return nil
@ -365,7 +365,7 @@ func (f *Frame) CreateField(field *Field) error {
// addField adds a single field to fields.
func (f *Frame) addField(field *Field) error {
if err := ValidateField(field); err != nil {
return err
return errors.Wrap(err, "validating field")
} else if f.HasField(field.Name) {
return ErrFieldExists
}
@ -388,7 +388,7 @@ func (f *Frame) GetFields() ([]*Field, error) {
err := f.loadMeta()
if err != nil {
return nil, err
return nil, errors.Wrap(err, "loading meta")
}
return f.fields, nil
@ -410,9 +410,9 @@ func (f *Frame) DeleteField(name string) error {
delete(f.views, viewName)
if err := view.Close(); err != nil {
return err
return errors.Wrap(err, "closing view")
} else if err := os.RemoveAll(view.Path()); err != nil {
return err
return errors.Wrap(err, "deleting directory")
}
}
@ -453,7 +453,7 @@ func (f *Frame) SetTimeQuantum(q TimeQuantum) error {
// Persist meta data to disk.
if err := f.saveMeta(); err != nil {
return err
return errors.Wrap(err, "saving meta")
}
return nil
@ -522,7 +522,7 @@ func (f *Frame) CreateViewIfNotExists(name string) (*View, error) {
View: name,
})
if err != nil {
return nil, err
return nil, errors.Wrap(err, "sending CreateView message")
}
}
@ -547,7 +547,7 @@ func (f *Frame) createViewIfNotExistsBase(name string) (*View, bool, error) {
view := f.newView(f.ViewPath(name), name)
if err := view.Open(); err != nil {
return nil, false, err
return nil, false, errors.Wrap(err, "opening view")
}
view.RowAttrStore = f.rowAttrStore
f.views[view.Name()] = view
@ -574,12 +574,12 @@ func (f *Frame) DeleteView(name string) error {
// Close data files before deletion.
if err := view.Close(); err != nil {
return err
return errors.Wrap(err, "closing view")
}
// Delete view directory.
if err := os.RemoveAll(view.Path()); err != nil {
return err
return errors.Wrap(err, "deleting directory")
}
delete(f.views, name)
@ -597,12 +597,12 @@ func (f *Frame) SetBit(name string, rowID, colID uint64, t *time.Time) (changed
// Retrieve view. Exit if it doesn't exist.
view, err := f.CreateViewIfNotExists(name)
if err != nil {
return changed, err
return changed, errors.Wrap(err, "creating view")
}
// Set non-time bit.
if v, err := view.SetBit(rowID, colID); err != nil {
return changed, err
return changed, errors.Wrap(err, "setting on view")
} else if v {
changed = v
}
@ -616,11 +616,11 @@ func (f *Frame) SetBit(name string, rowID, colID uint64, t *time.Time) (changed
for _, subname := range ViewsByTime(name, *t, f.TimeQuantum()) {
view, err := f.CreateViewIfNotExists(subname)
if err != nil {
return changed, err
return changed, errors.Wrapf(err, "creating view %s", subname)
}
if c, err := view.SetBit(rowID, colID); err != nil {
return changed, err
return changed, errors.Wrapf(err, "setting on view %s", subname)
} else if c {
changed = true
}
@ -639,12 +639,12 @@ func (f *Frame) ClearBit(name string, rowID, colID uint64, t *time.Time) (change
// Retrieve view. Exit if it doesn't exist.
view, err := f.CreateViewIfNotExists(name)
if err != nil {
return changed, err
return changed, errors.Wrap(err, "creating view")
}
// Clear non-time bit.
if v, err := view.ClearBit(rowID, colID); err != nil {
return changed, err
return changed, errors.Wrap(err, "setting on view")
} else if v {
changed = v
}
@ -658,11 +658,11 @@ func (f *Frame) ClearBit(name string, rowID, colID uint64, t *time.Time) (change
for _, subname := range ViewsByTime(name, *t, f.TimeQuantum()) {
view, err := f.CreateViewIfNotExists(subname)
if err != nil {
return changed, err
return changed, errors.Wrapf(err, "creating view %s", subname)
}
if c, err := view.ClearBit(rowID, colID); err != nil {
return changed, err
return changed, errors.Wrapf(err, "setting on view %s", subname)
} else if c {
changed = true
}
@ -708,7 +708,7 @@ func (f *Frame) SetFieldValue(columnID uint64, name string, value int64) (change
// Fetch target view.
view, err := f.CreateViewIfNotExists(ViewFieldPrefix + name)
if err != nil {
return false, err
return false, errors.Wrap(err, "creating view")
}
// Determine base value to store.
@ -834,7 +834,11 @@ func (f *Frame) Import(rowIDs, columnIDs []uint64, timestamps []*time.Time) erro
// Split import data by fragment.
dataByFragment := make(map[importKey]importData)
for i := range rowIDs {
rowID, columnID, timestamp := rowIDs[i], columnIDs[i], timestamps[i]
rowID, columnID := rowIDs[i], columnIDs[i]
var timestamp *time.Time
if len(timestamps) > i {
timestamp = timestamps[i]
}
var standard, inverse []string
if timestamp == nil {
@ -886,12 +890,12 @@ func (f *Frame) Import(rowIDs, columnIDs []uint64, timestamps []*time.Time) erro
view, err := f.CreateViewIfNotExists(key.View)
if err != nil {
return err
return errors.Wrap(err, "creating view")
}
frag, err := view.CreateFragmentIfNotExists(key.Slice)
if err != nil {
return err
return errors.Wrap(err, "creating view")
}
if err := frag.Import(data.RowIDs, data.ColumnIDs); err != nil {
@ -938,12 +942,12 @@ func (f *Frame) ImportValue(fieldName string, columnIDs []uint64, values []int64
// because we need to know bitDepth (based on min/max value).
view, err := f.CreateViewIfNotExists(key.View)
if err != nil {
return err
return errors.Wrap(err, "creating view")
}
frag, err := view.CreateFragmentIfNotExists(key.Slice)
if err != nil {
return err
return errors.Wrap(err, "creating fragment")
}
baseValues := make([]uint64, len(data.Values))

6
gossip/gossip.go
View file

@ -169,12 +169,14 @@ func WithLogger(logger *log.Logger) GossipMemberSetOption {
// NewGossipMemberSet returns a new instance of GossipMemberSet based on options.
func NewGossipMemberSet(name string, host string, cfg Config, ger *GossipEventReceiver, sh pilosa.StatusHandler, options ...GossipMemberSetOption) (*GossipMemberSet, error) {
g := &GossipMemberSet{}
g := &GossipMemberSet{
Logger: pilosa.NopLogger,
}
// options
for _, opt := range options {
if err := opt(g); err != nil {
return nil, err
return nil, errors.Wrap(err, "executing option")
}
}

30
handler.go
View file

@ -126,6 +126,7 @@ func NewRouter(handler *Handler) *mux.Router {
router.HandleFunc("/schema", handler.handleGetSchema).Methods("GET")
router.HandleFunc("/slices/max", handler.handleGetSlicesMax).Methods("GET") // TODO: deprecate, but it's being used by the client (for backups)
router.HandleFunc("/status", handler.handleGetStatus).Methods("GET")
router.HandleFunc("/info", handler.handleGetInfo).Methods("GET")
router.HandleFunc("/version", handler.handleGetVersion).Methods("GET")
router.HandleFunc("/cluster/resize/abort", handler.handlePostClusterResizeAbort).Methods("POST")
@ -252,6 +253,13 @@ func (h *Handler) handleGetStatus(w http.ResponseWriter, r *http.Request) {
}
}
func (h *Handler) handleGetInfo(w http.ResponseWriter, r *http.Request) {
info := h.API.Info()
if err := json.NewEncoder(w).Encode(info); err != nil {
h.Logger.Printf("write info response error: %s", err)
}
}
type getSchemaResponse struct {
Indexes []*IndexInfo `json:"indexes"`
}
@ -349,7 +357,7 @@ func (p *postIndexRequest) UnmarshalJSON(b []byte) error {
// m is an overflow map used to capture additional, unexpected keys.
m := make(map[string]interface{})
if err := json.Unmarshal(b, &m); err != nil {
return err
return errors.Wrap(err, "unmarshalling unexpected values")
}
validIndexOptions := getValidOptions(IndexOptions{})
@ -360,7 +368,7 @@ func (p *postIndexRequest) UnmarshalJSON(b []byte) error {
// Unmarshal expected values.
var _p _postIndexRequest
if err := json.Unmarshal(b, &_p); err != nil {
return err
return errors.Wrap(err, "unmarshalling expected values")
}
p.Options = _p.Options
@ -526,7 +534,7 @@ func (p *postFrameRequest) UnmarshalJSON(b []byte) error {
// m is an overflow map used to capture additional, unexpected keys.
m := make(map[string]interface{})
if err := json.Unmarshal(b, &m); err != nil {
return err
return errors.Wrap(err, "unmarshaling unexpected keys")
}
validFrameOptions := getValidOptions(FrameOptions{})
@ -538,7 +546,7 @@ func (p *postFrameRequest) UnmarshalJSON(b []byte) error {
// Unmarshal expected values.
var _p _postFrameRequest
if err := json.Unmarshal(b, &_p); err != nil {
return err
return errors.Wrap(err, "unmarshalling expected keys")
}
p.Options = _p.Options
@ -813,13 +821,13 @@ func (h *Handler) readProtobufQueryRequest(r *http.Request) (*QueryRequest, erro
// Slurp the body.
body, err := ioutil.ReadAll(r.Body)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "reading")
}
// Unmarshal into object.
var req internal.QueryRequest
if err := proto.Unmarshal(body, &req); err != nil {
return nil, err
return nil, errors.Wrap(err, "unmarshalling")
}
return decodeQueryRequest(&req), nil
@ -832,7 +840,7 @@ func (h *Handler) readURLQueryRequest(r *http.Request) (*QueryRequest, error) {
// Parse query string.
buf, err := ioutil.ReadAll(r.Body)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "reading")
}
query := string(buf)
@ -877,9 +885,9 @@ func (h *Handler) writeQueryResponse(w http.ResponseWriter, r *http.Request, res
// writeProtobufQueryResponse writes the response from the executor to w as protobuf.
func (h *Handler) writeProtobufQueryResponse(w http.ResponseWriter, resp *QueryResponse) error {
if buf, err := proto.Marshal(encodeQueryResponse(resp)); err != nil {
return err
return errors.Wrap(err, "marshalling")
} else if _, err := w.Write(buf); err != nil {
return err
return errors.Wrap(err, "writing")
}
return nil
}
@ -1327,7 +1335,7 @@ func parseUint64Slice(s string) ([]uint64, error) {
// Parse number.
num, err := strconv.ParseUint(str, 10, 64)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "parsing int")
}
a = append(a, num)
}
@ -1591,7 +1599,7 @@ func GetTimeStamp(data map[string]interface{}, timeField string) (int64, error)
v, err := time.Parse(TimeFormat, timestamp)
if err != nil {
return 0, err
return 0, errors.Wrap(err, "parsing timestamp")
}
return v.Unix(), nil

29
handler_test.go
View file

@ -19,6 +19,7 @@ import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"io/ioutil"
"net/http"
@ -154,6 +155,20 @@ func TestHandler_Status(t *testing.T) {
}
}
func TestHandler_Info(t *testing.T) {
s := test.NewServer()
defer s.Close()
h := test.NewHandler()
w := httptest.NewRecorder()
h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/info", nil))
if w.Code != http.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != fmt.Sprintf("{\"sliceWidth\":%d}\n", SliceWidth) {
t.Fatalf("unexpected body: %s", body)
}
}
// Ensure the handler can abort a cluster resize.
func TestHandler_ClusterResizeAbort(t *testing.T) {
@ -624,7 +639,7 @@ func TestHandler_Query_Err_JSON(t *testing.T) {
h.ServeHTTP(w, test.MustNewHTTPRequest("POST", "/index/i/query", strings.NewReader(`Bitmap(id=100)`)))
if w.Code != http.StatusBadRequest {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"error":"marker"}`+"\n" {
} else if body := w.Body.String(); body != `{"error":"executing: marker"}`+"\n" {
t.Fatalf("unexpected body: %q", body)
}
}
@ -652,7 +667,7 @@ func TestHandler_Query_Err_Protobuf(t *testing.T) {
var resp internal.QueryResponse
if err := proto.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
} else if s := resp.Err; s != `marker` {
} else if s := resp.Err; s != `executing: marker` {
t.Fatalf("unexpected error: %s", s)
}
}
@ -684,7 +699,7 @@ func TestHandler_Query_ErrParse(t *testing.T) {
h.ServeHTTP(w, test.MustNewHTTPRequest("POST", "/index/idx0/query?slices=0,1", strings.NewReader("bad_fn(")))
if w.Code != http.StatusBadRequest {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"error":"expected comma, right paren, or identifier, found \"\" occurred at line 1, char 8"}`+"\n" {
} else if body := w.Body.String(); body != `{"error":"parsing: expected comma, right paren, or identifier, found \"\" occurred at line 1, char 8"}`+"\n" {
t.Fatalf("unexpected body: %s", body)
}
}
@ -894,7 +909,7 @@ func TestHandler_Frame_AddField(t *testing.T) {
)
if err != nil {
t.Fatal(err)
} else if body := MustReadAll(resp.Body); string(body) != `invalid field type`+"\n" {
} else if body := MustReadAll(resp.Body); string(body) != `creating field: validating field: invalid field type`+"\n" {
t.Fatalf("unexpected body: %q", body)
} else if err := resp.Body.Close(); err != nil {
t.Fatal(err)
@ -916,7 +931,7 @@ func TestHandler_Frame_AddField(t *testing.T) {
)
if err != nil {
t.Fatal(err)
} else if body := MustReadAll(resp.Body); string(body) != `invalid field range`+"\n" {
} else if body := MustReadAll(resp.Body); string(body) != `creating field: validating field: invalid field range`+"\n" {
t.Fatalf("unexpected body: %q", body)
} else if err := resp.Body.Close(); err != nil {
t.Fatal(err)
@ -940,7 +955,7 @@ func TestHandler_Frame_AddField(t *testing.T) {
)
if err != nil {
t.Fatal(err)
} else if body := MustReadAll(resp.Body); string(body) != `field already exists`+"\n" {
} else if body := MustReadAll(resp.Body); string(body) != `creating field: field already exists`+"\n" {
t.Fatalf("unexpected body: %q", body)
} else if err := resp.Body.Close(); err != nil {
t.Fatal(err)
@ -1006,7 +1021,7 @@ func TestHandler_Frame_DeleteField(t *testing.T) {
t.Fatal(err)
} else if body, err := ioutil.ReadAll(resp.Body); err != nil {
t.Fatal(err)
} else if strings.TrimSpace(string(body)) != `field not found` {
} else if strings.TrimSpace(string(body)) != `deleting field: field not found` {
t.Fatalf("unexpected body: %q", body)
} else if err := resp.Body.Close(); err != nil {
t.Fatal(err)

54
holder.go
View file

@ -96,19 +96,19 @@ func (h *Holder) Open() error {
h.Logger.Printf("open holder path: %s", h.Path)
if err := os.MkdirAll(h.Path, 0777); err != nil {
return err
return errors.Wrap(err, "creating directory")
}
// Open path to read all index directories.
f, err := os.Open(h.Path)
if err != nil {
return err
return errors.Wrap(err, "opening directory")
}
defer f.Close()
fis, err := f.Readdir(0)
if err != nil {
return err
return errors.Wrap(err, "reading directory")
}
for _, fi := range fis {
@ -123,7 +123,7 @@ func (h *Holder) Open() error {
h.Logger.Printf("ERROR opening index: %s, err=%s", fi.Name(), err)
continue
} else if err != nil {
return err
return errors.Wrap(err, "opening index")
}
if err := index.Open(); err != nil {
if err == ErrName {
@ -158,7 +158,7 @@ func (h *Holder) Close() error {
for _, index := range h.indexes {
if err := index.Close(); err != nil {
return err
return errors.Wrap(err, "closing index")
}
}
return nil
@ -245,20 +245,20 @@ func (h *Holder) ApplySchema(schema *internal.Schema) error {
opt := IndexOptions{}
idx, err := h.CreateIndexIfNotExists(index.Name, opt)
if err != nil {
return err
return errors.Wrap(err, "creating index")
}
// Create frames that don't exist.
for _, f := range index.Frames {
opt := decodeFrameOptions(f.Meta)
frame, err := idx.CreateFrameIfNotExists(f.Name, *opt)
if err != nil {
return err
return errors.Wrap(err, "creating frame")
}
// Create views that don't exist.
for _, v := range f.Views {
_, err := frame.CreateViewIfNotExists(v)
if err != nil {
return err
return errors.Wrap(err, "creating view")
}
}
}
@ -347,11 +347,11 @@ func (h *Holder) createIndex(name string, opt IndexOptions) (*Index, error) {
// Otherwise create a new index.
index, err := h.newIndex(h.IndexPath(name), name)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "creating")
}
if err := index.Open(); err != nil {
return nil, err
return nil, errors.Wrap(err, "opening")
}
// Update options.
@ -387,12 +387,12 @@ func (h *Holder) DeleteIndex(name string) error {
// Close index.
if err := index.Close(); err != nil {
return err
return errors.Wrap(err, "closing")
}
// Delete index directory.
if err := os.RemoveAll(h.IndexPath(name)); err != nil {
return err
return errors.Wrap(err, "removing directory")
}
// Remove reference.
@ -528,7 +528,7 @@ func (h *Holder) loadNodeID() (string, error) {
nodeID := ""
h.Logger.Printf("load NodeID: %s", idPath)
if err := os.MkdirAll(h.Path, 0777); err != nil {
return "", err
return "", errors.Wrap(err, "creating directory")
}
nodeIDBytes, err := ioutil.ReadFile(idPath)
@ -538,10 +538,10 @@ func (h *Holder) loadNodeID() (string, error) {
nodeID = uuid.NewV4().String()
err = ioutil.WriteFile(idPath, []byte(nodeID), 0600)
if err != nil {
return "", err
return "", errors.Wrap(err, "writing file")
}
} else if err != nil {
return "", err
return "", errors.Wrap(err, "reading file")
}
return nodeID, nil
@ -667,7 +667,7 @@ func (s *HolderSyncer) syncIndex(index string) error {
// Read block checksums.
blks, err := idx.ColumnAttrStore().Blocks()
if err != nil {
return err
return errors.Wrap(err, "getting blocks")
}
s.Stats.CountWithCustomTags("ColumnAttrStoreBlocks", int64(len(blks)), 1.0, []string{indexTag})
@ -679,7 +679,7 @@ func (s *HolderSyncer) syncIndex(index string) error {
// Skip update and recomputation if no attributes have changed.
m, err := client.ColumnAttrDiff(context.Background(), index, blks)
if err != nil {
return err
return errors.Wrap(err, "getting differing blocks")
} else if len(m) == 0 {
continue
}
@ -687,13 +687,13 @@ func (s *HolderSyncer) syncIndex(index string) error {
// Update local copy.
if err := idx.ColumnAttrStore().SetBulkAttrs(m); err != nil {
return err
return errors.Wrap(err, "setting attrs")
}
// Recompute blocks.
blks, err = idx.ColumnAttrStore().Blocks()
if err != nil {
return err
return errors.Wrap(err, "recomputing blocks")
}
}
@ -713,7 +713,7 @@ func (s *HolderSyncer) syncFrame(index, name string) error {
// Read block checksums.
blks, err := f.RowAttrStore().Blocks()
if err != nil {
return err
return errors.Wrap(err, "getting blocks")
}
s.Stats.CountWithCustomTags("RowAttrStoreBlocks", int64(len(blks)), 1.0, []string{indexTag, frameTag})
@ -727,7 +727,7 @@ func (s *HolderSyncer) syncFrame(index, name string) error {
if err == ErrFrameNotFound {
continue // frame not created remotely yet, skip
} else if err != nil {
return err
return errors.Wrap(err, "getting differing blocks")
} else if len(m) == 0 {
continue
}
@ -735,13 +735,13 @@ func (s *HolderSyncer) syncFrame(index, name string) error {
// Update local copy.
if err := f.RowAttrStore().SetBulkAttrs(m); err != nil {
return err
return errors.Wrap(err, "setting attrs")
}
// Recompute blocks.
blks, err = f.RowAttrStore().Blocks()
if err != nil {
return err
return errors.Wrap(err, "recomputing blocks")
}
}
@ -759,13 +759,13 @@ func (s *HolderSyncer) syncFragment(index, frame, view string, slice uint64) err
// Ensure view exists locally.
v, err := f.CreateViewIfNotExists(view)
if err != nil {
return err
return errors.Wrap(err, "creating view")
}
// Ensure fragment exists locally.
frag, err := v.CreateFragmentIfNotExists(slice)
if err != nil {
return err
return errors.Wrap(err, "creating fragment")
}
// Sync fragments together.
@ -777,7 +777,7 @@ func (s *HolderSyncer) syncFragment(index, frame, view string, slice uint64) err
RemoteClient: s.RemoteClient,
}
if err := fs.SyncFragment(); err != nil {
return err
return errors.Wrap(err, "syncing fragment")
}
return nil
@ -827,7 +827,7 @@ func (c *HolderCleaner) CleanHolder() error {
}
// Delete fragment.
if err := view.DeleteFragment(fragSlice); err != nil {
return err
return errors.Wrap(err, "deleting fragment")
}
}
}

8
holder_test.go
View file

@ -86,7 +86,7 @@ func TestHolder_Open(t *testing.T) {
t.Fatal(err)
}
if err := h.Reopen(); err == nil || !strings.Contains(err.Error(), "open index: name=test, err=opening attrstore: invalid database") {
if err := h.Reopen(); err == nil || !strings.Contains(err.Error(), "open index: name=test, err=opening attrstore: opening storage: invalid database") {
t.Fatalf("unexpected error: %s", err)
}
})
@ -127,7 +127,7 @@ func TestHolder_Open(t *testing.T) {
t.Fatal(err)
}
if err := h.Reopen(); err == nil || !strings.Contains(err.Error(), "open index: name=foo, err=opening frames: open frame: name=bar, err=unexpected EOF") {
if err := h.Reopen(); err == nil || !strings.Contains(err.Error(), "open index: name=foo, err=opening frames: open frame: name=bar, err=loading meta: unmarshaling: unexpected EOF") {
t.Fatalf("unexpected error: %s", err)
}
})
@ -145,7 +145,7 @@ func TestHolder_Open(t *testing.T) {
t.Fatal(err)
}
if err := h.Reopen(); err == nil || !strings.Contains(err.Error(), "open index: name=foo, err=opening frames: open frame: name=bar, err=invalid database") {
if err := h.Reopen(); err == nil || !strings.Contains(err.Error(), "open index: name=foo, err=opening frames: open frame: name=bar, err=opening attrstore: opening storage: invalid database") {
t.Fatalf("unexpected error: %s", err)
}
})
@ -243,7 +243,7 @@ func TestHolder_Open(t *testing.T) {
t.Fatal(err)
}
if err := h.Reopen(); err == nil || !strings.Contains(err.Error(), "open fragment: slice=0, err=unmarshal storage") {
if err := h.Reopen(); err == nil || !strings.Contains(err.Error(), "open fragment: slice=0, err=opening storage: unmarshal storage") {
t.Fatalf("unexpected error: %s", err)
}
})

8
index.go
View file

@ -642,7 +642,8 @@ func (i *Index) openInputDefinitions() error {
// InputBits Process the []Bit though the Frame import process
func (i *Index) InputBits(frame string, bits []*Bit) error {
var rowIDs, columnIDs []uint64
timestamps := make([]*time.Time, len(bits))
var timestamps []*time.Time
f := i.Frame(frame)
if f == nil {
return fmt.Errorf("Frame not found: %s", frame)
@ -657,6 +658,11 @@ func (i *Index) InputBits(frame string, bits []*Bit) error {
// Convert timestamps to time.Time.
if bit.Timestamp > 0 {
// Don't create a full timestamps slice unless
// at least one bit contains a timestamp.
if len(timestamps) == 0 {
timestamps = make([]*time.Time, len(bits))
}
t := time.Unix(bit.Timestamp, 0)
timestamps[i] = &t
}

27
server.go
View file

@ -22,6 +22,7 @@ import (
"net/http"
"os"
"os/exec"
"path/filepath"
"runtime"
"strconv"
"strings"
@ -77,6 +78,7 @@ type Server struct {
maxWritesPerRequest int
defaultClient InternalClient
dataDir string
}
// ServerOption is a functional option type for pilosa.Server
@ -98,8 +100,7 @@ func OptServerReplicaN(n int) ServerOption {
func OptServerDataDir(dir string) ServerOption {
return func(s *Server) error {
s.Cluster.Path = dir
s.Holder.Path = dir
s.dataDir = dir
return nil
}
}
@ -231,9 +232,16 @@ func NewServer(opts ...ServerOption) (*Server, error) {
}
}
path, err := expandDirName(s.dataDir)
if err != nil {
return nil, err
}
s.Holder.Path = path
s.Holder.Logger = s.logger
s.Holder.Stats.SetLogger(s.logger)
s.Cluster.Path = path
s.Cluster.Logger = s.logger
s.Cluster.Holder = s.Holder
@ -292,7 +300,6 @@ func (s *Server) Open() error {
s.handler.API.Broadcaster = s.Broadcaster
s.handler.API.BroadcastHandler = s
s.handler.API.StatusHandler = s
s.handler.API.URI = s.URI
s.handler.API.Cluster = s.Cluster
// Initialize Holder.
@ -622,7 +629,7 @@ func (s *Server) mergeRemoteStatus(ns *internal.NodeStatus) error {
// Sync schema.
if err := s.Holder.ApplySchema(ns.Schema); err != nil {
return err
return errors.Wrap(err, "applying schema")
}
// Sync maxSlices (standard).
@ -784,3 +791,15 @@ type StatusHandler interface {
ClusterStatus() (proto.Message, error)
HandleRemoteStatus(proto.Message) error
}
func expandDirName(path string) (string, error) {
prefix := "~" + string(filepath.Separator)
if strings.HasPrefix(path, prefix) {
HomeDir := os.Getenv("HOME")
if HomeDir == "" {
return "", errors.New("data directory not specified and no home dir available")
}
return filepath.Join(HomeDir, strings.TrimPrefix(path, prefix)), nil
}
return path, nil
}

35
server/server.go
View file

@ -28,9 +28,7 @@ import (
"net/http"
"os"
"os/signal"
"path/filepath"
"strconv"
"strings"
"syscall"
"time"
@ -93,30 +91,22 @@ func NewCommand(stdin io.Reader, stdout, stderr io.Writer) *Command {
// Start starts the pilosa server - it returns once the server is running.
func (m *Command) Start() (err error) {
defer close(m.Started)
prefix := "~" + string(filepath.Separator)
if strings.HasPrefix(m.Config.DataDir, prefix) {
HomeDir := os.Getenv("HOME")
if HomeDir == "" {
return errors.New("data directory not specified and no home dir available")
}
m.Config.DataDir = filepath.Join(HomeDir, strings.TrimPrefix(m.Config.DataDir, prefix))
}
// SetupServer
err = m.SetupServer()
if err != nil {
return err
return errors.Wrap(err, "setting up server")
}
// SetupNetworking
err = m.SetupNetworking()
if err != nil {
return err
return errors.Wrap(err, "setting up networking")
}
// Initialize server.
if err = m.Server.Open(); err != nil {
return fmt.Errorf("server.Open: %v", err)
return errors.Wrap(err, "opening server")
}
m.logger.Printf("Listening as %s\n", m.Server.URI)
@ -271,7 +261,7 @@ func (m *Command) SetupNetworking() error {
for _, address := range m.Config.Cluster.Hosts {
uri, err := pilosa.NewURIFromAddress(address)
if err != nil {
return err
return errors.Wrap(err, "getting URI")
}
m.Server.Cluster.Nodes = append(m.Server.Cluster.Nodes, &pilosa.Node{
URI: *uri,
@ -287,7 +277,7 @@ func (m *Command) SetupNetworking() error {
gossipPort, err := strconv.Atoi(m.Config.Gossip.Port)
if err != nil {
return err
return errors.Wrap(err, "parsing port")
}
// get the host portion of addr to use for binding
@ -298,7 +288,7 @@ func (m *Command) SetupNetworking() error {
} else {
transport, err = gossip.NewTransport(gossipHost, gossipPort, m.logger.Logger())
if err != nil {
return err
return errors.Wrap(err, "getting transport")
}
}
@ -310,10 +300,19 @@ func (m *Command) SetupNetworking() error {
gossipEventReceiver := gossip.NewGossipEventReceiver(m.logger)
m.Server.Cluster.EventReceiver = gossipEventReceiver
gossipMemberSet, err := gossip.NewGossipMemberSet(m.Server.NodeID, m.Server.URI.Host(), m.Config.Gossip, gossipEventReceiver, m.Server, gossip.WithLogger(m.logger.Logger()), gossip.WithTransport(transport))
gossipMemberSet, err := gossip.NewGossipMemberSet(
m.Server.NodeID,
m.Server.URI.Host(),
m.Config.Gossip,
gossipEventReceiver,
m.Server,
gossip.WithLogger(m.logger.Logger()),
gossip.WithTransport(transport),
)
if err != nil {
return err
return errors.Wrap(err, "getting memberset")
}
gossipMemberSet.Logger = m.logger
m.Server.Cluster.MemberSet = gossipMemberSet
m.Server.Broadcaster = m.Server
m.Server.BroadcastReceiver = gossipMemberSet

4
server/server_test.go
View file

@ -52,10 +52,10 @@ func TestMain_Set_Quick(t *testing.T) {
// Execute SetBit() commands.
for _, cmd := range cmds {
if err := client.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists {
if err := client.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && !strings.Contains(err.Error(), "index already exists") {
t.Fatal(err)
}
if err := client.CreateFrame(context.Background(), "i", cmd.Frame, pilosa.FrameOptions{}); err != nil && err != pilosa.ErrFrameExists {
if err := client.CreateFrame(context.Background(), "i", cmd.Frame, pilosa.FrameOptions{}); err != nil && !strings.Contains(err.Error(), "frame already exists") {
t.Fatal(err)
}
if _, err := m.Query("i", "", fmt.Sprintf(`SetBit(row=%d, frame=%q, col=%d)`, cmd.ID, cmd.Frame, cmd.ColumnID)); err != nil {

7
test/handler.go
View file

@ -75,13 +75,6 @@ func NewServer() *Server {
}
s.Server = httptest.NewServer(s.Handler.Handler)
// Update handler to use hostname.
uri, err := pilosa.NewURIFromAddress(s.Host())
if err != nil {
panic(err)
}
s.Handler.API.URI = *uri
// Handler test messages can no-op.
s.Handler.API.Broadcaster = pilosa.NopBroadcaster
// Create a default cluster on the handler

4
uri.go
View file

@ -16,13 +16,13 @@ package pilosa
import (
"encoding/json"
"errors"
"fmt"
"regexp"
"strconv"
"strings"
"github.com/pilosa/pilosa/internal"
"github.com/pkg/errors"
)
var schemeRegexp = regexp.MustCompile("^[+a-z]+$")
@ -72,7 +72,7 @@ func NewURIFromHostPort(host string, port uint16) (*URI, error) {
uri := DefaultURI()
err := uri.SetHost(host)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "setting uri host")
}
uri.SetPort(port)
return uri, nil

21
view.go
View file

@ -24,6 +24,7 @@ import (
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/pql"
"github.com/pkg/errors"
)
// View layout modes.
@ -105,13 +106,13 @@ func (v *View) Open() error {
if err := func() error {
// Ensure the view's path exists.
if err := os.MkdirAll(v.path, 0777); err != nil {
return err
return errors.Wrap(err, "creating view directory")
} else if err := os.MkdirAll(filepath.Join(v.path, "fragments"), 0777); err != nil {
return err
return errors.Wrap(err, "creating fragments directory")
}
if err := v.openFragments(); err != nil {
return err
return errors.Wrap(err, "opening fragments")
}
return nil
@ -129,13 +130,13 @@ func (v *View) openFragments() error {
if os.IsNotExist(err) {
return nil
} else if err != nil {
return err
return errors.Wrap(err, "opening fragments directory")
}
defer file.Close()
fis, err := file.Readdir(0)
if err != nil {
return err
return errors.Wrap(err, "reading fragments directory")
}
for _, fi := range fis {
@ -168,7 +169,7 @@ func (v *View) Close() error {
// Close all fragments.
for _, frag := range v.fragments {
if err := frag.Close(); err != nil {
return err
return errors.Wrap(err, "closing fragment")
}
}
v.fragments = make(map[uint64]*Fragment)
@ -240,7 +241,7 @@ func (v *View) createFragmentIfNotExists(slice uint64) (*Fragment, error) {
// Initialize and open fragment.
frag := v.newFragment(v.FragmentPath(slice), slice)
if err := frag.Open(); err != nil {
return nil, err
return nil, errors.Wrap(err, "opening fragment")
}
frag.RowAttrStore = v.RowAttrStore
@ -256,7 +257,7 @@ func (v *View) createFragmentIfNotExists(slice uint64) (*Fragment, error) {
IsInverse: IsInverseView(v.name),
})
if err != nil {
return nil, err
return nil, errors.Wrap(err, "sending message")
}
}
@ -286,12 +287,12 @@ func (v *View) DeleteFragment(slice uint64) error {
// Close data files before deletion.
if err := fragment.Close(); err != nil {
return err
return errors.Wrap(err, "closing fragment")
}
// Delete fragment file.
if err := os.Remove(fragment.Path()); err != nil {
return err
return errors.Wrap(err, "deleting fragment file")
}
// Delete fragment cache file.