Merge pull request #728 from raskle/712-internal-hosts

712 Configuration validation

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"
 	`,

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
 }

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)
+	}
+}

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)
 	}
 }
 

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
+}

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)
+	}
+}

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