diff --git a/api.go b/api.go index 9f7356cb8..3f3e8bd50 100644 --- a/api.go +++ b/api.go @@ -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 , 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. diff --git a/boltdb/attrstore.go b/boltdb/attrstore.go index ebb539903..3604de5b6 100644 --- a/boltdb/attrstore.go +++ b/boltdb/attrstore.go @@ -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 } diff --git a/broadcast.go b/broadcast.go index de43f3b85..2be5e08db 100644 --- a/broadcast.go +++ b/broadcast.go @@ -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 } diff --git a/broadcast_test.go b/broadcast_test.go index 5654f7c28..970a249cb 100644 --- a/broadcast_test.go +++ b/broadcast_test.go @@ -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) } diff --git a/client.go b/client.go index 5997cf4ef..72225aef2 100644 --- a/client.go +++ b/client.go @@ -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 } diff --git a/client_test.go b/client_test.go index 88dc3accb..e2b806412 100644 --- a/client_test.go +++ b/client_test.go @@ -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 diff --git a/cluster.go b/cluster.go index b02bd50d2..9dfeac094 100644 --- a/cluster.go +++ b/cluster.go @@ -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") } } diff --git a/cluster_internal_test.go b/cluster_internal_test.go index da9989760..6b1f6e72b 100644 --- a/cluster_internal_test.go +++ b/cluster_internal_test.go @@ -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 { diff --git a/ctl/backup.go b/ctl/backup.go index 1b16391e0..54a760695 100644 --- a/ctl/backup.go +++ b/ctl/backup.go @@ -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 diff --git a/ctl/backup_test.go b/ctl/backup_test.go index d80b72475..a073bc8a8 100644 --- a/ctl/backup_test.go +++ b/ctl/backup_test.go @@ -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 diff --git a/ctl/bench.go b/ctl/bench.go index b9743bb5e..425894738 100644 --- a/ctl/bench.go +++ b/ctl/bench.go @@ -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 { diff --git a/ctl/bench_test.go b/ctl/bench_test.go index c3014e6ca..4790ccb44 100644 --- a/ctl/bench_test.go +++ b/ctl/bench_test.go @@ -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) } diff --git a/ctl/check.go b/ctl/check.go index 34b7033fc..04f947eb4 100644 --- a/ctl/check.go +++ b/ctl/check.go @@ -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. diff --git a/ctl/check_test.go b/ctl/check_test.go index 33a19feb3..56d70f97a 100644 --- a/ctl/check_test.go +++ b/ctl/check_test.go @@ -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 diff --git a/ctl/common.go b/ctl/common.go index 3629e233e..11042a704 100644 --- a/ctl/common.go +++ b/ctl/common.go @@ -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 } diff --git a/ctl/export.go b/ctl/export.go index 8c219a078..546d1ae1b 100644 --- a/ctl/export.go +++ b/ctl/export.go @@ -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") } } diff --git a/ctl/export_test.go b/ctl/export_test.go index b414f711f..48a90a065 100644 --- a/ctl/export_test.go +++ b/ctl/export_test.go @@ -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) } } diff --git a/ctl/import.go b/ctl/import.go index c3c117eb3..004e47e96 100644 --- a/ctl/import.go +++ b/ctl/import.go @@ -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") } } diff --git a/ctl/import_test.go b/ctl/import_test.go index 6cbc8271a..08eea8323 100644 --- a/ctl/import_test.go +++ b/ctl/import_test.go @@ -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 diff --git a/ctl/inspect.go b/ctl/inspect.go index f5825c35a..0c38c3fb5 100644 --- a/ctl/inspect.go +++ b/ctl/inspect.go @@ -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)) diff --git a/ctl/restore.go b/ctl/restore.go index 89ee8ac8d..ef863096e 100644 --- a/ctl/restore.go +++ b/ctl/restore.go @@ -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 diff --git a/ctl/restore_test.go b/ctl/restore_test.go index 04cb30415..896db57c5 100644 --- a/ctl/restore_test.go +++ b/ctl/restore_test.go @@ -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 diff --git a/diagnostics.go b/diagnostics.go index bfebbb495..46ffc2110 100644 --- a/diagnostics.go +++ b/diagnostics.go @@ -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() diff --git a/docs/query-language.md b/docs/query-language.md index 8ed0d5439..7d9f22886 100644 --- a/docs/query-language.md +++ b/docs/query-language.md @@ -160,8 +160,8 @@ Delete url value for repo 10. **Spec:** ``` -SetBit(, , , - [timestamp=TIMESTAMP]) +ClearBit(, , , + [timestamp=TIMESTAMP]) ``` **Description:** diff --git a/executor.go b/executor.go index 59e40511d..b6dc08d41 100644 --- a/executor.go +++ b/executor.go @@ -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. diff --git a/fragment.go b/fragment.go index 7168f4443..6b5021c99 100644 --- a/fragment.go +++ b/fragment.go @@ -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< 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) + } + } +} diff --git a/frame.go b/frame.go index 0c87210bf..0cf71f918 100644 --- a/frame.go +++ b/frame.go @@ -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)) diff --git a/gossip/gossip.go b/gossip/gossip.go index 25dd40cc4..2da6e3e4c 100644 --- a/gossip/gossip.go +++ b/gossip/gossip.go @@ -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") } } diff --git a/handler.go b/handler.go index 7bf11af1b..b3903ef6f 100644 --- a/handler.go +++ b/handler.go @@ -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 diff --git a/handler_test.go b/handler_test.go index 4187b3303..5877c449c 100644 --- a/handler_test.go +++ b/handler_test.go @@ -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) diff --git a/holder.go b/holder.go index c7f9dabf4..9f06f20bf 100644 --- a/holder.go +++ b/holder.go @@ -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") } } } diff --git a/holder_test.go b/holder_test.go index 45873959c..419cdc3f2 100644 --- a/holder_test.go +++ b/holder_test.go @@ -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) } }) diff --git a/index.go b/index.go index 8e227af83..d00ae4f5d 100644 --- a/index.go +++ b/index.go @@ -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 } diff --git a/server.go b/server.go index 52999b1c5..e739fe5f0 100644 --- a/server.go +++ b/server.go @@ -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 +} diff --git a/server/server.go b/server/server.go index 8fd699ce5..eb660fc23 100644 --- a/server/server.go +++ b/server/server.go @@ -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 diff --git a/server/server_test.go b/server/server_test.go index 75e0e2242..b78e8b79f 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -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 { diff --git a/test/handler.go b/test/handler.go index c325d3243..50b081812 100644 --- a/test/handler.go +++ b/test/handler.go @@ -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 diff --git a/uri.go b/uri.go index beb5060b5..dfb82baef 100644 --- a/uri.go +++ b/uri.go @@ -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 diff --git a/view.go b/view.go index dbcc73b8d..4f2ae1cd7 100644 --- a/view.go +++ b/view.go @@ -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.