mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-11 15:21:02 +00:00
update config to support etcd arguments
This commit is contained in:
Travis
parent
e459c9a77b
commit
457194f6a8
12 changed files with 111 additions and 129 deletions
|
|
@ -49,7 +49,7 @@ func TestServerConfig(t *testing.T) {
|
|||
tests := []commandTest{
|
||||
// TEST 0
|
||||
{
|
||||
args: []string{"server", "--data-dir", actualDataDir, "--cluster.hosts", "localhost:42454,localhost:10110", "--bind", "localhost:42454", "--bind-grpc", "localhost:30112", "--translation.map-size", "100000"},
|
||||
args: []string{"server", "--data-dir", actualDataDir, "--bind", "localhost:42454", "--bind-grpc", "localhost:30112", "--translation.map-size", "100000"},
|
||||
env: map[string]string{
|
||||
"PILOSA_DATA_DIR": "/tmp/myEnvDatadir",
|
||||
"PILOSA_LONG_QUERY_TIME": "1m30s",
|
||||
|
|
@ -66,11 +66,7 @@ func TestServerConfig(t *testing.T) {
|
|||
long-query-time = "1m10s"
|
||||
|
||||
[cluster]
|
||||
disabled = true
|
||||
replicas = 2
|
||||
hosts = [
|
||||
"localhost:19444",
|
||||
]
|
||||
long-query-time = "1m10s"
|
||||
[profile]
|
||||
block-rate = 100
|
||||
|
|
@ -81,7 +77,6 @@ func TestServerConfig(t *testing.T) {
|
|||
v.Check(cmd.Server.Config.DataDir, actualDataDir)
|
||||
v.Check(cmd.Server.Config.Bind, "localhost:42454")
|
||||
v.Check(cmd.Server.Config.Cluster.ReplicaN, 2)
|
||||
v.Check(cmd.Server.Config.Cluster.Hosts, []string{"localhost:42454", "localhost:10110"})
|
||||
v.Check(cmd.Server.Config.LongQueryTime, toml.Duration(time.Second*90))
|
||||
v.Check(cmd.Server.Config.Cluster.LongQueryTime, toml.Duration(time.Second*90))
|
||||
v.Check(cmd.Server.Config.MaxWritesPerRequest, 2000)
|
||||
|
|
@ -109,18 +104,12 @@ func TestServerConfig(t *testing.T) {
|
|||
bind = ` + nextPort() + `
|
||||
bind-grpc = ` + nextPort() + `
|
||||
data-dir = "` + actualDataDir + `"
|
||||
[cluster]
|
||||
disabled = true
|
||||
hosts = [
|
||||
"localhost:19444",
|
||||
]
|
||||
[profile]
|
||||
block-rate = 100
|
||||
mutex-fraction = 10
|
||||
`,
|
||||
validation: func() error {
|
||||
v := validator{}
|
||||
v.Check(cmd.Server.Config.Cluster.Hosts, []string{"localhost:1110", "localhost:1111"})
|
||||
v.Check(cmd.Server.Config.AntiEntropy.Interval, toml.Duration(time.Minute*9))
|
||||
v.Check(cmd.Server.Config.Translation.MapSize, 100000)
|
||||
v.Check(cmd.Server.Config.Profile.BlockRate, 4832)
|
||||
|
|
@ -136,10 +125,6 @@ func TestServerConfig(t *testing.T) {
|
|||
bind = "localhost:19444"
|
||||
bind-grpc = "localhost:29444"
|
||||
data-dir = "` + actualDataDir + `"
|
||||
[cluster]
|
||||
hosts = [
|
||||
"localhost:19444",
|
||||
]
|
||||
[anti-entropy]
|
||||
interval = "11m0s"
|
||||
[metric]
|
||||
|
|
@ -152,7 +137,6 @@ func TestServerConfig(t *testing.T) {
|
|||
`,
|
||||
validation: func() error {
|
||||
v := validator{}
|
||||
v.Check(cmd.Server.Config.Cluster.Hosts, []string{"localhost:19444"})
|
||||
v.Check(cmd.Server.Config.AntiEntropy.Interval, toml.Duration(time.Minute*11))
|
||||
v.Check(cmd.Server.Config.LogPath, logFile.Name())
|
||||
v.Check(cmd.Server.Config.Metric.Service, "statsd")
|
||||
|
|
|
|||
|
|
@ -26,77 +26,76 @@ 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.StringVar(&srv.Config.Name, "name", srv.Config.Name, "Name of the node in the cluster.")
|
||||
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.StringVar(&srv.Config.BindGRPC, "bind-grpc", srv.Config.BindGRPC, "URI on which pilosa should listen for gRPC requests.")
|
||||
flags.StringVar(&srv.Config.Advertise, "advertise", srv.Config.Advertise, "Address to advertise externally.")
|
||||
flags.StringVar(&srv.Config.AdvertiseGRPC, "advertise-grpc", srv.Config.AdvertiseGRPC, "Address to advertise externally for gRPC.")
|
||||
flags.IntVarP(&srv.Config.MaxWritesPerRequest, "max-writes-per-request", "", srv.Config.MaxWritesPerRequest, "Number of write commands per request.")
|
||||
flags.IntVar(&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")
|
||||
flags.BoolVar(&srv.Config.Verbose, "verbose", srv.Config.Verbose, "Enable verbose logging")
|
||||
flags.Uint64Var(&srv.Config.MaxMapCount, "max-map-count", srv.Config.MaxMapCount, "Limits the maximum number of active mmaps. Pilosa will fall back to reading files once this is exhausted. Set below your system's vm.max_map_count.")
|
||||
flags.Uint64Var(&srv.Config.MaxFileCount, "max-file-count", srv.Config.MaxFileCount, "Soft limit on the maximum number of fragment files Pilosa keeps open simultaneously.")
|
||||
flags.DurationVarP((*time.Duration)(&srv.Config.LongQueryTime), "long-query-time", "", time.Duration(srv.Config.LongQueryTime), "Duration that will trigger log and stat messages for slow queries. Zero to disable.")
|
||||
flags.DurationVar((*time.Duration)(&srv.Config.LongQueryTime), "long-query-time", time.Duration(srv.Config.LongQueryTime), "Duration that will trigger log and stat messages for slow queries. Zero to disable.")
|
||||
flags.IntVar(&srv.Config.QueryHistoryLength, "query-history-length", srv.Config.QueryHistoryLength, "Number of queries to remember in history.")
|
||||
|
||||
// TLS
|
||||
SetTLSConfig(flags, "", &srv.Config.TLS.CertificatePath, &srv.Config.TLS.CertificateKeyPath, &srv.Config.TLS.CACertPath, &srv.Config.TLS.SkipVerify, &srv.Config.TLS.EnableClientVerification)
|
||||
|
||||
// Handler
|
||||
flags.StringSliceVarP(&srv.Config.Handler.AllowedOrigins, "handler.allowed-origins", "", []string{}, "Comma separated list of allowed origin URIs (for CORS/Web UI).")
|
||||
flags.StringSliceVar(&srv.Config.Handler.AllowedOrigins, "handler.allowed-origins", []string{}, "Comma separated list of allowed origin URIs (for CORS/Web UI).")
|
||||
|
||||
// Cluster
|
||||
flags.BoolVarP(&srv.Config.Cluster.Disabled, "cluster.disabled", "", srv.Config.Cluster.Disabled, "Disabled multi-node cluster communication (used for testing)")
|
||||
flags.BoolVarP(&srv.Config.Cluster.Coordinator, "cluster.coordinator", "", srv.Config.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.StringSliceVarP(&srv.Config.Cluster.Hosts, "cluster.hosts", "", []string{}, "Comma separated list of hosts in cluster. Only used for testing.")
|
||||
flags.DurationVarP((*time.Duration)(&srv.Config.Cluster.LongQueryTime), "cluster.long-query-time", "", time.Duration(srv.Config.Cluster.LongQueryTime), "RENAMED TO 'long-query-time': Duration that will trigger log and stat messages for slow queries.") // negative duration indicates invalid value because 0 is meaningful
|
||||
flags.BoolVar(&srv.Config.Cluster.Coordinator, "cluster.coordinator", srv.Config.Cluster.Coordinator, "Host that will act as cluster coordinator during startup and resizing.")
|
||||
flags.IntVar(&srv.Config.Cluster.ReplicaN, "cluster.replicas", 1, "Number of hosts each piece of data should be stored on.")
|
||||
flags.DurationVar((*time.Duration)(&srv.Config.Cluster.LongQueryTime), "cluster.long-query-time", time.Duration(srv.Config.Cluster.LongQueryTime), "RENAMED TO 'long-query-time': Duration that will trigger log and stat messages for slow queries.") // negative duration indicates invalid value because 0 is meaningful
|
||||
flags.StringVar(&srv.Config.Cluster.Name, "cluster.name", srv.Config.Cluster.Name, "Human-readable name for the cluster.")
|
||||
|
||||
// Translation
|
||||
flags.StringVarP(&srv.Config.Translation.PrimaryURL, "translation.primary-url", "", srv.Config.Translation.PrimaryURL, "DEPRECATED: URL for primary translation node for replication.")
|
||||
flags.IntVarP(&srv.Config.Translation.MapSize, "translation.map-size", "", srv.Config.Translation.MapSize, "Size in bytes of mmap to allocate for key translation.")
|
||||
flags.StringVar(&srv.Config.Translation.PrimaryURL, "translation.primary-url", srv.Config.Translation.PrimaryURL, "DEPRECATED: URL for primary translation node for replication.")
|
||||
flags.IntVar(&srv.Config.Translation.MapSize, "translation.map-size", srv.Config.Translation.MapSize, "Size in bytes of mmap to allocate for key translation.")
|
||||
|
||||
// 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.AdvertiseHost, "gossip.advertise-host", "", srv.Config.Gossip.AdvertiseHost, "Host on which memberlist should advertise.")
|
||||
flags.StringVarP(&srv.Config.Gossip.AdvertisePort, "gossip.advertise-port", "", srv.Config.Gossip.AdvertisePort, "Port on which memberlist should advertise.")
|
||||
flags.StringVar(&srv.Config.Gossip.Port, "gossip.port", srv.Config.Gossip.Port, "Port to which pilosa should bind for internal state sharing.")
|
||||
flags.StringVar(&srv.Config.Gossip.AdvertiseHost, "gossip.advertise-host", srv.Config.Gossip.AdvertiseHost, "Host on which memberlist should advertise.")
|
||||
flags.StringVar(&srv.Config.Gossip.AdvertisePort, "gossip.advertise-port", srv.Config.Gossip.AdvertisePort, "Port on which memberlist should advertise.")
|
||||
|
||||
flags.StringSliceVarP(&srv.Config.Gossip.Seeds, "gossip.seeds", "", srv.Config.Gossip.Seeds, "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.Nodes, "gossip.nodes", "", srv.Config.Gossip.Nodes, "Number of random nodes to send gossip messages to per GossipInterval.")
|
||||
flags.DurationVarP((*time.Duration)(&srv.Config.Gossip.Interval), "gossip.interval", "", (time.Duration)(srv.Config.Gossip.Interval), "Interval between sending messages that need to be gossiped that haven't piggybacked on probing messages.")
|
||||
flags.DurationVarP((*time.Duration)(&srv.Config.Gossip.ToTheDeadTime), "gossip.to-the-dead-time", "", (time.Duration)(srv.Config.Gossip.ToTheDeadTime), "Interval after which a node has died that we will still try to gossip to it.")
|
||||
flags.StringSliceVar(&srv.Config.Gossip.Seeds, "gossip.seeds", srv.Config.Gossip.Seeds, "Host with which to seed the gossip membership.")
|
||||
flags.StringVar(&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.DurationVar((*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.IntVar(&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.DurationVar((*time.Duration)(&srv.Config.Gossip.PushPullInterval), "gossip.push-pull-interval", (time.Duration)(srv.Config.Gossip.PushPullInterval), "Interval between complete state syncs.")
|
||||
flags.DurationVar((*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.DurationVar((*time.Duration)(&srv.Config.Gossip.ProbeInterval), "gossip.probe-interval", (time.Duration)(srv.Config.Gossip.ProbeInterval), "Interval between random node probes.")
|
||||
flags.IntVar(&srv.Config.Gossip.Nodes, "gossip.nodes", srv.Config.Gossip.Nodes, "Number of random nodes to send gossip messages to per GossipInterval.")
|
||||
flags.DurationVar((*time.Duration)(&srv.Config.Gossip.Interval), "gossip.interval", (time.Duration)(srv.Config.Gossip.Interval), "Interval between sending messages that need to be gossiped that haven't piggybacked on probing messages.")
|
||||
flags.DurationVar((*time.Duration)(&srv.Config.Gossip.ToTheDeadTime), "gossip.to-the-dead-time", (time.Duration)(srv.Config.Gossip.ToTheDeadTime), "Interval after which a node has died that we will still try to gossip to it.")
|
||||
|
||||
// Etcd
|
||||
flags.StringVarP(&srv.Config.Etcd.Name, "etcd.name", "", srv.Config.Etcd.Name, "Name of node in Etcd.")
|
||||
flags.StringVarP(&srv.Config.Etcd.Dir, "etcd.dir", "", srv.Config.Etcd.Dir, "Directory to use for Etcd.")
|
||||
flags.StringVarP(&srv.Config.Etcd.LClientURL, "etcd.listen-client-addr", "", srv.Config.Etcd.LClientURL, "Listen client address.")
|
||||
flags.StringVarP(&srv.Config.Etcd.AClientURL, "etcd.advertise-client-addr", "", srv.Config.Etcd.AClientURL, "Advertise client address.")
|
||||
flags.StringVarP(&srv.Config.Etcd.LPeerURL, "etcd.listen-peer-addr", "", srv.Config.Etcd.LPeerURL, "Listen peer address.")
|
||||
flags.StringVarP(&srv.Config.Etcd.APeerURL, "etcd.advertise-peer-addr", "", srv.Config.Etcd.APeerURL, "Advertise peer address.")
|
||||
flags.StringVarP(&srv.Config.Etcd.ClusterURL, "etcd.cluster-url", "", srv.Config.Etcd.ClusterURL, "Cluster URL to join.")
|
||||
flags.StringVarP(&srv.Config.Etcd.ClusterName, "etcd.cluster-name", "", srv.Config.Etcd.ClusterName, "Cluster name.")
|
||||
flags.StringVarP(&srv.Config.Etcd.InitCluster, "etcd.initial-cluster", "", srv.Config.Etcd.InitCluster, "Initial cluster name1=apurl1,name2=apurl2")
|
||||
// Etcd.Name used Config.Name for it's value.
|
||||
// Etcd.Dir defaults to a directory under the pilosa data directory.
|
||||
flags.StringVar(&srv.Config.Etcd.LClientURL, "etcd.listen-client-address", srv.Config.Etcd.LClientURL, "Listen client address.")
|
||||
flags.StringVar(&srv.Config.Etcd.AClientURL, "etcd.advertise-client-address", srv.Config.Etcd.AClientURL, "Advertise client address. If not provided, uses the listen client address.")
|
||||
flags.StringVar(&srv.Config.Etcd.LPeerURL, "etcd.listen-peer-address", srv.Config.Etcd.LPeerURL, "Listen peer address.")
|
||||
flags.StringVar(&srv.Config.Etcd.APeerURL, "etcd.advertise-peer-address", srv.Config.Etcd.APeerURL, "Advertise peer address. If not provided, uses the listen peer address.")
|
||||
flags.StringVar(&srv.Config.Etcd.ClusterURL, "etcd.cluster-url", srv.Config.Etcd.ClusterURL, "Cluster URL to join.")
|
||||
// Etcd.ClusterName uses Cluster.Name for its value.
|
||||
flags.StringVar(&srv.Config.Etcd.InitCluster, "etcd.initial-cluster", srv.Config.Etcd.InitCluster, "Initial cluster name1=apurl1,name2=apurl2")
|
||||
|
||||
// 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.")
|
||||
flags.DurationVar((*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, "Where to send stats: can be expvar (in-memory served at /debug/vars), prometheus, statsd or none.")
|
||||
flags.StringVarP(&srv.Config.Metric.Host, "metric.host", "", srv.Config.Metric.Host, "URI to send metrics when metric.service is statsd.")
|
||||
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.")
|
||||
flags.StringVar(&srv.Config.Metric.Service, "metric.service", srv.Config.Metric.Service, "Where to send stats: can be expvar (in-memory served at /debug/vars), prometheus, statsd or none.")
|
||||
flags.StringVar(&srv.Config.Metric.Host, "metric.host", srv.Config.Metric.Host, "URI to send metrics when metric.service is statsd.")
|
||||
flags.DurationVar((*time.Duration)(&srv.Config.Metric.PollInterval), "metric.poll-interval", (time.Duration)(srv.Config.Metric.PollInterval), "Polling interval metrics.")
|
||||
flags.BoolVar((&srv.Config.Metric.Diagnostics), "metric.diagnostics", srv.Config.Metric.Diagnostics, "Enabled diagnostics reporting.")
|
||||
|
||||
// Tracing
|
||||
flags.StringVarP(&srv.Config.Tracing.AgentHostPort, "tracing.agent-host-port", "", srv.Config.Tracing.AgentHostPort, "Jaeger agent host:port.")
|
||||
flags.StringVarP(&srv.Config.Tracing.SamplerType, "tracing.sampler-type", "", srv.Config.Tracing.SamplerType, "Jaeger sampler type (remote, const, probabilistic, ratelimiting) or 'off' to disable tracing completely.")
|
||||
flags.Float64VarP(&srv.Config.Tracing.SamplerParam, "tracing.sampler-param", "", srv.Config.Tracing.SamplerParam, "Jaeger sampler parameter.")
|
||||
flags.StringVar(&srv.Config.Tracing.AgentHostPort, "tracing.agent-host-port", srv.Config.Tracing.AgentHostPort, "Jaeger agent host:port.")
|
||||
flags.StringVar(&srv.Config.Tracing.SamplerType, "tracing.sampler-type", srv.Config.Tracing.SamplerType, "Jaeger sampler type (remote, const, probabilistic, ratelimiting) or 'off' to disable tracing completely.")
|
||||
flags.Float64Var(&srv.Config.Tracing.SamplerParam, "tracing.sampler-param", srv.Config.Tracing.SamplerParam, "Jaeger sampler parameter.")
|
||||
|
||||
// Profiling
|
||||
flags.IntVar(&srv.Config.Profile.BlockRate, "profile.block-rate", srv.Config.Profile.BlockRate, "Sampling rate for goroutine blocking profiler. One sample per <rate> ns.")
|
||||
|
|
@ -112,7 +111,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
|
|||
flags.BoolVar(&srv.Config.Storage.FsyncEnabled, "storage.fsync", true, "enable fsync fully safe flush-to-disk")
|
||||
|
||||
// RowcacheOn
|
||||
flags.BoolVarP((&srv.Config.RowcacheOn), "rowcache-on", "", srv.Config.RowcacheOn, "turn on the rowcache for all backends (may speed some queries)")
|
||||
flags.BoolVar((&srv.Config.RowcacheOn), "rowcache-on", srv.Config.RowcacheOn, "turn on the rowcache for all backends (may speed some queries)")
|
||||
|
||||
// RBF specific flags. See pilosa/rbf/cfg/cfg.go for definitions.
|
||||
srv.Config.RBFConfig.DefineFlags(flags)
|
||||
|
|
@ -125,5 +124,4 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
|
|||
flags.DurationVar((*time.Duration)(&srv.Config.Postgres.WriteTimeout), "postgres.write-timeout", time.Duration(srv.Config.Postgres.WriteTimeout), "Timeout for writes on a postgres connection. (set 0 to disable)")
|
||||
flags.Uint32Var(&srv.Config.Postgres.MaxStartupSize, "postgres.max-startup-size", srv.Config.Postgres.MaxStartupSize, "Maximum acceptable size of a postgres startup packet, in bytes. (set 0 to disable)")
|
||||
flags.Uint16Var(&srv.Config.Postgres.ConnectionLimit, "postgres.connection-limit", srv.Config.Postgres.ConnectionLimit, "Maximum number of simultaneous postgres connections to allow. (set 0 to disable)")
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -41,10 +41,10 @@ import (
|
|||
type Options struct {
|
||||
Name string `toml:"name"`
|
||||
Dir string `toml:"dir"`
|
||||
LClientURL string `toml:"listen-client-addr"`
|
||||
AClientURL string `toml:"advertise-client-addr"`
|
||||
LPeerURL string `toml:"listen-peer-addr"`
|
||||
APeerURL string `toml:"advertise-peer-addr"`
|
||||
LClientURL string `toml:"listen-client-address"`
|
||||
AClientURL string `toml:"advertise-client-address"`
|
||||
LPeerURL string `toml:"listen-peer-address"`
|
||||
APeerURL string `toml:"advertise-peer-address"`
|
||||
InitCluster string `toml:"initial-cluster"`
|
||||
ClusterURL string `toml:"cluster-url"`
|
||||
ClusterName string `toml:"cluster-name"`
|
||||
|
|
@ -127,9 +127,17 @@ func parseOptions(opt Options) *embed.Config {
|
|||
cfg.Dir = opt.Dir
|
||||
cfg.InitialClusterToken = opt.ClusterName
|
||||
cfg.LCUrls = types.MustNewURLs([]string{opt.LClientURL})
|
||||
cfg.ACUrls = types.MustNewURLs([]string{opt.AClientURL})
|
||||
if opt.AClientURL != "" {
|
||||
cfg.ACUrls = types.MustNewURLs([]string{opt.AClientURL})
|
||||
} else {
|
||||
cfg.ACUrls = cfg.LCUrls
|
||||
}
|
||||
cfg.LPUrls = types.MustNewURLs([]string{opt.LPeerURL})
|
||||
cfg.APUrls = types.MustNewURLs([]string{opt.APeerURL})
|
||||
if opt.APeerURL != "" {
|
||||
cfg.APUrls = types.MustNewURLs([]string{opt.APeerURL})
|
||||
} else {
|
||||
cfg.APUrls = cfg.LPUrls
|
||||
}
|
||||
|
||||
lps := make([]*net.TCPListener, len(opt.LPeerSocket))
|
||||
copy(lps, opt.LPeerSocket)
|
||||
|
|
|
|||
12
server.go
12
server.go
|
|
@ -62,8 +62,6 @@ type Server struct { // nolint: maligned
|
|||
diagnostics *diagnosticsCollector
|
||||
executor *executor
|
||||
executorPoolSize int
|
||||
hosts []string
|
||||
clusterDisabled bool
|
||||
serializer Serializer
|
||||
|
||||
// Distributed Consensus
|
||||
|
|
@ -281,16 +279,6 @@ func OptServerGRPCURI(uri *pnet.URI) ServerOption {
|
|||
}
|
||||
}
|
||||
|
||||
// OptServerClusterDisabled tells the server whether to use a static cluster with the
|
||||
// defined hosts. Mostly used for testing.
|
||||
func OptServerClusterDisabled(disabled bool, hosts []string) ServerOption {
|
||||
return func(s *Server) error {
|
||||
s.hosts = hosts
|
||||
s.clusterDisabled = disabled
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// OptServerClusterName sets the human-readable cluster name.
|
||||
func OptServerClusterName(name string) ServerOption {
|
||||
return func(s *Server) error {
|
||||
|
|
|
|||
|
|
@ -190,6 +190,8 @@ func TestClusterResize_AddNode(t *testing.T) {
|
|||
|
||||
m1.Config.Gossip.Port = portsCfg[0].Gossip.Port
|
||||
m1.Config.Etcd = portsCfg[0].Etcd
|
||||
m1.Config.Name = portsCfg[0].Name
|
||||
m1.Config.Cluster.Name = portsCfg[0].Cluster.Name
|
||||
m1.Config.BindGRPC = portsCfg[0].BindGRPC
|
||||
|
||||
return m1.Start()
|
||||
|
|
@ -247,6 +249,8 @@ func TestClusterResize_AddNode(t *testing.T) {
|
|||
|
||||
m1.Config.Gossip.Port = portsCfg[0].Gossip.Port
|
||||
m1.Config.Etcd = portsCfg[0].Etcd
|
||||
m1.Config.Name = portsCfg[0].Name
|
||||
m1.Config.Cluster.Name = portsCfg[0].Cluster.Name
|
||||
m1.Config.BindGRPC = portsCfg[0].BindGRPC
|
||||
|
||||
return m1.Start()
|
||||
|
|
@ -303,6 +307,8 @@ func TestClusterResize_AddNode(t *testing.T) {
|
|||
|
||||
m1.Config.Gossip.Port = portsCfg[0].Gossip.Port
|
||||
m1.Config.Etcd = portsCfg[0].Etcd
|
||||
m1.Config.Name = portsCfg[0].Name
|
||||
m1.Config.Cluster.Name = portsCfg[0].Cluster.Name
|
||||
m1.Config.BindGRPC = portsCfg[0].BindGRPC
|
||||
|
||||
return m1.Start()
|
||||
|
|
@ -365,6 +371,8 @@ func TestClusterResize_AddNode(t *testing.T) {
|
|||
|
||||
m1.Config.Gossip.Port = portsCfg[0].Gossip.Port
|
||||
m1.Config.Etcd = portsCfg[0].Etcd
|
||||
m1.Config.Name = portsCfg[0].Name
|
||||
m1.Config.Cluster.Name = portsCfg[0].Cluster.Name
|
||||
m1.Config.BindGRPC = portsCfg[0].BindGRPC
|
||||
|
||||
return m1.Start()
|
||||
|
|
@ -421,6 +429,8 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) {
|
|||
|
||||
m1.Config.Gossip.Port = portsCfg[0].Gossip.Port
|
||||
m1.Config.Etcd = portsCfg[0].Etcd
|
||||
m1.Config.Name = portsCfg[0].Name
|
||||
m1.Config.Cluster.Name = portsCfg[0].Cluster.Name
|
||||
m1.Config.BindGRPC = portsCfg[0].BindGRPC
|
||||
return m1.Start()
|
||||
}, 4, 10); err != nil {
|
||||
|
|
@ -479,6 +489,8 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) {
|
|||
|
||||
m1.Config.Gossip.Port = portsCfg[0].Gossip.Port
|
||||
m1.Config.Etcd = portsCfg[0].Etcd
|
||||
m1.Config.Name = portsCfg[0].Name
|
||||
m1.Config.Cluster.Name = portsCfg[0].Cluster.Name
|
||||
m1.Config.BindGRPC = portsCfg[0].BindGRPC
|
||||
return m1.Start()
|
||||
}, 4, 10); err != nil {
|
||||
|
|
@ -543,6 +555,8 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) {
|
|||
|
||||
m1.Config.Gossip.Port = portsCfg[0].Gossip.Port
|
||||
m1.Config.Etcd = portsCfg[0].Etcd
|
||||
m1.Config.Name = portsCfg[0].Name
|
||||
m1.Config.Cluster.Name = portsCfg[0].Cluster.Name
|
||||
m1.Config.BindGRPC = portsCfg[0].BindGRPC
|
||||
|
||||
errc := make(chan error, 1)
|
||||
|
|
@ -605,6 +619,8 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) {
|
|||
|
||||
m1.Config.Gossip.Port = portsCfg[0].Gossip.Port
|
||||
m1.Config.Etcd = portsCfg[0].Etcd
|
||||
m1.Config.Name = portsCfg[0].Name
|
||||
m1.Config.Cluster.Name = portsCfg[0].Cluster.Name
|
||||
m1.Config.BindGRPC = portsCfg[0].BindGRPC
|
||||
|
||||
errc := make(chan error, 1)
|
||||
|
|
|
|||
|
|
@ -53,6 +53,9 @@ type TLSConfig struct {
|
|||
|
||||
// Config represents the configuration for the command.
|
||||
type Config struct {
|
||||
// Name a unique name for this node in the cluster.
|
||||
Name string `toml:"name"`
|
||||
|
||||
// DataDir is the directory where Pilosa stores both indexed data and
|
||||
// running state such as cluster topology information.
|
||||
DataDir string `toml:"data-dir"`
|
||||
|
|
@ -120,12 +123,9 @@ type Config struct {
|
|||
ImportWorkerPoolSize int `toml:"-"`
|
||||
|
||||
Cluster struct {
|
||||
// Disabled controls whether clustering functionality is enabled.
|
||||
Disabled bool `toml:"disabled"`
|
||||
Coordinator bool `toml:"coordinator"`
|
||||
ReplicaN int `toml:"replicas"`
|
||||
Hosts []string `toml:"hosts"`
|
||||
Name string `toml:"name"`
|
||||
Coordinator bool `toml:"coordinator"`
|
||||
ReplicaN int `toml:"replicas"`
|
||||
Name string `toml:"name"`
|
||||
// This LongQueryTime is deprecated but still exists for backward compatibility
|
||||
LongQueryTime toml.Duration `toml:"long-query-time"`
|
||||
} `toml:"cluster"`
|
||||
|
|
@ -231,7 +231,6 @@ func (c *Config) MustValidate() {
|
|||
}
|
||||
}
|
||||
|
||||
// validate ...
|
||||
func (c *Config) validate() error {
|
||||
hostPort := []string{
|
||||
"Bind", c.Bind, // :10101
|
||||
|
|
@ -239,9 +238,9 @@ func (c *Config) validate() error {
|
|||
"Advertise", c.Advertise, // on hp = 'http://localhost:63002'
|
||||
"AdvertiseGRPC", c.AdvertiseGRPC, // on hp = 'http://localhost:63003'
|
||||
"Etcd.LClientURL", c.Etcd.LClientURL, // on hp = ':14000'
|
||||
//c.Etcd.AClientURL, // hardcoded to same as LClientURL
|
||||
"Etcd.AClientURL", c.Etcd.AClientURL, // ""
|
||||
"Etcd.LPeerURL", c.Etcd.LPeerURL, // ":"
|
||||
//c.Etcd.APeerURL, // hardcoded to same as LPeerURL
|
||||
"Etcd.APeerURL", c.Etcd.APeerURL, // ""
|
||||
"Etcd.ClusterURL", c.Etcd.ClusterURL,
|
||||
"Gossip.Port", fmt.Sprintf(":%v", c.Gossip.Port),
|
||||
"Gossip.AdvertisePort", fmt.Sprintf(":%v", c.Gossip.AdvertisePort),
|
||||
|
|
@ -290,6 +289,7 @@ func (c *Config) validate() error {
|
|||
// NewConfig returns an instance of Config with default options.
|
||||
func NewConfig() *Config {
|
||||
c := &Config{
|
||||
Name: "pilosa0",
|
||||
DataDir: "~/.pilosa",
|
||||
Bind: ":" + defaultBindPort,
|
||||
BindGRPC: ":" + defaultBindGRPCPort,
|
||||
|
|
@ -317,9 +317,8 @@ func NewConfig() *Config {
|
|||
}
|
||||
|
||||
// Cluster config.
|
||||
c.Cluster.Disabled = false
|
||||
c.Cluster.Name = "cluster0"
|
||||
c.Cluster.ReplicaN = 1
|
||||
c.Cluster.Hosts = []string{}
|
||||
c.Cluster.LongQueryTime = toml.Duration(-time.Minute) //TODO remove this once cluster.longQueryTime is fully deprecated
|
||||
|
||||
// Gossip config.
|
||||
|
|
@ -355,13 +354,14 @@ func NewConfig() *Config {
|
|||
c.Postgres.WriteTimeout = toml.Duration(10 * time.Second)
|
||||
// we don't really need a connection limit
|
||||
|
||||
c.Etcd.AClientURL = "http://localhost:10301"
|
||||
c.Etcd.AClientURL = ""
|
||||
c.Etcd.LClientURL = "http://localhost:10301"
|
||||
c.Etcd.APeerURL = "http://localhost:10401"
|
||||
c.Etcd.APeerURL = ""
|
||||
c.Etcd.LPeerURL = "http://localhost:10401"
|
||||
c.Etcd.Dir = ""
|
||||
c.Etcd.Name = "nodeName"
|
||||
c.Etcd.ClusterName = "clusterName"
|
||||
c.Etcd.Name = ""
|
||||
c.Etcd.ClusterName = ""
|
||||
c.Etcd.InitCluster = c.Name + "=" + c.Etcd.LPeerURL
|
||||
|
||||
return c
|
||||
}
|
||||
|
|
@ -372,34 +372,34 @@ func NewConfig() *Config {
|
|||
// completely empty, or have both a host part and a port part
|
||||
// separated by a colon. In the latter case either can be empty to
|
||||
// indicate it's left unspecified.
|
||||
func (cfg *Config) validateAddrs(ctx context.Context) error {
|
||||
func (c *Config) validateAddrs(ctx context.Context) error {
|
||||
// Validate the advertise address.
|
||||
advScheme, advHost, advPort, err := validateAdvertiseAddr(ctx, cfg.Advertise, cfg.Bind, defaultBindPort)
|
||||
advScheme, advHost, advPort, err := validateAdvertiseAddr(ctx, c.Advertise, c.Bind, defaultBindPort)
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "validating advertise address")
|
||||
}
|
||||
cfg.Advertise = schemeHostPortString(advScheme, advHost, advPort)
|
||||
c.Advertise = schemeHostPortString(advScheme, advHost, advPort)
|
||||
|
||||
// Validate the listen address.
|
||||
listenScheme, listenHost, listenPort, err := validateListenAddr(ctx, cfg.Bind, defaultBindPort)
|
||||
listenScheme, listenHost, listenPort, err := validateListenAddr(ctx, c.Bind, defaultBindPort)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "validating listen address")
|
||||
}
|
||||
cfg.Bind = schemeHostPortString(listenScheme, listenHost, listenPort)
|
||||
c.Bind = schemeHostPortString(listenScheme, listenHost, listenPort)
|
||||
|
||||
// Validate the gRPC advertise address.
|
||||
_, grpcAdvHost, grpcAdvPort, err := validateAdvertiseAddr(ctx, cfg.AdvertiseGRPC, cfg.BindGRPC, defaultBindGRPCPort)
|
||||
_, grpcAdvHost, grpcAdvPort, err := validateAdvertiseAddr(ctx, c.AdvertiseGRPC, c.BindGRPC, defaultBindGRPCPort)
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "validating grpc advertise address")
|
||||
}
|
||||
cfg.AdvertiseGRPC = schemeHostPortString("grpc", grpcAdvHost, grpcAdvPort)
|
||||
c.AdvertiseGRPC = schemeHostPortString("grpc", grpcAdvHost, grpcAdvPort)
|
||||
|
||||
// Validate the gRPC listen address.
|
||||
_, grpcListenHost, grpcListenPort, err := validateListenAddr(ctx, cfg.BindGRPC, defaultBindGRPCPort)
|
||||
_, grpcListenHost, grpcListenPort, err := validateListenAddr(ctx, c.BindGRPC, defaultBindGRPCPort)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "validating grpc listen address")
|
||||
}
|
||||
cfg.BindGRPC = schemeHostPortString("grpc", grpcListenHost, grpcListenPort)
|
||||
c.BindGRPC = schemeHostPortString("grpc", grpcListenHost, grpcListenPort)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -23,14 +23,6 @@ import (
|
|||
"github.com/pilosa/pilosa/v2/toml"
|
||||
)
|
||||
|
||||
func Test_NewConfig(t *testing.T) {
|
||||
c := server.NewConfig()
|
||||
|
||||
if c.Cluster.Disabled {
|
||||
t.Fatalf("unexpected Cluster.Disabled: %v", c.Cluster.Disabled)
|
||||
}
|
||||
}
|
||||
|
||||
func Test_ValidateConfig(t *testing.T) {
|
||||
c := server.NewConfig()
|
||||
c.MustValidate()
|
||||
|
|
|
|||
|
|
@ -393,6 +393,15 @@ func (m *Command) SetupServer() error {
|
|||
coordinatorOpt = pilosa.OptServerIsCoordinator(true)
|
||||
}
|
||||
|
||||
// Use other config parameters to set Etcd parameters which we don't want to
|
||||
// expose in the user-facing config.
|
||||
//
|
||||
// Use cluster.name for etcd.cluster-name
|
||||
m.Config.Etcd.ClusterName = m.Config.Cluster.Name
|
||||
//
|
||||
// Use name for etcd.name
|
||||
m.Config.Etcd.Name = m.Config.Name
|
||||
//
|
||||
// If an Etcd.Dir is not provided, nest a default under the pilosa data dir.
|
||||
if m.Config.Etcd.Dir == "" {
|
||||
path, err := expandDirName(m.Config.DataDir)
|
||||
|
|
@ -425,7 +434,6 @@ func (m *Command) SetupServer() error {
|
|||
pilosa.OptServerURI(advertiseURI),
|
||||
pilosa.OptServerGRPCURI(advertiseGRPCURI),
|
||||
pilosa.OptServerInternalClient(http.NewInternalClientFromURI(uri, c)),
|
||||
pilosa.OptServerClusterDisabled(m.Config.Cluster.Disabled, m.Config.Cluster.Hosts),
|
||||
pilosa.OptServerClusterName(m.Config.Cluster.Name),
|
||||
pilosa.OptServerSerializer(proto.Serializer{}),
|
||||
pilosa.OptServerStorageConfig(m.Config.Storage),
|
||||
|
|
@ -477,10 +485,6 @@ func (m *Command) SetupServer() error {
|
|||
|
||||
// setupNetworking sets up internode communication based on the configuration.
|
||||
func (m *Command) setupNetworking() error {
|
||||
if m.Config.Cluster.Disabled {
|
||||
return nil
|
||||
}
|
||||
|
||||
gossipPort, err := strconv.Atoi(m.Config.Gossip.Port)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "parsing port")
|
||||
|
|
|
|||
|
|
@ -17,12 +17,9 @@ package test
|
|||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
"math"
|
||||
"net"
|
||||
"path"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
|
@ -423,6 +420,8 @@ func (c *Cluster) Start() error {
|
|||
for i, cc := range c.Nodes {
|
||||
cc := cc
|
||||
cc.Config.Etcd = portsCfg[i].Etcd
|
||||
cc.Config.Name = portsCfg[i].Name
|
||||
cc.Config.Cluster.Name = portsCfg[i].Cluster.Name
|
||||
cc.Config.BindGRPC = portsCfg[i].BindGRPC
|
||||
|
||||
eg.Go(func() error {
|
||||
|
|
@ -554,17 +553,12 @@ func newCluster(tb testing.TB, size int, opts ...[]server.CommandOption) (*Clust
|
|||
}
|
||||
|
||||
cluster := &Cluster{Nodes: make([]*Command, size)}
|
||||
name := tb.Name()
|
||||
for i := 0; i < size; i++ {
|
||||
var commandOpts []server.CommandOption
|
||||
if len(opts) > 0 {
|
||||
commandOpts = opts[i%len(opts)]
|
||||
}
|
||||
m := NewCommandNode(tb, i == 0, commandOpts...)
|
||||
err := ioutil.WriteFile(path.Join(m.Config.DataDir, ".id"), []byte(name+"__"+strconv.Itoa(i)), 0600)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "writing node id")
|
||||
}
|
||||
cluster.Nodes[i] = m
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -52,6 +52,7 @@ func GenPortsConfig(ports []Ports) []*server.Config {
|
|||
clusterURLs := make([]string, len(ports))
|
||||
for i := range cfgs {
|
||||
name := fmt.Sprintf("server%d", i)
|
||||
clusterName := "cluster-abc123"
|
||||
|
||||
lsnC, portC := ports[i].LsnC, ports[i].PortC
|
||||
lClientURL := fmt.Sprintf("http://localhost:%d", portC)
|
||||
|
|
@ -59,19 +60,18 @@ func GenPortsConfig(ports []Ports) []*server.Config {
|
|||
lPeerURL := fmt.Sprintf("http://localhost:%d", portP)
|
||||
|
||||
discoDir := ""
|
||||
if d, err := ioutil.TempDir("/tmp", "disco."); err == nil {
|
||||
if d, err := ioutil.TempDir("", "disco."); err == nil {
|
||||
discoDir = d
|
||||
}
|
||||
|
||||
cfgs[i] = &server.Config{
|
||||
Name: name,
|
||||
Gossip: gossip.Config{
|
||||
Port: fmt.Sprint(ports[i].Gossip),
|
||||
},
|
||||
BindGRPC: fmt.Sprintf(":%d", ports[i].Grpc),
|
||||
Etcd: etcd.Options{
|
||||
Name: name,
|
||||
Dir: discoDir,
|
||||
ClusterName: "bartholemuuuuu",
|
||||
LClientURL: lClientURL,
|
||||
AClientURL: lClientURL,
|
||||
LPeerURL: lPeerURL,
|
||||
|
|
@ -81,10 +81,9 @@ func GenPortsConfig(ports []Ports) []*server.Config {
|
|||
LClientSocket: []*net.TCPListener{lsnC},
|
||||
},
|
||||
}
|
||||
cfgs[i].Cluster.Name = clusterName
|
||||
|
||||
clusterURLs[i] = fmt.Sprintf("%s=%s", name, lPeerURL)
|
||||
fmt.Printf("\ndebug test/disco.go: on i=%v, GenPortsConfig Gossip: %v, Etcd.Client: %v, Etcd.Peer: %v, BindGRPC: %v\n",
|
||||
i, ports[i].Gossip, portC, portP, ports[i].Grpc)
|
||||
}
|
||||
for i := range cfgs {
|
||||
cfgs[i].Etcd.InitCluster = strings.Join(clusterURLs, ",")
|
||||
|
|
|
|||
|
|
@ -96,7 +96,6 @@ func NewCommandNode(tb testing.TB, isCoordinator bool, opts ...server.CommandOpt
|
|||
// has been specified, it will override this one.
|
||||
opts = prependTestServerOpts(opts)
|
||||
m := newCommand(tb, opts...)
|
||||
m.Config.Cluster.Disabled = false
|
||||
m.Config.Cluster.Coordinator = isCoordinator
|
||||
return m
|
||||
}
|
||||
|
|
|
|||
|
|
@ -23,7 +23,7 @@ import (
|
|||
|
||||
// Node represents a node in the cluster.
|
||||
type Node struct {
|
||||
Mu sync.Mutex
|
||||
Mu sync.Mutex `json:"-"` // TODO: we really need to get rid of this
|
||||
|
||||
ID string `json:"id"`
|
||||
URI net.URI `json:"uri"`
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue