scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-10-07 11:27:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'develop' into http-inversion-2

Browse source
This commit is contained in:
Cody Soyland 2018-06-13 21:38:30 -05:00 • committed by GitHub
commit 415b4de110
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
24 changed files with 312 additions and 293 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
attr.go
View file

@ -24,10 +24,10 @@ import (
// Attribute data type enum.
const (
AttrTypeString = 1
AttrTypeInt = 2
AttrTypeBool = 3
AttrTypeFloat = 4
attrTypeString = 1
attrTypeInt = 2
attrTypeBool = 3
attrTypeFloat = 4
)
// AttrStore represents an interface for handling row/column attributes.
@ -165,19 +165,19 @@ func encodeAttr(key string, value interface{}) *internal.Attr {
pb := &internal.Attr{Key: key}
switch value := value.(type) {
case string:
pb.Type = AttrTypeString
pb.Type = attrTypeString
pb.StringValue = value
case float64:
pb.Type = AttrTypeFloat
pb.Type = attrTypeFloat
pb.FloatValue = value
case uint64:
pb.Type = AttrTypeInt
pb.Type = attrTypeInt
pb.IntValue = int64(value)
case int64:
pb.Type = AttrTypeInt
pb.Type = attrTypeInt
pb.IntValue = value
case bool:
pb.Type = AttrTypeBool
pb.Type = attrTypeBool
pb.BoolValue = value
}
return pb
@ -186,13 +186,13 @@ func encodeAttr(key string, value interface{}) *internal.Attr {
// decodeAttr converts from an Attr internal representation to a key/value pair.
func decodeAttr(attr *internal.Attr) (key string, value interface{}) {
switch attr.Type {
case AttrTypeString:
case attrTypeString:
return attr.Key, attr.StringValue
case AttrTypeInt:
case attrTypeInt:
return attr.Key, attr.IntValue
case AttrTypeBool:
case attrTypeBool:
return attr.Key, attr.BoolValue
case AttrTypeFloat:
case attrTypeFloat:
return attr.Key, attr.FloatValue
default:
return attr.Key, nil

90
broadcast.go
View file

@ -120,21 +120,21 @@ func (n *nopGossiper) SendAsync(pb proto.Message) error {
// Broadcast message types.
const (
MessageTypeCreateSlice = iota
MessageTypeCreateIndex
MessageTypeDeleteIndex
MessageTypeCreateField
MessageTypeDeleteField
MessageTypeCreateView
MessageTypeDeleteView
MessageTypeClusterStatus
MessageTypeResizeInstruction
MessageTypeResizeInstructionComplete
MessageTypeSetCoordinator
MessageTypeUpdateCoordinator
MessageTypeNodeState
MessageTypeRecalculateCaches
MessageTypeNodeEvent
messageTypeCreateSlice = iota
messageTypeCreateIndex
messageTypeDeleteIndex
messageTypeCreateField
messageTypeDeleteField
messageTypeCreateView
messageTypeDeleteView
messageTypeClusterStatus
messageTypeResizeInstruction
messageTypeResizeInstructionComplete
messageTypeSetCoordinator
messageTypeUpdateCoordinator
messageTypeNodeState
messageTypeRecalculateCaches
messageTypeNodeEvent
)
// MarshalMessage encodes the protobuf message into a byte slice.
@ -142,35 +142,35 @@ func MarshalMessage(m proto.Message) ([]byte, error) {
var typ uint8
switch obj := m.(type) {
case *internal.CreateSliceMessage:
typ = MessageTypeCreateSlice
typ = messageTypeCreateSlice
case *internal.CreateIndexMessage:
typ = MessageTypeCreateIndex
typ = messageTypeCreateIndex
case *internal.DeleteIndexMessage:
typ = MessageTypeDeleteIndex
typ = messageTypeDeleteIndex
case *internal.CreateFieldMessage:
typ = MessageTypeCreateField
typ = messageTypeCreateField
case *internal.DeleteFieldMessage:
typ = MessageTypeDeleteField
typ = messageTypeDeleteField
case *internal.CreateViewMessage:
typ = MessageTypeCreateView
typ = messageTypeCreateView
case *internal.DeleteViewMessage:
typ = MessageTypeDeleteView
typ = messageTypeDeleteView
case *internal.ClusterStatus:
typ = MessageTypeClusterStatus
typ = messageTypeClusterStatus
case *internal.ResizeInstruction:
typ = MessageTypeResizeInstruction
typ = messageTypeResizeInstruction
case *internal.ResizeInstructionComplete:
typ = MessageTypeResizeInstructionComplete
typ = messageTypeResizeInstructionComplete
case *internal.SetCoordinatorMessage:
typ = MessageTypeSetCoordinator
typ = messageTypeSetCoordinator
case *internal.UpdateCoordinatorMessage:
typ = MessageTypeUpdateCoordinator
typ = messageTypeUpdateCoordinator
case *internal.NodeStateMessage:
typ = MessageTypeNodeState
typ = messageTypeNodeState
case *internal.RecalculateCaches:
typ = MessageTypeRecalculateCaches
typ = messageTypeRecalculateCaches
case *internal.NodeEventMessage:
typ = MessageTypeNodeEvent
typ = messageTypeNodeEvent
default:
return nil, fmt.Errorf("message type not implemented for marshalling: %s", reflect.TypeOf(obj))
}
@ -187,35 +187,35 @@ func UnmarshalMessage(buf []byte) (proto.Message, error) {
var m proto.Message
switch typ {
case MessageTypeCreateSlice:
case messageTypeCreateSlice:
m = &internal.CreateSliceMessage{}
case MessageTypeCreateIndex:
case messageTypeCreateIndex:
m = &internal.CreateIndexMessage{}
case MessageTypeDeleteIndex:
case messageTypeDeleteIndex:
m = &internal.DeleteIndexMessage{}
case MessageTypeCreateField:
case messageTypeCreateField:
m = &internal.CreateFieldMessage{}
case MessageTypeDeleteField:
case messageTypeDeleteField:
m = &internal.DeleteFieldMessage{}
case MessageTypeCreateView:
case messageTypeCreateView:
m = &internal.CreateViewMessage{}
case MessageTypeDeleteView:
case messageTypeDeleteView:
m = &internal.DeleteViewMessage{}
case MessageTypeClusterStatus:
case messageTypeClusterStatus:
m = &internal.ClusterStatus{}
case MessageTypeResizeInstruction:
case messageTypeResizeInstruction:
m = &internal.ResizeInstruction{}
case MessageTypeResizeInstructionComplete:
case messageTypeResizeInstructionComplete:
m = &internal.ResizeInstructionComplete{}
case MessageTypeSetCoordinator:
case messageTypeSetCoordinator:
m = &internal.SetCoordinatorMessage{}
case MessageTypeUpdateCoordinator:
case messageTypeUpdateCoordinator:
m = &internal.UpdateCoordinatorMessage{}
case MessageTypeNodeState:
case messageTypeNodeState:
m = &internal.NodeStateMessage{}
case MessageTypeRecalculateCaches:
case messageTypeRecalculateCaches:
m = &internal.RecalculateCaches{}
case MessageTypeNodeEvent:
case messageTypeNodeEvent:
m = &internal.NodeEventMessage{}
default:
return nil, fmt.Errorf("invalid message type: %d", typ)

6
cache.go
View file

@ -27,8 +27,8 @@ import (
)
const (
// ThresholdFactor is used to calculate the threshold for new items entering the cache
ThresholdFactor = 1.1
// thresholdFactor is used to calculate the threshold for new items entering the cache
thresholdFactor = 1.1
)
// Cache represents a cache of counts.
@ -158,7 +158,7 @@ type RankCache struct {
func NewRankCache(maxEntries uint32) *RankCache {
return &RankCache{
maxEntries: maxEntries,
thresholdBuffer: int(ThresholdFactor * float64(maxEntries)),
thresholdBuffer: int(thresholdFactor * float64(maxEntries)),
entries: make(map[uint64]uint64),
stats: NopStatsClient,
}

4
cluster.go
View file

@ -939,14 +939,14 @@ func (c *Cluster) markAsJoined() {
}
func (c *Cluster) needTopologyAgreement() bool {
return c.State() == ClusterStateStarting && !StringSlicesAreEqual(c.Topology.NodeIDs, c.nodeIDs())
return c.State() == ClusterStateStarting && !stringSlicesAreEqual(c.Topology.NodeIDs, c.nodeIDs())
}
func (c *Cluster) haveTopologyAgreement() bool {
if c.Static {
return true
}
return StringSlicesAreEqual(c.Topology.NodeIDs, c.nodeIDs())
return stringSlicesAreEqual(c.Topology.NodeIDs, c.nodeIDs())
}
func (c *Cluster) allNodesReady() bool {

18
executor.go
View file

@ -25,13 +25,13 @@ import (
"github.com/pkg/errors"
)
// DefaultField is the field used if one is not specified.
// defaultField is the field used if one is not specified.
const (
DefaultField = "general"
defaultField = "general"
// MinThreshold is the lowest count to use in a Top-N operation when
// defaultMinThreshold is the lowest count to use in a Top-N operation when
// looking for additional id/count pairs.
MinThreshold = 1
defaultMinThreshold = 1
columnLabel = "col"
rowLabel = "row"
@ -588,7 +588,7 @@ func (e *Executor) executeTopNSlice(ctx context.Context, index string, c *pql.Ca
// Set default field.
if field == "" {
field = DefaultField
field = defaultField
}
f := e.Holder.Fragment(index, field, ViewStandard, slice)
@ -597,7 +597,7 @@ func (e *Executor) executeTopNSlice(ctx context.Context, index string, c *pql.Ca
}
if minThreshold <= 0 {
minThreshold = MinThreshold
minThreshold = defaultMinThreshold
}
if tanimotoThreshold > 100 {
@ -646,7 +646,7 @@ func (e *Executor) executeBitmapSlice(ctx context.Context, index string, c *pql.
// Fetch field & row label based on argument.
field, _ := c.Args["field"].(string)
if field == "" {
field = DefaultField
field = defaultField
}
f := e.Holder.Field(index, field)
if f == nil {
@ -700,7 +700,7 @@ func (e *Executor) executeRangeSlice(ctx context.Context, index string, c *pql.C
// Parse field, use default if unset.
field, _ := c.Args["field"].(string)
if field == "" {
field = DefaultField
field = defaultField
}
// Retrieve column label.
@ -752,7 +752,7 @@ func (e *Executor) executeRangeSlice(ctx context.Context, index string, c *pql.C
// Union bitmaps across all time-based views.
row := &Row{}
for _, view := range ViewsByTimeRange(ViewStandard, startTime, endTime, q) {
for _, view := range viewsByTimeRange(ViewStandard, startTime, endTime, q) {
f := e.Holder.Fragment(index, field, view, slice)
if f == nil {
continue

24
field.go
View file

@ -33,10 +33,10 @@ import (
const (
DefaultFieldType = FieldTypeSet
DefaultCacheType = CacheTypeRanked
defaultCacheType = CacheTypeRanked
// Default ranked field cache
DefaultCacheSize = 50000
defaultCacheSize = 50000
)
// Field types.
@ -82,7 +82,7 @@ func OptFieldFieldOptions(o FieldOptions) FieldOption {
// NewField returns a new instance of field.
func NewField(path, index, name string, opts ...FieldOption) (*Field, error) {
err := ValidateName(name)
err := validateName(name)
if err != nil {
return nil, err
}
@ -101,8 +101,8 @@ func NewField(path, index, name string, opts ...FieldOption) (*Field, error) {
options: FieldOptions{
Type: DefaultFieldType,
CacheType: DefaultCacheType,
CacheSize: DefaultCacheSize,
CacheType: defaultCacheType,
CacheSize: defaultCacheSize,
},
Logger: NopLogger,
@ -645,7 +645,7 @@ func (f *Field) ViewRow(viewName string, rowID uint64) (*Row, error) {
// SetBit sets a bit on a view within the field.
func (f *Field) SetBit(name string, rowID, colID uint64, t *time.Time) (changed bool, err error) {
// Validate view name.
if !IsValidView(name) {
if !isValidView(name) {
return false, ErrInvalidView
}
@ -668,7 +668,7 @@ func (f *Field) SetBit(name string, rowID, colID uint64, t *time.Time) (changed
}
// If a timestamp is specified then set bits across all views for the quantum.
for _, subname := range ViewsByTime(name, *t, f.TimeQuantum()) {
for _, subname := range viewsByTime(name, *t, f.TimeQuantum()) {
view, err := f.CreateViewIfNotExists(subname)
if err != nil {
return changed, errors.Wrapf(err, "creating view %s", subname)
@ -687,7 +687,7 @@ func (f *Field) SetBit(name string, rowID, colID uint64, t *time.Time) (changed
// ClearBit clears a bit within the field.
func (f *Field) ClearBit(name string, rowID, colID uint64, t *time.Time) (changed bool, err error) {
// Validate view name.
if !IsValidView(name) {
if !isValidView(name) {
return false, ErrInvalidView
}
@ -710,7 +710,7 @@ func (f *Field) ClearBit(name string, rowID, colID uint64, t *time.Time) (change
}
// If a timestamp is specified then clear bits across all views for the quantum.
for _, subname := range ViewsByTime(name, *t, f.TimeQuantum()) {
for _, subname := range viewsByTime(name, *t, f.TimeQuantum()) {
view, err := f.CreateViewIfNotExists(subname)
if err != nil {
return changed, errors.Wrapf(err, "creating view %s", subname)
@ -899,7 +899,7 @@ func (f *Field) Import(rowIDs, columnIDs []uint64, timestamps []*time.Time) erro
if timestamp == nil {
standard = []string{ViewStandard}
} else {
standard = ViewsByTime(ViewStandard, *timestamp, q)
standard = viewsByTime(ViewStandard, *timestamp, q)
// In order to match the logic of `SetBit()`, we want bits
// with timestamps to write to both time and standard views.
standard = append(standard, ViewStandard)
@ -1233,8 +1233,8 @@ const (
CacheTypeNone = "none"
)
// IsValidCacheType returns true if v is a valid cache type.
func IsValidCacheType(v string) bool {
// isValidCacheType returns true if v is a valid cache type.
func isValidCacheType(v string) bool {
switch v {
case CacheTypeLRU, CacheTypeRanked, CacheTypeNone:
return true

30
fragment.go
View file

@ -47,22 +47,20 @@ const (
// SliceWidth is the number of column IDs in a slice.
SliceWidth = 1048576
// SnapshotExt is the file extension used for an in-process snapshot.
SnapshotExt = ".snapshotting"
// snapshotExt is the file extension used for an in-process snapshot.
snapshotExt = ".snapshotting"
// CopyExt is the file extension used for the temp file used while copying.
CopyExt = ".copying"
// copyExt is the file extension used for the temp file used while copying.
copyExt = ".copying"
// CacheExt is the file extension for persisted cache ids.
CacheExt = ".cache"
// cacheExt is the file extension for persisted cache ids.
cacheExt = ".cache"
// HashBlockSize is the number of rows in a merkle hash block.
HashBlockSize = 100
)
const (
// DefaultFragmentMaxOpN is the default value for Fragment.MaxOpN.
DefaultFragmentMaxOpN = 2000
// defaultFragmentMaxOpN is the default value for Fragment.MaxOpN.
defaultFragmentMaxOpN = 2000
)
// Fragment represents the intersection of a field and slice in an index.
@ -119,18 +117,18 @@ func NewFragment(path, index, field, view string, slice uint64) *Fragment {
field: field,
view: view,
slice: slice,
CacheType: DefaultCacheType,
CacheSize: DefaultCacheSize,
CacheType: defaultCacheType,
CacheSize: defaultCacheSize,
Logger: NopLogger,
MaxOpN: DefaultFragmentMaxOpN,
MaxOpN: defaultFragmentMaxOpN,
stats: NopStatsClient,
}
}
// cachePath returns the path to the fragment's cache data.
func (f *Fragment) cachePath() string { return f.path + CacheExt }
func (f *Fragment) cachePath() string { return f.path + cacheExt }
// Open opens the underlying storage.
func (f *Fragment) Open() error {
@ -1431,7 +1429,7 @@ func (f *Fragment) snapshot() error {
defer track(start, completeMessage, f.stats, f.Logger)
// Create a temporary file to snapshot to.
snapshotPath := f.path + SnapshotExt
snapshotPath := f.path + snapshotExt
file, err := os.Create(snapshotPath)
if err != nil {
return fmt.Errorf("create snapshot file: %s", err)
@ -1635,7 +1633,7 @@ func (f *Fragment) ReadFrom(r io.Reader) (n int64, err error) {
func (f *Fragment) readStorageFromArchive(r io.Reader) error {
// Create a temporary file to copy into.
path := f.path + CopyExt
path := f.path + copyExt
file, err := os.Create(path)
if err != nil {
return errors.Wrap(err, "creating directory")

2
fragment_internal_test.go
View file

@ -1245,7 +1245,7 @@ func mustOpenFragment(index, field, view string, slice uint64, cacheType string)
file.Close()
if cacheType == "" {
cacheType = DefaultCacheType
cacheType = defaultCacheType
}
f := NewFragment(file.Name(), index, field, view, slice)

21
gossip/gossip.go
View file

@ -18,6 +18,7 @@ import (
"fmt"
"io/ioutil"
"log"
"net"
"strconv"
"strings"
"sync"
@ -213,7 +214,7 @@ func NewGossipMemberSet(name string, host string, cfg Config, ger *GossipEventRe
conf.BindAddr = host
conf.BindPort = port
conf.AdvertisePort = port
conf.AdvertiseAddr = pilosa.HostToIP(host)
conf.AdvertiseAddr = hostToIP(host)
//
conf.TCPTimeout = time.Duration(cfg.StreamTimeout)
conf.SuspicionMult = cfg.SuspicionMult
@ -580,3 +581,21 @@ type Config struct {
Nodes int `toml:"nodes"`
ToTheDeadTime toml.Duration `toml:"to-the-dead-time"`
}
// hostToIP converts host to an IP4 address based on net.LookupIP().
func hostToIP(host string) string {
// if host is not an IP addr, check net.LookupIP()
if net.ParseIP(host) == nil {
hosts, err := net.LookupIP(host)
if err != nil {
return host
}
for _, h := range hosts {
// this restricts pilosa to IP4
if h.To4() != nil {
return h.String()
}
}
}
return host
}

6
holder.go
View file

@ -33,8 +33,8 @@ import (
)
const (
// DefaultCacheFlushInterval is the default value for Fragment.CacheFlushInterval.
DefaultCacheFlushInterval = 1 * time.Minute
// defaultCacheFlushInterval is the default value for Fragment.CacheFlushInterval.
defaultCacheFlushInterval = 1 * time.Minute
// FileLimit is the maximum open file limit (ulimit -n) to automatically set.
FileLimit = 262144 // (512^2)
@ -83,7 +83,7 @@ func NewHolder() *Holder {
NewAttrStore: NewNopAttrStore,
CacheFlushInterval: DefaultCacheFlushInterval,
CacheFlushInterval: defaultCacheFlushInterval,
Logger: NopLogger,
}

4
index.go
View file

@ -53,7 +53,7 @@ type Index struct {
// NewIndex returns a new instance of Index.
func NewIndex(path, name string) (*Index, error) {
err := ValidateName(name)
err := validateName(name)
if err != nil {
return nil, errors.Wrap(err, "validating name")
}
@ -295,7 +295,7 @@ func (i *Index) CreateFieldIfNotExists(name string, opt FieldOptions) (*Field, e
func (i *Index) createField(name string, opt FieldOptions) (*Field, error) {
if name == "" {
return nil, errors.New("field name required")
} else if opt.CacheType != "" && !IsValidCacheType(opt.CacheType) {
} else if opt.CacheType != "" && !isValidCacheType(opt.CacheType) {
return nil, ErrInvalidCacheType
}

68
pilosa.go
View file

@ -16,9 +16,7 @@ package pilosa
import (
"errors"
"net"
"regexp"
"strings"
"github.com/pilosa/pilosa/internal"
)
@ -108,26 +106,16 @@ func EncodeColumnAttrSet(set *ColumnAttrSet) *internal.ColumnAttrSet {
// TimeFormat is the go-style time format used to parse string dates.
const TimeFormat = "2006-01-02T15:04"
// ValidateName ensures that the name is a valid format.
func ValidateName(name string) error {
// validateName ensures that the name is a valid format.
func validateName(name string) error {
if !nameRegexp.Match([]byte(name)) {
return ErrName
}
return nil
}
// StringInSlice checks for substring a in the slice.
func StringInSlice(a string, list []string) bool {
for _, b := range list {
if b == a {
return true
}
}
return false
}
// StringSlicesAreEqual determines if two string slices are equal.
func StringSlicesAreEqual(a, b []string) bool {
// stringSlicesAreEqual determines if two string slices are equal.
func stringSlicesAreEqual(a, b []string) bool {
if a == nil && b == nil {
return true
@ -150,54 +138,6 @@ func StringSlicesAreEqual(a, b []string) bool {
return true
}
// SliceDiff returns the difference between two uint64 slices.
func SliceDiff(a, b []uint64) []uint64 {
m := make(map[uint64]uint64)
for _, y := range b {
m[y]++
}
var ret []uint64
for _, x := range a {
if m[x] > 0 {
m[x]--
continue
}
ret = append(ret, x)
}
return ret
}
// ContainsSubstring checks to see if substring a is contained in any string in the slice.
func ContainsSubstring(a string, list []string) bool {
for _, b := range list {
if strings.Contains(b, a) {
return true
}
}
return false
}
// HostToIP converts host to an IP4 address based on net.LookupIP().
func HostToIP(host string) string {
// if host is not an IP addr, check net.LookupIP()
if net.ParseIP(host) == nil {
hosts, err := net.LookupIP(host)
if err != nil {
return host
}
for _, h := range hosts {
// this restricts pilosa to IP4
if h.To4() != nil {
return h.String()
}
}
}
return host
}
// AddressWithDefaults converts addr into a valid address,
// using defaults when necessary.
func AddressWithDefaults(addr string) (*URI, error) {

43
pilosa_internal_test.go Normal file
View file

@ -0,0 +1,43 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package pilosa
import (
"testing"
)
func TestValidateName(t *testing.T) {
names := []string{
"a", "ab", "ab1", "b-c", "d_e",
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
}
for _, name := range names {
if validateName(name) != nil {
t.Fatalf("Should be valid index name: %s", name)
}
}
}
func TestValidateNameInvalid(t *testing.T) {
names := []string{
"", "'", "^", "/", "\\", "A", "*", "a:b", "valid?no", "yüce", "1", "_", "-",
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa1",
}
for _, name := range names {
if validateName(name) == nil {
t.Fatalf("Should be invalid index name: %s", name)
}
}
}

48
pilosa_test.go
View file

@ -22,54 +22,6 @@ import (
_ "github.com/pilosa/pilosa/test"
)
func TestValidateName(t *testing.T) {
names := []string{
"a", "ab", "ab1", "b-c", "d_e",
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
}
for _, name := range names {
if pilosa.ValidateName(name) != nil {
t.Fatalf("Should be valid index name: %s", name)
}
}
}
func TestValidateNameInvalid(t *testing.T) {
names := []string{
"", "'", "^", "/", "\\", "A", "*", "a:b", "valid?no", "yüce", "1", "_", "-",
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa1",
}
for _, name := range names {
if pilosa.ValidateName(name) == nil {
t.Fatalf("Should be invalid index name: %s", name)
}
}
}
func TestStringInSlice(t *testing.T) {
list := []string{"localhost:10101", "localhost:10102", "localhost:10103"}
substr := "localhost:10101"
if !pilosa.StringInSlice(substr, list) {
t.Fatalf("Expected substring %s in %v", substr, list)
}
substr = "10101"
if pilosa.StringInSlice(substr, list) {
t.Fatalf("Expected substring %s not in %v", substr, list)
}
}
func TestContainsSubstring(t *testing.T) {
list := []string{"localhost:10101", "localhost:10102", "localhost:10103"}
substr := "10101"
if !pilosa.ContainsSubstring(substr, list) {
t.Fatalf("Expected substring %s contained in %v", substr, list)
}
substr = "4000"
if pilosa.ContainsSubstring(substr, list) {
t.Fatalf("Expected substring %s in not contained in %v", substr, list)
}
}
func TestAddressWithDefaults(t *testing.T) {
tests := []struct {
addr string

12
server.go
View file

@ -637,7 +637,7 @@ func (s *Server) monitorDiagnostics() {
// Flush the diagnostics metrics at startup, then on each tick interval
flush := func() {
openFiles, err := CountOpenFiles()
openFiles, err := countOpenFiles()
if err == nil {
s.diagnostics.Set("OpenFiles", openFiles)
}
@ -694,7 +694,7 @@ func (s *Server) monitorRuntime() {
// Record the number of go routines.
s.Holder.Stats.Gauge("goroutines", float64(runtime.NumGoroutine()), 1.0)
openFiles, err := CountOpenFiles()
openFiles, err := countOpenFiles()
// Open File handles.
if err == nil {
s.Holder.Stats.Gauge("OpenFiles", float64(openFiles), 1.0)
@ -710,8 +710,8 @@ func (s *Server) monitorRuntime() {
}
}
// CountOpenFiles on operating systems that support lsof.
func CountOpenFiles() (int, error) {
// countOpenFiles on operating systems that support lsof.
func countOpenFiles() (int, error) {
switch runtime.GOOS {
case "darwin", "linux", "unix", "freebsd":
// -b option avoid kernel blocks
@ -725,9 +725,9 @@ func CountOpenFiles() (int, error) {
return len(lines), nil
case "windows":
// TODO: count open file handles on windows
return 0, errors.New("CountOpenFiles() on Windows is not supported")
return 0, errors.New("countOpenFiles() on Windows is not supported")
default:
return 0, errors.New("CountOpenFiles() on this OS is not supported")
return 0, errors.New("countOpenFiles() on this OS is not supported")
}
}

16
server/server_test.go
View file

@ -21,7 +21,6 @@ import (
"io/ioutil"
"math/rand"
"reflect"
"runtime"
"sort"
"strings"
"testing"
@ -263,21 +262,6 @@ func tempMkdir(t *testing.T) string {
return dir
}
// Ensure the file handle count is working
func TestCountOpenFiles(t *testing.T) {
// Windows is not supported yet
if runtime.GOOS == "windows" {
t.Skip("Skipping unsupported CountOpenFiles test on Windows.")
}
count, err := pilosa.CountOpenFiles()
if err != nil {
t.Errorf("CountOpenFiles failed: %s", err)
}
if count == 0 {
t.Error("CountOpenFiles returned invalid value 0.")
}
}
func TestMain_RecalculateHashes(t *testing.T) {
const clusterSize = 5
cluster := test.MustRunMainWithCluster(t, clusterSize)

35
server_internal_test.go Normal file
View file

@ -0,0 +1,35 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package pilosa
import (
"runtime"
"testing"
)
// Ensure the file handle count is working
func TestCountOpenFiles(t *testing.T) {
// Windows is not supported yet
if runtime.GOOS == "windows" {
t.Skip("Skipping unsupported countOpenFiles test on Windows.")
}
count, err := countOpenFiles()
if err != nil {
t.Errorf("countOpenFiles failed: %s", err)
}
if count == 0 {
t.Error("countOpenFiles returned invalid value 0.")
}
}

14
server_test.go
View file

@ -1,3 +1,17 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package pilosa_test
import (

6
stats.go
View file

@ -110,7 +110,7 @@ func (c *ExpvarStatsClient) WithTags(tags ...string) StatsClient {
return &ExpvarStatsClient{
m: m,
tags: UnionStringSlice(c.tags, tags),
tags: unionStringSlice(c.tags, tags),
}
}
@ -249,8 +249,8 @@ func (a MultiStatsClient) Close() error {
return nil
}
// UnionStringSlice returns a sorted set of tags which combine a & b.
func UnionStringSlice(a, b []string) []string {
// unionStringSlice returns a sorted set of tags which combine a & b.
func unionStringSlice(a, b []string) []string {
// Sort both sets first.
sort.Strings(a)
sort.Strings(b)

38
statsd/statsd.go
View file

@ -15,6 +15,7 @@
package statsd
import (
"sort"
"time"
"github.com/DataDog/datadog-go/statsd"
@ -72,7 +73,7 @@ func (c *StatsClient) Tags() []string {
func (c *StatsClient) WithTags(tags ...string) pilosa.StatsClient {
return &StatsClient{
client: c.client,
tags: pilosa.UnionStringSlice(c.tags, tags),
tags: unionStringSlice(c.tags, tags),
logger: c.logger,
}
}
@ -124,3 +125,38 @@ func (c *StatsClient) Timing(name string, value time.Duration, rate float64) {
func (c *StatsClient) SetLogger(logger pilosa.Logger) {
c.logger = logger
}
// unionStringSlice returns a sorted set of tags which combine a & b.
func unionStringSlice(a, b []string) []string {
// Sort both sets first.
sort.Strings(a)
sort.Strings(b)
// Find size of largest slice.
n := len(a)
if len(b) > n {
n = len(b)
}
// Exit if both sets are empty.
if n == 0 {
return nil
}
// Iterate over both in order and merge.
other := make([]string, 0, n)
for len(a) > 0 || len(b) > 0 {
if len(a) == 0 {
other, b = append(other, b[0]), b[1:]
} else if len(b) == 0 {
other, a = append(other, a[0]), a[1:]
} else if a[0] < b[0] {
other, a = append(other, a[0]), a[1:]
} else if b[0] < a[0] {
other, b = append(other, b[0]), b[1:]
} else {
other, a, b = append(other, a[0]), a[1:], b[1:]
}
}
return other
}

28
time.go
View file

@ -79,8 +79,8 @@ func ParseTimeQuantum(v string) (TimeQuantum, error) {
return q, nil
}
// ViewByTimeUnit returns the view name for time with a given quantum unit.
func ViewByTimeUnit(name string, t time.Time, unit rune) string {
// viewByTimeUnit returns the view name for time with a given quantum unit.
func viewByTimeUnit(name string, t time.Time, unit rune) string {
switch unit {
case 'Y':
return fmt.Sprintf("%s_%s", name, t.Format("2006"))
@ -95,11 +95,11 @@ func ViewByTimeUnit(name string, t time.Time, unit rune) string {
}
}
// ViewsByTime returns a list of views for a given timestamp.
func ViewsByTime(name string, t time.Time, q TimeQuantum) []string {
// viewsByTime returns a list of views for a given timestamp.
func viewsByTime(name string, t time.Time, q TimeQuantum) []string {
a := make([]string, 0, len(q))
for _, unit := range q {
view := ViewByTimeUnit(name, t, unit)
view := viewByTimeUnit(name, t, unit)
if view == "" {
continue
}
@ -108,8 +108,8 @@ func ViewsByTime(name string, t time.Time, q TimeQuantum) []string {
return a
}
// ViewsByTimeRange returns a list of views to traverse to query a time range.
func ViewsByTimeRange(name string, start, end time.Time, q TimeQuantum) []string {
// viewsByTimeRange returns a list of views to traverse to query a time range.
func viewsByTimeRange(name string, start, end time.Time, q TimeQuantum) []string {
t := start
// Save flags for performance.
@ -127,7 +127,7 @@ func ViewsByTimeRange(name string, start, end time.Time, q TimeQuantum) []string
if !nextDayGTE(t, end) {
break
} else if t.Hour() != 0 {
results = append(results, ViewByTimeUnit(name, t, 'H'))
results = append(results, viewByTimeUnit(name, t, 'H'))
t = t.Add(time.Hour)
continue
}
@ -138,7 +138,7 @@ func ViewsByTimeRange(name string, start, end time.Time, q TimeQuantum) []string
if !nextMonthGTE(t, end) {
break
} else if t.Day() != 1 {
results = append(results, ViewByTimeUnit(name, t, 'D'))
results = append(results, viewByTimeUnit(name, t, 'D'))
t = t.AddDate(0, 0, 1)
continue
}
@ -148,7 +148,7 @@ func ViewsByTimeRange(name string, start, end time.Time, q TimeQuantum) []string
if !nextYearGTE(t, end) {
break
} else if t.Month() != 1 {
results = append(results, ViewByTimeUnit(name, t, 'M'))
results = append(results, viewByTimeUnit(name, t, 'M'))
t = t.AddDate(0, 1, 0)
continue
}
@ -164,16 +164,16 @@ func ViewsByTimeRange(name string, start, end time.Time, q TimeQuantum) []string
// Walk back down from largest units to smallest units.
for t.Before(end) {
if hasYear && nextYearGTE(t, end) {
results = append(results, ViewByTimeUnit(name, t, 'Y'))
results = append(results, viewByTimeUnit(name, t, 'Y'))
t = t.AddDate(1, 0, 0)
} else if hasMonth && nextMonthGTE(t, end) {
results = append(results, ViewByTimeUnit(name, t, 'M'))
results = append(results, viewByTimeUnit(name, t, 'M'))
t = t.AddDate(0, 1, 0)
} else if hasDay && nextDayGTE(t, end) {
results = append(results, ViewByTimeUnit(name, t, 'D'))
results = append(results, viewByTimeUnit(name, t, 'D'))
t = t.AddDate(0, 0, 1)
} else if hasHour {
results = append(results, ViewByTimeUnit(name, t, 'H'))
results = append(results, viewByTimeUnit(name, t, 'H'))
t = t.Add(time.Hour)
} else {
break

58
time_test.go → time_internal_test.go
View file

@ -12,28 +12,26 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package pilosa_test
package pilosa
import (
"reflect"
"testing"
"time"
"github.com/pilosa/pilosa"
)
// Ensure string can be parsed into time quantum.
func TestParseTimeQuantum(t *testing.T) {
t.Run("OK", func(t *testing.T) {
if q, err := pilosa.ParseTimeQuantum("YMDH"); err != nil {
if q, err := ParseTimeQuantum("YMDH"); err != nil {
t.Fatalf("unexpected error: %s", err)
} else if q != pilosa.TimeQuantum("YMDH") {
} else if q != TimeQuantum("YMDH") {
t.Fatalf("unexpected quantum: %#v", q)
}
})
t.Run("ErrInvalidTimeQuantum", func(t *testing.T) {
if _, err := pilosa.ParseTimeQuantum("BADQUANTUM"); err != pilosa.ErrInvalidTimeQuantum {
if _, err := ParseTimeQuantum("BADQUANTUM"); err != ErrInvalidTimeQuantum {
t.Fatalf("unexpected error: %s", err)
}
})
@ -44,22 +42,22 @@ func TestViewByTimeUnit(t *testing.T) {
ts := time.Date(2000, time.January, 2, 3, 4, 5, 6, time.UTC)
t.Run("Y", func(t *testing.T) {
if s := pilosa.ViewByTimeUnit("F", ts, 'Y'); s != "F_2000" {
if s := viewByTimeUnit("F", ts, 'Y'); s != "F_2000" {
t.Fatalf("unexpected name: %s", s)
}
})
t.Run("M", func(t *testing.T) {
if s := pilosa.ViewByTimeUnit("F", ts, 'M'); s != "F_200001" {
if s := viewByTimeUnit("F", ts, 'M'); s != "F_200001" {
t.Fatalf("unexpected name: %s", s)
}
})
t.Run("D", func(t *testing.T) {
if s := pilosa.ViewByTimeUnit("F", ts, 'D'); s != "F_20000102" {
if s := viewByTimeUnit("F", ts, 'D'); s != "F_20000102" {
t.Fatalf("unexpected name: %s", s)
}
})
t.Run("H", func(t *testing.T) {
if s := pilosa.ViewByTimeUnit("F", ts, 'H'); s != "F_2000010203" {
if s := viewByTimeUnit("F", ts, 'H'); s != "F_2000010203" {
t.Fatalf("unexpected name: %s", s)
}
})
@ -70,14 +68,14 @@ func TestViewsByTime(t *testing.T) {
ts := time.Date(2000, time.January, 2, 3, 4, 5, 6, time.UTC)
t.Run("YMDH", func(t *testing.T) {
a := pilosa.ViewsByTime("F", ts, MustParseTimeQuantum("YMDH"))
a := viewsByTime("F", ts, mustParseTimeQuantum("YMDH"))
if !reflect.DeepEqual(a, []string{"F_2000", "F_200001", "F_20000102", "F_2000010203"}) {
t.Fatalf("unexpected names: %+v", a)
}
})
t.Run("D", func(t *testing.T) {
a := pilosa.ViewsByTime("F", ts, MustParseTimeQuantum("D"))
a := viewsByTime("F", ts, mustParseTimeQuantum("D"))
if !reflect.DeepEqual(a, []string{"F_20000102"}) {
t.Fatalf("unexpected names: %+v", a)
}
@ -87,82 +85,82 @@ func TestViewsByTime(t *testing.T) {
// Ensure sets of fields can be returned for a given time range.
func TestViewsByTimeRange(t *testing.T) {
t.Run("Y", func(t *testing.T) {
a := pilosa.ViewsByTimeRange("F", MustParseTime("2000-01-01 00:00"), MustParseTime("2002-01-01 00:00"), MustParseTimeQuantum("Y"))
a := viewsByTimeRange("F", mustParseTime("2000-01-01 00:00"), mustParseTime("2002-01-01 00:00"), mustParseTimeQuantum("Y"))
if !reflect.DeepEqual(a, []string{"F_2000", "F_2001"}) {
t.Fatalf("unexpected fields: %#v", a)
}
})
t.Run("YM", func(t *testing.T) {
a := pilosa.ViewsByTimeRange("F", MustParseTime("2000-11-01 00:00"), MustParseTime("2003-03-01 00:00"), MustParseTimeQuantum("YM"))
a := viewsByTimeRange("F", mustParseTime("2000-11-01 00:00"), mustParseTime("2003-03-01 00:00"), mustParseTimeQuantum("YM"))
if !reflect.DeepEqual(a, []string{"F_200011", "F_200012", "F_2001", "F_2002", "F_200301", "F_200302"}) {
t.Fatalf("unexpected fields: %#v", a)
}
})
t.Run("YMD", func(t *testing.T) {
a := pilosa.ViewsByTimeRange("F", MustParseTime("2000-11-28 00:00"), MustParseTime("2003-03-02 00:00"), MustParseTimeQuantum("YMD"))
a := viewsByTimeRange("F", mustParseTime("2000-11-28 00:00"), mustParseTime("2003-03-02 00:00"), mustParseTimeQuantum("YMD"))
if !reflect.DeepEqual(a, []string{"F_20001128", "F_20001129", "F_20001130", "F_200012", "F_2001", "F_2002", "F_200301", "F_200302", "F_20030301"}) {
t.Fatalf("unexpected fields: %#v", a)
}
})
t.Run("YMDH", func(t *testing.T) {
a := pilosa.ViewsByTimeRange("F", MustParseTime("2000-11-28 22:00"), MustParseTime("2002-03-01 03:00"), MustParseTimeQuantum("YMDH"))
a := viewsByTimeRange("F", mustParseTime("2000-11-28 22:00"), mustParseTime("2002-03-01 03:00"), mustParseTimeQuantum("YMDH"))
if !reflect.DeepEqual(a, []string{"F_2000112822", "F_2000112823", "F_20001129", "F_20001130", "F_200012", "F_2001", "F_200201", "F_200202", "F_2002030100", "F_2002030101", "F_2002030102"}) {
t.Fatalf("unexpected fields: %#v", a)
}
})
t.Run("M", func(t *testing.T) {
a := pilosa.ViewsByTimeRange("F", MustParseTime("2000-01-01 00:00"), MustParseTime("2000-03-01 00:00"), MustParseTimeQuantum("M"))
a := viewsByTimeRange("F", mustParseTime("2000-01-01 00:00"), mustParseTime("2000-03-01 00:00"), mustParseTimeQuantum("M"))
if !reflect.DeepEqual(a, []string{"F_200001", "F_200002"}) {
t.Fatalf("unexpected fields: %#v", a)
}
})
t.Run("MD", func(t *testing.T) {
a := pilosa.ViewsByTimeRange("F", MustParseTime("2000-11-29 00:00"), MustParseTime("2002-02-03 00:00"), MustParseTimeQuantum("MD"))
a := viewsByTimeRange("F", mustParseTime("2000-11-29 00:00"), mustParseTime("2002-02-03 00:00"), mustParseTimeQuantum("MD"))
if !reflect.DeepEqual(a, []string{"F_20001129", "F_20001130", "F_200012", "F_200101", "F_200102", "F_200103", "F_200104", "F_200105", "F_200106", "F_200107", "F_200108", "F_200109", "F_200110", "F_200111", "F_200112", "F_200201", "F_20020201", "F_20020202"}) {
t.Fatalf("unexpected fields: %#v", a)
}
})
t.Run("MDH", func(t *testing.T) {
a := pilosa.ViewsByTimeRange("F", MustParseTime("2000-11-29 22:00"), MustParseTime("2002-03-02 03:00"), MustParseTimeQuantum("MDH"))
a := viewsByTimeRange("F", mustParseTime("2000-11-29 22:00"), mustParseTime("2002-03-02 03:00"), mustParseTimeQuantum("MDH"))
if !reflect.DeepEqual(a, []string{"F_2000112922", "F_2000112923", "F_20001130", "F_200012", "F_200101", "F_200102", "F_200103", "F_200104", "F_200105", "F_200106", "F_200107", "F_200108", "F_200109", "F_200110", "F_200111", "F_200112", "F_200201", "F_200202", "F_20020301", "F_2002030200", "F_2002030201", "F_2002030202"}) {
t.Fatalf("unexpected fields: %#v", a)
}
})
t.Run("D", func(t *testing.T) {
a := pilosa.ViewsByTimeRange("F", MustParseTime("2000-01-01 00:00"), MustParseTime("2000-01-04 00:00"), MustParseTimeQuantum("D"))
a := viewsByTimeRange("F", mustParseTime("2000-01-01 00:00"), mustParseTime("2000-01-04 00:00"), mustParseTimeQuantum("D"))
if !reflect.DeepEqual(a, []string{"F_20000101", "F_20000102", "F_20000103"}) {
t.Fatalf("unexpected fields: %#v", a)
}
})
t.Run("DH", func(t *testing.T) {
a := pilosa.ViewsByTimeRange("F", MustParseTime("2000-01-01 22:00"), MustParseTime("2000-03-01 02:00"), MustParseTimeQuantum("DH"))
a := viewsByTimeRange("F", mustParseTime("2000-01-01 22:00"), mustParseTime("2000-03-01 02:00"), mustParseTimeQuantum("DH"))
if !reflect.DeepEqual(a, []string{"F_2000010122", "F_2000010123", "F_20000102", "F_20000103", "F_20000104", "F_20000105", "F_20000106", "F_20000107", "F_20000108", "F_20000109", "F_20000110", "F_20000111", "F_20000112", "F_20000113", "F_20000114", "F_20000115", "F_20000116", "F_20000117", "F_20000118", "F_20000119", "F_20000120", "F_20000121", "F_20000122", "F_20000123", "F_20000124", "F_20000125", "F_20000126", "F_20000127", "F_20000128", "F_20000129", "F_20000130", "F_20000131", "F_20000201", "F_20000202", "F_20000203", "F_20000204", "F_20000205", "F_20000206", "F_20000207", "F_20000208", "F_20000209", "F_20000210", "F_20000211", "F_20000212", "F_20000213", "F_20000214", "F_20000215", "F_20000216", "F_20000217", "F_20000218", "F_20000219", "F_20000220", "F_20000221", "F_20000222", "F_20000223", "F_20000224", "F_20000225", "F_20000226", "F_20000227", "F_20000228", "F_20000229", "F_2000030100", "F_2000030101"}) {
t.Fatalf("unexpected fields: %#v", a)
}
})
t.Run("H", func(t *testing.T) {
a := pilosa.ViewsByTimeRange("F", MustParseTime("2000-01-01 00:00"), MustParseTime("2000-01-01 02:00"), MustParseTimeQuantum("H"))
a := viewsByTimeRange("F", mustParseTime("2000-01-01 00:00"), mustParseTime("2000-01-01 02:00"), mustParseTimeQuantum("H"))
if !reflect.DeepEqual(a, []string{"F_2000010100", "F_2000010101"}) {
t.Fatalf("unexpected fields: %#v", a)
}
})
}
// DefaultTimeLayout is the time layout used by the tests.
const DefaultTimeLayout = "2006-01-02 15:04"
// defaultTimeLayout is the time layout used by the tests.
const defaultTimeLayout = "2006-01-02 15:04"
// MustParseTime parses value using DefaultTimeLayout. Panic on error.
func MustParseTime(value string) time.Time {
v, err := time.Parse(DefaultTimeLayout, value)
// mustParseTime parses value using DefaultTimeLayout. Panic on error.
func mustParseTime(value string) time.Time {
v, err := time.Parse(defaultTimeLayout, value)
if err != nil {
panic(err)
}
return v
}
// MustParseTimeQuantum parses v into a time quantum. Panic on error.
func MustParseTimeQuantum(v string) pilosa.TimeQuantum {
q, err := pilosa.ParseTimeQuantum(v)
// mustParseTimeQuantum parses v into a time quantum. Panic on error.
func mustParseTimeQuantum(v string) TimeQuantum {
q, err := ParseTimeQuantum(v)
if err != nil {
panic(err)
}

6
view.go
View file

@ -34,8 +34,8 @@ const (
viewBSIGroupPrefix = "bsig_"
)
// IsValidView returns true if name is valid.
func IsValidView(name string) bool {
// isValidView returns true if name is valid.
func isValidView(name string) bool {
return name == ViewStandard
}
@ -73,7 +73,7 @@ func NewView(path, index, field, name string, cacheSize uint32) *View {
name: name,
cacheSize: cacheSize,
cacheType: DefaultCacheType,
cacheType: defaultCacheType,
fragments: make(map[uint64]*Fragment),
broadcaster: NopBroadcaster,

2
view_internal_test.go
View file

@ -26,7 +26,7 @@ func mustOpenView(index, field, name string) *View {
panic(err)
}
v := NewView(path, index, field, name, DefaultCacheSize)
v := NewView(path, index, field, name, defaultCacheSize)
if err := v.open(); err != nil {
panic(err)
}