scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-10 15:01:03 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix some lint warnings raised in VS-Code

Browse source
This commit is contained in:
Travis Turner 2019-04-17 18:10:05 -05:00
parent 4cc1505b2f
commit b46ff7b990
No known key found for this signature in database
GPG key ID: 7F08008DFD9314C9
16 changed files with 169 additions and 29 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
api.go
View file

@ -96,7 +96,7 @@ func (api *API) validate(f apiMethod) error {
if _, ok := validAPIMethods[state][f]; ok {
return nil
}
return newApiMethodNotAllowedError(errors.Errorf("api method %s not allowed in state %s", f, state))
return newAPIMethodNotAllowedError(errors.Errorf("api method %s not allowed in state %s", f, state))
}
// Query parses a PQL query out of the request and executes it.
@ -617,7 +617,7 @@ func (api *API) RecalculateCaches(ctx context.Context) error {
return nil
}
// PostClusterMessage is for internal use. It decodes a protobuf message out of
// ClusterMessage is for internal use. It decodes a protobuf message out of
// the body and forwards it to the BroadcastHandler.
func (api *API) ClusterMessage(ctx context.Context, reqBody io.Reader) error {
span, _ := tracing.StartSpanFromContext(ctx, "API.ClusterMessage")
@ -712,7 +712,7 @@ func (api *API) DeleteView(ctx context.Context, indexName string, fieldName stri
return errors.Wrap(err, "sending DeleteView message")
}
// IndexAttrDiff
// IndexAttrDiff determines the local column attribute data blocks which differ from those provided.
func (api *API) IndexAttrDiff(ctx context.Context, indexName string, blocks []AttrBlock) (map[uint64]map[string]interface{}, error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.IndexAttrDiff")
defer span.Finish()
@ -750,6 +750,7 @@ func (api *API) IndexAttrDiff(ctx context.Context, indexName string, blocks []At
return attrs, nil
}
// FieldAttrDiff determines the local row attribute data blocks which differ from those provided.
func (api *API) FieldAttrDiff(ctx context.Context, indexName string, fieldName string, blocks []AttrBlock) (map[uint64]map[string]interface{}, error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.FieldAttrDiff")
defer span.Finish()
@ -796,6 +797,8 @@ type ImportOptions struct {
// ImportOption is a functional option type for API.Import.
type ImportOption func(*ImportOptions) error
// OptImportOptionsClear is a functional option on ImportOption
// used to specify whether the import is a set or clear operation.
func OptImportOptionsClear(c bool) ImportOption {
return func(o *ImportOptions) error {
o.Clear = c
@ -803,6 +806,8 @@ func OptImportOptionsClear(c bool) ImportOption {
}
}
// OptImportOptionsIgnoreKeyCheck is a functional option on ImportOption
// used to specify whether key check should be ignored.
func OptImportOptionsIgnoreKeyCheck(b bool) ImportOption {
return func(o *ImportOptions) error {
o.IgnoreKeyCheck = b
@ -1175,7 +1180,7 @@ func (api *API) Version() string {
return strings.TrimPrefix(Version, "v")
}
// Info returns information about this server instance
// Info returns information about this server instance.
func (api *API) Info() serverInfo {
si := api.server.systemInfo
// we don't report errors on failures to get this information
@ -1192,6 +1197,7 @@ func (api *API) Info() serverInfo {
}
}
// TranslateKeys handles a TranslateKeyRequest.
func (api *API) TranslateKeys(body io.Reader) ([]byte, error) {
reqBytes, err := ioutil.ReadAll(body)
if err != nil {

1
client.go
View file

@ -72,6 +72,7 @@ type InternalClient interface {
//===============
// InternalQueryClient is the internal interface for querying a node.
type InternalQueryClient interface {
QueryNode(ctx context.Context, uri *URI, index string, queryRequest *QueryRequest) (*QueryResponse, error)
}

25
cluster.go
View file

@ -1969,12 +1969,16 @@ func (c *cluster) setStatic(hosts []string) error {
return nil
}
// ClusterStatus describes the status of the cluster including its
// state and node topology.
type ClusterStatus struct {
ClusterID string
State string
Nodes []*Node
}
// ResizeInstruction contains the instruction provided to a node
// during a cluster resize operation.
type ResizeInstruction struct {
JobID int64
Node *Node
@ -1984,6 +1988,8 @@ type ResizeInstruction struct {
ClusterStatus *ClusterStatus
}
// ResizeSource is the source of data for a node acting on a
// ResizeInstruction.
type ResizeSource struct {
Node *Node `protobuf:"bytes,1,opt,name=Node" json:"Node,omitempty"`
Index string `protobuf:"bytes,2,opt,name=Index,proto3" json:"Index,omitempty"`
@ -2023,82 +2029,101 @@ func decodeTopology(topology *internal.Topology) (*Topology, error) {
return t, nil
}
// CreateShardMessage is an internal message indicating shard creation.
type CreateShardMessage struct {
Index string
Field string
Shard uint64
}
// CreateIndexMessage is an internal message indicating index creation.
type CreateIndexMessage struct {
Index string
Meta *IndexOptions
}
// DeleteIndexMessage is an internal message indicating index deletion.
type DeleteIndexMessage struct {
Index string
}
// CreateFieldMessage is an internal message indicating field creation.
type CreateFieldMessage struct {
Index string
Field string
Meta *FieldOptions
}
// DeleteFieldMessage is an internal message indicating field deletion.
type DeleteFieldMessage struct {
Index string
Field string
}
// DeleteAvailableShardMessage is an internal message indicating available shard deletion.
type DeleteAvailableShardMessage struct {
Index string
Field string
ShardID uint64
}
// CreateViewMessage is an internal message indicating view creation.
type CreateViewMessage struct {
Index string
Field string
View string
}
// DeleteViewMessage is an internal message indicating view deletion.
type DeleteViewMessage struct {
Index string
Field string
View string
}
// ResizeInstructionComplete is an internal message to the coordinator indicating
// that the resize instructions performed on a single node have completed.
type ResizeInstructionComplete struct {
JobID int64
Node *Node
Error string
}
// SetCoordinatorMessage is an internal message instructing nodes to honor a new coordinator.
type SetCoordinatorMessage struct {
New *Node
}
// UpdateCoordinatorMessage is an internal message for reassigning the coordinator.
type UpdateCoordinatorMessage struct {
New *Node
}
// NodeStateMessage is an internal message for broadcasting a node's state.
type NodeStateMessage struct {
NodeID string `protobuf:"bytes,1,opt,name=NodeID,proto3" json:"NodeID,omitempty"`
State string `protobuf:"bytes,2,opt,name=State,proto3" json:"State,omitempty"`
}
// NodeStatus is an internal message representing the contents of a node.
type NodeStatus struct {
Node *Node
Indexes []*IndexStatus
Schema *Schema
}
// IndexStatus is an internal message representing the contents of an index.
type IndexStatus struct {
Name string
Fields []*FieldStatus
}
// FieldStatus is an internal message representing the contents of a field.
type FieldStatus struct {
Name string
AvailableShards *roaring.Bitmap
}
// RecalculateCaches is an internal message for recalculating all caches
// within a holder.
type RecalculateCaches struct{}

6
diagnostics.go
View file

@ -230,11 +230,11 @@ func (d *diagnosticsCollector) EnrichWithSchemaProperties() {
for _, index := range d.server.holder.Indexes() {
numShards += index.AvailableShards().Count()
numIndexes += 1
numIndexes++
for _, field := range index.Fields() {
numFields += 1
numFields++
if field.Type() == FieldTypeInt {
bsiFieldCount += 1
bsiFieldCount++
}
if field.TimeQuantum() != "" {
timeQuantumEnabled = true

1
event.go
View file

@ -17,6 +17,7 @@ package pilosa
// NodeEventType are the types of node events.
type NodeEventType int
// Constant node event types.
const (
NodeJoin NodeEventType = iota
NodeLeave

27
executor.go
View file

@ -987,6 +987,8 @@ type FieldRow struct {
RowKey string `json:"rowKey,omitempty"`
}
// MarshalJSON marshals FieldRow to JSON such that
// either a Key or an ID is included.
func (fr FieldRow) MarshalJSON() ([]byte, error) {
if fr.RowKey != "" {
return json.Marshal(struct {
@ -1006,10 +1008,12 @@ func (fr FieldRow) MarshalJSON() ([]byte, error) {
})
}
// String is the FieldRow stringer.
func (fr FieldRow) String() string {
return fmt.Sprintf("%s.%d.%s", fr.Field, fr.RowID, fr.RowKey)
}
// GroupCount represents a result item for a group by query.
type GroupCount struct {
Group []FieldRow `json:"group"`
Count uint64 `json:"count"`
@ -1048,6 +1052,7 @@ func mergeGroupCounts(a, b []GroupCount, limit int) []GroupCount {
return ret
}
// Compare is used in ordering two GroupCount objects.
func (g GroupCount) Compare(o GroupCount) int {
for i := range g.Group {
if g.Group[i].RowID < o.Group[i].RowID {
@ -1158,7 +1163,7 @@ func (e *executor) executeRowsShard(_ context.Context, index string, fieldName s
// views contains the list of views to inspect (and merge)
// in order to represent `Rows` for the field.
var views []string = []string{viewStandard}
var views = []string{viewStandard}
// Handle `time` fields.
if f.Type() == FieldTypeTime {
@ -1706,11 +1711,11 @@ func (e *executor) executeClearBitField(ctx context.Context, index string, c *pq
}
// Forward call to remote node otherwise.
if res, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil); err != nil {
res, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil)
if err != nil {
return false, err
} else {
ret = res[0].(bool)
}
ret = res[0].(bool)
}
return ret, nil
}
@ -1982,11 +1987,11 @@ func (e *executor) executeSetBitField(ctx context.Context, index string, c *pql.
}
// Forward call to remote node otherwise.
if res, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil); err != nil {
res, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil)
if err != nil {
return false, err
} else {
ret = res[0].(bool)
}
ret = res[0].(bool)
}
return ret, nil
}
@ -2017,11 +2022,11 @@ func (e *executor) executeSetValueField(ctx context.Context, index string, c *pq
}
// Forward call to remote node otherwise.
if res, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil); err != nil {
res, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil)
if err != nil {
return false, err
} else {
ret = res[0].(bool)
}
ret = res[0].(bool)
}
return ret, nil
}
@ -2881,7 +2886,7 @@ func newGroupByIterator(rowIDs []RowIDs, children []*pql.Call, filter *Row, inde
return nil, errors.Wrap(err, "getting previous")
} else if hasPrev && !ignorePrev {
if i == len(children)-1 {
prev += 1
prev++
}
gbi.rowIters[i].Seek(prev)
}

24
field.go
View file

@ -92,6 +92,8 @@ type Field struct {
// FieldOption is a functional option type for pilosa.fieldOptions.
type FieldOption func(fo *FieldOptions) error
// OptFieldKeys is a functional option on FieldOptions
// used to specify whether keys are used for this field.
func OptFieldKeys() FieldOption {
return func(fo *FieldOptions) error {
fo.Keys = true
@ -99,6 +101,8 @@ func OptFieldKeys() FieldOption {
}
}
// OptFieldTypeDefault is a functional option on FieldOptions
// used to set the field type and cache setting to the default values.
func OptFieldTypeDefault() FieldOption {
return func(fo *FieldOptions) error {
if fo.Type != "" {
@ -111,6 +115,9 @@ func OptFieldTypeDefault() FieldOption {
}
}
// OptFieldTypeSet is a functional option on FieldOptions
// used to specify the field as being type `set` and to
// provide any respective configuration values.
func OptFieldTypeSet(cacheType string, cacheSize uint32) FieldOption {
return func(fo *FieldOptions) error {
if fo.Type != "" {
@ -123,6 +130,9 @@ func OptFieldTypeSet(cacheType string, cacheSize uint32) FieldOption {
}
}
// OptFieldTypeInt is a functional option on FieldOptions
// used to specify the field as being type `int` and to
// provide any respective configuration values.
func OptFieldTypeInt(min, max int64) FieldOption {
return func(fo *FieldOptions) error {
if fo.Type != "" {
@ -138,7 +148,9 @@ func OptFieldTypeInt(min, max int64) FieldOption {
}
}
// OptFieldTypeTime sets the field type to time.
// OptFieldTypeTime is a functional option on FieldOptions
// used to specify the field as being type `time` and to
// provide any respective configuration values.
// Pass true to skip creation of the standard view.
func OptFieldTypeTime(timeQuantum TimeQuantum, opt ...bool) FieldOption {
return func(fo *FieldOptions) error {
@ -155,6 +167,9 @@ func OptFieldTypeTime(timeQuantum TimeQuantum, opt ...bool) FieldOption {
}
}
// OptFieldTypeMutex is a functional option on FieldOptions
// used to specify the field as being type `mutex` and to
// provide any respective configuration values.
func OptFieldTypeMutex(cacheType string, cacheSize uint32) FieldOption {
return func(fo *FieldOptions) error {
if fo.Type != "" {
@ -167,6 +182,9 @@ func OptFieldTypeMutex(cacheType string, cacheSize uint32) FieldOption {
}
}
// OptFieldTypeBool is a functional option on FieldOptions
// used to specify the field as being type `bool` and to
// provide any respective configuration values.
func OptFieldTypeBool() FieldOption {
return func(fo *FieldOptions) error {
if fo.Type != "" {
@ -1036,6 +1054,7 @@ func (f *Field) Max(filter *Row, name string) (max, count int64, err error) {
return int64(vmax) + bsig.Min, int64(vcount), nil
}
// Range performs a conditional operation on Field.
func (f *Field) Range(name string, op pql.Token, predicate int64) (*Row, error) {
// Retrieve and validate bsiGroup.
bsig := f.bsiGroup(name)
@ -1283,6 +1302,9 @@ func encodeFieldOptions(o *FieldOptions) *internal.FieldOptions {
}
}
// MarshalJSON marshals FieldOptions to JSON such that
// only those attributes associated to the field type
// are included.
func (o *FieldOptions) MarshalJSON() ([]byte, error) {
switch o.Type {
case FieldTypeSet:

3
fragment.go
View file

@ -50,7 +50,6 @@ import (
const (
// ShardWidth is the number of column IDs in a shard. It must be a power of 2 greater than or equal to 16.
// shardWidthExponent = 20 // set in shardwidthNN.go files
ShardWidth = 1 << shardwidth.Exponent
// shardVsContainerExponent is the power of 2 of ShardWith minus the power
@ -2299,7 +2298,7 @@ func (ri *rowIterator) Next() (r *Row, rowID uint64, wrapped bool) {
}
rowID = ri.rowIDs[ri.cur]
r = ri.f.row(rowID)
ri.cur += 1
ri.cur++
return r, rowID, wrapped
}

2
fragment_internal_test.go
View file

@ -2757,7 +2757,7 @@ func calcTop(rowIDs, colIDs []uint64) []Pair {
func calcExpected(inputs ...[]uint64) [][]uint64 {
// create map of row id to set of column values in that row.
rows := make(map[uint64]map[uint64]struct{})
var maxrow uint64 = 0
var maxrow uint64
for _, input := range inputs {
for _, val := range input {
row := val / ShardWidth

2
gc.go
View file

@ -32,6 +32,6 @@ type nopGCNotifier struct{}
func (n *nopGCNotifier) Close() {}
// AfterGC is a no-op implementation of GCNotifier AfterGC method.
func (c *nopGCNotifier) AfterGC() <-chan struct{} {
func (n *nopGCNotifier) AfterGC() <-chan struct{} {
return nil
}

18
handler.go
View file

@ -74,6 +74,8 @@ func (resp *QueryResponse) MarshalJSON() ([]byte, error) {
})
}
// Handler is the interface for the data handler, a wrapper around
// Pilosa's data store.
type Handler interface {
Serve() error
Close() error
@ -89,8 +91,11 @@ func (n nopHandler) Close() error {
return nil
}
// NopHandler is a no-op implementation of the Handler interface.
var NopHandler Handler = nopHandler{}
// ImportValueRequest describes the import request structure
// for a value (BSI) import.
type ImportValueRequest struct {
Index string
Field string
@ -100,6 +105,8 @@ type ImportValueRequest struct {
Values []int64
}
// ImportRequest describes the import request structure
// for an import.
type ImportRequest struct {
Index string
Field string
@ -111,15 +118,20 @@ type ImportRequest struct {
Timestamps []int64
}
// ImportRoaringRequest describes the import request structure
// for an import containing roaring-encoded data.
type ImportRoaringRequest struct {
Clear bool
Views map[string][]byte
}
// ImportResponse is the structured response of an import.
type ImportResponse struct {
Err string
}
// BlockDataRequest describes the structure of a request
// for fragment block data.
type BlockDataRequest struct {
Index string
Field string
@ -128,17 +140,23 @@ type BlockDataRequest struct {
Block uint64
}
// BlockDataResponse is the structured response of a block
// data request.
type BlockDataResponse struct {
RowIDs []uint64
ColumnIDs []uint64
}
// TranslateKeysRequest describes the structure of a request
// for a batch of key translations.
type TranslateKeysRequest struct {
Index string
Field string
Keys []string
}
// TranslateKeysResponse is the structured response of a key
// translation request.
type TranslateKeysResponse struct {
IDs []uint64
}

6
pilosa.go
View file

@ -77,8 +77,8 @@ type apiMethodNotAllowedError struct {
error
}
// newApiMethodNotAllowedError returns err wrapped in an ApiMethodNotAllowedError.
func newApiMethodNotAllowedError(err error) apiMethodNotAllowedError {
// newAPIMethodNotAllowedError returns err wrapped in an ApiMethodNotAllowedError.
func newAPIMethodNotAllowedError(err error) apiMethodNotAllowedError {
return apiMethodNotAllowedError{err}
}
@ -128,6 +128,8 @@ type ColumnAttrSet struct {
Attrs map[string]interface{} `json:"attrs,omitempty"`
}
// MarshalJSON marshals the ColumnAttrSet to JSON such that
// either a Key or an ID is included.
func (cas ColumnAttrSet) MarshalJSON() ([]byte, error) {
if cas.Key != "" {
return json.Marshal(struct {

1
row.go
View file

@ -43,6 +43,7 @@ func NewRow(columns ...uint64) *Row {
return r
}
// IsEmpty returns true if the row doesn't contain any set bits.
func (r *Row) IsEmpty() bool {
if len(r.segments) == 0 {
return true

51
server.go
View file

@ -75,6 +75,7 @@ type Server struct { // nolint: maligned
dataDir string
}
// Holder returns the holder for server.
// TODO: have this return an interface for Holder instead of concrete object?
func (s *Server) Holder() *Holder {
return s.holder
@ -83,6 +84,8 @@ func (s *Server) Holder() *Holder {
// ServerOption is a functional option type for pilosa.Server
type ServerOption func(s *Server) error
// OptServerLogger is a functional option on Server
// used to set the logger.
func OptServerLogger(l logger.Logger) ServerOption {
return func(s *Server) error {
s.logger = l
@ -90,6 +93,8 @@ func OptServerLogger(l logger.Logger) ServerOption {
}
}
// OptServerReplicaN is a functional option on Server
// used to set the number of replicas.
func OptServerReplicaN(n int) ServerOption {
return func(s *Server) error {
s.cluster.ReplicaN = n
@ -97,6 +102,8 @@ func OptServerReplicaN(n int) ServerOption {
}
}
// OptServerDataDir is a functional option on Server
// used to set the data directory.
func OptServerDataDir(dir string) ServerOption {
return func(s *Server) error {
s.dataDir = dir
@ -104,6 +111,9 @@ func OptServerDataDir(dir string) ServerOption {
}
}
// OptServerAttrStoreFunc is a functional option on Server
// used to provide the function to use to generate a new
// attribute store.
func OptServerAttrStoreFunc(af func(string) AttrStore) ServerOption {
return func(s *Server) error {
s.holder.NewAttrStore = af
@ -111,6 +121,8 @@ func OptServerAttrStoreFunc(af func(string) AttrStore) ServerOption {
}
}
// OptServerAntiEntropyInterval is a functional option on Server
// used to set the anti-entropy interval.
func OptServerAntiEntropyInterval(interval time.Duration) ServerOption {
return func(s *Server) error {
s.antiEntropyInterval = interval
@ -118,6 +130,8 @@ func OptServerAntiEntropyInterval(interval time.Duration) ServerOption {
}
}
// OptServerLongQueryTime is a functional option on Server
// used to set long query duration.
func OptServerLongQueryTime(dur time.Duration) ServerOption {
return func(s *Server) error {
s.cluster.longQueryTime = dur
@ -125,6 +139,8 @@ func OptServerLongQueryTime(dur time.Duration) ServerOption {
}
}
// OptServerMaxWritesPerRequest is a functional option on Server
// used to set the maximum number of writes allowed per request.
func OptServerMaxWritesPerRequest(n int) ServerOption {
return func(s *Server) error {
s.maxWritesPerRequest = n
@ -132,6 +148,8 @@ func OptServerMaxWritesPerRequest(n int) ServerOption {
}
}
// OptServerMetricInterval is a functional option on Server
// used to set the interval between metric samples.
func OptServerMetricInterval(dur time.Duration) ServerOption {
return func(s *Server) error {
s.metricInterval = dur
@ -139,6 +157,8 @@ func OptServerMetricInterval(dur time.Duration) ServerOption {
}
}
// OptServerSystemInfo is a functional option on Server
// used to set the system information source.
func OptServerSystemInfo(si SystemInfo) ServerOption {
return func(s *Server) error {
s.systemInfo = si
@ -146,6 +166,8 @@ func OptServerSystemInfo(si SystemInfo) ServerOption {
}
}
// OptServerGCNotifier is a functional option on Server
// used to set the garbage collection notification source.
func OptServerGCNotifier(gcn GCNotifier) ServerOption {
return func(s *Server) error {
s.gcNotifier = gcn
@ -153,6 +175,8 @@ func OptServerGCNotifier(gcn GCNotifier) ServerOption {
}
}
// OptServerInternalClient is a functional option on Server
// used to set the implementation of InternalClient.
func OptServerInternalClient(c InternalClient) ServerOption {
return func(s *Server) error {
s.executor = newExecutor(optExecutorInternalQueryClient(c))
@ -162,7 +186,7 @@ func OptServerInternalClient(c InternalClient) ServerOption {
}
}
// DEPRECATED
// OptServerPrimaryTranslateStore has been deprecated.
func OptServerPrimaryTranslateStore(store TranslateStore) ServerOption {
return func(s *Server) error {
s.logger.Printf("DEPRECATED: OptServerPrimaryTranslateStore")
@ -170,6 +194,9 @@ func OptServerPrimaryTranslateStore(store TranslateStore) ServerOption {
}
}
// OptServerPrimaryTranslateStoreFunc is a functional option on Server
// used to specify the function used to create a new primary translate
// store.
func OptServerPrimaryTranslateStoreFunc(tf func(interface{}) TranslateStore) ServerOption {
return func(s *Server) error {
@ -178,6 +205,8 @@ func OptServerPrimaryTranslateStoreFunc(tf func(interface{}) TranslateStore) Ser
}
}
// OptServerStatsClient is a functional option on Server
// used to specify the stats client.
func OptServerStatsClient(sc stats.StatsClient) ServerOption {
return func(s *Server) error {
s.holder.Stats = sc
@ -185,6 +214,8 @@ func OptServerStatsClient(sc stats.StatsClient) ServerOption {
}
}
// OptServerDiagnosticsInterval is a functional option on Server
// used to specify the duration between diagnostic checks.
func OptServerDiagnosticsInterval(dur time.Duration) ServerOption {
return func(s *Server) error {
s.diagnosticInterval = dur
@ -192,6 +223,8 @@ func OptServerDiagnosticsInterval(dur time.Duration) ServerOption {
}
}
// OptServerURI is a functional option on Server
// used to set the server URI.
func OptServerURI(uri *URI) ServerOption {
return func(s *Server) error {
s.uri = *uri
@ -199,7 +232,7 @@ func OptServerURI(uri *URI) ServerOption {
}
}
// OptClusterDisabled tells the server whether to use a static cluster with the
// OptServerClusterDisabled tells the server whether to use a static cluster with the
// defined hosts. Mostly used for testing.
func OptServerClusterDisabled(disabled bool, hosts []string) ServerOption {
return func(s *Server) error {
@ -209,6 +242,8 @@ func OptServerClusterDisabled(disabled bool, hosts []string) ServerOption {
}
}
// OptServerSerializer is a functional option on Server
// used to set the serializer.
func OptServerSerializer(ser Serializer) ServerOption {
return func(s *Server) error {
s.serializer = ser
@ -216,6 +251,8 @@ func OptServerSerializer(ser Serializer) ServerOption {
}
}
// OptServerIsCoordinator is a functional option on Server
// used to specify whether or not this server is the coordinator.
func OptServerIsCoordinator(is bool) ServerOption {
return func(s *Server) error {
s.isCoordinator = is
@ -223,6 +260,8 @@ func OptServerIsCoordinator(is bool) ServerOption {
}
}
// OptServerNodeID is a functional option on Server
// used to set the server node ID.
func OptServerNodeID(nodeID string) ServerOption {
return func(s *Server) error {
s.nodeID = nodeID
@ -230,6 +269,9 @@ func OptServerNodeID(nodeID string) ServerOption {
}
}
// OptServerClusterHasher is a functional option on Server
// used to specify the consistent hash algorithm for data
// location within the cluster.
func OptServerClusterHasher(h Hasher) ServerOption {
return func(s *Server) error {
s.cluster.Hasher = h
@ -237,6 +279,8 @@ func OptServerClusterHasher(h Hasher) ServerOption {
}
}
// OptServerTranslateFileMapSize is a functional option on Server
// used to specify the size of the translate file.
func OptServerTranslateFileMapSize(mapSize int) ServerOption {
return func(s *Server) error {
s.holder.translateFile = NewTranslateFile(OptTranslateFileMapSize(mapSize))
@ -681,9 +725,8 @@ func (s *Server) monitorDiagnostics() {
if s.diagnosticInterval < time.Minute {
s.logger.Printf("diagnostics disabled")
return
} else {
s.logger.Printf("Pilosa is currently configured to send small diagnostics reports to our team every %v. More information here: https://www.pilosa.com/docs/latest/administration/#diagnostics", s.diagnosticInterval)
}
s.logger.Printf("Pilosa is currently configured to send small diagnostics reports to our team every %v. More information here: https://www.pilosa.com/docs/latest/administration/#diagnostics", s.diagnosticInterval)
s.diagnostics.Logger = s.logger
s.diagnostics.SetVersion(Version)

13
translate.go
View file

@ -33,6 +33,7 @@ import (
"github.com/pkg/errors"
)
// Log entry type constants.
const (
LogEntryTypeInsertColumn = 1
LogEntryTypeInsertRow = 2
@ -42,6 +43,7 @@ const (
defaultReplicationRetryInterval = 1 * time.Second
)
// Translate store errors.
var (
ErrTranslateStoreClosed = errors.New("pilosa: translate store closed")
ErrTranslateStoreReaderClosed = errors.New("pilosa: translate store reader closed")
@ -99,12 +101,17 @@ type TranslateFile struct {
// TranslateFileOption is a functional option type for pilosa.TranslateFile
type TranslateFileOption func(f *TranslateFile) error
// OptTranslateFileMapSize is a functional option on TranslateFile
// used to set the map size.
func OptTranslateFileMapSize(mapSize int) TranslateFileOption {
return func(f *TranslateFile) error {
f.mapSize = mapSize
return nil
}
}
// OptTranslateFileLogger is a functional option on TranslateFile
// used to set the file logger.
func OptTranslateFileLogger(l logger.Logger) TranslateFileOption {
return func(s *TranslateFile) error {
s.logger = l
@ -151,6 +158,7 @@ func NewTranslateFile(opts ...TranslateFileOption) *TranslateFile {
return f
}
// Open opens the translate file.
func (s *TranslateFile) Open() (err error) {
// Open writer & buffered writer.
if err := os.MkdirAll(filepath.Dir(s.Path), 0777); err != nil {
@ -232,6 +240,7 @@ func (s *TranslateFile) handlePrimaryStoreEvent(ev primaryStoreEvent) error {
return nil
}
// Close closes the translate file.
func (s *TranslateFile) Close() (err error) {
s.once.Do(func() {
close(s.closing)
@ -588,6 +597,7 @@ func (s *TranslateFile) TranslateColumnToString(index string, value uint64) (str
return "", nil
}
// TranslateRowsToUint64 converts a slice of row keys to a slice of row IDs.
func (s *TranslateFile) TranslateRowsToUint64(index, field string, values []string) ([]uint64, error) {
key := fieldKey{index, field}
@ -679,6 +689,7 @@ func (s *TranslateFile) TranslateRowsToUint64(index, field string, values []stri
return ret, nil
}
// TranslateRowToString translates a row ID to a string key.
func (s *TranslateFile) TranslateRowToString(index, field string, id uint64) (string, error) {
s.mu.RLock()
if idx := s.rows[fieldKey{index, field}]; idx != nil {
@ -700,6 +711,8 @@ func (s *TranslateFile) Reader(ctx context.Context, offset int64) (io.ReadCloser
return rc, nil
}
// LogEntry is a batch of Key/ID mappings which is replicated to other nodes
// for read-only key translation.
type LogEntry struct {
Type uint8
Index []byte

4
uri.go
View file

@ -56,8 +56,11 @@ func defaultURI() *URI {
}
}
// URIs is a convenience type representing a slice of URI.
type URIs []URI
// HostPortStrings returns a slice of host:port strings
// based on the slice of URI.
func (u URIs) HostPortStrings() []string {
s := make([]string, len(u))
for i, a := range u {
@ -199,6 +202,7 @@ func (u *URI) MarshalJSON() ([]byte, error) {
return json.Marshal(output)
}
// UnmarshalJSON unmarshals a byte slice to a URI.
func (u *URI) UnmarshalJSON(b []byte) error {
var input struct {
Scheme string `json:"scheme,omitempty"`