Merge pull request #1099 from travisturner/cluster-disabled

Changes configuration cluster.type (string) to cluster.disabled (bool)

cmd/server_test.go

@@ -52,7 +52,7 @@ func TestServerConfig(t *testing.T) {
 	max-writes-per-request = 3000
 
 	[cluster]
-		type = "static"
+		disabled = true
 		replicas = 2
 		hosts = [
 			"localhost:19444",
@@ -78,7 +78,7 @@ func TestServerConfig(t *testing.T) {
 	bind = "localhost:0"
 	data-dir = "` + actualDataDir + `"
 	[cluster]
-		type = "static"
+		disabled = true
 		hosts = [
 			"localhost:19444",
 		]
@@ -92,7 +92,7 @@ func TestServerConfig(t *testing.T) {
 		},
 		// TEST 2
 		{
-			args: []string{"server", "--log-path", logFile.Name(), "--cluster.type", "static"},
+			args: []string{"server", "--log-path", logFile.Name(), "--cluster.disabled", "true"},
 			env:  map[string]string{"PILOSA_PROFILE_CPU_TIME": "1m"},
 			cfgFileContent: `
 	bind = "localhost:19444"

config.go

@@ -35,8 +35,8 @@ const (
 	// DefaultPort is the default port to use with the hostname.
 	DefaultPort = "10101"
 
-	// DefaultClusterType sets the node intercommunication method.
-	DefaultClusterType = ClusterGossip
+	// DefaultClusterDisabled sets the node intercommunication method.
+	DefaultClusterDisabled = false
 
 	// DefaultMetrics sets the internal metrics to no-op.
 	DefaultMetrics = "nop"
@@ -113,9 +113,6 @@ const (
 	DefaultMetricPollInterval = 0 * time.Minute
 )
 
-// ClusterTypes set of cluster types.
-var ClusterTypes = []string{ClusterNone, ClusterStatic, ClusterGossip}
-
 // TLSConfig contains TLS configuration
 type TLSConfig struct {
 	// CertificatePath contains the path to the certificate (.crt or .pem file)
@@ -145,9 +142,9 @@ type Config struct {
 	TLS TLSConfig
 
 	Cluster struct {
+		Disabled      bool     `toml:"disabled"`
 		Coordinator   string   `toml:"coordinator"`
 		ReplicaN      int      `toml:"replicas"`
-		Type          string   `toml:"type"`
 		Hosts         []string `toml:"hosts"`
 		LongQueryTime Duration `toml:"long-query-time"`
 	} `toml:"cluster"`
@@ -189,9 +186,9 @@ func NewConfig() *Config {
 	}
 
 	// Cluster config.
+	c.Cluster.Disabled = DefaultClusterDisabled
 	// c.Cluster.Coordinator = ""
 	c.Cluster.ReplicaN = DefaultReplicaN
-	c.Cluster.Type = DefaultClusterType
 	c.Cluster.Hosts = []string{}
 	c.Cluster.LongQueryTime = Duration(time.Minute)
 
@@ -222,38 +219,12 @@ func NewConfig() *Config {
 
 // Validate that all configuration permutations are compatible with each other.
 func (c *Config) Validate() error {
-	if !StringInSlice(c.Cluster.Type, ClusterTypes) {
-		return ErrConfigClusterTypeInvalid
+	if !c.Cluster.Disabled && len(c.Cluster.Hosts) > 0 {
+		return ErrConfigClusterEnabledHosts
 	}
-
-	if c.Cluster.Type == ClusterGossip {
-		if len(c.Cluster.Hosts) > 0 {
-			bindWithDefaults, err := AddressWithDefaults(c.Bind)
-			if err != nil {
-				return err
-			}
-			if !c.foundHost(bindWithDefaults) {
-				return ErrConfigHostsMissing
-			}
-		}
-	}
-
 	return nil
 }
 
-func (c *Config) foundHost(host *URI) bool {
-	for _, clusterHost := range c.Cluster.Hosts {
-		uri, err := NewURIFromAddress(clusterHost)
-		if err != nil {
-			continue
-		}
-		if host.Equals(uri) {
-			return true
-		}
-	}
-	return false
-}
-
 // Duration is a TOML wrapper type for time.Duration.
 type Duration time.Duration
 

config_test.go

@@ -11,27 +11,15 @@ import (
 func Test_NewConfig(t *testing.T) {
 	c := pilosa.NewConfig()
 
+	if c.Cluster.Disabled != pilosa.DefaultClusterDisabled {
+		t.Fatalf("unexpected Cluster.Disabled: %v", c.Cluster.Disabled)
+	}
+
+	// Ensure that hosts can't be specificed on a non-disabled cluster.
 	c.Cluster.Hosts = []string{c.Bind, "localhost:10102"}
 
 	// Change cluster type from the default (gossip) to an invalid string.
-	c.Cluster.Type = "invalid-type"
-	if err := c.Validate(); err != pilosa.ErrConfigClusterTypeInvalid {
-		t.Fatal(err)
-	}
-
-	// Change cluster type back to gossip.
-	c.Cluster.Type = pilosa.ClusterGossip
-
-	// Check for bind address in cluster hosts.
-	c.Bind = "localhost:1"
-	if err := c.Validate(); err != pilosa.ErrConfigHostsMissing {
-		t.Fatal(err)
-	}
-
-	c.Bind = "localhost:10101"
-	c.Cluster.ReplicaN = 2
-	c.GossipSeed = "localhost:14000"
-	if err := c.Validate(); err != nil {
+	if err := c.Validate(); err != pilosa.ErrConfigClusterEnabledHosts {
 		t.Fatal(err)
 	}
 }

ctl/server.go

@@ -35,9 +35,9 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
 	SetTLSConfig(flags, &srv.Config.TLS.CertificatePath, &srv.Config.TLS.CertificateKeyPath, &srv.Config.TLS.SkipVerify)
 
 	// Cluster
+	flags.BoolVarP(&srv.Config.Cluster.Disabled, "cluster.disabled", "", srv.Config.Cluster.Disabled, "Disabled multi-node cluster communication (used for testing)")
 	flags.StringVarP(&srv.Config.Cluster.Coordinator, "cluster.coordinator", "", "", "Host that will act as cluster coordinator during startup and resizing.")
 	flags.IntVarP(&srv.Config.Cluster.ReplicaN, "cluster.replicas", "", 1, "Number of hosts each piece of data should be stored on.")
-	flags.StringVarP(&srv.Config.Cluster.Type, "cluster.type", "", "gossip", "Determine how the cluster handles membership and state sharing. Choose from [static, gossip]")
 	flags.StringSliceVarP(&srv.Config.Cluster.Hosts, "cluster.hosts", "", []string{}, "Comma separated list of hosts in cluster.")
 	flags.DurationVarP((*time.Duration)(&srv.Config.Cluster.LongQueryTime), "cluster.long-query-time", "", time.Minute, "Duration that will trigger log and stat messages for slow queries.")
 

pilosa.go

@@ -73,8 +73,7 @@ var (
 	ErrQueryRequired    = errors.New("query required")
 	ErrTooManyWrites    = errors.New("too many write commands")
 
-	ErrConfigClusterTypeInvalid = errors.New("invalid cluster type")
-	ErrConfigHostsMissing       = errors.New("missing bind address in cluster hosts")
+	ErrConfigClusterEnabledHosts = errors.New("providing hosts to a non-disabled cluster is not allowed")
 )
 
 // Regular expression to validate index and frame names.

server/cluster_test.go

@@ -217,7 +217,7 @@ func TestClusterResize_EmptyNode(t *testing.T) {
 // Ensure that a cluster of empty nodes comes up in a NORMAL state.
 func TestClusterResize_EmptyNodes(t *testing.T) {
 	// Configure node0
-	m0 := test.NewMain()
+	m0 := test.NewMainWithCluster()
 	defer m0.Close()
 
 	gossipHost := "localhost"
@@ -228,7 +228,7 @@ func TestClusterResize_EmptyNodes(t *testing.T) {
 	}
 
 	// Configure node1
-	m1 := test.NewMain()
+	m1 := test.NewMainWithCluster()
 	defer m1.Close()
 
 	seed, coord, err = m1.RunWithTransport(gossipHost, gossipPort, seed, &coord)
@@ -247,7 +247,7 @@ func TestClusterResize_EmptyNodes(t *testing.T) {
 func TestClusterResize_AddNode(t *testing.T) {
 	t.Run("NoData", func(t *testing.T) {
 		// Configure node0
-		m0 := test.NewMain()
+		m0 := test.NewMainWithCluster()
 		defer m0.Close()
 
 		seed, coord, err := m0.RunWithTransport("localhost", 0, "", nil)
@@ -256,7 +256,7 @@ func TestClusterResize_AddNode(t *testing.T) {
 		}
 
 		// Configure node1
-		m1 := test.NewMain()
+		m1 := test.NewMainWithCluster()
 		defer m1.Close()
 
 		var eg errgroup.Group
@@ -281,7 +281,7 @@ func TestClusterResize_AddNode(t *testing.T) {
 	})
 	t.Run("WithIndex", func(t *testing.T) {
 		// Configure node0
-		m0 := test.NewMain()
+		m0 := test.NewMainWithCluster()
 		defer m0.Close()
 
 		seed, coord, err := m0.RunWithTransport("localhost", 0, "", nil)
@@ -300,7 +300,7 @@ func TestClusterResize_AddNode(t *testing.T) {
 		}
 
 		// Configure node1
-		m1 := test.NewMain()
+		m1 := test.NewMainWithCluster()
 		defer m1.Close()
 
 		var eg errgroup.Group
@@ -327,7 +327,7 @@ func TestClusterResize_AddNode(t *testing.T) {
 	t.Run("ContinuousSlices", func(t *testing.T) {
 
 		// Configure node0
-		m0 := test.NewMain()
+		m0 := test.NewMainWithCluster()
 		defer m0.Close()
 
 		seed, coord, err := m0.RunWithTransport("localhost", 0, "", nil)
@@ -355,7 +355,7 @@ func TestClusterResize_AddNode(t *testing.T) {
 		}
 
 		// Configure node1
-		m1 := test.NewMain()
+		m1 := test.NewMainWithCluster()
 		defer m1.Close()
 
 		var eg errgroup.Group
@@ -382,7 +382,7 @@ func TestClusterResize_AddNode(t *testing.T) {
 	t.Run("SkippedSlice", func(t *testing.T) {
 
 		// Configure node0
-		m0 := test.NewMain()
+		m0 := test.NewMainWithCluster()
 		defer m0.Close()
 
 		seed, coord, err := m0.RunWithTransport("localhost", 0, "", nil)
@@ -410,7 +410,7 @@ func TestClusterResize_AddNode(t *testing.T) {
 		}
 
 		// Configure node1
-		m1 := test.NewMain()
+		m1 := test.NewMainWithCluster()
 		defer m1.Close()
 
 		var eg errgroup.Group

server/server.go

@@ -211,46 +211,7 @@ func (m *Command) SetupServer() error {
 
 // SetupNetworking sets up internode communication based on the configuration.
 func (m *Command) SetupNetworking() error {
-	switch m.Config.Cluster.Type {
-	case pilosa.ClusterGossip:
-		// Set internal port (string).
-		gossipPortStr := pilosa.DefaultGossipPort
-		// Config.GossipPort is deprecated, so Config.Gossip.Port has priority
-		if m.Config.Gossip.Port != "" {
-			gossipPortStr = m.Config.Gossip.Port
-		} else if m.Config.GossipPort != "" {
-			gossipPortStr = m.Config.GossipPort
-		}
-
-		gossipPort, err := strconv.Atoi(gossipPortStr)
-		if err != nil {
-			return err
-		}
-
-		// get the host portion of addr to use for binding
-		gossipHost := m.Server.URI.Host()
-		var transport *gossip.Transport
-		if m.GossipTransport != nil {
-			transport = m.GossipTransport
-		} else {
-			transport, err = gossip.NewTransport(gossipHost, gossipPort)
-			if err != nil {
-				return err
-			}
-		}
-
-		m.Server.NodeID = m.Server.LoadNodeID()
-
-		m.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver()
-		gossipMemberSet, err := gossip.NewGossipMemberSetWithTransport(m.Server.NodeID, m.Config, transport, m.Server)
-		if err != nil {
-			return err
-		}
-		m.Server.Cluster.MemberSet = gossipMemberSet
-		m.Server.Broadcaster = m.Server
-		m.Server.BroadcastReceiver = gossipMemberSet
-		m.Server.Gossiper = gossipMemberSet
-	case pilosa.ClusterStatic, pilosa.ClusterNone:
+	if m.Config.Cluster.Disabled {
 		m.Server.Cluster.Static = true
 		for _, address := range m.Config.Cluster.Hosts {
 			uri, err := pilosa.NewURIFromAddress(address)
@@ -270,9 +231,46 @@ func (m *Command) SetupNetworking() error {
 		if err != nil {
 			return err
 		}
-	default:
-		return fmt.Errorf("'%v' is not a supported value for broadcaster type", m.Config.Cluster.Type)
+		return nil
 	}
+
+	// Set internal port (string).
+	gossipPortStr := pilosa.DefaultGossipPort
+	// Config.GossipPort is deprecated, so Config.Gossip.Port has priority
+	if m.Config.Gossip.Port != "" {
+		gossipPortStr = m.Config.Gossip.Port
+	} else if m.Config.GossipPort != "" {
+		gossipPortStr = m.Config.GossipPort
+	}
+
+	gossipPort, err := strconv.Atoi(gossipPortStr)
+	if err != nil {
+		return err
+	}
+
+	// get the host portion of addr to use for binding
+	gossipHost := m.Server.URI.Host()
+	var transport *gossip.Transport
+	if m.GossipTransport != nil {
+		transport = m.GossipTransport
+	} else {
+		transport, err = gossip.NewTransport(gossipHost, gossipPort)
+		if err != nil {
+			return err
+		}
+	}
+
+	m.Server.NodeID = m.Server.LoadNodeID()
+
+	m.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver()
+	gossipMemberSet, err := gossip.NewGossipMemberSetWithTransport(m.Server.NodeID, m.Config, transport, m.Server)
+	if err != nil {
+		return err
+	}
+	m.Server.Cluster.MemberSet = gossipMemberSet
+	m.Server.Broadcaster = m.Server
+	m.Server.BroadcastReceiver = gossipMemberSet
+	m.Server.Gossiper = gossipMemberSet
 	return nil
 }
 

server/server_test.go

@@ -269,7 +269,7 @@ func TestMain_SetColumnAttrsWithColumnOption(t *testing.T) {
 
 // Ensure program can set bits on one cluster and then restore to a second cluster.
 func TestMain_FrameRestore(t *testing.T) {
-	mains1 := test.NewMainArrayWithCluster(2)
+	mains1 := test.MustRunMainWithCluster(t, 2)
 	m10 := mains1[0]
 	m11 := mains1[1]
 
@@ -303,7 +303,7 @@ func TestMain_FrameRestore(t *testing.T) {
 	}
 
 	// Start second cluster.
-	mains2 := test.NewMainArrayWithCluster(2)
+	mains2 := test.MustRunMainWithCluster(t, 2)
 	m20 := mains2[0]
 	defer m20.Close()
 	m21 := mains2[1]

test/pilosa.go

@@ -37,7 +37,7 @@ func NewMain() *Main {
 	m.Server.Network = *Network
 	m.Config.DataDir = path
 	m.Config.Bind = "localhost:0"
-	m.Config.Cluster.Type = "static"
+	m.Config.Cluster.Disabled = true
 	m.Command.Stdin = &m.Stdin
 	m.Command.Stdout = &m.Stdout
 	m.Command.Stderr = &m.Stderr
@@ -50,16 +50,50 @@ func NewMain() *Main {
 	return m
 }
 
-func NewMainArrayWithCluster(size int) []*Main {
-	cluster, err := NewServerCluster(size)
+// NewMainWithCluster returns a new instance of Main with clustering enabled.
+func NewMainWithCluster() *Main {
+	m := NewMain()
+	m.Config.Cluster.Disabled = false
+	return m
+}
+
+// MustRunMainWithCluster ruturns a running array of *Main where
+// all nodes are joined via memberlist (i.e. clustering enabled).
+func MustRunMainWithCluster(t *testing.T, size int) []*Main {
+	ma, err := runMainWithCluster(size)
 	if err != nil {
-		panic(err)
+		t.Fatalf("new main array with cluster: %v", err)
 	}
-	mainArray := make([]*Main, size)
+	return ma
+}
+
+// runMainWithCluster runs an array of *Main where all nodes are
+// joined via memberlist (i.e. clustering enabled).
+func runMainWithCluster(size int) ([]*Main, error) {
+	if size == 0 {
+		return nil, errors.New("cluster must contain at least one node")
+	}
+
+	mains := make([]*Main, size)
+
+	gossipHost := "localhost"
+	gossipPort := 0
+	var err error
+	var gossipSeed string
+	var coordinator pilosa.URI
+
 	for i := 0; i < size; i++ {
-		mainArray[i] = cluster.Servers[i]
+		m := NewMainWithCluster()
+
+		gossipSeed, coordinator, err = m.RunWithTransport(gossipHost, gossipPort, gossipSeed, &coordinator)
+		if err != nil {
+			return nil, errors.Wrap(err, "RunWithTransport")
+		}
+
+		mains[i] = m
 	}
-	return mainArray
+
+	return mains, nil
 }
 
 // MustRunMain returns a new, running Main. Panic on error.
@@ -101,8 +135,6 @@ func (m *Main) Reopen() error {
 func (m *Main) RunWithTransport(host string, bindPort int, joinSeed string, coordinator *pilosa.URI) (seed string, coord pilosa.URI, err error) {
 	defer close(m.Started)
 
-	m.Config.Cluster.Type = "gossip"
-
 	/*
 	   TEST:
 	   - SetupServer (just static settings from config)
@@ -202,46 +234,6 @@ func (m *Main) CreateDefinition(index, def, query string) (string, error) {
 
 ////////////////////////////////////////////////////////////////////////////////////
 
-type Cluster struct {
-	Servers []*Main
-}
-
-func MustNewServerCluster(t *testing.T, size int) *Cluster {
-	cluster, err := NewServerCluster(size)
-	if err != nil {
-		t.Fatalf("new cluster: %v", err)
-	}
-	return cluster
-}
-
-func NewServerCluster(size int) (cluster *Cluster, err error) {
-	if size == 0 {
-		return nil, errors.New("cluster must contain at least one node")
-	}
-
-	cluster = &Cluster{
-		Servers: make([]*Main, size),
-	}
-
-	gossipHost := "localhost"
-	gossipPort := 0
-	var gossipSeed string
-	var coordinator pilosa.URI
-
-	for i := 0; i < size; i++ {
-		m := NewMain()
-
-		gossipSeed, coordinator, err = m.RunWithTransport(gossipHost, gossipPort, gossipSeed, &coordinator)
-		if err != nil {
-			return nil, errors.Wrap(err, "RunWithTransport")
-		}
-
-		cluster.Servers[i] = m
-	}
-
-	return cluster, nil
-}
-
 // MustDo executes http.Do() with an http.NewRequest(). Panic on error.
 func MustDo(method, urlStr string, body string) *httpResponse {
 	req, err := http.NewRequest(method, urlStr, strings.NewReader(body))

test/pilosa_test.go

@@ -10,9 +10,9 @@ import (
 )
 
 func TestNewCluster(t *testing.T) {
-	cluster := test.MustNewServerCluster(t, 3)
+	cluster := test.MustRunMainWithCluster(t, 3)
 
-	response, err := http.Get("http://" + cluster.Servers[0].Server.Addr().String() + "/status")
+	response, err := http.Get("http://" + cluster[0].Server.Addr().String() + "/status")
 	if err != nil {
 		t.Fatalf("getting schema: %v", err)
 	}