diff --git a/config.go b/config.go index 1b4a4485a..9dfbb53ac 100644 --- a/config.go +++ b/config.go @@ -35,14 +35,77 @@ 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 ) // ClusterTypes set of cluster types. @@ -68,9 +131,17 @@ type Config struct { GossipSeed string `toml:"gossip-seed"` Gossip struct { - Port string `toml:"port"` - Seed string `toml:"seed"` - Key string `toml:"key"` + 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"` Cluster struct { @@ -115,6 +186,17 @@ func NewConfig() *Config { c.Metric.Service = DefaultMetrics c.Metric.Diagnostics = true c.TLS = TLSConfig{} + + // Gossip related config. + 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) + return c } diff --git a/ctl/server.go b/ctl/server.go index 07837caa0..ba49aded1 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -28,9 +28,20 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags.StringVarP(&srv.Config.Bind, "bind", "b", ":10101", "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.") + // gossip 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.DurationVarP((*time.Duration)(&srv.Config.Gossip.StreamTimeout), "gossip.stream-timeout", "", 10*time.Second, "Timeout for establishing a stream connection with a remote node for a full state sync.") + flags.IntVarP(&srv.Config.Gossip.SuspicionMult, "gossip.suspicion-mult", "", 4, "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", "", 30*time.Second, "Interval between complete state syncs.") + flags.DurationVarP((*time.Duration)(&srv.Config.Gossip.ProbeTimeout), "gossip.probe-timeout", "", 500*time.Millisecond, "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", "", 1*time.Second, "Interval between random node probes.") + flags.IntVarP(&srv.Config.Gossip.GossipNodes, "gossip.gossip-nodes", "", 3, "Number of random nodes to send gossip messages to per GossipInterval.") + flags.DurationVarP((*time.Duration)(&srv.Config.Gossip.GossipInterval), "gossip.gossip-interval", "", 200*time.Millisecond, "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", "", 30*time.Second, "Interval after which a node has died that we will still try to gossip to it.") + 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.IntVarP(&srv.Config.Cluster.ReplicaN, "cluster.replicas", "", 1, "Number of hosts each piece of data should be stored on.") diff --git a/gossip/gossip.go b/gossip/gossip.go index 35f0056c8..1ee93263a 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" @@ -154,51 +157,82 @@ 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, err + } + + var gossipKey []byte + if cfg.Gossip.Key != "" { + gossipKey, err = ioutil.ReadFile(cfg.Gossip.Key) + if err != nil { + return nil, 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 + } + host, _, err := net.SplitHostPort(cfg.Bind) if err != nil { return nil, err } - return NewGossipMemberSetWithTransport(name, gossipHost, transport, gossipSeed, server, secretKey) + // Set up the transport. + transport, err := NewTransport(host, port) + if err != nil { + return nil, err + } + + return NewGossipMemberSetWithTransport(name, cfg, transport, server) } // SendSync implementation of the Broadcaster interface. @@ -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/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..27c3c2e4c 100644 --- a/server/server.go +++ b/server/server.go @@ -22,7 +22,6 @@ import ( "errors" "fmt" "io" - "io/ioutil" "math/rand" "os" "path/filepath" @@ -236,21 +235,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 +252,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