diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 62113a697..d15c42ed7 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -15,6 +15,8 @@ Before you start working on new features, you should [open a new issue][1] to le you're doing before you start working, otherwise you run the risk of duplicating effort. This also gives others an opportunity to provide input for your feature. +If you want to help but you aren't sure where to start, check out our [github label for low-effort issues][6]. + - Fork the [Pilosa repository][2] and then clone your fork: ```shell @@ -57,3 +59,4 @@ gives others an opportunity to provide input for your feature. [3]: https://github.com/pilosa/pilosa/compare/ [4]: https://github.com/pilosa/general/blob/master/proposal.md [5]: https://github.com/pilosa/pilosa/issues +[6]: https://github.com/pilosa/pilosa/issues?q=is%3Aopen+is%3Aissue+label%3Anewcomer \ No newline at end of file diff --git a/attr.go b/attr.go index 6f12ffdc0..ab29fc9c0 100644 --- a/attr.go +++ b/attr.go @@ -39,21 +39,56 @@ const ( AttrTypeFloat = 4 ) +// AttrCache represents a cache for attributes. +type AttrCache struct { + mu sync.RWMutex + attrs map[uint64]map[string]interface{} +} + +// Get returns the cached attributes for a given id. +func (c *AttrCache) Get(id uint64) map[string]interface{} { + c.mu.RLock() + defer c.mu.RUnlock() + attrs := c.attrs[id] + if attrs == nil { + return nil + } + + // Make a copy for safety + ret := make(map[string]interface{}) + for k, v := range attrs { + ret[k] = v + } + return ret +} + +// Set updates the cached attributes for a given id. +func (c *AttrCache) Set(id uint64, attrs map[string]interface{}) { + c.mu.Lock() + defer c.mu.Unlock() + c.attrs[id] = attrs +} + // AttrStore represents a storage layer for attributes. type AttrStore struct { - mu sync.RWMutex - path string - db *bolt.DB + mu sync.RWMutex + path string + db *bolt.DB + attrCache *AttrCache +} - // in-memory cache - attrs map[uint64]map[string]interface{} +// NewAttrCache returns a new instance of AttrCache. +func NewAttrCache() *AttrCache { + return &AttrCache{ + attrs: make(map[uint64]map[string]interface{}), + } } // NewAttrStore returns a new instance of AttrStore. func NewAttrStore(path string) *AttrStore { return &AttrStore{ - path: path, - attrs: make(map[uint64]map[string]interface{}), + path: path, + attrCache: NewAttrCache(), } } @@ -96,7 +131,7 @@ func (s *AttrStore) Attrs(id uint64) (m map[string]interface{}, err error) { defer s.mu.RUnlock() // Check cache for map. - if m = s.attrs[id]; m != nil { + if m = s.attrCache.Get(id); m != nil { return m, nil } @@ -112,7 +147,7 @@ func (s *AttrStore) Attrs(id uint64) (m map[string]interface{}, err error) { } // Add to cache. - s.attrs[id] = m + s.attrCache.Set(id, m) return } @@ -149,7 +184,7 @@ func (s *AttrStore) SetAttrs(id uint64, m map[string]interface{}) error { } // Swap attributes map in cache. - s.attrs[id] = attr + s.attrCache.Set(id, attr) return nil } @@ -184,7 +219,7 @@ func (s *AttrStore) SetBulkAttrs(m map[uint64]map[string]interface{}) error { // Swap attributes map in cache. for id, attr := range attrs { - s.attrs[id] = attr + s.attrCache.Set(id, attr) } return nil diff --git a/cmd/server_test.go b/cmd/server_test.go index 10851e45b..25d0d7fd1 100644 --- a/cmd/server_test.go +++ b/cmd/server_test.go @@ -52,18 +52,23 @@ func TestServerConfig(t *testing.T) { [cluster] poll-interval = "45s" + type = "http" replicas = 2 hosts = [ "localhost:19444", ] + internal-hosts = [ + "localhost:19500", + "localhost:19501", + ] `, validation: func() error { v := validator{} v.Check(cmd.Server.Config.DataDir, actualDataDir) - v.Check(cmd.Server.Config.Host, "example.com:10111") + v.Check(cmd.Server.Config.Bind, "example.com:10111") v.Check(cmd.Server.Config.Cluster.ReplicaN, 2) v.Check(cmd.Server.Config.Cluster.Hosts, []string{"example.com:10111", "example.com:10110"}) - v.Check(cmd.Server.Config.Cluster.PollingInterval, pilosa.Duration(time.Second*182)) + v.Check(cmd.Server.Config.Cluster.PollInterval, pilosa.Duration(time.Second*182)) return v.Error() }, }, @@ -75,9 +80,14 @@ func TestServerConfig(t *testing.T) { bind = "localhost:0" data-dir = "` + actualDataDir + `" [cluster] + type = "http" hosts = [ "localhost:19444", ] + internal-hosts = [ + "localhost:19500", + "localhost:19501", + ] [plugins] path = "/var/sloth" `, @@ -113,7 +123,7 @@ func TestServerConfig(t *testing.T) { validation: func() error { v := validator{} v.Check(cmd.Server.Config.Cluster.Hosts, []string{"localhost:19444"}) - v.Check(cmd.Server.Config.Cluster.PollingInterval, pilosa.Duration(time.Minute*2)) + v.Check(cmd.Server.Config.Cluster.PollInterval, pilosa.Duration(time.Minute*2)) v.Check(cmd.Server.Config.AntiEntropy.Interval, pilosa.Duration(time.Minute*11)) v.Check(cmd.Server.CPUProfile, profFile.Name()) v.Check(cmd.Server.CPUTime, time.Minute) diff --git a/config.go b/config.go index 439e9c3ed..30c1ebdd3 100644 --- a/config.go +++ b/config.go @@ -14,7 +14,17 @@ package pilosa -import "time" +import ( + "time" +) + +// Cluster types. +const ( + ClusterNone = "" + ClusterStatic = "static" + ClusterHTTP = "http" + ClusterGossip = "gossip" +) const ( // DefaultHost is the default hostname to use. @@ -24,7 +34,7 @@ const ( DefaultPort = "10101" // DefaultClusterType sets the node intercommunication method. - DefaultClusterType = "static" + DefaultClusterType = ClusterStatic // DefaultInternalPort the port the nodes intercommunicate on. DefaultInternalPort = "14000" @@ -36,20 +46,23 @@ const ( DefaultMaxWritesPerRequest = 5000 ) +// ClusterTypes set of cluster types. +var ClusterTypes = []string{ClusterNone, ClusterStatic, ClusterHTTP, ClusterGossip} + // Config represents the configuration for the command. type Config struct { DataDir string `toml:"data-dir"` - Host string `toml:"host"` + Bind string `toml:"bind"` Cluster struct { - ReplicaN int `toml:"replicas"` - Type string `toml:"type"` - Hosts []string `toml:"hosts"` - InternalHosts []string `toml:"internal-hosts"` - PollingInterval Duration `toml:"polling-interval"` - InternalPort string `toml:"internal-port"` - GossipSeed string `toml:"gossip-seed"` - LongQueryTime Duration `toml:"long-query-time"` + ReplicaN int `toml:"replicas"` + Type string `toml:"type"` + Hosts []string `toml:"hosts"` + InternalHosts []string `toml:"internal-hosts"` + PollInterval Duration `toml:"poll-interval"` + InternalPort string `toml:"internal-port"` + GossipSeed string `toml:"gossip-seed"` + LongQueryTime Duration `toml:"long-query-time"` } `toml:"cluster"` Plugins struct { @@ -67,21 +80,21 @@ type Config struct { LogPath string `toml:"log-path"` Metric struct { - Service string `toml:"service"` - Host string `toml:"host"` - PollingInterval Duration `toml:"interval"` - } `toml:"metrics"` + Service string `toml:"service"` + Host string `toml:"host"` + PollInterval Duration `toml:"poll-interval"` + } `toml:"metric"` } // NewConfig returns an instance of Config with default options. func NewConfig() *Config { c := &Config{ - Host: DefaultHost + ":" + DefaultPort, + Bind: DefaultHost + ":" + DefaultPort, MaxWritesPerRequest: DefaultMaxWritesPerRequest, } c.Cluster.ReplicaN = DefaultReplicaN c.Cluster.Type = DefaultClusterType - c.Cluster.PollingInterval = Duration(DefaultPollingInterval) + c.Cluster.PollInterval = Duration(DefaultPollingInterval) c.Cluster.Hosts = []string{} c.Cluster.InternalHosts = []string{} c.AntiEntropy.Interval = Duration(DefaultAntiEntropyInterval) @@ -89,6 +102,35 @@ func NewConfig() *Config { return c } +// Validate that all configuration permutations are compatible with each other. +func (c *Config) Validate() error { + if !StringInSlice(c.Cluster.Type, ClusterTypes) { + return ErrConfigClusterTypeInvalid + } + if len(c.Cluster.Hosts) > 1 && !(c.Cluster.Type == ClusterHTTP || c.Cluster.Type == ClusterGossip) { + return ErrConfigClusterTypeMissing + } + if c.Cluster.Type == ClusterHTTP || c.Cluster.Type == ClusterGossip { + if c.Cluster.ReplicaN > len(c.Cluster.Hosts) { + return ErrConfigReplicaNInvalid + } + if len(c.Cluster.Hosts) != len(c.Cluster.InternalHosts) { + return ErrConfigHostsMismatch + } + if !foundItem(c.Cluster.Hosts, c.Bind) { + return ErrConfigHostsMissing + } + if !ContainsSubstring(c.Cluster.InternalPort, c.Cluster.InternalHosts) { + return ErrConfigBroadcastPort + } + } + if c.Cluster.Type == ClusterGossip && !StringInSlice(c.Cluster.GossipSeed, c.Cluster.InternalHosts) { + return ErrConfigGossipSeed + } + + return nil +} + // Duration is a TOML wrapper type for time.Duration. type Duration time.Duration @@ -111,6 +153,7 @@ func (d Duration) MarshalText() (text []byte, err error) { return []byte(d.String()), nil } +// MarshalTOML write duration into valid TOML. func (d Duration) MarshalTOML() ([]byte, error) { return []byte(d.String()), nil } diff --git a/config_test.go b/config_test.go new file mode 100644 index 000000000..78c4a6127 --- /dev/null +++ b/config_test.go @@ -0,0 +1,88 @@ +package pilosa_test + +import ( + "reflect" + "testing" + "time" + + "github.com/pilosa/pilosa" +) + +func Test_NewConfig(t *testing.T) { + c := pilosa.NewConfig() + + c.Cluster.Hosts = []string{c.Bind, "localhost:10102"} + if err := c.Validate(); err != pilosa.ErrConfigClusterTypeMissing { + t.Fatal(err) + } + + c.Cluster.Type = "test" + if err := c.Validate(); err != pilosa.ErrConfigClusterTypeInvalid { + t.Fatal(err) + } + + c.Cluster.Type = pilosa.ClusterHTTP + if err := c.Validate(); err != pilosa.ErrConfigHostsMismatch { + t.Fatal(err) + } + + c.Cluster.InternalPort = pilosa.DefaultInternalPort + c.Cluster.InternalHosts = []string{"localhost:14004", "localhost:14001"} + if err := c.Validate(); err != pilosa.ErrConfigBroadcastPort { + t.Fatal(err) + } + + c.Cluster.InternalHosts = []string{"localhost:14000", "localhost:14001"} + c.Bind = "localhost:1" + // Check for bind addres in cluster hosts + if err := c.Validate(); err != pilosa.ErrConfigHostsMissing { + t.Fatal(err) + } + + c.Bind = "localhost:10101" + c.Cluster.ReplicaN = 3 + if err := c.Validate(); err != pilosa.ErrConfigReplicaNInvalid { + t.Fatal(err) + } + + c.Cluster.ReplicaN = 2 + c.Cluster.Type = pilosa.ClusterGossip + c.Cluster.GossipSeed = "localhost:10101" + if err := c.Validate(); err != pilosa.ErrConfigGossipSeed { + t.Fatal(err) + } + + c.Cluster.GossipSeed = "localhost:14000" + if err := c.Validate(); err != nil { + t.Fatal(err) + } +} + +func TestDuration(t *testing.T) { + d := pilosa.Duration(time.Second * 182) + if d.String() != "3m2s" { + t.Fatalf("Unexpected time Duration %s", d) + } + + b := []byte{51, 109, 50, 115} + v, _ := d.MarshalText() + if !reflect.DeepEqual(b, v) { + t.Fatalf("Unexpected marshalled value %v", v) + } + + v, _ = d.MarshalTOML() + if !reflect.DeepEqual(b, v) { + t.Fatalf("Unexpected marshalled value %v", v) + } + + err := d.UnmarshalText([]byte("5")) + if err.Error() != "time: missing unit in duration 5" { + t.Fatalf("expected time: missing unit in duration: %s", err) + } + + err = d.UnmarshalText([]byte("3m2s")) + v, _ = d.MarshalText() + if !reflect.DeepEqual(b, v) { + t.Fatalf("Unexpected marshalled value %v", v) + } +} diff --git a/ctl/server.go b/ctl/server.go index 6186443f8..cde11424b 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -25,12 +25,13 @@ import ( func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags := cmd.Flags() flags.StringVarP(&srv.Config.DataDir, "data-dir", "d", "~/.pilosa", "Directory to store pilosa data files.") - flags.StringVarP(&srv.Config.Host, "bind", "b", ":10101", "Default URI on which pilosa should listen.") + flags.StringVarP(&srv.Config.Bind, "bind", "b", ":10101", "Default URI on which pilosa should listen.") flags.IntVarP(&srv.Config.MaxWritesPerRequest, "max-writes-per-request", "", srv.Config.MaxWritesPerRequest, "Number of write commands per request.") flags.IntVarP(&srv.Config.Cluster.ReplicaN, "cluster.replicas", "", 1, "Number of hosts each piece of data should be stored on.") flags.StringSliceVarP(&srv.Config.Cluster.Hosts, "cluster.hosts", "", []string{}, "Comma separated list of hosts in cluster.") flags.StringSliceVarP(&srv.Config.Cluster.InternalHosts, "cluster.internal-hosts", "", []string{}, "Comma separated list of hosts in cluster used for internal communication.") - flags.DurationVarP((*time.Duration)(&srv.Config.Cluster.PollingInterval), "cluster.poll-interval", "", time.Minute, "Polling interval for cluster.") // TODO what actually is this? + flags.DurationVarP((*time.Duration)(&srv.Config.Cluster.PollInterval), "cluster.poll-interval", "", time.Minute, "Polling interval for cluster.") // TODO what actually is this? + flags.DurationVarP((*time.Duration)(&srv.Config.Cluster.LongQueryTime), "cluster.long-query-time", "", time.Minute, "Long Query Time.") flags.StringVarP(&srv.Config.Plugins.Path, "plugins.path", "", "", "Path to plugin directory.") flags.StringVar(&srv.Config.LogPath, "log-path", "", "Log path") flags.DurationVarP((*time.Duration)(&srv.Config.AntiEntropy.Interval), "anti-entropy.interval", "", time.Minute*10, "Interval at which to run anti-entropy routine.") @@ -41,5 +42,5 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags.StringVarP(&srv.Config.Cluster.InternalPort, "cluster.internal-port", "", "", "Port to which pilosa should bind for internal state sharing.") flags.StringVarP(&srv.Config.Metric.Service, "metric.service", "", "nop", "Default URI on which pilosa should listen.") flags.StringVarP(&srv.Config.Metric.Host, "metric.host", "", "", "Default URI to send metrics.") - flags.DurationVarP((*time.Duration)(&srv.Config.Metric.PollingInterval), "metric.poll-interval", "", time.Minute*0, "Polling interval metrics.") + flags.DurationVarP((*time.Duration)(&srv.Config.Metric.PollInterval), "metric.poll-interval", "", time.Minute*0, "Polling interval metrics.") } diff --git a/docs/configuration.md b/docs/configuration.md index f93647cb5..5eecde849 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -144,6 +144,19 @@ Any flag that has a value that is a comma separated list on the command line bec data-dir = "~/.pilosa" ``` +#### Gossip Seed + +* Description: When using the gossip [Cluster Type]({{< ref "#cluster-type" >}}), this specifies which internal host should be used to initialize membership in the cluster. Typcially this can be the address of any available host in the cluster. For example, when starting a three-node cluster made up of `node0`, `node1`, and `node2`, the `gossip-seed` for all three nodes can be configured to be the address of `node0`. +* Flag: `--gossip-seed="localhost:11101"` +* Env: `PILOSA_GOSSIP_SEED="localhost:11101"` +* Config: + + ```toml + [cluster] + type = "gossip" + gossip-seed = "localhost:11101" + ``` + #### Profile CPU * Description: If this is set to a path, collect a cpu profile and store it there. @@ -199,4 +212,4 @@ Any flag that has a value that is a comma separated list on the command line bec ```toml [metric] poll-interval = "0m15s" - ``` \ No newline at end of file + ``` diff --git a/docs/query-language.md b/docs/query-language.md index 5400c8ead..93ddb846a 100644 --- a/docs/query-language.md +++ b/docs/query-language.md @@ -341,6 +341,13 @@ have the attribute specified by `field` with one of the values specified in **Result Type:** array of key/count objects +**Caveats:** +* Performing a TopN() query on a frame with cache type ranked will return the top bitmaps sorted by count in descending order. +* Frames with cache type lru will maintain an LRU (Least Recently Used) cache, thus a TopN() query on this type of frame will return bitmaps sorted in order of most recently set bit. +* The frame's cache size determines the number of sorted bitmaps to maintain in the cache for purposes of TopN() queries. There is a tradeoff between performance and accuracy; increasing the cache size will improve accuracy of results at the cost of performance. +* Once full, the cache will truncate the set of bitmaps according to the frame option CacheSize. Bitmaps that straddle the limit and have the same count will be truncated in no particular order. +* The TopN() query's attribute filter is applied to the existing sorted cache of bitmaps. Bitmaps that fall outside of the sorted cache range, even if they would normally pass the filter, are ignored. + **Examples:** ``` diff --git a/executor.go b/executor.go index 6cb0e9e7a..d58be42f1 100644 --- a/executor.go +++ b/executor.go @@ -168,6 +168,8 @@ func (e *Executor) executeCall(ctx context.Context, index string, c *pql.Call, s return e.executeCount(ctx, index, c, slices, opt) case "SetBit": return e.executeSetBit(ctx, index, c, opt) + case "SetFieldValue": + return nil, e.executeSetFieldValue(ctx, index, c, opt) case "SetRowAttrs": return nil, e.executeSetRowAttrs(ctx, index, c, opt) case "SetColumnAttrs": @@ -839,6 +841,77 @@ func (e *Executor) executeSetBitView(ctx context.Context, index string, c *pql.C return ret, nil } +// executeSetFieldValue executes a SetFieldValue() call. +func (e *Executor) executeSetFieldValue(ctx context.Context, index string, c *pql.Call, opt *ExecOptions) error { + frameName, ok := c.Args["frame"].(string) + if !ok { + return errors.New("SetFieldValue() frame required") + } + + // Retrieve column label. + idx := e.Holder.Index(index) + if idx == nil { + return ErrIndexNotFound + } + columnLabel := idx.ColumnLabel() + + // Retrieve frame. + frame := e.Holder.Frame(index, frameName) + if frame == nil { + return ErrFrameNotFound + } + + // Parse labels. + columnID, ok, err := c.UintArg(columnLabel) + if err != nil { + return fmt.Errorf("reading SetFieldValue() column: %v", err) + } else if !ok { + return fmt.Errorf("SetFieldValue() column field '%v' required", columnLabel) + } + + // Copy args and remove reserved fields. + args := pql.CopyArgs(c.Args) + delete(args, "frame") + delete(args, columnLabel) + + // Set values. + for name, value := range args { + switch value := value.(type) { + case int64: + if _, err := frame.SetFieldValue(columnID, name, value); err != nil { + return err + } + default: + return ErrInvalidFieldValueType + } + } + frame.Stats.Count("SetFieldValue", 1, 1.0) + + // Do not forward call if this is already being forwarded. + if opt.Remote { + return nil + } + + // Execute on remote nodes in parallel. + nodes := Nodes(e.Cluster.Nodes).FilterHost(e.Host) + resp := make(chan error, len(nodes)) + for _, node := range nodes { + go func(node *Node) { + _, err := e.exec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil, opt) + resp <- err + }(node) + } + + // Return first error. + for range nodes { + if err := <-resp; err != nil { + return err + } + } + + return nil +} + // executeSetRowAttrs executes a SetRowAttrs() call. func (e *Executor) executeSetRowAttrs(ctx context.Context, index string, c *pql.Call, opt *ExecOptions) error { frameName, ok := c.Args["frame"].(string) diff --git a/executor_test.go b/executor_test.go index d4714b714..3c0f7e8ad 100644 --- a/executor_test.go +++ b/executor_test.go @@ -234,6 +234,103 @@ func TestExecutor_Execute_SetBit(t *testing.T) { } } +// Ensure a SetFieldValue() query can be executed. +func TestExecutor_Execute_SetFieldValue(t *testing.T) { + t.Run("OK", func(t *testing.T) { + hldr := test.MustOpenHolder() + defer hldr.Close() + + // Create frames. + index := hldr.MustCreateIndexIfNotExists("i", pilosa.IndexOptions{}) + if _, err := index.CreateFrameIfNotExists("f", pilosa.FrameOptions{ + RangeEnabled: true, + Fields: []*pilosa.Field{ + {Name: "field0", Type: pilosa.FieldTypeInt, Min: 0, Max: 50}, + {Name: "field1", Type: pilosa.FieldTypeInt, Min: 1, Max: 2}, + }, + }); err != nil { + t.Fatal(err) + } else if _, err := index.CreateFrameIfNotExists("xxx", pilosa.FrameOptions{}); err != nil { + t.Fatal(err) + } + + // Set field values. + e := test.NewExecutor(hldr.Holder, test.NewCluster(1)) + if _, err := e.Execute(context.Background(), "i", test.MustParse(`SetFieldValue(columnID=10, frame=f, field0=25, field1=2)`), nil, nil); err != nil { + t.Fatal(err) + } else if _, err := e.Execute(context.Background(), "i", test.MustParse(`SetFieldValue(columnID=100, frame=f, field0=10)`), nil, nil); err != nil { + t.Fatal(err) + } + + f := hldr.Frame("i", "f") + if value, exists, err := f.FieldValue(10, "field0"); err != nil { + t.Fatal(err) + } else if !exists { + t.Fatal("expected value to exist") + } else if value != 25 { + t.Fatal("unexpected value: %v", value) + } + + if value, exists, err := f.FieldValue(10, "field1"); err != nil { + t.Fatal(err) + } else if !exists { + t.Fatal("expected value to exist") + } else if value != 2 { + t.Fatal("unexpected value: %v", value) + } + + if value, exists, err := f.FieldValue(100, "field0"); err != nil { + t.Fatal(err) + } else if !exists { + t.Fatal("expected value to exist") + } else if value != 10 { + t.Fatal("unexpected value: %v", value) + } + }) + + t.Run("", func(t *testing.T) { + hldr := test.MustOpenHolder() + defer hldr.Close() + index := hldr.MustCreateIndexIfNotExists("i", pilosa.IndexOptions{}) + if _, err := index.CreateFrameIfNotExists("f", pilosa.FrameOptions{ + RangeEnabled: true, + Fields: []*pilosa.Field{ + {Name: "field0", Type: pilosa.FieldTypeInt, Min: 0, Max: 100}, + }, + }); err != nil { + t.Fatal(err) + } + + t.Run("ErrFrameRequired", func(t *testing.T) { + e := test.NewExecutor(hldr.Holder, test.NewCluster(1)) + if _, err := e.Execute(context.Background(), "i", test.MustParse(`SetFieldValue(columnID=10, field0=100)`), nil, nil); err == nil || err.Error() != `SetFieldValue() frame required` { + t.Fatalf("unexpected error: %s", err) + } + }) + + t.Run("ErrColumnFieldRequired", func(t *testing.T) { + e := test.NewExecutor(hldr.Holder, test.NewCluster(1)) + if _, err := e.Execute(context.Background(), "i", test.MustParse(`SetFieldValue(invalid_column_name=10, frame=f, field0=100)`), nil, nil); err == nil || err.Error() != `SetFieldValue() column field 'columnID' required` { + t.Fatalf("unexpected error: %s", err) + } + }) + + t.Run("ErrColumnFieldValue", func(t *testing.T) { + e := test.NewExecutor(hldr.Holder, test.NewCluster(1)) + if _, err := e.Execute(context.Background(), "i", test.MustParse(`SetFieldValue(invalid_column_name="bad_column", frame=f, field0=100)`), nil, nil); err == nil || err.Error() != `SetFieldValue() column field 'columnID' required` { + t.Fatalf("unexpected error: %s", err) + } + }) + + t.Run("ErrInvalidFieldValueType", func(t *testing.T) { + e := test.NewExecutor(hldr.Holder, test.NewCluster(1)) + if _, err := e.Execute(context.Background(), "i", test.MustParse(`SetFieldValue(columnID=10, frame=f, field0="hello")`), nil, nil); err == nil || err.Error() != `invalid field value type` { + t.Fatalf("unexpected error: %s", err) + } + }) + }) +} + // Ensure a SetRowAttrs() query can be executed. func TestExecutor_Execute_SetRowAttrs(t *testing.T) { hldr := test.MustOpenHolder() diff --git a/fragment_test.go b/fragment_test.go index 50e169fe4..52092eae9 100644 --- a/fragment_test.go +++ b/fragment_test.go @@ -176,7 +176,8 @@ func TestFragment_SetFieldValue(t *testing.T) { if err := quick.Check(func(bitDepth uint, columnN uint64, values []uint64) bool { // Limit bit depth & maximum values. bitDepth = (bitDepth % 62) + 1 - columnN = (columnN % 100) + 1 + columnN = (columnN % 99) + 1 + for i := range values { values[i] = values[i] % (1 << bitDepth) } @@ -497,7 +498,7 @@ func TestFragment_Checksum(t *testing.T) { // Ensure new checksum is different. if chksum := f.Checksum(); bytes.Equal(chksum, orig) { - t.Fatalf("expected checksum to change: %x", chksum, orig) + t.Fatalf("expected checksum to change: %x - %x", chksum, orig) } } diff --git a/pilosa.go b/pilosa.go index bdce7a3d8..9f9ca7c59 100644 --- a/pilosa.go +++ b/pilosa.go @@ -17,6 +17,7 @@ package pilosa import ( "errors" "regexp" + "strings" "github.com/pilosa/pilosa/internal" ) @@ -53,6 +54,7 @@ var ( ErrInverseRangeNotAllowed = errors.New("inverse range not allowed") ErrRangeCacheNotAllowed = errors.New("range cache not allowed") ErrFrameFieldsNotAllowed = errors.New("frame fields not allowed") + ErrInvalidFieldValueType = errors.New("invalid field value type") ErrFieldValueTooLow = errors.New("field value too low") ErrFieldValueTooHigh = errors.New("field value too high") @@ -66,6 +68,14 @@ var ( ErrFragmentNotFound = errors.New("fragment not found") ErrQueryRequired = errors.New("query required") ErrTooManyWrites = errors.New("too many write commands") + + ErrConfigClusterTypeInvalid = errors.New("invalid cluster type") + ErrConfigClusterTypeMissing = errors.New("missing cluster type") + ErrConfigHostsMissing = errors.New("missing bind address in cluster hosts") + ErrConfigBroadcastPort = errors.New("internal-port not found in internal-hosts") + ErrConfigHostsMismatch = errors.New("hosts and internal-hosts length mismatch") + ErrConfigReplicaNInvalid = errors.New("replica number must be <= hosts") + ErrConfigGossipSeed = errors.New("invalid gossip seed") ) // Regular expression to validate index and frame names. @@ -142,3 +152,23 @@ func ValidateLabel(label string) error { } return nil } + +// StringInSlice checks is substring a is in the slice +func StringInSlice(a string, list []string) bool { + for _, b := range list { + if b == a { + return true + } + } + return false +} + +// ContainsSubstring checks is substring a is contained in the slice +func ContainsSubstring(a string, list []string) bool { + for _, b := range list { + if strings.Contains(b, a) { + return true + } + } + return false +} diff --git a/pilosa_test.go b/pilosa_test.go index 7f43fefbf..3b136dd56 100644 --- a/pilosa_test.go +++ b/pilosa_test.go @@ -54,3 +54,27 @@ func TestValidateLabelInvalid(t *testing.T) { } } } + +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) + } +} diff --git a/server/server.go b/server/server.go index f66b88870..6e51c4e74 100644 --- a/server/server.go +++ b/server/server.go @@ -107,7 +107,11 @@ func (m *Command) Run(args ...string) (err error) { // SetupServer use the cluster configuration to setup this server func (m *Command) SetupServer() error { - var err error + err := m.Config.Validate() + if err != nil { + return err + } + cluster := pilosa.NewCluster() cluster.ReplicaN = m.Config.Cluster.ReplicaN @@ -131,7 +135,7 @@ func (m *Command) SetupServer() error { // Configure holder. m.Server.Logger().Printf("Using data from: %s\n", m.Config.DataDir) m.Server.Holder.Path = m.Config.DataDir - m.Server.MetricInterval = time.Duration(m.Config.Metric.PollingInterval) + m.Server.MetricInterval = time.Duration(m.Config.Metric.PollInterval) m.Server.Holder.Stats, err = NewStatsClient(m.Config.Metric.Service, m.Config.Metric.Host) if err != nil { return err @@ -142,7 +146,7 @@ func (m *Command) SetupServer() error { // Copy configuration flags. m.Server.MaxWritesPerRequest = m.Config.MaxWritesPerRequest - m.Server.Host, err = normalizeHost(m.Config.Host) + m.Server.Host, err = normalizeHost(m.Config.Bind) if err != nil { return err } @@ -154,7 +158,7 @@ func (m *Command) SetupServer() error { } switch m.Config.Cluster.Type { - case "http": + case pilosa.ClusterHTTP: m.Server.Broadcaster = httpbroadcast.NewHTTPBroadcaster(m.Server, internalPortStr) m.Server.BroadcastReceiver = httpbroadcast.NewHTTPBroadcastReceiver(internalPortStr, m.Server.LogOutput) m.Server.Cluster.NodeSet = httpbroadcast.NewHTTPNodeSet() @@ -162,7 +166,7 @@ func (m *Command) SetupServer() error { if err != nil { return err } - case "gossip": + case pilosa.ClusterGossip: gossipPort, err := strconv.Atoi(internalPortStr) if err != nil { return err @@ -172,15 +176,15 @@ func (m *Command) SetupServer() error { gossipSeed = m.Config.Cluster.GossipSeed } // get the host portion of addr to use for binding - gossipHost, _, err := net.SplitHostPort(m.Config.Host) + gossipHost, _, err := net.SplitHostPort(m.Config.Bind) if err != nil { - gossipHost = m.Config.Host + gossipHost = m.Config.Bind } - gossipNodeSet := gossip.NewGossipNodeSet(m.Config.Host, gossipHost, gossipPort, gossipSeed, m.Server) + gossipNodeSet := gossip.NewGossipNodeSet(m.Config.Bind, gossipHost, gossipPort, gossipSeed, m.Server) m.Server.Cluster.NodeSet = gossipNodeSet m.Server.Broadcaster = gossipNodeSet m.Server.BroadcastReceiver = gossipNodeSet - case "static", "": + case pilosa.ClusterStatic, pilosa.ClusterNone: m.Server.Broadcaster = pilosa.NopBroadcaster m.Server.Cluster.NodeSet = pilosa.NewStaticNodeSet() m.Server.BroadcastReceiver = pilosa.NopBroadcastReceiver diff --git a/server/server_test.go b/server/server_test.go index b6af8c423..37b6f2f38 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -346,10 +346,10 @@ func TestMain_FrameRestore(t *testing.T) { // Ensure the host can be parsed. func TestConfig_Parse_Host(t *testing.T) { - if c, err := ParseConfig(`host = "local"`); err != nil { + if c, err := ParseConfig(`bind = "local"`); err != nil { t.Fatal(err) - } else if c.Host != "local" { - t.Fatalf("unexpected host: %s", c.Host) + } else if c.Bind != "local" { + t.Fatalf("unexpected host: %s", c.Bind) } } @@ -606,7 +606,7 @@ func NewMain() *Main { m := &Main{Command: server.NewCommand(os.Stdin, os.Stdout, os.Stderr)} m.Server.Network = *test.Network m.Config.DataDir = path - m.Config.Host = "localhost:0" + m.Config.Bind = "localhost:0" m.Command.Stdin = &m.Stdin m.Command.Stdout = &m.Stdout m.Command.Stderr = &m.Stderr