diff --git a/cmd/server_test.go b/cmd/server_test.go index b650e316a..c9b900bde 100644 --- a/cmd/server_test.go +++ b/cmd/server_test.go @@ -45,7 +45,7 @@ func TestServerConfig(t *testing.T) { // TEST 0 { args: []string{"server", "--data-dir", actualDataDir, "--cluster.hosts", "localhost:10111,localhost:10110", "--bind", "localhost:10111"}, - env: map[string]string{"PILOSA_DATA_DIR": "/tmp/myEnvDatadir", "PILOSA_CLUSTER_POLL_INTERVAL": "3m2s", "PILOSA_CLUSTER_LONG_QUERY_TIME": "1m30s", "PILOSA_MAX_WRITES_PER_REQUEST": "2000"}, + env: map[string]string{"PILOSA_DATA_DIR": "/tmp/myEnvDatadir", "PILOSA_CLUSTER_LONG_QUERY_TIME": "1m30s", "PILOSA_MAX_WRITES_PER_REQUEST": "2000"}, cfgFileContent: ` data-dir = "/tmp/myFileDatadir" bind = "localhost:0" @@ -65,7 +65,6 @@ func TestServerConfig(t *testing.T) { v.Check(cmd.Server.Config.Bind, "localhost:10111") v.Check(cmd.Server.Config.Cluster.ReplicaN, 2) v.Check(cmd.Server.Config.Cluster.Hosts, []string{"localhost:10111", "localhost:10110"}) - v.Check(cmd.Server.Config.Cluster.PollInterval, pilosa.Duration(time.Second*182)) v.Check(cmd.Server.Config.Cluster.LongQueryTime, pilosa.Duration(time.Second*90)) v.Check(cmd.Server.Config.MaxWritesPerRequest, 2000) return v.Error() diff --git a/config.go b/config.go index 1b4a4485a..a43963f93 100644 --- a/config.go +++ b/config.go @@ -26,6 +26,9 @@ const ( ) const ( + // DefaultDataDir is the default data directory. + DefaultDataDir = "~/.pilosa" + // DefaultHost is the default hostname to use. DefaultHost = "localhost" @@ -35,14 +38,79 @@ const ( // DefaultClusterType sets the node intercommunication method. DefaultClusterType = ClusterGossip - // DefaultGossipPort indicates the port to which pilosa should bind for internal state sharing. - DefaultGossipPort = "14000" - // DefaultMetrics sets the internal metrics to no-op. DefaultMetrics = "nop" // DefaultMaxWritesPerRequest is the default number of writes per request. DefaultMaxWritesPerRequest = 5000 + + // Gossip config based on memberlist.Config. + + // Port indicates the port to which pilosa should bind for internal state sharing. + DefaultGossipPort = "14000" + + // StreamTimeout is the timeout for establishing a stream connection with + // a remote node for a full state sync, and for stream read and write + // operations. Maps to memberlist TCPTimeout. + DefaultGossipStreamTimeout = 10 * time.Second + + // SuspicionMult is the multiplier for determining the time an + // inaccessible node is considered suspect before declaring it dead. + // The actual timeout is calculated using the formula: + // + // SuspicionTimeout = SuspicionMult * log(N+1) * ProbeInterval + // + // This allows the timeout to scale properly with expected propagation + // delay with a larger cluster size. The higher the multiplier, the longer + // an inaccessible node is considered part of the cluster before declaring + // it dead, giving that suspect node more time to refute if it is indeed + // still alive. + DefaultGossipSuspicionMult = 4 + + // PushPullInterval is the interval between complete state syncs. + // Complete state syncs are done with a single node over TCP and are + // quite expensive relative to standard gossiped messages. Setting this + // to zero will disable state push/pull syncs completely. + // + // Setting this interval lower (more frequent) will increase convergence + // speeds across larger clusters at the expense of increased bandwidth + // usage. + DefaultGossipPushPullInterval = 30 * time.Second + + // ProbeInterval and ProbeTimeout are used to configure probing behavior + // for memberlist. + // + // ProbeInterval is the interval between random node probes. Setting + // this lower (more frequent) will cause the memberlist cluster to detect + // failed nodes more quickly at the expense of increased bandwidth usage. + // + // ProbeTimeout is the timeout to wait for an ack from a probed node + // before assuming it is unhealthy. This should be set to 99-percentile + // of RTT (round-trip time) on your network. + DefaultGossipProbeInterval = 1 * time.Second + DefaultGossipProbeTimeout = 500 * time.Millisecond + + // GossipInterval and GossipNodes are used to configure the gossip + // behavior of memberlist. + // + // GossipInterval is the interval between sending messages that need + // to be gossiped that haven't been able to piggyback on probing messages. + // If this is set to zero, non-piggyback gossip is disabled. By lowering + // this value (more frequent) gossip messages are propagated across + // the cluster more quickly at the expense of increased bandwidth. + // + // GossipNodes is the number of random nodes to send gossip messages to + // per GossipInterval. Increasing this number causes the gossip messages + // to propagate across the cluster more quickly at the expense of + // increased bandwidth. + // + // GossipToTheDeadTime is the interval after which a node has died that + // we will still try to gossip to it. This gives it a chance to refute. + DefaultGossipGossipInterval = 200 * time.Millisecond + DefaultGossipGossipNodes = 3 + DefaultGossipGossipToTheDeadTime = 30 * time.Second + + DefaultMetricPollInterval = 0 * time.Minute ) // ClusterTypes set of cluster types. @@ -67,54 +135,88 @@ type Config struct { // GossipSeed DEPRECATED GossipSeed string `toml:"gossip-seed"` - Gossip struct { - Port string `toml:"port"` - Seed string `toml:"seed"` - Key string `toml:"key"` - } `toml:"gossip"` + // Limits the number of mutating commands that can be in a single request to + // the server. This includes SetBit, ClearBit, SetRowAttrs & SetColumnAttrs. + MaxWritesPerRequest int `toml:"max-writes-per-request"` + + LogPath string `toml:"log-path"` + + // TLS + TLS TLSConfig Cluster struct { Coordinator string `toml:"coordinator"` ReplicaN int `toml:"replicas"` Type string `toml:"type"` Hosts []string `toml:"hosts"` - PollInterval Duration `toml:"poll-interval"` LongQueryTime Duration `toml:"long-query-time"` } `toml:"cluster"` + Gossip struct { + Port string `toml:"port"` + Seed string `toml:"seed"` + Key string `toml:"key"` + StreamTimeout Duration `toml:"stream-timeout"` + SuspicionMult int `toml:"suspicion-mult"` + PushPullInterval Duration `toml:"push-pull-interval"` + ProbeTimeout Duration `toml:"probe-timeout"` + ProbeInterval Duration `toml:"probe-interval"` + GossipNodes int `toml:"gossip-nodes"` + GossipInterval Duration `toml:"gossip-interval"` + GossipToTheDeadTime Duration `toml:"gossip-to-the-dead-time"` + } `toml:"gossip"` + AntiEntropy struct { Interval Duration `toml:"interval"` } `toml:"anti-entropy"` - // Limits the number of mutating commands that can be in a single request to - // the server. This includes SetBit, ClearBit, SetRowAttrs & SetColumnAttrs. - MaxWritesPerRequest int `toml:"max-writes-per-request"` - - LogPath string `toml:"log-path"` - Metric struct { Service string `toml:"service"` Host string `toml:"host"` PollInterval Duration `toml:"poll-interval"` Diagnostics bool `toml:"diagnostics"` } `toml:"metric"` - - TLS TLSConfig } // NewConfig returns an instance of Config with default options. func NewConfig() *Config { c := &Config{ - Bind: DefaultHost + ":" + DefaultPort, + DataDir: DefaultDataDir, + Bind: ":" + DefaultPort, MaxWritesPerRequest: DefaultMaxWritesPerRequest, + // LogPath: "", + TLS: TLSConfig{}, } + + // Cluster config. + // c.Cluster.Coordinator = "" c.Cluster.ReplicaN = DefaultReplicaN c.Cluster.Type = DefaultClusterType c.Cluster.Hosts = []string{} + c.Cluster.LongQueryTime = Duration(time.Minute) + + // Gossip config. + // c.Gossip.Port = "" + // c.Gossip.Seed = "" + // c.Gossip.Key = "" + c.Gossip.StreamTimeout = Duration(DefaultGossipStreamTimeout) + c.Gossip.SuspicionMult = DefaultGossipSuspicionMult + c.Gossip.PushPullInterval = Duration(DefaultGossipPushPullInterval) + c.Gossip.ProbeTimeout = Duration(DefaultGossipProbeTimeout) + c.Gossip.ProbeInterval = Duration(DefaultGossipProbeInterval) + c.Gossip.GossipNodes = DefaultGossipGossipNodes + c.Gossip.GossipInterval = Duration(DefaultGossipGossipInterval) + c.Gossip.GossipToTheDeadTime = Duration(DefaultGossipGossipToTheDeadTime) + + // AntiEntropy config. c.AntiEntropy.Interval = Duration(DefaultAntiEntropyInterval) + + // Metric config. c.Metric.Service = DefaultMetrics + // c.Metric.Host = "" + c.Metric.PollInterval = Duration(DefaultMetricPollInterval) c.Metric.Diagnostics = true - c.TLS = TLSConfig{} + return c } diff --git a/ctl/config_test.go b/ctl/config_test.go index a3bd035ac..b54446e34 100644 --- a/ctl/config_test.go +++ b/ctl/config_test.go @@ -17,11 +17,12 @@ package ctl import ( "bytes" "context" - "github.com/pilosa/pilosa" "io" "os" "strings" "testing" + + "github.com/pilosa/pilosa" ) func TestConfigCommand_Run(t *testing.T) { @@ -38,7 +39,7 @@ func TestConfigCommand_Run(t *testing.T) { if err != nil { t.Fatalf("Config Run doesn't work: %s", err) - } else if !strings.Contains(buf.String(), pilosa.DefaultHost) { - t.Fatalf("Unexpected config: %s", buf.String()) + } else if !strings.Contains(buf.String(), ":10101") { + t.Fatalf("Unexpected config: \n%s", buf.String()) } } diff --git a/ctl/server.go b/ctl/server.go index 07837caa0..1dddeb0d7 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -24,27 +24,46 @@ import ( // BuildServerFlags attaches a set of flags to the command for a server instance. 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.DataDir, "data-dir", "d", srv.Config.DataDir, "Directory to store pilosa data files.") + flags.StringVarP(&srv.Config.Bind, "bind", "b", srv.Config.Bind, "Default URI on which pilosa should listen.") flags.StringVarP(&srv.Config.GossipPort, "gossip-port", "", "", "(DEPRECATED) Port to which pilosa should bind for internal state sharing.") flags.StringVarP(&srv.Config.GossipSeed, "gossip-seed", "", "", "(DEPRECATED) Host with which to seed the gossip membership.") - flags.StringVarP(&srv.Config.Gossip.Port, "gossip.port", "", "", "Port to which pilosa should bind for internal state sharing.") - flags.StringVarP(&srv.Config.Gossip.Seed, "gossip.seed", "", "", "Host with which to seed the gossip membership.") - flags.StringVarP(&srv.Config.Gossip.Key, "gossip.key", "", "", "The path to file of the encryption key for gossip. The contents of the file should be either 16, 24, or 32 bytes to select AES-128, AES-192, or AES-256.") - flags.StringVarP(&srv.Config.Cluster.Coordinator, "cluster.coordinator", "", "", "Host that will act as cluster coordinator during startup and resizing.") flags.IntVarP(&srv.Config.MaxWritesPerRequest, "max-writes-per-request", "", srv.Config.MaxWritesPerRequest, "Number of write commands per request.") + flags.StringVar(&srv.Config.LogPath, "log-path", srv.Config.LogPath, "Log path") + + // TLS + SetTLSConfig(flags, &srv.Config.TLS.CertificatePath, &srv.Config.TLS.CertificateKeyPath, &srv.Config.TLS.SkipVerify) + + // Cluster + 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.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, "Duration that will trigger log and stat messages for slow queries.") - flags.StringVar(&srv.Config.LogPath, "log-path", "", "Log path") - flags.DurationVarP((*time.Duration)(&srv.Config.AntiEntropy.Interval), "anti-entropy.interval", "", time.Minute*10, "Interval at which to run anti-entropy routine.") + + // Gossip + flags.StringVarP(&srv.Config.Gossip.Port, "gossip.port", "", srv.Config.Gossip.Port, "Port to which pilosa should bind for internal state sharing.") + flags.StringVarP(&srv.Config.Gossip.Seed, "gossip.seed", "", srv.Config.Gossip.Seed, "Host with which to seed the gossip membership.") + flags.StringVarP(&srv.Config.Gossip.Key, "gossip.key", "", srv.Config.Gossip.Key, "The path to file of the encryption key for gossip. The contents of the file should be either 16, 24, or 32 bytes to select AES-128, AES-192, or AES-256.") + flags.DurationVarP((*time.Duration)(&srv.Config.Gossip.StreamTimeout), "gossip.stream-timeout", "", (time.Duration)(srv.Config.Gossip.StreamTimeout), "Timeout for establishing a stream connection with a remote node for a full state sync.") + flags.IntVarP(&srv.Config.Gossip.SuspicionMult, "gossip.suspicion-mult", "", srv.Config.Gossip.SuspicionMult, "Multiplier for determining the time an inaccessible node is considered suspect before declaring it dead.") + flags.DurationVarP((*time.Duration)(&srv.Config.Gossip.PushPullInterval), "gossip.push-pull-interval", "", (time.Duration)(srv.Config.Gossip.PushPullInterval), "Interval between complete state syncs.") + flags.DurationVarP((*time.Duration)(&srv.Config.Gossip.ProbeTimeout), "gossip.probe-timeout", "", (time.Duration)(srv.Config.Gossip.ProbeTimeout), "Timeout to wait for an ack from a probed node before assuming it is unhealthy.") + flags.DurationVarP((*time.Duration)(&srv.Config.Gossip.ProbeInterval), "gossip.probe-interval", "", (time.Duration)(srv.Config.Gossip.ProbeInterval), "Interval between random node probes.") + flags.IntVarP(&srv.Config.Gossip.GossipNodes, "gossip.gossip-nodes", "", srv.Config.Gossip.GossipNodes, "Number of random nodes to send gossip messages to per GossipInterval.") + flags.DurationVarP((*time.Duration)(&srv.Config.Gossip.GossipInterval), "gossip.gossip-interval", "", (time.Duration)(srv.Config.Gossip.GossipInterval), "Interval between sending messages that need to be gossiped that haven't piggybacked on probing messages.") + flags.DurationVarP((*time.Duration)(&srv.Config.Gossip.GossipToTheDeadTime), "gossip.gossip-to-the-dead-time", "", (time.Duration)(srv.Config.Gossip.GossipToTheDeadTime), "Interval after which a node has died that we will still try to gossip to it.") + + // AntiEntropy + flags.DurationVarP((*time.Duration)(&srv.Config.AntiEntropy.Interval), "anti-entropy.interval", "", (time.Duration)(srv.Config.AntiEntropy.Interval), "Interval at which to run anti-entropy routine.") + + // Metric + flags.StringVarP(&srv.Config.Metric.Service, "metric.service", "", srv.Config.Metric.Service, "Default URI on which pilosa should listen.") + flags.StringVarP(&srv.Config.Metric.Host, "metric.host", "", srv.Config.Metric.Host, "Default URI to send metrics.") + flags.DurationVarP((*time.Duration)(&srv.Config.Metric.PollInterval), "metric.poll-interval", "", (time.Duration)(srv.Config.Metric.PollInterval), "Polling interval metrics.") + flags.BoolVarP((&srv.Config.Metric.Diagnostics), "metric.diagnostics", "", srv.Config.Metric.Diagnostics, "Enabled diagnostics reporting.") + + // CPU Profiling 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", "", "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.BoolVarP((&srv.Config.Metric.Diagnostics), "metric.diagnostics", "", true, "Enabled diagnostics reporting.") - flags.DurationVarP((*time.Duration)(&srv.Config.Metric.PollInterval), "metric.poll-interval", "", time.Minute*0, "Polling interval metrics.") - SetTLSConfig(flags, &srv.Config.TLS.CertificatePath, &srv.Config.TLS.CertificateKeyPath, &srv.Config.TLS.SkipVerify) } diff --git a/gossip/gossip.go b/gossip/gossip.go index 35f0056c8..84daad8b3 100644 --- a/gossip/gossip.go +++ b/gossip/gossip.go @@ -17,8 +17,11 @@ package gossip import ( "fmt" "io" + "io/ioutil" "log" + "net" "os" + "strconv" "strings" "sync" "time" @@ -79,7 +82,7 @@ func (g *GossipMemberSet) Seed() string { // Open implements the MemberSet interface to start network activity. func (g *GossipMemberSet) Open() error { if g.handler == nil { - return fmt.Errorf("opening GossipMemberSet: you must call Start(pilosa.BroadcastHandler) before calling Open()") + return fmt.Errorf("must call Start(pilosa.BroadcastHandler) before calling Open()") } err := error(nil) @@ -87,7 +90,7 @@ func (g *GossipMemberSet) Open() error { g.memberlist, err = memberlist.Create(g.config.memberlistConfig) g.mu.Unlock() if err != nil { - return err + return fmt.Errorf("creating memberlist: %s", err) } g.broadcasts = &memberlist.TransmitLimitedQueue{ @@ -101,7 +104,7 @@ func (g *GossipMemberSet) Open() error { uri, err := pilosa.NewURIFromAddress(g.config.gossipSeed) if err != nil { - return err + return fmt.Errorf("new uri from address: %s", err) } // attach to gossip seed node @@ -111,7 +114,7 @@ func (g *GossipMemberSet) Open() error { err = g.joinWithRetry(pilosa.NodeSet(pilosa.Nodes(nodes).URIs()).ToHostPortStrings()) g.mu.RUnlock() if err != nil { - return err + return fmt.Errorf("joining member set: %s", err) } return nil } @@ -154,58 +157,89 @@ type gossipConfig struct { } // NewGossipMemberSetWithTransport returns a new instance of GossipMemberSet given a Transport. -func NewGossipMemberSetWithTransport(name string, gossipHost string, transport *Transport, gossipSeed string, server *pilosa.Server, secretKey []byte) (*GossipMemberSet, error) { - port := transport.Net.GetAutoBindPort() +func NewGossipMemberSetWithTransport(name string, cfg *pilosa.Config, transport *Transport, server *pilosa.Server) (*GossipMemberSet, error) { g := &GossipMemberSet{ LogOutput: server.LogOutput, } + port := transport.Net.GetAutoBindPort() + host, _, err := net.SplitHostPort(cfg.Bind) + if err != nil { + return nil, fmt.Errorf("split host port: %s", err) + } + + var gossipKey []byte + if cfg.Gossip.Key != "" { + gossipKey, err = ioutil.ReadFile(cfg.Gossip.Key) + if err != nil { + return nil, fmt.Errorf("reading gossip key: %s", err) + } + } + // memberlist config conf := memberlist.DefaultLocalConfig() conf.Transport = transport.Net + conf.Name = name + conf.BindAddr = host conf.BindPort = port conf.AdvertisePort = port - conf.Name = name - conf.BindAddr = gossipHost - conf.AdvertiseAddr = pilosa.HostToIP(gossipHost) - //conf.PushPullInterval = 0 * time.Second // Default is 15s in DefaultLocalConfig. + conf.AdvertiseAddr = pilosa.HostToIP(host) + // + conf.TCPTimeout = time.Duration(cfg.Gossip.StreamTimeout) + conf.SuspicionMult = cfg.Gossip.SuspicionMult + conf.PushPullInterval = time.Duration(cfg.Gossip.PushPullInterval) + conf.ProbeTimeout = time.Duration(cfg.Gossip.ProbeTimeout) + conf.ProbeInterval = time.Duration(cfg.Gossip.ProbeInterval) + conf.GossipNodes = cfg.Gossip.GossipNodes + conf.GossipInterval = time.Duration(cfg.Gossip.GossipInterval) + conf.GossipToTheDeadTime = time.Duration(cfg.Gossip.GossipToTheDeadTime) + // conf.Delegate = g - conf.SecretKey = secretKey + conf.SecretKey = gossipKey conf.Events = server.Cluster.EventReceiver.(memberlist.EventDelegate) - //TODO: pull memberlist config from pilosa.cfg file g.config = &gossipConfig{ memberlistConfig: conf, - gossipSeed: gossipSeed, + gossipSeed: cfg.Gossip.Seed, } g.statusHandler = server // If no gossipSeed is provided, use local host:port. - if gossipSeed == "" { - g.config.gossipSeed = fmt.Sprintf("%s:%d", gossipHost, port) + if cfg.Gossip.Seed == "" { + g.config.gossipSeed = fmt.Sprintf("%s:%d", host, port) } return g, nil } // NewGossipMemberSet returns a new instance of GossipMemberSet given a gossip port. -func NewGossipMemberSet(name string, gossipHost string, gossipPort int, gossipSeed string, server *pilosa.Server, secretKey []byte) (*GossipMemberSet, error) { - // set up the transport - transport, err := NewTransport(gossipHost, gossipPort) +func NewGossipMemberSet(name string, cfg *pilosa.Config, server *pilosa.Server) (*GossipMemberSet, error) { + + port, err := strconv.Atoi(cfg.Gossip.Port) if err != nil { - return nil, err + return nil, fmt.Errorf("convert port: %s", err) + } + host, _, err := net.SplitHostPort(cfg.Bind) + if err != nil { + return nil, fmt.Errorf("split host port: %s", err) } - return NewGossipMemberSetWithTransport(name, gossipHost, transport, gossipSeed, server, secretKey) + // Set up the transport. + transport, err := NewTransport(host, port) + if err != nil { + return nil, fmt.Errorf("new tranport: %s", err) + } + + return NewGossipMemberSetWithTransport(name, cfg, transport, server) } // SendSync implementation of the Broadcaster interface. func (g *GossipMemberSet) SendSync(pb proto.Message) error { msg, err := pilosa.MarshalMessage(pb) if err != nil { - return err + return fmt.Errorf("marshal message: %s", err) } mlist := g.memberlist @@ -233,7 +267,7 @@ func (g *GossipMemberSet) SendSync(pb proto.Message) error { func (g *GossipMemberSet) SendAsync(pb proto.Message) error { msg, err := pilosa.MarshalMessage(pb) if err != nil { - return err + return fmt.Errorf("marshal message: %s", err) } b := &broadcast{ @@ -404,12 +438,12 @@ func NewTransport(host string, port int) (*Transport, error) { net, err := newTransport(conf) if err != nil { - return nil, err + return nil, fmt.Errorf("new transport: %s", err) } uri, err := pilosa.NewURIFromHostPort(host, uint16(net.GetAutoBindPort())) if err != nil { - return nil, err + return nil, fmt.Errorf("new uri from host port: %s", err) } return &Transport{ @@ -473,12 +507,6 @@ func newTransport(conf *memberlist.Config) (*memberlist.NetTransport, error) { if err != nil { return nil, fmt.Errorf("Could not set up network transport: %v", err) } - if conf.BindPort == 0 { - port := nt.GetAutoBindPort() - conf.BindPort = port - conf.AdvertisePort = port - logger.Printf("[DEBUG] Using dynamic bind port %d", port) - } return nt, nil } diff --git a/server.go b/server.go index dc25103ff..aea2c8518 100644 --- a/server.go +++ b/server.go @@ -42,7 +42,6 @@ import ( // Default server settings. const ( DefaultAntiEntropyInterval = 10 * time.Minute - DefaultPollingInterval = 60 * time.Second DefaultDiagnosticServer = "https://diagnostics.pilosa.com/v0/diagnostics" ) diff --git a/server/cluster_test.go b/server/cluster_test.go index bc7b1cd91..fdf6fdc88 100644 --- a/server/cluster_test.go +++ b/server/cluster_test.go @@ -27,7 +27,7 @@ import ( ) // Ensure program can send/receive broadcast messages. -func TestMain_XSendReceiveMessage(t *testing.T) { +func TestMain_SendReceiveMessage(t *testing.T) { m0 := MustRunMain() defer m0.Close() @@ -45,14 +45,13 @@ func TestMain_XSendReceiveMessage(t *testing.T) { // Configure node0 // get the host portion of addr to use for binding - gossipHost := m0.Server.URI.Host() - gossipPort := 0 - gossipSeed := "" + m0.Config.Gossip.Port = "0" + m0.Config.Gossip.Seed = "" m0.Server.Cluster.Coordinator = m0.Server.URI m0.Server.Cluster.Topology = &pilosa.Topology{NodeSet: []pilosa.URI{m0.Server.URI, m1.Server.URI}} m0.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver() - gossipMemberSet0, err := gossip.NewGossipMemberSet(m0.Server.URI.HostPort(), gossipHost, gossipPort, gossipSeed, m0.Server, nil) + gossipMemberSet0, err := gossip.NewGossipMemberSet(m0.Server.URI.HostPort(), m0.Config, m0.Server) if err != nil { t.Fatal(err) } @@ -74,13 +73,12 @@ func TestMain_XSendReceiveMessage(t *testing.T) { // Configure node1 // get the host portion of addr to use for binding - gossipHost = m1.Server.URI.Host() - gossipPort = 0 - gossipSeed = gossipMemberSet0.Seed() + m1.Config.Gossip.Port = "0" + m1.Config.Gossip.Seed = gossipMemberSet0.Seed() m1.Server.Cluster.Coordinator = m0.Server.URI m1.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver() - gossipMemberSet1, err := gossip.NewGossipMemberSet(m1.Server.URI.HostPort(), gossipHost, gossipPort, gossipSeed, m1.Server, nil) + gossipMemberSet1, err := gossip.NewGossipMemberSet(m1.Server.URI.HostPort(), m1.Config, m1.Server) if err != nil { t.Fatal(err) } diff --git a/server/server.go b/server/server.go index 045079229..d238d1c2f 100644 --- a/server/server.go +++ b/server/server.go @@ -22,7 +22,6 @@ import ( "errors" "fmt" "io" - "io/ioutil" "math/rand" "os" "path/filepath" @@ -42,9 +41,6 @@ func init() { } const ( - // DefaultDataDir is the default data directory. - DefaultDataDir = "~/.pilosa" - // DefaultDiagnosticsInterval is the default sync frequency diagnostic metrics. DefaultDiagnosticsInterval = 1 * time.Hour ) @@ -236,21 +232,6 @@ func (m *Command) SetupNetworking() error { if err != nil { return err } - gossipSeed := pilosa.DefaultHost + ":" + pilosa.DefaultGossipPort - // Config.GossipSeed is deprecated, so Config.Gossip.Seed has priority - if m.Config.Gossip.Seed != "" { - gossipSeed = m.Config.Gossip.Seed - } else if m.Config.GossipSeed != "" { - gossipSeed = m.Config.GossipSeed - } - - var gossipKey []byte - if m.Config.Gossip.Key != "" { - gossipKey, err = ioutil.ReadFile(m.Config.Gossip.Key) - if err != nil { - return err - } - } // get the host portion of addr to use for binding gossipHost := m.Server.URI.Host() @@ -268,7 +249,7 @@ func (m *Command) SetupNetworking() error { if m.Server.Name == "" { return fmt.Errorf("must provide a valid name for gossip membership") } - gossipMemberSet, err := gossip.NewGossipMemberSetWithTransport(m.Server.Name, gossipHost, transport, gossipSeed, m.Server, gossipKey) + gossipMemberSet, err := gossip.NewGossipMemberSetWithTransport(m.Server.Name, m.Config, transport, m.Server) if err != nil { return err } diff --git a/server/server_test.go b/server/server_test.go index 4c9eb52bf..b96dbec26 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -30,7 +30,6 @@ import ( "strings" "testing" "testing/quick" - "time" "github.com/BurntSushi/toml" "github.com/pilosa/pilosa" @@ -384,189 +383,6 @@ func TestCountOpenFiles(t *testing.T) { } } -// Ensure program can send/receive broadcast messages. -func TestMain_SendReceiveMessage(t *testing.T) { - - m0 := MustRunMain() - defer m0.Close() - - m1 := MustRunMain() - defer m1.Close() - - // Update cluster config - m0.Server.Cluster.Nodes = []*pilosa.Node{ - {URI: m0.Server.URI}, - {URI: m1.Server.URI}, - } - m1.Server.Cluster.Nodes = m0.Server.Cluster.Nodes - - // Configure node0 - - // get the host portion of addr to use for binding - gossipHost := m0.Server.URI.Host() - gossipPort := 0 - gossipSeed := "" - - topology := &pilosa.Topology{NodeSet: []pilosa.URI{m0.Server.URI, m1.Server.URI}} - - m0.Server.Cluster.Coordinator = m0.Server.URI - m0.Server.Cluster.Topology = topology - m0.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver() - - gossipMemberSet0, err := gossip.NewGossipMemberSet(m0.Server.URI.HostPort(), gossipHost, gossipPort, gossipSeed, m0.Server, nil) - if err != nil { - t.Fatal(err) - } - m0.Server.Cluster.MemberSet = gossipMemberSet0 - m0.Server.Broadcaster = m0.Server - m0.Server.Gossiper = gossipMemberSet0 - m0.Server.Handler.Broadcaster = m0.Server.Broadcaster - m0.Server.Holder.Broadcaster = m0.Server.Broadcaster - m0.Server.BroadcastReceiver = gossipMemberSet0 - - if err := m0.Server.BroadcastReceiver.Start(m0.Server); err != nil { - t.Fatal(err) - } - // Open Cluster management. - if err := m0.Server.Cluster.Open(); err != nil { - t.Fatal(err) - } - - // Configure node1 - - // get the host portion of addr to use for binding - gossipHost = m1.Server.URI.Host() - gossipPort = 0 - gossipSeed = gossipMemberSet0.Seed() - - m1.Server.Cluster.Coordinator = m0.Server.URI - m1.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver() - - gossipMemberSet1, err := gossip.NewGossipMemberSet(m1.Server.URI.HostPort(), gossipHost, gossipPort, gossipSeed, m1.Server, nil) - if err != nil { - t.Fatal(err) - } - m1.Server.Cluster.MemberSet = gossipMemberSet1 - m1.Server.Broadcaster = m1.Server - m1.Server.Gossiper = gossipMemberSet1 - m1.Server.Handler.Broadcaster = m1.Server.Broadcaster - m1.Server.Holder.Broadcaster = m1.Server.Broadcaster - m1.Server.BroadcastReceiver = gossipMemberSet1 - - if err := m1.Server.BroadcastReceiver.Start(m1.Server); err != nil { - t.Fatal(err) - } - // Open Cluster management. - if err := m1.Server.Cluster.Open(); err != nil { - t.Fatal(err) - } - - //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// - - // Expected indexes and Frames - expected := map[string][]string{ - "i": []string{"f"}, - } - - // Create a client for each node. - client0 := m0.Client() - client1 := m1.Client() - - // Create indexes and frames on one node. - if err := client0.CreateIndex(context.Background(), "i", pilosa.IndexOptions{}); err != nil && err != pilosa.ErrIndexExists { - t.Fatal(err) - } else if err := client0.CreateFrame(context.Background(), "i", "f", pilosa.FrameOptions{}); err != nil { - t.Fatal(err) - } - - // Make sure node0 knows about the index and frame created. - schema0, err := client0.Schema(context.Background()) - if err != nil { - t.Fatal(err) - } - received0 := map[string][]string{} - for _, idx := range schema0 { - received0[idx.Name] = []string{} - for _, frame := range idx.Frames { - received0[idx.Name] = append(received0[idx.Name], frame.Name) - } - } - if !reflect.DeepEqual(received0, expected) { - t.Fatalf("unexpected schema on node0: %s", received0) - } - - // Make sure node1 knows about the index and frame created. - schema1, err := client1.Schema(context.Background()) - if err != nil { - t.Fatal(err) - } - received1 := map[string][]string{} - for _, idx := range schema1 { - received1[idx.Name] = []string{} - for _, frame := range idx.Frames { - received1[idx.Name] = append(received1[idx.Name], frame.Name) - } - } - if !reflect.DeepEqual(received1, expected) { - t.Fatalf("unexpected schema on node1: %s", received1) - } - - // 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 { - 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 { - t.Fatal(err) - } - if maxSlices0["i"] != 2 { - t.Fatalf("unexpected maxSlice on node0: %d", maxSlices0["i"]) - } - - // Make sure node1 knows about the latest MaxSlice. - maxSlices1, err := client1.MaxSliceByIndex(context.Background()) - if err != nil { - t.Fatal(err) - } - if maxSlices1["i"] != 2 { - t.Fatalf("unexpected maxSlice on node1: %d", maxSlices1["i"]) - } - - // 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 { - t.Fatal(err) - } - - // We have to wait for the broadcast message to be sent before checking state. - time.Sleep(1 * time.Second) - - frame0 := m0.Server.Holder.Frame("i", "event-time") - if frame0 == nil { - t.Fatal("frame not found") - } - frame1 := m1.Server.Holder.Frame("i", "event-time") - if frame1 == nil { - t.Fatal("frame not found") - } -} - // Main represents a test wrapper for main.Main. type Main struct { *server.Command