diff --git a/cmd/server_test.go b/cmd/server_test.go index ff19458a2..b99d88ab9 100644 --- a/cmd/server_test.go +++ b/cmd/server_test.go @@ -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") diff --git a/ctl/server.go b/ctl/server.go index e279230d3..1344e7d37 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -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 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)") - } diff --git a/etcd/embed.go b/etcd/embed.go index e778c3b8a..57492d3f5 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -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) diff --git a/server.go b/server.go index b3af0a97a..5338b56de 100644 --- a/server.go +++ b/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 { diff --git a/server/cluster_test.go b/server/cluster_test.go index a590da062..da6cf611f 100644 --- a/server/cluster_test.go +++ b/server/cluster_test.go @@ -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) diff --git a/server/config.go b/server/config.go index 105f45c5c..8cfdd9c30 100644 --- a/server/config.go +++ b/server/config.go @@ -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 } diff --git a/server/config_test.go b/server/config_test.go index ed0501c5e..db2b83b29 100644 --- a/server/config_test.go +++ b/server/config_test.go @@ -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() diff --git a/server/server.go b/server/server.go index 527404007..aa1ae1d82 100644 --- a/server/server.go +++ b/server/server.go @@ -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") diff --git a/test/cluster.go b/test/cluster.go index 445565ed2..3dcced632 100644 --- a/test/cluster.go +++ b/test/cluster.go @@ -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 } diff --git a/test/disco.go b/test/disco.go index 936026995..46328a24e 100644 --- a/test/disco.go +++ b/test/disco.go @@ -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, ",") diff --git a/test/pilosa.go b/test/pilosa.go index 7a3250ef7..81538b679 100644 --- a/test/pilosa.go +++ b/test/pilosa.go @@ -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 } diff --git a/topology/node.go b/topology/node.go index 816fa7545..e5bf0a51e 100644 --- a/topology/node.go +++ b/topology/node.go @@ -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"`