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

move pilosa.Config to pilosa/server.Config

Browse source 
step 1 of #1203

The Config object is really just a specification of the options to pilosa
server, so it makes sense to have it in that package.
This commit is contained in:
Matthew Jaffee 2018-04-19 14:51:42 -05:00
parent 5a740eb1d9
commit 876ed56e30
No known key found for this signature in database
GPG key ID: 51C676AF9FFCDB87
21 changed files with 308 additions and 307 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
broadcast_test.go
View file

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

5
cluster.go
View file

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

8
cmd/server_test.go
View file

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

226
config.go
View file

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

5
ctl/backup.go
View file

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

5
ctl/bench.go
View file

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

3
ctl/common.go
View file

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

3
ctl/config.go
View file

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

4
ctl/config_test.go
View file

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

5
ctl/export.go
View file

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

5
ctl/import.go
View file

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

5
ctl/restore.go
View file

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

105
gossip/gossip.go
View file

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

4
pilosa.go
View file

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

25
server.go
View file

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

4
server/cluster_test.go
View file

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

132
server/config.go Normal file
View file

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

17
config_test.go → server/config_test.go
View file

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

14
server/server.go
View file

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

5
server/server_test.go
View file

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

30
toml/toml.go Normal file
View file

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