scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Add the rest of the available memberlist configuration options

Browse source 
into pilosa.Config.Gossip.
This commit is contained in:
Travis Turner 2017-12-15 12:37:51 -06:00
parent 32c5b20748
commit 78cddbd0c7
No known key found for this signature in database
GPG key ID: 7F08008DFD9314C9
6 changed files with 156 additions and 237 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

94
config.go
View file

@ -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
}

11
ctl/server.go
View file

@ -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.")

70
gossip/gossip.go
View file

@ -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
}

16
server/cluster_test.go
View file

@ -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)
}

18
server/server.go
View file

@ -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
}

184
server/server_test.go
View file

@ -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