scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-15 08:41:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1014 from travisturner/cluster-resize-gossip-config

Browse source 
Cluster resize gossip config
This commit is contained in:
Travis Turner 2018-01-03 11:20:20 -06:00 • committed by GitHub
commit 3631cfc314
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
9 changed files with 226 additions and 283 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
cmd/server_test.go
View file

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

140
config.go
View file

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

7
ctl/config_test.go
View file

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

49
ctl/server.go
View file

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

88
gossip/gossip.go
View file

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

1
server.go
View file

@ -42,7 +42,6 @@ import (
// Default server settings.
const (
DefaultAntiEntropyInterval = 10 * time.Minute
DefaultPollingInterval = 60 * time.Second
DefaultDiagnosticServer = "https://diagnostics.pilosa.com/v0/diagnostics"
)

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

21
server/server.go
View file

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

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