diff --git a/cmd/server_test.go b/cmd/server_test.go index a0f6096c8..25d0d7fd1 100644 --- a/cmd/server_test.go +++ b/cmd/server_test.go @@ -52,10 +52,15 @@ 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{} @@ -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" `, diff --git a/config.go b/config.go index b6961850b..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,6 +46,9 @@ 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"` @@ -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/fragment_test.go b/fragment_test.go index 5d3dd7271..01f4a46bc 100644 --- a/fragment_test.go +++ b/fragment_test.go @@ -497,7 +497,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 a334a2be1..5ed50e8dd 100644 --- a/pilosa.go +++ b/pilosa.go @@ -17,6 +17,7 @@ package pilosa import ( "errors" "regexp" + "strings" "github.com/pilosa/pilosa/internal" ) @@ -56,6 +57,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. @@ -132,3 +141,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 fd4462be9..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 @@ -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 @@ -180,7 +184,7 @@ func (m *Command) SetupServer() error { 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