diff --git a/cmd/server_test.go b/cmd/server_test.go index c9b900bde..b1f39e921 100644 --- a/cmd/server_test.go +++ b/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" diff --git a/config.go b/config.go index a43963f93..05f132db2 100644 --- a/config.go +++ b/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 diff --git a/config_test.go b/config_test.go index c409fd80f..38a3fc640 100644 --- a/config_test.go +++ b/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) } } diff --git a/ctl/server.go b/ctl/server.go index 1dddeb0d7..3c5b1a058 100644 --- a/ctl/server.go +++ b/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.") diff --git a/pilosa.go b/pilosa.go index 6e7777804..98c67203f 100644 --- a/pilosa.go +++ b/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. diff --git a/server/cluster_test.go b/server/cluster_test.go index f8789f169..114a25b0c 100644 --- a/server/cluster_test.go +++ b/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 diff --git a/server/server.go b/server/server.go index 8379ef581..ef07ee381 100644 --- a/server/server.go +++ b/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 } diff --git a/server/server_test.go b/server/server_test.go index 352fc6121..c821ccdf8 100644 --- a/server/server_test.go +++ b/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] diff --git a/test/pilosa.go b/test/pilosa.go index cbe259040..a467a2f5c 100644 --- a/test/pilosa.go +++ b/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)) diff --git a/test/pilosa_test.go b/test/pilosa_test.go index 8478e6bba..0d0f168d5 100644 --- a/test/pilosa_test.go +++ b/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) }