From a3b1b2ee1a475545d8a0d45ddb4e59034839744e Mon Sep 17 00:00:00 2001 From: Travis Date: Wed, 26 Jul 2017 09:16:39 -0500 Subject: [PATCH] DefaultClusterType = ClusterGossip This commit removes the `httpbroadcast` NodeSet, sets the default cluster type to `gossip`, and uses the `static` cluster type for most test cases. --- cluster.go | 3 +- cluster_test.go | 5 +- cmd/root_test.go | 2 +- cmd/server_test.go | 14 +-- config.go | 44 ++------ config_test.go | 29 ++---- ctl/server.go | 7 +- ctl/server_test.go | 8 +- docs/api-reference.md | 2 +- docs/configuration.md | 20 +--- gossip/gossip.go | 28 +++++- handler_test.go | 2 +- httpbroadcast/messenger.go | 201 ------------------------------------- pilosa.go | 4 - server/server.go | 29 ++---- server/server_test.go | 94 ++++++++++++----- 16 files changed, 138 insertions(+), 354 deletions(-) delete mode 100644 httpbroadcast/messenger.go diff --git a/cluster.go b/cluster.go index 981e8bdc2..6b5dddd38 100644 --- a/cluster.go +++ b/cluster.go @@ -38,8 +38,7 @@ const ( // Node represents a node in the cluster. type Node struct { - Host string `json:"host"` - InternalHost string `json:"internalHost"` + Host string `json:"host"` status *internal.NodeStatus `json:"status"` } diff --git a/cluster_test.go b/cluster_test.go index d6b9dd487..8ab1ab461 100644 --- a/cluster_test.go +++ b/cluster_test.go @@ -22,7 +22,6 @@ import ( "github.com/davecgh/go-spew/spew" "github.com/pilosa/pilosa" - "github.com/pilosa/pilosa/httpbroadcast" "github.com/pilosa/pilosa/test" ) @@ -110,10 +109,10 @@ func TestCluster_NodeStates(t *testing.T) { {Host: "serverB:1000"}, {Host: "serverC:1000"}, }, - NodeSet: &httpbroadcast.HTTPNodeSet{}, + NodeSet: &pilosa.StaticNodeSet{}, } - err := c.NodeSet.(*httpbroadcast.HTTPNodeSet).Join([]*pilosa.Node{ + err := c.NodeSet.(*pilosa.StaticNodeSet).Join([]*pilosa.Node{ &pilosa.Node{Host: "serverA:1000"}, &pilosa.Node{Host: "serverC:1000"}, &pilosa.Node{Host: "serverD:1000"}, diff --git a/cmd/root_test.go b/cmd/root_test.go index e1aaaf31d..93711a5a1 100644 --- a/cmd/root_test.go +++ b/cmd/root_test.go @@ -96,7 +96,7 @@ func (v *validator) Check(actual, expected interface{}) { // Error returns the validator's error value if any v.Check call found an error. func (v *validator) Error() error { return v.err } -// commandTest represents all possible ways to configure a a pilosa command, as +// commandTest represents all possible ways to configure a pilosa command, as // well as a function for validating whether the command worked as expected. // args should be set to everything that comes after "pilosa" on the comand // line. See tests like backup_test.go for examples. diff --git a/cmd/server_test.go b/cmd/server_test.go index 25d0d7fd1..88daf251c 100644 --- a/cmd/server_test.go +++ b/cmd/server_test.go @@ -52,15 +52,11 @@ func TestServerConfig(t *testing.T) { [cluster] poll-interval = "45s" - type = "http" + type = "static" replicas = 2 hosts = [ "localhost:19444", ] - internal-hosts = [ - "localhost:19500", - "localhost:19501", - ] `, validation: func() error { v := validator{} @@ -80,14 +76,10 @@ func TestServerConfig(t *testing.T) { bind = "localhost:0" data-dir = "` + actualDataDir + `" [cluster] - type = "http" + type = "static" hosts = [ "localhost:19444", ] - internal-hosts = [ - "localhost:19500", - "localhost:19501", - ] [plugins] path = "/var/sloth" `, @@ -101,7 +93,7 @@ func TestServerConfig(t *testing.T) { }, // TEST 2 { - args: []string{"server", "--log-path", logFile.Name()}, + args: []string{"server", "--log-path", logFile.Name(), "--cluster.type", "static"}, env: map[string]string{"PILOSA_PROFILE.CPU_TIME": "1m"}, cfgFileContent: ` bind = "localhost:19444" diff --git a/config.go b/config.go index 3b248878a..7c4956fce 100644 --- a/config.go +++ b/config.go @@ -20,7 +20,6 @@ import "time" const ( ClusterNone = "" ClusterStatic = "static" - ClusterHTTP = "http" ClusterGossip = "gossip" ) @@ -32,12 +31,12 @@ const ( DefaultPort = "10101" // DefaultClusterType sets the node intercommunication method. - DefaultClusterType = ClusterStatic + DefaultClusterType = ClusterGossip - // DefaultInternalPort the port the nodes intercommunicate on. - DefaultInternalPort = "14000" + // DefaultGossipPort indicates the port to which pilosa should bind for internal state sharing. + DefaultGossipPort = "14000" - // DefaultMetrics sets the internal metrics to no op + // DefaultMetrics sets the internal metrics to no-op. DefaultMetrics = "nop" // DefaultMaxWritesPerRequest is the default number of writes per request. @@ -45,21 +44,20 @@ const ( ) // ClusterTypes set of cluster types. -var ClusterTypes = []string{ClusterNone, ClusterStatic, ClusterHTTP, ClusterGossip} +var ClusterTypes = []string{ClusterNone, ClusterStatic, ClusterGossip} // Config represents the configuration for the command. type Config struct { - DataDir string `toml:"data-dir"` - Bind string `toml:"bind"` - InternalPort string `toml:"internal-port"` + DataDir string `toml:"data-dir"` + Bind string `toml:"bind"` + GossipPort string `toml:"gossip-port"` + GossipSeed string `toml:"gossip-seed"` Cluster struct { ReplicaN int `toml:"replicas"` Type string `toml:"type"` Hosts []string `toml:"hosts"` - InternalHosts []string `toml:"internal-hosts"` PollInterval Duration `toml:"poll-interval"` - GossipSeed string `toml:"gossip-seed"` LongQueryTime Duration `toml:"long-query-time"` } `toml:"cluster"` @@ -94,7 +92,6 @@ func NewConfig() *Config { c.Cluster.Type = DefaultClusterType c.Cluster.PollInterval = Duration(DefaultPollingInterval) c.Cluster.Hosts = []string{} - c.Cluster.InternalHosts = []string{} c.AntiEntropy.Interval = Duration(DefaultAntiEntropyInterval) c.Metric.Service = DefaultMetrics return c @@ -105,31 +102,12 @@ 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 c.Cluster.Type == ClusterGossip { if !foundItem(c.Cluster.Hosts, c.Bind) { return ErrConfigHostsMissing } } - if c.Cluster.Type == ClusterHTTP { - if len(c.Cluster.Hosts) != len(c.Cluster.InternalHosts) { - return ErrConfigHostsMismatch - } - // TODO: this seems like an odd check; it's just ensuring that InternalPort - // matches any one substring from any of the InternalHosts. - // I suggest we either remove this completely or make it actually check - // the port portion of the address for this node. (note that this only applies - // to the http broadcaster, so if we simply use gossip for all implementations - // we can remove this). - if !ContainsSubstring(c.InternalPort, c.Cluster.InternalHosts) { - return ErrConfigBroadcastPort - } - } return nil } diff --git a/config_test.go b/config_test.go index 10febfa05..c409fd80f 100644 --- a/config_test.go +++ b/config_test.go @@ -12,42 +12,25 @@ 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" + // 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) } - c.Cluster.Type = pilosa.ClusterHTTP - if err := c.Validate(); err != pilosa.ErrConfigHostsMismatch { - t.Fatal(err) - } + // Change cluster type back to gossip. + c.Cluster.Type = pilosa.ClusterGossip - c.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"} + // Check for bind address in cluster hosts. 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:14000" + c.GossipSeed = "localhost:14000" if err := c.Validate(); err != nil { t.Fatal(err) } diff --git a/ctl/server.go b/ctl/server.go index 70f43bc41..e3a2c4156 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -26,11 +26,11 @@ 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.Bind, "bind", "b", ":10101", "Default URI on which pilosa should listen.") - flags.StringVarP(&srv.Config.InternalPort, "internal-port", "", "", "Port to which pilosa should bind for internal state sharing.") + flags.StringVarP(&srv.Config.GossipPort, "gossip-port", "", "", "Port to which pilosa should bind for internal state sharing.") + flags.StringVarP(&srv.Config.GossipSeed, "gossip-seed", "", "", "Host with which to seed the gossip membership.") 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.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.") @@ -38,8 +38,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags.DurationVarP((*time.Duration)(&srv.Config.AntiEntropy.Interval), "anti-entropy.interval", "", time.Minute*10, "Interval at which to run anti-entropy routine.") flags.StringVarP(&srv.CPUProfile, "profile.cpu", "", "", "Where to store CPU profile.") flags.DurationVarP(&srv.CPUTime, "profile.cpu-time", "", 30*time.Second, "CPU profile duration.") - flags.StringVarP(&srv.Config.Cluster.Type, "cluster.type", "", "static", "Determine how the cluster handles membership and state sharing. Choose from [static, http, gossip]") - flags.StringVarP(&srv.Config.Cluster.GossipSeed, "cluster.gossip-seed", "", "", "Host with which to seed the gossip membership.") + flags.StringVarP(&srv.Config.Cluster.Type, "cluster.type", "", "gossip", "Determine how the cluster handles membership and state sharing. Choose from [static, gossip]") 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.PollInterval), "metric.poll-interval", "", time.Minute*0, "Polling interval metrics.") diff --git a/ctl/server_test.go b/ctl/server_test.go index 16dd26916..dc31426ac 100644 --- a/ctl/server_test.go +++ b/ctl/server_test.go @@ -28,13 +28,13 @@ func TestBuildServerFlags(t *testing.T) { stdin, stdout, stderr := GetIO(buf) Server := server.NewCommand(stdin, stdout, stderr) BuildServerFlags(cm, Server) - if cm.Flags().Lookup("internal-port").Name == "" { - t.Fatal("internal-port flag is missed ") + if cm.Flags().Lookup("gossip-port").Name == "" { + t.Fatal("gossip-port flag is required") } if cm.Flags().Lookup("data-dir").Name == "" { - t.Fatal("data-dir flag is missed ") + t.Fatal("data-dir flag is required") } if cm.Flags().Lookup("log-path").Name == "" { - t.Fatal("log-path is missed ") + t.Fatal("log-path flag is required") } } diff --git a/docs/api-reference.md b/docs/api-reference.md index 8a9eb7c46..0d7006c9a 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -352,7 +352,7 @@ curl -XGET localhost:10101/hosts Response: ``` -[{"host":":10101","internalHost":""}] +[{"host":":10101"}] ``` ### Get version diff --git a/docs/configuration.md b/docs/configuration.md index b87d546fe..08ff7c4a3 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -58,15 +58,15 @@ Any flag that has a value that is a comma separated list on the command line bec bind = localhost:10101 ``` -#### Internal Port +#### Gossip Port * Description: Port to which Pilosa should bind for internal communication. -* Flag: `--internal-port=11101` -* Env: `PILOSA_INTERNAL_PORT=11101` +* Flag: `--gossip-port=11101` +* Env: `PILOSA_GOSSIP_PORT=11101` * Config: ```toml - internal-port = 11101 + gossip-port = 11101 ``` #### Cluster Hosts @@ -81,18 +81,6 @@ Any flag that has a value that is a comma separated list on the command line bec hosts = ["localhost:10101"] ``` -#### Cluster Internal Hosts - -* Description: List of hosts in the cluster used for internal communication. Multiple hosts should be comma separated in the flag and env forms. -* Flag: `--cluster.internal-hosts="localhost:11101"` -* Env: `PILOSA_CLUSTER.INTERNAL_HOSTS="localhost:11101"` -* Config: - - ```toml - [cluster] - internal-hosts = ["localhost:11101"] - ``` - #### Cluster Poll Interval * Description: Polling interval for cluster. diff --git a/gossip/gossip.go b/gossip/gossip.go index 92ec94c97..95282a658 100644 --- a/gossip/gossip.go +++ b/gossip/gossip.go @@ -18,6 +18,7 @@ import ( "fmt" "io" "log" + "time" "golang.org/x/sync/errgroup" @@ -71,7 +72,7 @@ func (g *GossipNodeSet) Open() error { // attach to gossip seed node nodes := []*pilosa.Node{&pilosa.Node{Host: g.config.gossipSeed}} //TODO: support a list of seeds - _, err = g.memberlist.Join(pilosa.Nodes(nodes).Hosts()) + err = g.joinWithRetry(pilosa.Nodes(nodes).Hosts()) if err != nil { return err } @@ -84,6 +85,31 @@ func (g *GossipNodeSet) Open() error { return nil } +// joinWithRetry wraps the standard memberlist Join function in a retry. +func (g *GossipNodeSet) joinWithRetry(hosts []string) error { + err := retry(60, 2*time.Second, func() error { + _, err := g.memberlist.Join(hosts) + return err + }) + return err +} + +// retry periodically retries function fn a specified number of attempts. +func retry(attempts int, sleep time.Duration, fn func() error) (err error) { + for i := 0; ; i++ { + err = fn() + if err == nil { + return + } + if i >= (attempts - 1) { + break + } + time.Sleep(sleep) + log.Println("retrying after error:", err) + } + return fmt.Errorf("after %d attempts, last error: %s", attempts, err) +} + // logger returns a logger for the GossipNodeSet. func (g *GossipNodeSet) logger() *log.Logger { return log.New(g.LogOutput, "", log.LstdFlags) diff --git a/handler_test.go b/handler_test.go index 6e13bf6ba..22a1fd4f3 100644 --- a/handler_test.go +++ b/handler_test.go @@ -942,7 +942,7 @@ func TestHandler_Fragment_Nodes(t *testing.T) { h.ServeHTTP(w, r) if w.Code != http.StatusOK { t.Fatalf("unexpected status code: %d", w.Code) - } else if w.Body.String() != `[{"host":"host2","internalHost":""},{"host":"host0","internalHost":""}]`+"\n" { + } else if w.Body.String() != `[{"host":"host2"},{"host":"host0"}]`+"\n" { t.Fatalf("unexpected body: %q", w.Body.String()) } } diff --git a/httpbroadcast/messenger.go b/httpbroadcast/messenger.go deleted file mode 100644 index 43a7e3ec7..000000000 --- a/httpbroadcast/messenger.go +++ /dev/null @@ -1,201 +0,0 @@ -// Copyright 2017 Pilosa Corp. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package httpbroadcast - -import ( - "bytes" - "errors" - "fmt" - "io" - "io/ioutil" - "net/http" - "net/url" - - "golang.org/x/sync/errgroup" - - "github.com/gogo/protobuf/proto" - "github.com/pilosa/pilosa" -) - -// HTTPBroadcaster represents a NodeSet that broadcasts messages over HTTP. -type HTTPBroadcaster struct { - server *pilosa.Server - internalPort string -} - -// NewHTTPBroadcaster returns a new instance of HTTPBroadcaster. -func NewHTTPBroadcaster(s *pilosa.Server, internalPort string) *HTTPBroadcaster { - return &HTTPBroadcaster{server: s, internalPort: internalPort} -} - -// SendSync sends a protobuf message to all nodes simultaneously. -// It waits for all nodes to respond before the function returns (and returns any errors). -func (h *HTTPBroadcaster) SendSync(pb proto.Message) error { - // Marshal the pb to []byte - buf, err := pilosa.MarshalMessage(pb) - if err != nil { - return err - } - - nodes, err := h.nodes() - if err != nil { - return err - } - - var g errgroup.Group - for _, n := range nodes { - // Don't send the message to the local node. - if n.Host == h.server.Host { - continue - } - node := n - g.Go(func() error { - return h.sendNodeMessage(node, buf) - }) - } - return g.Wait() -} - -// SendAsync exists to implement the Broadcaster interface, but just calls -// SendSync. -func (h *HTTPBroadcaster) SendAsync(pb proto.Message) error { - return h.SendSync(pb) -} - -func (h *HTTPBroadcaster) nodes() ([]*pilosa.Node, error) { - if h.server == nil { - return nil, errors.New("HTTPBroadcaster has no reference to Server") - } - nodeset, ok := h.server.Cluster.NodeSet.(*HTTPNodeSet) - if !ok { - return nil, errors.New("NodeSet cannot be caste to HTTPNodeSet") - } - return nodeset.Nodes(), nil -} - -func (h *HTTPBroadcaster) sendNodeMessage(node *pilosa.Node, msg []byte) error { - var client *http.Client - client = http.DefaultClient - - // Create HTTP request. - req, err := http.NewRequest("POST", (&url.URL{ - Scheme: "http", - Host: node.InternalHost, - }).String(), bytes.NewReader(msg)) - - // Require protobuf encoding. - req.Header.Set("Content-Type", "application/x-protobuf") - req.Header.Set("User-Agent", "pilosa/"+pilosa.Version) - - // Send request to remote node. - resp, err := client.Do(req) - if err != nil { - return err - } - defer resp.Body.Close() - - // Read response into buffer. - body, err := ioutil.ReadAll(resp.Body) - - if err != nil { - return err - } - - // Check status code. - if resp.StatusCode != http.StatusOK { - return fmt.Errorf("invalid status sendNodeMessage: code=%d, err=%s, req=%v", resp.StatusCode, body, req) - } - - return nil -} - -// HTTPBroadcastReceiver unmarshals incoming messages over HTTP and passes them on to the handler. -type HTTPBroadcastReceiver struct { - port string - handler pilosa.BroadcastHandler - logOutput io.Writer -} - -// NewHTTPBroadcastReceiver returns a new instance of HTTPBroadcastReceiver. -func NewHTTPBroadcastReceiver(port string, logOutput io.Writer) *HTTPBroadcastReceiver { - return &HTTPBroadcastReceiver{ - port: port, - logOutput: logOutput, - } -} - -// Start implements the BroadcastReceiver interface and starts listening for broadcast messages. -func (rec *HTTPBroadcastReceiver) Start(b pilosa.BroadcastHandler) error { - rec.handler = b - go func() { - err := http.ListenAndServe(":"+rec.port, rec) - if err != nil { - fmt.Fprintf(rec.logOutput, "Error listening on %v for HTTPBroadcastReceiver: %v\n", ":"+rec.port, err) - } - }() - return nil -} - -func (rec *HTTPBroadcastReceiver) ServeHTTP(w http.ResponseWriter, r *http.Request) { - if r.Header.Get("Content-Type") != "application/x-protobuf" { - http.Error(w, "Unsupported media type", http.StatusUnsupportedMediaType) - return - } - - // Read entire body. - body, err := ioutil.ReadAll(r.Body) - if err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return - } - - // Unmarshal message to specific proto type. - m, err := pilosa.UnmarshalMessage(body) - if err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return - } - - if err := rec.handler.ReceiveMessage(m); err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return - } -} - -// HTTPNodeSet represents a NodeSet that broadcasts messages over HTTP. -type HTTPNodeSet struct { - nodes []*pilosa.Node -} - -// NewHTTPNodeSet returns a new instance of HTTPNodeSet. -func NewHTTPNodeSet() *HTTPNodeSet { - return &HTTPNodeSet{} -} - -// Nodes implements the NodeSet interface and returns a list of nodes in the cluster. -func (h *HTTPNodeSet) Nodes() []*pilosa.Node { - return h.nodes -} - -// Open implements the NodeSet interface to start network activity, but for a HTTPNodeSet it does nothing. -func (h *HTTPNodeSet) Open() error { - return nil -} - -// Join sets the NodeSet nodes to the slice of Nodes passed in. -func (h *HTTPNodeSet) Join(nodes []*pilosa.Node) error { - h.nodes = nodes - return nil -} diff --git a/pilosa.go b/pilosa.go index faba8d55a..af4c37ba9 100644 --- a/pilosa.go +++ b/pilosa.go @@ -71,11 +71,7 @@ var ( 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") ) // Regular expression to validate index and frame names. diff --git a/server/server.go b/server/server.go index 1553917bb..9dc80eff2 100644 --- a/server/server.go +++ b/server/server.go @@ -32,7 +32,6 @@ import ( "github.com/pilosa/pilosa" "github.com/pilosa/pilosa/gossip" - "github.com/pilosa/pilosa/httpbroadcast" "github.com/pilosa/pilosa/statsd" ) @@ -105,7 +104,7 @@ func (m *Command) Run(args ...string) (err error) { return nil } -// SetupServer use the cluster configuration to setup this server +// SetupServer uses the cluster configuration to set up this server. func (m *Command) SetupServer() error { err := m.Config.Validate() if err != nil { @@ -118,12 +117,6 @@ func (m *Command) SetupServer() error { for _, hostport := range m.Config.Cluster.Hosts { cluster.Nodes = append(cluster.Nodes, &pilosa.Node{Host: hostport}) } - // TODO: if InternalHosts is not provided then pilosa.Node.InternalHost is empty. - // This will throw an error when trying to Broadcast messages over HTTP. - // One option may be to fall back to using host from hostport + config.InternalPort. - for i, internalhostport := range m.Config.Cluster.InternalHosts { - cluster.Nodes[i].InternalHost = internalhostport - } m.Server.Cluster = cluster // Setup logging output. @@ -152,28 +145,20 @@ func (m *Command) SetupServer() error { } // Set internal port (string). - internalPortStr := pilosa.DefaultInternalPort - if m.Config.InternalPort != "" { - internalPortStr = m.Config.InternalPort + gossipPortStr := pilosa.DefaultGossipPort + if m.Config.GossipPort != "" { + gossipPortStr = m.Config.GossipPort } switch m.Config.Cluster.Type { - 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() - err := m.Server.Cluster.NodeSet.(*httpbroadcast.HTTPNodeSet).Join(m.Server.Cluster.Nodes) - if err != nil { - return err - } case pilosa.ClusterGossip: - gossipPort, err := strconv.Atoi(internalPortStr) + gossipPort, err := strconv.Atoi(gossipPortStr) if err != nil { return err } gossipSeed := pilosa.DefaultHost - if m.Config.Cluster.GossipSeed != "" { - gossipSeed = m.Config.Cluster.GossipSeed + if m.Config.GossipSeed != "" { + gossipSeed = m.Config.GossipSeed } // get the host portion of addr to use for binding gossipHost, _, err := net.SplitHostPort(m.Config.Bind) diff --git a/server/server_test.go b/server/server_test.go index 37b6f2f38..3d48cf752 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -33,10 +33,11 @@ import ( "strings" "testing" "testing/quick" + "time" "github.com/BurntSushi/toml" "github.com/pilosa/pilosa" - "github.com/pilosa/pilosa/httpbroadcast" + "github.com/pilosa/pilosa/gossip" "github.com/pilosa/pilosa/server" "github.com/pilosa/pilosa/test" ) @@ -426,38 +427,73 @@ func TestMain_SendReceiveMessage(t *testing.T) { // Update cluster config m0.Server.Cluster.Nodes = []*pilosa.Node{ - {Host: m0.Server.Host, InternalHost: "localhost:" + freePorts[0]}, - {Host: m1.Server.Host, InternalHost: "localhost:" + freePorts[1]}, + {Host: m0.Server.Host}, + {Host: m1.Server.Host}, } m1.Server.Cluster.Nodes = m0.Server.Cluster.Nodes // Configure node0 - m0.Server.Broadcaster = httpbroadcast.NewHTTPBroadcaster(m0.Server, freePorts[0]) - m0.Server.Handler.Broadcaster = m0.Server.Broadcaster - m0.Server.Holder.Broadcaster = m0.Server.Broadcaster - m0.Server.BroadcastReceiver = httpbroadcast.NewHTTPBroadcastReceiver(freePorts[0], nil) - m0.Server.Cluster.NodeSet = httpbroadcast.NewHTTPNodeSet() - err = m0.Server.Cluster.NodeSet.(*httpbroadcast.HTTPNodeSet).Join(m0.Server.Cluster.Nodes) + + // get the host portion of addr to use for binding + gossipHost, _, err := net.SplitHostPort(m0.Server.Host) + if err != nil { + gossipHost = m0.Server.Host + } + if gossipHost == "localhost" { + gossipHost = "127.0.0.1" + } + gossipPort, err := strconv.Atoi(freePorts[0]) if err != nil { t.Fatal(err) } + gossipSeed := gossipHost + ":" + freePorts[0] + + gossipNodeSet0 := gossip.NewGossipNodeSet(m0.Server.Host, gossipHost, gossipPort, gossipSeed, m0.Server) + m0.Server.Cluster.NodeSet = gossipNodeSet0 + m0.Server.Broadcaster = gossipNodeSet0 + m0.Server.Handler.Broadcaster = m0.Server.Broadcaster + m0.Server.Holder.Broadcaster = m0.Server.Broadcaster + m0.Server.BroadcastReceiver = gossipNodeSet0 + if err := m0.Server.BroadcastReceiver.Start(m0.Server); err != nil { t.Fatal(err) } + // Open NodeSet communication + if err := m0.Server.Cluster.NodeSet.Open(); err != nil { + t.Fatal(err) + } // Configure node1 - m1.Server.Broadcaster = httpbroadcast.NewHTTPBroadcaster(m1.Server, freePorts[1]) - m1.Server.Handler.Broadcaster = m1.Server.Broadcaster - m1.Server.Holder.Broadcaster = m1.Server.Broadcaster - m1.Server.BroadcastReceiver = httpbroadcast.NewHTTPBroadcastReceiver(freePorts[1], nil) - m1.Server.Cluster.NodeSet = httpbroadcast.NewHTTPNodeSet() - err = m1.Server.Cluster.NodeSet.(*httpbroadcast.HTTPNodeSet).Join(m1.Server.Cluster.Nodes) + + // get the host portion of addr to use for binding + gossipHost, _, err = net.SplitHostPort(m1.Server.Host) + if err != nil { + gossipHost = m1.Server.Host + } + if gossipHost == "localhost" { + gossipHost = "127.0.0.1" + } + gossipPort, err = strconv.Atoi(freePorts[1]) if err != nil { t.Fatal(err) } + + gossipNodeSet1 := gossip.NewGossipNodeSet(m1.Server.Host, gossipHost, gossipPort, gossipSeed, m1.Server) + m1.Server.Cluster.NodeSet = gossipNodeSet1 + m1.Server.Broadcaster = gossipNodeSet1 + m1.Server.Handler.Broadcaster = m1.Server.Broadcaster + m1.Server.Holder.Broadcaster = m1.Server.Broadcaster + m1.Server.BroadcastReceiver = gossipNodeSet1 + if err := m1.Server.BroadcastReceiver.Start(m1.Server); err != nil { t.Fatal(err) } + // Open NodeSet communication + if err := m1.Server.Cluster.NodeSet.Open(); err != nil { + t.Fatal(err) + } + + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// // Expected indexes and Frames expected := map[string][]string{ @@ -509,12 +545,15 @@ func TestMain_SendReceiveMessage(t *testing.T) { // Write data on first node. if _, err := m0.Query("i", "", ` - SetBit(rowID=1, frame="f", columnID=1) - SetBit(rowID=1, frame="f", columnID=2400000) - `); err != nil { + SetBit(rowID=1, frame="f", columnID=1) + SetBit(rowID=1, frame="f", columnID=2400000) + `); err != nil { t.Fatal(err) } + // We have to wait for the broadcast message to be sent before checking state. + time.Sleep(1 * time.Second) + // Make sure node0 knows about the latest MaxSlice. maxSlices0, err := client0.MaxSliceByIndex(context.Background()) if err != nil { @@ -535,15 +574,15 @@ func TestMain_SendReceiveMessage(t *testing.T) { // Write input definition to the first node. if _, err := m0.CreateDefinition("i", "test", `{ - "frames": [{"name": "event-time", - "options": { - "cacheType": "ranked", - "timeQuantum": "YMD" - }}], - "fields": [{"name": "columnID", - "primaryKey": true - }]} - `); err != nil { + "frames": [{"name": "event-time", + "options": { + "cacheType": "ranked", + "timeQuantum": "YMD" + }}], + "fields": [{"name": "columnID", + "primaryKey": true + }]} + `); err != nil { t.Fatal(err) } @@ -607,6 +646,7 @@ func NewMain() *Main { m.Server.Network = *test.Network m.Config.DataDir = path m.Config.Bind = "localhost:0" + m.Config.Cluster.Type = "static" m.Command.Stdin = &m.Stdin m.Command.Stdout = &m.Stdout m.Command.Stderr = &m.Stderr