diff --git a/broadcast_test.go b/broadcast_test.go index 8f0245c88..f4ff4c0fa 100644 --- a/broadcast_test.go +++ b/broadcast_test.go @@ -53,7 +53,10 @@ func testMessageMarshal(t *testing.T, m proto.Message) { // Ensure that BroadcastReceiver can register a BroadcastHandler. func TestBroadcast_BroadcastReceiver(t *testing.T) { - s := pilosa.NewServer() + s, err := pilosa.NewServer() + if err != nil { + t.Fatalf("getting new server: %v", err) + } sbr := NewSimpleBroadcastReceiver() sbh := NewSimpleBroadcastHandler() diff --git a/cluster.go b/cluster.go index 7e871698d..be668a6bc 100644 --- a/cluster.go +++ b/cluster.go @@ -40,9 +40,6 @@ const ( // DefaultPartitionN is the default number of partitions in a cluster. DefaultPartitionN = 256 - // DefaultReplicaN is the default number of replicas per partition. - DefaultReplicaN = 1 - // ClusterState represents the state returned in the /status endpoint. ClusterStateStarting = "STARTING" ClusterStateNormal = "NORMAL" @@ -276,7 +273,7 @@ func NewCluster() *Cluster { return &Cluster{ Hasher: &jmphasher{}, PartitionN: DefaultPartitionN, - ReplicaN: DefaultReplicaN, + ReplicaN: 1, EventReceiver: NopEventReceiver, joiningLeavingNodes: make(chan nodeAction, 10), // buffered channel diff --git a/cmd/server_test.go b/cmd/server_test.go index b1f39e921..e69ce1134 100644 --- a/cmd/server_test.go +++ b/cmd/server_test.go @@ -21,9 +21,9 @@ import ( "testing" "time" - "github.com/pilosa/pilosa" "github.com/pilosa/pilosa/cmd" _ "github.com/pilosa/pilosa/test" + "github.com/pilosa/pilosa/toml" ) func TestServerHelp(t *testing.T) { @@ -65,7 +65,7 @@ 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.LongQueryTime, pilosa.Duration(time.Second*90)) + v.Check(cmd.Server.Config.Cluster.LongQueryTime, toml.Duration(time.Second*90)) v.Check(cmd.Server.Config.MaxWritesPerRequest, 2000) return v.Error() }, @@ -86,7 +86,7 @@ func TestServerConfig(t *testing.T) { validation: func() error { v := validator{} v.Check(cmd.Server.Config.Cluster.Hosts, []string{"localhost:1110", "localhost:1111"}) - v.Check(cmd.Server.Config.AntiEntropy.Interval, pilosa.Duration(time.Minute*9)) + v.Check(cmd.Server.Config.AntiEntropy.Interval, toml.Duration(time.Minute*9)) return v.Error() }, }, @@ -113,7 +113,7 @@ 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, pilosa.Duration(time.Minute*11)) + v.Check(cmd.Server.Config.AntiEntropy.Interval, toml.Duration(time.Minute*11)) v.Check(cmd.Server.CPUProfile, profFile.Name()) v.Check(cmd.Server.CPUTime, time.Minute) v.Check(cmd.Server.Config.LogPath, logFile.Name()) diff --git a/config.go b/config.go deleted file mode 100644 index 483965916..000000000 --- a/config.go +++ /dev/null @@ -1,226 +0,0 @@ -// Copyright 2017 Pilosa Corp. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package pilosa - -import ( - "time" -) - -// Cluster types. -const ( - ClusterNone = "" - ClusterStatic = "static" - ClusterGossip = "gossip" -) - -// TLSConfig contains TLS configuration -type TLSConfig struct { - // CertificatePath contains the path to the certificate (.crt or .pem file) - CertificatePath string `toml:"certificate-path"` - // CertificateKeyPath contains the path to the certificate key (.key file) - CertificateKeyPath string `toml:"certificate-key-path"` - // SkipVerify disables verification for self-signed certificates - SkipVerify bool `toml:"skip-verify"` -} - -// Config represents the configuration for the command. -type Config struct { - // DataDir is the directory where Pilosa stores both indexed data and - // running state such as cluster topology information. - DataDir string `toml:"data-dir"` - // Bind is the host:port on which Pilosa will listen. - Bind string `toml:"bind"` - - // MaxWritesPerRequest 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 configures where Pilosa will write logs. - LogPath string `toml:"log-path"` - - // Verbose toggles verbose logging which can be useful for debugging. - Verbose bool `toml:"verbose"` - - // TLS - TLS TLSConfig - - 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"` - LongQueryTime Duration `toml:"long-query-time"` - } `toml:"cluster"` - - // Gossip config is based around memberlist.Config. - Gossip struct { - // Port indicates the port to which pilosa should bind for internal state sharing. - Port string `toml:"port"` - Seeds []string `toml:"seeds"` - Key string `toml:"key"` - // 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. - StreamTimeout Duration `toml:"stream-timeout"` - // 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. - SuspicionMult int `toml:"suspicion-mult"` - // 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. - PushPullInterval Duration `toml:"push-pull-interval"` - // 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. - ProbeInterval Duration `toml:"probe-interval"` - ProbeTimeout Duration `toml:"probe-timeout"` - - // Interval and Nodes are used to configure the gossip - // behavior of memberlist. - // - // Interval 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. - // - // Nodes is the number of random nodes to send gossip messages to - // per Interval. Increasing this number causes the gossip messages - // to propagate across the cluster more quickly at the expense of - // increased bandwidth. - // - // ToTheDeadTime 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. - Interval Duration `toml:"interval"` - Nodes int `toml:"nodes"` - ToTheDeadTime Duration `toml:"to-the-dead-time"` - } `toml:"gossip"` - - AntiEntropy struct { - Interval Duration `toml:"interval"` - } `toml:"anti-entropy"` - - Metric struct { - // Service can be statsd, expvar, or none. - Service string `toml:"service"` - // Host tells the statsd client where to write. - Host string `toml:"host"` - PollInterval Duration `toml:"poll-interval"` - // Diagnostics toggles sending some limited diagnostic information to - // Pilosa's developers. - Diagnostics bool `toml:"diagnostics"` - } `toml:"metric"` -} - -// NewConfig returns an instance of Config with default options. -func NewConfig() *Config { - c := &Config{ - DataDir: "~/.pilosa", - Bind: ":10101", - MaxWritesPerRequest: 5000, - // LogPath: "", - // Verbose: false, - TLS: TLSConfig{}, - } - - // Cluster config. - c.Cluster.Disabled = false - // c.Cluster.Coordinator = false - c.Cluster.ReplicaN = DefaultReplicaN - c.Cluster.Hosts = []string{} - c.Cluster.LongQueryTime = Duration(time.Minute) - - // Gossip config. - c.Gossip.Port = "14000" - // c.Gossip.Seeds = []string{} - // c.Gossip.Key = "" - c.Gossip.StreamTimeout = Duration(10 * time.Second) - c.Gossip.SuspicionMult = 4 - c.Gossip.PushPullInterval = Duration(30 * time.Second) - c.Gossip.ProbeInterval = Duration(1 * time.Second) - c.Gossip.ProbeTimeout = Duration(500 * time.Millisecond) - c.Gossip.Interval = Duration(200 * time.Millisecond) - c.Gossip.Nodes = 3 - c.Gossip.ToTheDeadTime = Duration(30 * time.Second) - - // AntiEntropy config. - c.AntiEntropy.Interval = Duration(10 * time.Minute) - - // Metric config. - c.Metric.Service = "none" - // c.Metric.Host = "" - c.Metric.PollInterval = Duration(0 * time.Minute) - c.Metric.Diagnostics = true - - return c -} - -// Validate that all configuration permutations are compatible with each other. -func (c *Config) Validate() error { - if !c.Cluster.Disabled && len(c.Cluster.Hosts) > 0 { - return ErrConfigClusterEnabledHosts - } - return nil -} - -// Duration is a TOML wrapper type for time.Duration. -type Duration time.Duration - -// String returns the string representation of the duration. -func (d Duration) String() string { return time.Duration(d).String() } - -// UnmarshalText parses a TOML value into a duration value. -func (d *Duration) UnmarshalText(text []byte) error { - v, err := time.ParseDuration(string(text)) - if err != nil { - return err - } - - *d = Duration(v) - return nil -} - -// MarshalText writes duration value in text format. -func (d Duration) MarshalText() (text []byte, err error) { - return []byte(d.String()), nil -} - -// MarshalTOML write duration into valid TOML. -func (d Duration) MarshalTOML() ([]byte, error) { - return []byte(d.String()), nil -} diff --git a/ctl/backup.go b/ctl/backup.go index 8e725b41e..1b16391e0 100644 --- a/ctl/backup.go +++ b/ctl/backup.go @@ -21,6 +21,7 @@ import ( "os" "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/server" ) // BackupCommand represents a command for backing up a view. @@ -39,7 +40,7 @@ type BackupCommand struct { // Standard input/output *pilosa.CmdIO - TLS pilosa.TLSConfig + TLS server.TLSConfig } // NewBackupCommand returns a new instance of BackupCommand. @@ -88,6 +89,6 @@ func (cmd *BackupCommand) TLSHost() string { return cmd.Host } -func (cmd *BackupCommand) TLSConfiguration() pilosa.TLSConfig { +func (cmd *BackupCommand) TLSConfiguration() server.TLSConfig { return cmd.TLS } diff --git a/ctl/bench.go b/ctl/bench.go index 01e07cc14..9e37fb704 100644 --- a/ctl/bench.go +++ b/ctl/bench.go @@ -24,6 +24,7 @@ import ( "github.com/pilosa/pilosa" "github.com/pilosa/pilosa/internal" + "github.com/pilosa/pilosa/server" ) // BenchCommand represents a command for benchmarking index operations. @@ -42,7 +43,7 @@ type BenchCommand struct { // Standard input/output *pilosa.CmdIO - TLS pilosa.TLSConfig + TLS server.TLSConfig } // NewBenchCommand returns a new instance of BenchCommand. @@ -110,6 +111,6 @@ func (cmd *BenchCommand) TLSHost() string { return cmd.Host } -func (cmd *BenchCommand) TLSConfiguration() pilosa.TLSConfig { +func (cmd *BenchCommand) TLSConfiguration() server.TLSConfig { return cmd.TLS } diff --git a/ctl/common.go b/ctl/common.go index 9bba68657..6ebcd3132 100644 --- a/ctl/common.go +++ b/ctl/common.go @@ -18,13 +18,14 @@ import ( "crypto/tls" "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/server" "github.com/spf13/pflag" ) // CommandWithTLSSupport is the interface for commands which has TLS settings type CommandWithTLSSupport interface { TLSHost() string - TLSConfiguration() pilosa.TLSConfig + TLSConfiguration() server.TLSConfig } // SetTLSConfig creates common TLS flags diff --git a/ctl/config.go b/ctl/config.go index 2047941c7..7277f74ea 100644 --- a/ctl/config.go +++ b/ctl/config.go @@ -21,12 +21,13 @@ import ( toml "github.com/pelletier/go-toml" "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/server" ) // ConfigCommand represents a command for printing a default config. type ConfigCommand struct { *pilosa.CmdIO - Config *pilosa.Config + Config *server.Config } // NewConfigCommand returns a new instance of ConfigCommand. diff --git a/ctl/config_test.go b/ctl/config_test.go index b54446e34..ca08c273b 100644 --- a/ctl/config_test.go +++ b/ctl/config_test.go @@ -22,7 +22,7 @@ import ( "strings" "testing" - "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/server" ) func TestConfigCommand_Run(t *testing.T) { @@ -30,7 +30,7 @@ func TestConfigCommand_Run(t *testing.T) { stdin := bytes.NewReader(rder) r, w, _ := os.Pipe() cm := NewConfigCommand(stdin, w, os.Stderr) - cm.Config = pilosa.NewConfig() + cm.Config = server.NewConfig() err := cm.Run(context.Background()) w.Close() diff --git a/ctl/export.go b/ctl/export.go index 84b91edec..8c219a078 100644 --- a/ctl/export.go +++ b/ctl/export.go @@ -21,6 +21,7 @@ import ( "os" "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/server" ) // ExportCommand represents a command for bulk exporting data from a server. @@ -38,7 +39,7 @@ type ExportCommand struct { // Standard input/output *pilosa.CmdIO - TLS pilosa.TLSConfig + TLS server.TLSConfig } // NewExportCommand returns a new instance of ExportCommand. @@ -114,6 +115,6 @@ func (cmd *ExportCommand) TLSHost() string { return cmd.Host } -func (cmd *ExportCommand) TLSConfiguration() pilosa.TLSConfig { +func (cmd *ExportCommand) TLSConfiguration() server.TLSConfig { return cmd.TLS } diff --git a/ctl/import.go b/ctl/import.go index 8103bf673..c3c117eb3 100644 --- a/ctl/import.go +++ b/ctl/import.go @@ -27,6 +27,7 @@ import ( "time" "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/server" ) // ImportCommand represents a command for bulk importing data. @@ -66,7 +67,7 @@ type ImportCommand struct { // Standard input/output *pilosa.CmdIO - TLS pilosa.TLSConfig + TLS server.TLSConfig } // NewImportCommand returns a new instance of ImportCommand. @@ -447,6 +448,6 @@ func (cmd *ImportCommand) TLSHost() string { return cmd.Host } -func (cmd *ImportCommand) TLSConfiguration() pilosa.TLSConfig { +func (cmd *ImportCommand) TLSConfiguration() server.TLSConfig { return cmd.TLS } diff --git a/ctl/restore.go b/ctl/restore.go index 38528c2b5..89ee8ac8d 100644 --- a/ctl/restore.go +++ b/ctl/restore.go @@ -21,6 +21,7 @@ import ( "os" "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/server" ) // RestoreCommand represents a command for restoring a frame from a backup. @@ -39,7 +40,7 @@ type RestoreCommand struct { // Standard input/output *pilosa.CmdIO - TLS pilosa.TLSConfig + TLS server.TLSConfig } // NewRestoreCommand returns a new instance of RestoreCommand. @@ -81,6 +82,6 @@ func (cmd *RestoreCommand) TLSHost() string { return cmd.Host } -func (cmd *RestoreCommand) TLSConfiguration() pilosa.TLSConfig { +func (cmd *RestoreCommand) TLSConfiguration() server.TLSConfig { return cmd.TLS } diff --git a/gossip/gossip.go b/gossip/gossip.go index a27da24c8..8ad00a8a7 100644 --- a/gossip/gossip.go +++ b/gossip/gossip.go @@ -29,6 +29,7 @@ import ( "github.com/hashicorp/memberlist" "github.com/pilosa/pilosa" "github.com/pilosa/pilosa/internal" + "github.com/pilosa/pilosa/toml" "github.com/pkg/errors" ) @@ -167,7 +168,7 @@ func WithLogger(logger *log.Logger) func(*GossipMemberSet) error { } // NewGossipMemberSet returns a new instance of GossipMemberSet based on options. -func NewGossipMemberSet(name string, cfg *pilosa.Config, server *pilosa.Server, options ...GossipMemberSetOption) (*GossipMemberSet, error) { +func NewGossipMemberSet(name string, host string, cfg Config, server *pilosa.Server, options ...GossipMemberSetOption) (*GossipMemberSet, error) { g := &GossipMemberSet{ Logger: server.Logger, @@ -181,17 +182,11 @@ func NewGossipMemberSet(name string, cfg *pilosa.Config, server *pilosa.Server, } if g.transport == nil { - port, err := strconv.Atoi(cfg.Gossip.Port) + port, err := strconv.Atoi(cfg.Port) if err != nil { return nil, fmt.Errorf("convert port: %s", err) } - bindURI, err := pilosa.NewURIFromAddress(cfg.Bind) - if err != nil { - return nil, fmt.Errorf("getting uri from bind address: %s", err) - } - host := bindURI.Host() - // Set up the transport. transport, err := NewTransport(host, port, g.logger) if err != nil { @@ -203,15 +198,10 @@ func NewGossipMemberSet(name string, cfg *pilosa.Config, server *pilosa.Server, port := g.transport.Net.GetAutoBindPort() - bindURI, err := pilosa.NewURIFromAddress(cfg.Bind) - if err != nil { - return nil, fmt.Errorf("getting uri from bind address (with transport): %s", err) - } - host := bindURI.Host() - var gossipKey []byte - if cfg.Gossip.Key != "" { - gossipKey, err = ioutil.ReadFile(cfg.Gossip.Key) + var err error + if cfg.Key != "" { + gossipKey, err = ioutil.ReadFile(cfg.Key) if err != nil { return nil, fmt.Errorf("reading gossip key: %s", err) } @@ -226,14 +216,14 @@ func NewGossipMemberSet(name string, cfg *pilosa.Config, server *pilosa.Server, conf.AdvertisePort = port 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.Nodes - conf.GossipInterval = time.Duration(cfg.Gossip.Interval) - conf.GossipToTheDeadTime = time.Duration(cfg.Gossip.ToTheDeadTime) + conf.TCPTimeout = time.Duration(cfg.StreamTimeout) + conf.SuspicionMult = cfg.SuspicionMult + conf.PushPullInterval = time.Duration(cfg.PushPullInterval) + conf.ProbeTimeout = time.Duration(cfg.ProbeTimeout) + conf.ProbeInterval = time.Duration(cfg.ProbeInterval) + conf.GossipNodes = cfg.Nodes + conf.GossipInterval = time.Duration(cfg.Interval) + conf.GossipToTheDeadTime = time.Duration(cfg.ToTheDeadTime) // conf.Delegate = g conf.SecretKey = gossipKey @@ -242,7 +232,7 @@ func NewGossipMemberSet(name string, cfg *pilosa.Config, server *pilosa.Server, g.config = &gossipConfig{ memberlistConfig: conf, - gossipSeeds: cfg.Gossip.Seeds, + gossipSeeds: cfg.Seeds, } g.statusHandler = server @@ -526,3 +516,68 @@ func newTransport(conf *memberlist.Config) (*memberlist.NetTransport, error) { return nt, nil } + +// Config holds toml-friendly memberlist configuration. +type Config struct { + // Port indicates the port to which pilosa should bind for internal state sharing. + Port string `toml:"port"` + Seeds []string `toml:"seeds"` + Key string `toml:"key"` + // 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. + StreamTimeout toml.Duration `toml:"stream-timeout"` + // 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. + SuspicionMult int `toml:"suspicion-mult"` + // 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. + PushPullInterval toml.Duration `toml:"push-pull-interval"` + // 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. + ProbeInterval toml.Duration `toml:"probe-interval"` + ProbeTimeout toml.Duration `toml:"probe-timeout"` + + // Interval and Nodes are used to configure the gossip + // behavior of memberlist. + // + // Interval 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. + // + // Nodes is the number of random nodes to send gossip messages to + // per Interval. Increasing this number causes the gossip messages + // to propagate across the cluster more quickly at the expense of + // increased bandwidth. + // + // ToTheDeadTime 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. + Interval toml.Duration `toml:"interval"` + Nodes int `toml:"nodes"` + ToTheDeadTime toml.Duration `toml:"to-the-dead-time"` +} diff --git a/pilosa.go b/pilosa.go index 2d98b910a..62eb88113 100644 --- a/pilosa.go +++ b/pilosa.go @@ -68,10 +68,6 @@ var ( ErrQueryRequired = errors.New("query required") ErrTooManyWrites = errors.New("too many write commands") - ErrConfigClusterEnabledHosts = errors.New("providing hosts to a non-disabled cluster is not allowed") - ErrConfigClusterTypeInvalid = errors.New("invalid cluster type") - ErrConfigHostsMissing = errors.New("missing bind address in cluster hosts") - ErrClusterDoesNotOwnSlice = errors.New("cluster does not own slice") ErrNodeIDNotExists = errors.New("node with provided ID does not exist") diff --git a/server.go b/server.go index f28ca3864..6d25e3320 100644 --- a/server.go +++ b/server.go @@ -17,7 +17,6 @@ package pilosa import ( "context" "crypto/tls" - "errors" "fmt" "net" "net/http" @@ -31,6 +30,7 @@ import ( "github.com/gogo/protobuf/proto" "github.com/pilosa/pilosa/internal" + "github.com/pkg/errors" "golang.org/x/sync/errgroup" ) @@ -45,6 +45,16 @@ var _ Broadcaster = &Server{} var _ BroadcastHandler = &Server{} var _ StatusHandler = &Server{} +// ServerOption is a functional option type for pilosa.Server +type ServerOption func(s *Server) error + +func OptServerLogger(l Logger) ServerOption { + return func(s *Server) error { + s.Logger = l + return nil + } +} + // Server represents a holder wrapped by a running HTTP server. type Server struct { ln net.Listener @@ -90,7 +100,7 @@ type Server struct { } // NewServer returns a new instance of Server. -func NewServer() *Server { +func NewServer(opts ...ServerOption) (*Server, error) { s := &Server{ closing: make(chan struct{}), @@ -107,16 +117,23 @@ func NewServer() *Server { NewAttrStore: NewNopAttrStore, - AntiEntropyInterval: time.Duration(NewConfig().AntiEntropy.Interval), + AntiEntropyInterval: time.Minute * 10, MetricInterval: 0, DiagnosticInterval: 0, Logger: NopLogger, } + for _, opt := range opts { + err := opt(s) + if err != nil { + return nil, errors.Wrap(err, "applying option") + } + } + s.Handler.API = NewAPI() s.Handler.API.Holder = s.Holder - return s + return s, nil } // Open opens and initializes the server. diff --git a/server/cluster_test.go b/server/cluster_test.go index 7f4886612..5b0321285 100644 --- a/server/cluster_test.go +++ b/server/cluster_test.go @@ -55,7 +55,7 @@ func TestMain_SendReceiveMessage(t *testing.T) { m0.Server.Cluster.Coordinator = m0.Server.NodeID m0.Server.Cluster.Topology = &pilosa.Topology{NodeIDs: []string{m0.Server.NodeID, m1.Server.NodeID}} m0.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver(m0.Server.Logger) - gossipMemberSet0, err := gossip.NewGossipMemberSet(m0.Server.URI.HostPort(), m0.Config, m0.Server) + gossipMemberSet0, err := gossip.NewGossipMemberSet(m0.Server.URI.HostPort(), m0.Server.URI.Host(), m0.Config.Gossip, m0.Server) if err != nil { t.Fatal(err) } @@ -82,7 +82,7 @@ func TestMain_SendReceiveMessage(t *testing.T) { m1.Server.Cluster.Coordinator = m0.Server.NodeID m1.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver(m1.Server.Logger) - gossipMemberSet1, err := gossip.NewGossipMemberSet(m1.Server.URI.HostPort(), m1.Config, m1.Server) + gossipMemberSet1, err := gossip.NewGossipMemberSet(m1.Server.URI.HostPort(), m1.Server.URI.Host(), m1.Config.Gossip, m1.Server) if err != nil { t.Fatal(err) } diff --git a/server/config.go b/server/config.go new file mode 100644 index 000000000..2e2feeeb2 --- /dev/null +++ b/server/config.go @@ -0,0 +1,132 @@ +// Copyright 2017 Pilosa Corp. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package server + +import ( + "time" + + "github.com/pilosa/pilosa/gossip" + "github.com/pilosa/pilosa/toml" +) + +// Cluster types. +const ( + ClusterNone = "" + ClusterStatic = "static" + ClusterGossip = "gossip" +) + +// TLSConfig contains TLS configuration +type TLSConfig struct { + // CertificatePath contains the path to the certificate (.crt or .pem file) + CertificatePath string `toml:"certificate-path"` + // CertificateKeyPath contains the path to the certificate key (.key file) + CertificateKeyPath string `toml:"certificate-key-path"` + // SkipVerify disables verification for self-signed certificates + SkipVerify bool `toml:"skip-verify"` +} + +// Config represents the configuration for the command. +type Config struct { + // DataDir is the directory where Pilosa stores both indexed data and + // running state such as cluster topology information. + DataDir string `toml:"data-dir"` + // Bind is the host:port on which Pilosa will listen. + Bind string `toml:"bind"` + + // MaxWritesPerRequest 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 configures where Pilosa will write logs. + LogPath string `toml:"log-path"` + + // Verbose toggles verbose logging which can be useful for debugging. + Verbose bool `toml:"verbose"` + + // TLS + TLS TLSConfig + + 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"` + LongQueryTime toml.Duration `toml:"long-query-time"` + } `toml:"cluster"` + + // Gossip config is based around memberlist.Config. + Gossip gossip.Config `toml:"gossip"` + + AntiEntropy struct { + Interval toml.Duration `toml:"interval"` + } `toml:"anti-entropy"` + + Metric struct { + // Service can be statsd, expvar, or none. + Service string `toml:"service"` + // Host tells the statsd client where to write. + Host string `toml:"host"` + PollInterval toml.Duration `toml:"poll-interval"` + // Diagnostics toggles sending some limited diagnostic information to + // Pilosa's developers. + Diagnostics bool `toml:"diagnostics"` + } `toml:"metric"` +} + +// NewConfig returns an instance of Config with default options. +func NewConfig() *Config { + c := &Config{ + DataDir: "~/.pilosa", + Bind: ":10101", + MaxWritesPerRequest: 5000, + // LogPath: "", + // Verbose: false, + TLS: TLSConfig{}, + } + + // Cluster config. + c.Cluster.Disabled = false + // c.Cluster.Coordinator = false + c.Cluster.ReplicaN = 1 + c.Cluster.Hosts = []string{} + c.Cluster.LongQueryTime = toml.Duration(time.Minute) + + // Gossip config. + c.Gossip.Port = "14000" + // c.Gossip.Seeds = []string{} + // c.Gossip.Key = "" + c.Gossip.StreamTimeout = toml.Duration(10 * time.Second) + c.Gossip.SuspicionMult = 4 + c.Gossip.PushPullInterval = toml.Duration(30 * time.Second) + c.Gossip.ProbeInterval = toml.Duration(1 * time.Second) + c.Gossip.ProbeTimeout = toml.Duration(500 * time.Millisecond) + c.Gossip.Interval = toml.Duration(200 * time.Millisecond) + c.Gossip.Nodes = 3 + c.Gossip.ToTheDeadTime = toml.Duration(30 * time.Second) + + // AntiEntropy config. + c.AntiEntropy.Interval = toml.Duration(10 * time.Minute) + + // Metric config. + c.Metric.Service = "none" + // c.Metric.Host = "" + c.Metric.PollInterval = toml.Duration(0 * time.Minute) + c.Metric.Diagnostics = true + + return c +} diff --git a/config_test.go b/server/config_test.go similarity index 78% rename from config_test.go rename to server/config_test.go index c07486ddb..2027ef35a 100644 --- a/config_test.go +++ b/server/config_test.go @@ -12,34 +12,27 @@ // See the License for the specific language governing permissions and // limitations under the License. -package pilosa_test +package server_test import ( "reflect" "testing" "time" - "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/server" + "github.com/pilosa/pilosa/toml" ) func Test_NewConfig(t *testing.T) { - c := pilosa.NewConfig() + c := server.NewConfig() if c.Cluster.Disabled { t.Fatalf("unexpected Cluster.Disabled: %v", c.Cluster.Disabled) } - - // Ensure that hosts can't be specificed on a non-disabled cluster. - c.Cluster.Hosts = []string{c.Bind, "localhost:10102"} - - // Change cluster type from the default (gossip) to an invalid string. - if err := c.Validate(); err != pilosa.ErrConfigClusterEnabledHosts { - t.Fatal(err) - } } func TestDuration(t *testing.T) { - d := pilosa.Duration(time.Second * 182) + d := toml.Duration(time.Second * 182) if d.String() != "3m2s" { t.Fatalf("Unexpected time Duration %s", d) } diff --git a/server/server.go b/server/server.go index 7f66d31ed..8f4ab6147 100644 --- a/server/server.go +++ b/server/server.go @@ -56,7 +56,7 @@ type Command struct { Server *pilosa.Server // Configuration. - Config *pilosa.Config + Config *Config // Profiling options. CPUProfile string @@ -80,9 +80,10 @@ type Command struct { // NewCommand returns a new instance of Main. func NewCommand(stdin io.Reader, stdout, stderr io.Writer) *Command { + s, _ := pilosa.NewServer() return &Command{ - Server: pilosa.NewServer(), - Config: pilosa.NewConfig(), + Server: s, + Config: NewConfig(), CmdIO: pilosa.NewCmdIO(stdin, stdout, stderr), @@ -150,11 +151,6 @@ func (m *Command) SetupLogger() error { // SetupServer uses the cluster configuration to set up this server. func (m *Command) SetupServer() error { - err := m.Config.Validate() - if err != nil { - return err - } - m.Server.Handler.Logger = m.Server.Logger m.Server.Holder.Logger = m.Server.Logger m.Server.Holder.Stats.SetLogger(m.Server.Logger) @@ -278,7 +274,7 @@ func (m *Command) SetupNetworking() error { } m.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver(m.Server.Logger) - gossipMemberSet, err := gossip.NewGossipMemberSet(m.Server.NodeID, m.Config, m.Server, gossip.WithLogger(m.logger), gossip.WithTransport(transport)) + gossipMemberSet, err := gossip.NewGossipMemberSet(m.Server.NodeID, m.Server.URI.Host(), m.Config.Gossip, m.Server, gossip.WithLogger(m.logger), gossip.WithTransport(transport)) if err != nil { return err } diff --git a/server/server_test.go b/server/server_test.go index 497ffd7da..532257c13 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -30,6 +30,7 @@ import ( "github.com/BurntSushi/toml" "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/server" "github.com/pilosa/pilosa/test" ) @@ -489,8 +490,8 @@ func GenerateSetCommands(n int, rand *rand.Rand) []SetCommand { } // ParseConfig parses s into a Config. -func ParseConfig(s string) (pilosa.Config, error) { - var c pilosa.Config +func ParseConfig(s string) (server.Config, error) { + var c server.Config _, err := toml.Decode(s, &c) return c, err } diff --git a/toml/toml.go b/toml/toml.go new file mode 100644 index 000000000..5193ad787 --- /dev/null +++ b/toml/toml.go @@ -0,0 +1,30 @@ +package toml + +import "time" + +// Duration is a TOML wrapper type for time.Duration. +type Duration time.Duration + +// String returns the string representation of the duration. +func (d Duration) String() string { return time.Duration(d).String() } + +// UnmarshalText parses a TOML value into a duration value. +func (d *Duration) UnmarshalText(text []byte) error { + v, err := time.ParseDuration(string(text)) + if err != nil { + return err + } + + *d = Duration(v) + return nil +} + +// MarshalText writes duration value in text format. +func (d Duration) MarshalText() (text []byte, err error) { + return []byte(d.String()), nil +} + +// MarshalTOML write duration into valid TOML. +func (d Duration) MarshalTOML() ([]byte, error) { + return []byte(d.String()), nil +}