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

refactoring pilosa/server

Browse source 
trying to separate internal an external concerns in pilosa.Server - it should
handle Cluster, Holder, etc. while pilosa/server handles things with external
deps - e.g. Logger, Stats, Handler, etc. Using functional options in
pilosa.Server now.
This commit is contained in:
Matthew Jaffee 2018-04-23 10:37:20 -05:00
parent b53db06a6e
commit 28acc29a10
No known key found for this signature in database
GPG key ID: 51C676AF9FFCDB87
16 changed files with 311 additions and 339 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
broadcast_test.go
View file

@ -15,12 +15,16 @@
package pilosa_test
import (
"bytes"
"reflect"
"testing"
"io/ioutil"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/server"
)
// Ensure a message can be marshaled and unmarshaled.
@ -52,11 +56,12 @@ func testMessageMarshal(t *testing.T, m proto.Message) {
// Ensure that BroadcastReceiver can register a BroadcastHandler.
func TestBroadcast_BroadcastReceiver(t *testing.T) {
s, err := pilosa.NewServer()
com := server.NewCommand(bytes.NewBuffer([]byte{}), ioutil.Discard, ioutil.Discard)
err := com.SetupServer() // this test shouldn't need to import pilosa/server just to set up the Server, but it really shouldn't need to setup the Server at all. The Server should not be the implementation of Broadcast* TODO
if err != nil {
t.Fatalf("getting new server: %v", err)
t.Fatalf("setting up server: %v", err)
}
s := com.Server
sbr := NewSimpleBroadcastReceiver()
sbh := NewSimpleBroadcastHandler()

3
client_test.go
View file

@ -26,6 +26,7 @@ import (
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/pql"
"github.com/pilosa/pilosa/server"
"github.com/pilosa/pilosa/test"
)
@ -47,7 +48,7 @@ func createCluster(c *pilosa.Cluster) ([]*test.Server, []*test.Holder) {
var defaultClient *http.Client
func init() {
defaultClient = pilosa.GetHTTPClient(nil)
defaultClient = server.GetHTTPClient(nil)
}

54
cmd/server.go
View file

@ -15,17 +15,11 @@
package cmd
import (
"fmt"
"io"
"os"
"os/signal"
"runtime/pprof"
"syscall"
"time"
"github.com/pkg/errors"
"github.com/spf13/cobra"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/ctl"
"github.com/pilosa/pilosa/server"
)
@ -45,52 +39,10 @@ It will load existing data from the configured
directory and start listening for client connections
on the configured port.`,
RunE: func(cmd *cobra.Command, args []string) error {
// Set up the logger.
if err := Server.SetupLogger(); err != nil {
return fmt.Errorf("error setting up the logger: %v", err)
}
logger := Server.Server.Logger
logger.Printf("Pilosa %s, build time %s\n", pilosa.Version, pilosa.BuildTime)
// Start CPU profiling.
if Server.CPUProfile != "" {
f, err := os.Create(Server.CPUProfile)
if err != nil {
return fmt.Errorf("create cpu profile: %v", err)
}
defer f.Close()
fmt.Fprintln(Server.Stderr, "Starting cpu profile")
pprof.StartCPUProfile(f)
time.AfterFunc(Server.CPUTime, func() {
fmt.Fprintln(Server.Stderr, "Stopping cpu profile")
pprof.StopCPUProfile()
f.Close()
})
}
// Execute the program.
if err := Server.Run(); err != nil {
return fmt.Errorf("error running server: %v", err)
return errors.Wrap(err, "running server")
}
// First SIGKILL causes server to shut down gracefully.
c := make(chan os.Signal, 2)
signal.Notify(c, os.Interrupt, syscall.SIGTERM)
select {
case sig := <-c:
logger.Printf("Received %s; gracefully shutting down...\n", sig.String())
// Second signal causes a hard shutdown.
go func() { <-c; os.Exit(1) }()
if err := Server.Close(); err != nil {
return err
}
case <-Server.Done:
logger.Printf("Server closed externally")
}
return nil
return errors.Wrap(Server.Wait(), "waiting on Server")
},
}

12
cmd/server_test.go
View file

@ -37,8 +37,6 @@ func TestServerHelp(t *testing.T) {
func TestServerConfig(t *testing.T) {
actualDataDir, err := ioutil.TempDir("", "")
failErr(t, err, "making data dir")
profFile, err := ioutil.TempFile("", "")
failErr(t, err, "making temp file")
logFile, err := ioutil.TempFile("", "")
failErr(t, err, "making log file")
tests := []commandTest{
@ -93,7 +91,7 @@ func TestServerConfig(t *testing.T) {
// TEST 2
{
args: []string{"server", "--log-path", logFile.Name(), "--cluster.disabled", "true"},
env: map[string]string{"PILOSA_PROFILE_CPU_TIME": "1m"},
env: map[string]string{},
cfgFileContent: `
bind = "localhost:19444"
data-dir = "` + actualDataDir + `"
@ -103,9 +101,6 @@ func TestServerConfig(t *testing.T) {
]
[anti-entropy]
interval = "11m0s"
[profile]
cpu = "` + profFile.Name() + `"
cpu-time = "35s"
[metric]
service = "statsd"
host = "127.0.0.1:8125"
@ -114,8 +109,6 @@ func TestServerConfig(t *testing.T) {
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.CPUProfile, profFile.Name())
v.Check(cmd.Server.CPUTime, time.Minute)
v.Check(cmd.Server.Config.LogPath, logFile.Name())
v.Check(cmd.Server.Config.Metric.Service, "statsd")
v.Check(cmd.Server.Config.Metric.Host, "127.0.0.1:8125")
@ -147,6 +140,9 @@ func TestServerConfig(t *testing.T) {
case <-cmd.Server.Started:
case <-executed:
}
if execErr != nil {
t.Fatalf("executing server command: %v", execErr)
}
err := cmd.Server.Close()
failErr(t, err, "closing pilosa server command")
<-executed

2
ctl/common.go
View file

@ -49,7 +49,7 @@ func CommandClient(cmd CommandWithTLSSupport) (*pilosa.InternalHTTPClient, error
InsecureSkipVerify: tlsConfig.SkipVerify,
}
}
client, err := pilosa.NewInternalHTTPClient(cmd.TLSHost(), pilosa.GetHTTPClient(TLSConfig))
client, err := pilosa.NewInternalHTTPClient(cmd.TLSHost(), server.GetHTTPClient(TLSConfig))
if err != nil {
return nil, err
}

4
ctl/server.go
View file

@ -61,8 +61,4 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
flags.StringVarP(&srv.Config.Metric.Host, "metric.host", "", srv.Config.Metric.Host, "Default URI to send metrics.")
flags.DurationVarP((*time.Duration)(&srv.Config.Metric.PollInterval), "metric.poll-interval", "", (time.Duration)(srv.Config.Metric.PollInterval), "Polling interval metrics.")
flags.BoolVarP((&srv.Config.Metric.Diagnostics), "metric.diagnostics", "", srv.Config.Metric.Diagnostics, "Enabled diagnostics reporting.")
// CPU Profiling
flags.StringVarP(&srv.CPUProfile, "profile.cpu", "", "", "Where to store CPU profile.")
flags.DurationVarP(&srv.CPUTime, "profile.cpu-time", "", 30*time.Second, "CPU profile duration.")
}

7
handler.go
View file

@ -15,7 +15,6 @@
package pilosa
import (
"context"
"encoding/json"
"expvar"
"fmt"
@ -35,7 +34,6 @@ import (
"github.com/gogo/protobuf/proto"
"github.com/gorilla/mux"
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/pql"
"github.com/pkg/errors"
)
@ -45,11 +43,6 @@ type Handler struct {
FileSystem FileSystem
// The execution engine for running queries.
Executor interface {
Execute(context context.Context, index string, query *pql.Query, slices []uint64, opt *ExecOptions) ([]interface{}, error)
}
Logger Logger
// Keeps the query argument validators for each handler

1
holder.go
View file

@ -530,7 +530,6 @@ func (h *Holder) setFileLimit() {
func (h *Holder) loadNodeID() (string, error) {
idPath := path.Join(h.Path, "ID")
nodeID := ""
h.Logger.Printf("load NodeID: %s", idPath)
if err := os.MkdirAll(h.Path, 0777); err != nil {
return "", err

5
holder_test.go
View file

@ -25,6 +25,7 @@ import (
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/pql"
"github.com/pilosa/pilosa/server"
"github.com/pilosa/pilosa/test"
)
@ -367,7 +368,7 @@ func TestHolder_DeleteIndex(t *testing.T) {
// Ensure holder can sync with a remote holder.
func TestHolderSyncer_SyncHolder(t *testing.T) {
cluster := test.NewCluster(2)
client := pilosa.GetHTTPClient(nil)
client := server.GetHTTPClient(nil)
// Create a local holder.
hldr0 := test.MustOpenHolder()
defer hldr0.Close()
@ -451,7 +452,7 @@ func TestHolderSyncer_SyncHolder(t *testing.T) {
Holder: hldr0.Holder,
Node: cluster.Nodes[0],
Cluster: cluster,
RemoteClient: pilosa.GetHTTPClient(nil),
RemoteClient: server.GetHTTPClient(nil),
Stats: pilosa.NopStatsClient,
}

228
server.go
View file

@ -55,55 +55,154 @@ func OptServerLogger(l Logger) ServerOption {
}
}
func OptServerReplicaN(n int) ServerOption {
return func(s *Server) error {
s.Cluster.ReplicaN = n
return nil
}
}
func OptServerDataDir(dir string) ServerOption {
return func(s *Server) error {
s.Cluster.Path = dir
s.Holder.Path = dir
return nil
}
}
func OptServerAttrStoreFunc(af func(string) AttrStore) ServerOption {
return func(s *Server) error {
s.NewAttrStore = af
s.Holder.NewAttrStore = af
return nil
}
}
func OptServerAntiEntropyInterval(interval time.Duration) ServerOption {
return func(s *Server) error {
s.AntiEntropyInterval = interval
return nil
}
}
func OptServerLongQueryTime(dur time.Duration) ServerOption {
return func(s *Server) error {
s.Cluster.LongQueryTime = dur
return nil
}
}
func OptServerHandler(h *Handler) ServerOption {
return func(s *Server) error {
s.Handler = h
return nil
}
}
func OptServerMaxWritesPerRequest(n int) ServerOption {
return func(s *Server) error {
s.MaxWritesPerRequest = n
return nil
}
}
func OptServerMetricInterval(dur time.Duration) ServerOption {
return func(s *Server) error {
s.MetricInterval = dur
return nil
}
}
func OptServerSystemInfo(si SystemInfo) ServerOption {
return func(s *Server) error {
s.SystemInfo = si
return nil
}
}
func OptServerGCNotifier(gcn GCNotifier) ServerOption {
return func(s *Server) error {
s.GCNotifier = gcn
return nil
}
}
func OptServerRemoteClient(c *http.Client) ServerOption {
return func(s *Server) error {
s.RemoteClient = c
s.Cluster.RemoteClient = c
return nil
}
}
func OptServerStatsClient(sc StatsClient) ServerOption {
return func(s *Server) error {
s.Holder.Stats = sc
return nil
}
}
func OptServerDiagnosticsInterval(dur time.Duration) ServerOption {
return func(s *Server) error {
s.DiagnosticInterval = dur
return nil
}
}
func OptServerListener(ln net.Listener) ServerOption {
return func(s *Server) error {
s.ln = ln
return nil
}
}
func OptServerURI(uri *URI) ServerOption {
return func(s *Server) error {
s.URI = *uri
return nil
}
}
// Server represents a holder wrapped by a running HTTP server.
type Server struct {
ln net.Listener
// Close management.
wg sync.WaitGroup
closing chan struct{}
// Data storage and HTTP interface.
Holder *Holder
// Internal
Holder *Holder
Cluster *Cluster
diagnostics *DiagnosticsCollector
// External
Handler *Handler
Broadcaster Broadcaster
BroadcastReceiver BroadcastReceiver
Gossiper Gossiper
RemoteClient *http.Client
SystemInfo SystemInfo
GCNotifier GCNotifier
NewAttrStore func(string) AttrStore
Logger Logger
TLS *tls.Config
ln net.Listener
// Cluster configuration.
Network string
NodeID string
URI URI
Cluster *Cluster
diagnostics *DiagnosticsCollector
SystemInfo SystemInfo
GCNotifier GCNotifier
NewAttrStore func(string) AttrStore
// Background monitoring intervals.
NodeID string
URI URI
AntiEntropyInterval time.Duration
MetricInterval time.Duration
DiagnosticInterval time.Duration
// TLS configuration
TLS *tls.Config
// Misc options.
MaxWritesPerRequest int
Logger Logger
defaultClient InternalClient
}
// NewServer returns a new instance of Server.
func NewServer(opts ...ServerOption) (*Server, error) {
s := &Server{
closing: make(chan struct{}),
closing: make(chan struct{}),
Cluster: NewCluster(),
Holder: NewHolder(),
Handler: NewHandler(),
Broadcaster: NopBroadcaster,
@ -111,8 +210,6 @@ func NewServer(opts ...ServerOption) (*Server, error) {
diagnostics: NewDiagnosticsCollector(DefaultDiagnosticServer),
SystemInfo: NewNopSystemInfo(),
Network: "tcp",
GCNotifier: NopGCNotifier,
NewAttrStore: NewNopAttrStore,
@ -131,20 +228,25 @@ func NewServer(opts ...ServerOption) (*Server, error) {
}
}
s.Handler.API = NewAPI()
s.Handler.API.Holder = s.Holder
s.Holder.Logger = s.Logger
s.Holder.Stats.SetLogger(s.Logger)
s.Cluster.Logger = s.Logger
s.Cluster.Holder = s.Holder
s.Cluster.RemoteClient = s.RemoteClient
// update URI port with actual listener port. TODO this should probably be done outside of here.
if s.URI.Port() == 0 {
s.URI.SetPort(uint16(s.ln.Addr().(*net.TCPAddr).Port))
}
return s, nil
}
// Open opens and initializes the server.
func (s *Server) Open() error {
s.Handler.API.Logger = s.Logger // TODO do this in NewServer with functional options
s.Logger.Printf("open server")
// s.ln can be configured prior to Open() via s.OpenListener().
if s.ln == nil {
if err := s.OpenListener(); err != nil {
return err
}
return errors.New("Must pass a listener option to NewServer")
}
// Get or create NodeID.
@ -181,13 +283,13 @@ func (s *Server) Open() error {
s.Cluster.MaxWritesPerRequest = s.MaxWritesPerRequest
// Initialize HTTP handler.
s.Handler.API.Holder = s.Holder
s.Handler.API.Broadcaster = s.Broadcaster
s.Handler.API.BroadcastHandler = s
s.Handler.API.StatusHandler = s
s.Handler.API.URI = s.URI
s.Handler.API.Cluster = s.Cluster
s.Handler.API.Executor = e
s.Handler.Executor = e
// Initialize Holder.
s.Holder.Broadcaster = s.Broadcaster
@ -234,43 +336,6 @@ func (s *Server) Open() error {
return nil
}
// OpenListener opens a listener for the Server.
func (s *Server) OpenListener() error {
s.Logger.Printf("open server listener: %s", s.URI)
if s.ln != nil {
return fmt.Errorf("a listener already exists for server: %s", s.URI)
}
var ln net.Listener
var err error
// If bind URI has the https scheme, enable TLS
if s.URI.Scheme() == "https" && s.TLS != nil {
ln, err = tls.Listen("tcp", s.URI.HostPort(), s.TLS)
if err != nil {
return err
}
} else if s.URI.Scheme() == "http" {
// Open HTTP listener to determine port (if specified as :0).
ln, err = net.Listen(s.Network, s.URI.HostPort())
if err != nil {
return fmt.Errorf("net.Listen: %v", err)
}
} else {
return fmt.Errorf("unsupported scheme: %s", s.URI.Scheme())
}
s.ln = ln
if s.URI.Port() == 0 {
// If the port is 0, it is set automatically.
// Find out automatically set port and update the host.
s.URI.SetPort(uint16(s.ln.Addr().(*net.TCPAddr).Port))
}
return nil
}
// Close closes the server and waits for it to shutdown.
func (s *Server) Close() error {
// Notify goroutines to stop.
@ -311,25 +376,6 @@ func (s *Server) Addr() net.Addr {
}
return s.ln.Addr()
}
func GetHTTPClient(t *tls.Config) *http.Client {
transport := &http.Transport{
Proxy: http.ProxyFromEnvironment,
DialContext: (&net.Dialer{
Timeout: 30 * time.Second,
KeepAlive: 30 * time.Second,
DualStack: true,
}).DialContext,
MaxIdleConns: 1000,
MaxIdleConnsPerHost: 200,
IdleConnTimeout: 90 * time.Second,
TLSHandshakeTimeout: 10 * time.Second,
ExpectContinueTimeout: 1 * time.Second,
}
if t != nil {
transport.TLSClientConfig = t
}
return &http.Client{Transport: transport}
}
func (s *Server) monitorAntiEntropy() {
ticker := time.NewTicker(s.AntiEntropyInterval)

69
server/cluster_test.go
View file

@ -26,81 +26,16 @@ import (
"golang.org/x/sync/errgroup"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/gossip"
"github.com/pilosa/pilosa/test"
)
// Ensure program can send/receive broadcast messages.
func TestMain_SendReceiveMessage(t *testing.T) {
m0 := test.MustRunMain()
ms := test.MustRunMainWithCluster(t, 2)
m0, m1 := ms[0], ms[1]
defer m0.Close()
m1 := test.MustRunMain()
defer m1.Close()
// Update cluster config
m0.Server.Cluster.Nodes = []*pilosa.Node{
{ID: m0.Server.NodeID, URI: m0.Server.URI},
{ID: m1.Server.NodeID, URI: m1.Server.URI},
}
m1.Server.Cluster.Nodes = m0.Server.Cluster.Nodes
// Configure node0
// get the host portion of addr to use for binding
m0.Config.Gossip.Port = "0"
m0.Config.Gossip.Seeds = []string{}
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.Server.URI.Host(), m0.Config.Gossip, m0.Server)
if err != nil {
t.Fatal(err)
}
m0.Server.Cluster.MemberSet = gossipMemberSet0
m0.Server.Broadcaster = m0.Server
m0.Server.Gossiper = gossipMemberSet0
m0.Server.Handler.API.Broadcaster = m0.Server.Broadcaster
m0.Server.Holder.Broadcaster = m0.Server.Broadcaster
m0.Server.BroadcastReceiver = gossipMemberSet0
if err := m0.Server.BroadcastReceiver.Start(m0.Server); err != nil {
t.Fatal(err)
}
// Open Cluster management.
if err := m0.Server.Cluster.Open(); err != nil {
t.Fatal(err)
}
// Configure node1
// get the host portion of addr to use for binding
m1.Config.Gossip.Port = "0"
m1.Config.Gossip.Seeds = []string{gossipMemberSet0.GetBindAddr()}
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.Server.URI.Host(), m1.Config.Gossip, m1.Server)
if err != nil {
t.Fatal(err)
}
m1.Server.Cluster.MemberSet = gossipMemberSet1
m1.Server.Broadcaster = m1.Server
m1.Server.Gossiper = gossipMemberSet1
m1.Server.Handler.API.Broadcaster = m1.Server.Broadcaster
m1.Server.Holder.Broadcaster = m1.Server.Broadcaster
m1.Server.BroadcastReceiver = gossipMemberSet1
if err := m1.Server.BroadcastReceiver.Start(m1.Server); err != nil {
t.Fatal(err)
}
// Open Cluster management.
if err := m1.Server.Cluster.Open(); err != nil {
t.Fatal(err)
}
m0.Server.Cluster.SetState(pilosa.ClusterStateNormal)
m1.Server.Cluster.SetState(pilosa.ClusterStateNormal)

215
server/server.go
View file

@ -24,10 +24,14 @@ import (
"io"
"log"
"math/rand"
"net"
"net/http"
"os"
"os/signal"
"path/filepath"
"strconv"
"strings"
"syscall"
"time"
"crypto/tls"
@ -46,10 +50,10 @@ func init() {
rand.Seed(time.Now().UTC().UnixNano())
}
const (
// DefaultDataDir is the default data directory.
DefaultDataDir = "~/.pilosa"
)
type loggerLogger interface {
pilosa.Logger
Logger() *log.Logger
}
// Command represents the state of the pilosa server command.
type Command struct {
@ -58,10 +62,6 @@ type Command struct {
// Configuration.
Config *Config
// Profiling options.
CPUProfile string
CPUTime time.Duration
// Gossip transport
GossipTransport *gossip.Transport
@ -75,18 +75,12 @@ type Command struct {
// Passed to the Gossip implementation.
logOutput io.Writer
logger *log.Logger
logger loggerLogger
}
// NewCommand returns a new instance of Main.
func NewCommand(stdin io.Reader, stdout, stderr io.Writer) *Command {
s, err := pilosa.NewServer()
if err != nil {
panic(err)
}
return &Command{
Server: s,
Config: NewConfig(),
CmdIO: pilosa.NewCmdIO(stdin, stdout, stderr),
@ -97,7 +91,7 @@ func NewCommand(stdin io.Reader, stdout, stderr io.Writer) *Command {
}
// Run executes the pilosa server.
func (m *Command) Run(args ...string) (err error) {
func (m *Command) Run(args ...string) (err error) { // TODO args WTF
defer close(m.Started)
prefix := "~" + string(filepath.Separator)
if strings.HasPrefix(m.Config.DataDir, prefix) {
@ -125,76 +119,72 @@ func (m *Command) Run(args ...string) (err error) {
return fmt.Errorf("server.Open: %v", err)
}
m.Server.Logger.Printf("Listening as %s\n", m.Server.URI)
m.logger.Printf("Listening as %s\n", m.Server.URI)
return nil
}
// Wait waits for the server to be closed or interrupted.
func (m *Command) Wait() error {
// First SIGKILL causes server to shut down gracefully.
c := make(chan os.Signal, 2)
signal.Notify(c, os.Interrupt, syscall.SIGTERM)
select {
case sig := <-c:
m.logger.Printf("Received %s; gracefully shutting down...\n", sig.String())
// Second signal causes a hard shutdown.
go func() { <-c; os.Exit(1) }()
return errors.Wrap(m.Close(), "closing command")
case <-m.Done:
m.logger.Printf("Server closed externally")
return nil
}
}
// SetupLogger sets up the logger based on the configuration.
func (m *Command) SetupLogger() error {
func (m *Command) SetupLogger() (pilosa.Logger, error) {
if m.logger != nil {
return m.logger, nil
}
var err error
if m.Config.LogPath == "" {
m.logOutput = m.Stderr
} else {
m.logOutput, err = os.OpenFile(m.Config.LogPath, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
if err != nil {
return err
return nil, errors.Wrap(err, "opening file")
}
}
if m.Config.Verbose {
vbl := pilosa.NewVerboseLogger(m.logOutput)
m.logger = vbl.Logger()
m.Server.Logger = vbl
m.logger = pilosa.NewVerboseLogger(m.logOutput)
} else {
sl := pilosa.NewStandardLogger(m.logOutput)
m.logger = sl.Logger()
m.Server.Logger = sl
m.logger = pilosa.NewStandardLogger(m.logOutput)
}
return nil
return m.logger, nil
}
// SetupServer uses the cluster configuration to set up this server.
func (m *Command) SetupServer() error {
m.Server.Handler.Logger = m.Server.Logger
m.Server.Holder.Logger = m.Server.Logger
m.Server.Holder.Stats.SetLogger(m.Server.Logger)
if m.logger == nil {
_, err := m.SetupLogger()
if err != nil {
return errors.Wrap(err, "setting up logger")
}
}
handler := pilosa.NewHandler()
handler.Logger = m.logger
handler.FileSystem = &statik.FileSystem{}
handler.API = pilosa.NewAPI()
handler.API.Logger = m.logger
uri, err := pilosa.AddressWithDefaults(m.Config.Bind)
if err != nil {
return err
return errors.Wrap(err, "processing bind address")
}
m.Server.URI = *uri
cluster := pilosa.NewCluster()
cluster.ReplicaN = m.Config.Cluster.ReplicaN
cluster.Holder = m.Server.Holder
cluster.Logger = m.Server.Logger
m.Server.Cluster = cluster
// Configure data directory (for Cluster .topology)
m.Server.Cluster.Path = m.Config.DataDir
m.Server.NewAttrStore = boltdb.NewAttrStore
m.Server.Holder.NewAttrStore = boltdb.NewAttrStore
// Configure holder.
m.Server.Logger.Printf("Using data from: %s\n", m.Config.DataDir)
m.Server.Holder.Path = m.Config.DataDir
m.Server.MetricInterval = time.Duration(m.Config.Metric.PollInterval)
if m.Config.Metric.Diagnostics {
m.Server.DiagnosticInterval = time.Duration(DefaultDiagnosticsInterval)
}
m.Server.SystemInfo = gopsutil.NewSystemInfo()
m.Server.GCNotifier = gcnotify.NewActiveGCNotifier()
m.Server.Holder.Stats, err = NewStatsClient(m.Config.Metric.Service, m.Config.Metric.Host)
if err != nil {
return err
}
// Copy configuration flags.
m.Server.MaxWritesPerRequest = m.Config.MaxWritesPerRequest
// Setup TLS
var TLSConfig *tls.Config
@ -207,27 +197,73 @@ func (m *Command) SetupServer() error {
}
cert, err := tls.LoadX509KeyPair(m.Config.TLS.CertificatePath, m.Config.TLS.CertificateKeyPath)
if err != nil {
return err
return errors.Wrap(err, "load x509 key pair")
}
m.Server.TLS = &tls.Config{
TLSConfig = &tls.Config{
Certificates: []tls.Certificate{cert},
InsecureSkipVerify: m.Config.TLS.SkipVerify,
}
TLSConfig = m.Server.TLS
}
c := pilosa.GetHTTPClient(TLSConfig)
m.Server.RemoteClient = c
m.Server.Handler.API.RemoteClient = c
m.Server.Cluster.RemoteClient = c
// Statik file system.
m.Server.Handler.FileSystem = &statik.FileSystem{}
diagnosticsInterval := time.Duration(0)
if m.Config.Metric.Diagnostics {
diagnosticsInterval = time.Duration(DefaultDiagnosticsInterval)
}
// Set configuration options.
m.Server.AntiEntropyInterval = time.Duration(m.Config.AntiEntropy.Interval)
m.Server.Cluster.LongQueryTime = time.Duration(m.Config.Cluster.LongQueryTime)
return nil
statsClient, err := NewStatsClient(m.Config.Metric.Service, m.Config.Metric.Host)
if err != nil {
return errors.Wrap(err, "new stats client")
}
ln, err := getListener(*uri, TLSConfig)
if err != nil {
return errors.Wrap(err, "getting listener")
}
c := GetHTTPClient(TLSConfig)
handler.API.RemoteClient = c
m.Server, err = pilosa.NewServer(
pilosa.OptServerAntiEntropyInterval(time.Duration(m.Config.AntiEntropy.Interval)),
pilosa.OptServerLongQueryTime(time.Duration(m.Config.Cluster.LongQueryTime)),
pilosa.OptServerDataDir(m.Config.DataDir),
pilosa.OptServerReplicaN(m.Config.Cluster.ReplicaN),
pilosa.OptServerMaxWritesPerRequest(m.Config.MaxWritesPerRequest),
pilosa.OptServerMetricInterval(time.Duration(m.Config.Metric.PollInterval)),
pilosa.OptServerDiagnosticsInterval(diagnosticsInterval),
pilosa.OptServerLogger(m.logger),
pilosa.OptServerAttrStoreFunc(boltdb.NewAttrStore),
pilosa.OptServerHandler(handler),
pilosa.OptServerSystemInfo(gopsutil.NewSystemInfo()),
pilosa.OptServerGCNotifier(gcnotify.NewActiveGCNotifier()),
pilosa.OptServerStatsClient(statsClient),
pilosa.OptServerListener(ln),
pilosa.OptServerURI(uri),
pilosa.OptServerRemoteClient(c),
)
return errors.Wrap(err, "new server")
}
func GetHTTPClient(t *tls.Config) *http.Client {
transport := &http.Transport{
Proxy: http.ProxyFromEnvironment,
DialContext: (&net.Dialer{
Timeout: 30 * time.Second,
KeepAlive: 30 * time.Second,
DualStack: true,
}).DialContext,
MaxIdleConns: 1000,
MaxIdleConnsPerHost: 200,
IdleConnTimeout: 90 * time.Second,
TLSHandshakeTimeout: 10 * time.Second,
ExpectContinueTimeout: 1 * time.Second,
}
if t != nil {
transport.TLSClientConfig = t
}
return &http.Client{Transport: transport}
}
// SetupNetworking sets up internode communication based on the configuration.
@ -266,7 +302,7 @@ func (m *Command) SetupNetworking() error {
if m.GossipTransport != nil {
transport = m.GossipTransport
} else {
transport, err = gossip.NewTransport(gossipHost, gossipPort, m.logger)
transport, err = gossip.NewTransport(gossipHost, gossipPort, m.logger.Logger())
if err != nil {
return err
}
@ -277,8 +313,8 @@ func (m *Command) SetupNetworking() error {
m.Server.Cluster.Coordinator = m.Server.NodeID
}
m.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver(m.Server.Logger)
gossipMemberSet, err := gossip.NewGossipMemberSet(m.Server.NodeID, m.Server.URI.Host(), m.Config.Gossip, m.Server, gossip.WithLogger(m.logger), gossip.WithTransport(transport))
m.Server.Cluster.EventReceiver = gossip.NewGossipEventReceiver(m.logger)
gossipMemberSet, err := gossip.NewGossipMemberSet(m.Server.NodeID, m.Server.URI.Host(), m.Config.Gossip, m.Server, gossip.WithLogger(m.logger.Logger()), gossip.WithTransport(transport))
if err != nil {
return err
}
@ -318,3 +354,24 @@ func NewStatsClient(name string, host string) (pilosa.StatsClient, error) {
return nil, errors.Errorf("'%v' not a valid stats client, choose from [expvar, statsd, none].")
}
}
// OpenListener opens a listener for the Server.
func getListener(uri pilosa.URI, tlsconf *tls.Config) (ln net.Listener, err error) {
// If bind URI has the https scheme, enable TLS
if uri.Scheme() == "https" && tlsconf != nil {
ln, err = tls.Listen("tcp", uri.HostPort(), tlsconf)
if err != nil {
return nil, errors.Wrap(err, "tls.Listener")
}
} else if uri.Scheme() == "http" {
// Open HTTP listener to determine port (if specified as :0).
ln, err = net.Listen("tcp", uri.HostPort())
if err != nil {
return nil, errors.Wrap(err, "net.Listen")
}
} else {
return nil, errors.Errorf("unsupported scheme: %s", uri.Scheme())
}
return ln, nil
}

6
server/server_test.go
View file

@ -45,7 +45,7 @@ func TestMain_Set_Quick(t *testing.T) {
defer m.Close()
// Create client.
client, err := pilosa.NewInternalHTTPClient(m.Server.URI.HostPort(), pilosa.GetHTTPClient(nil))
client, err := pilosa.NewInternalHTTPClient(m.Server.URI.HostPort(), server.GetHTTPClient(nil))
if err != nil {
t.Fatal(err)
}
@ -322,11 +322,11 @@ func TestMain_FrameRestore(t *testing.T) {
defer m21.Close()
// Import from first cluster.
client20, err := pilosa.NewInternalHTTPClient(m20.Server.URI.HostPort(), pilosa.GetHTTPClient(nil))
client20, err := pilosa.NewInternalHTTPClient(m20.Server.URI.HostPort(), server.GetHTTPClient(nil))
if err != nil {
t.Fatal("new client:", err)
}
client21, err := pilosa.NewInternalHTTPClient(m21.Server.URI.HostPort(), pilosa.GetHTTPClient(nil))
client21, err := pilosa.NewInternalHTTPClient(m21.Server.URI.HostPort(), server.GetHTTPClient(nil))
if err != nil {
t.Fatal("new client:", err)
}

3
test/executor.go
View file

@ -20,6 +20,7 @@ import (
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/pql"
"github.com/pilosa/pilosa/server"
)
// Executor represents a test wrapper for pilosa.Executor.
@ -30,7 +31,7 @@ type Executor struct {
var remoteClient *http.Client
func init() {
remoteClient = pilosa.GetHTTPClient(nil)
remoteClient = server.GetHTTPClient(nil)
}
// NewExecutor returns a new instance of Executor.

23
test/pilosa.go
View file

@ -49,15 +49,13 @@ func NewMain() *Main {
}
m := &Main{Command: server.NewCommand(os.Stdin, os.Stdout, os.Stderr)}
m.Server.Network = *Network
m.Server.NewAttrStore = NewAttrStore
m.Server.Holder.NewAttrStore = NewAttrStore
m.Config.DataDir = path
m.Config.Bind = "http://localhost:0"
m.Config.Cluster.Disabled = true
m.Command.Stdin = &m.Stdin
m.Command.Stdout = &m.Stdout
m.Command.Stderr = &m.Stderr
m.SetupServer()
if testing.Verbose() {
m.Command.Stdout = io.MultiWriter(os.Stdout, m.Command.Stdout)
@ -101,6 +99,7 @@ func runMainWithCluster(size int) ([]*Main, error) {
for i := 0; i < size; i++ {
m := NewMainWithCluster(i == 0)
m.Config.Cluster.Disabled = false
gossipSeeds[i], err = m.RunWithTransport(gossipHost, gossipPort, gossipSeeds[:i])
if err != nil {
@ -136,12 +135,16 @@ func (m *Main) Reopen() error {
}
// Create new main with the same config.
config := m.Config
config := m.Command.Config
m.Command = server.NewCommand(os.Stdin, os.Stdout, os.Stderr)
m.Server.Network = *Network
m.Command.Config = config
err := m.SetupServer()
if err != nil {
return errors.Wrap(err, "setting up server")
}
m.Server.NewAttrStore = boltdb.NewAttrStore
m.Server.Holder.NewAttrStore = m.Server.NewAttrStore
m.Config = config
// Run new program.
if err := m.Run(); err != nil {
@ -174,12 +177,6 @@ func (m *Main) RunWithTransport(host string, bindPort int, joinSeeds []string) (
return seed, err
}
// Open server listener.
err = m.Server.OpenListener()
if err != nil {
return seed, err
}
// Open gossip transport to use in SetupServer.
transport, err := gossip.NewTransport(host, bindPort, nil)
if err != nil {
@ -221,7 +218,7 @@ func (m *Main) URL() string { return "http://" + m.Server.Addr().String() }
// Client returns a client to connect to the program.
func (m *Main) Client() *pilosa.InternalHTTPClient {
client, err := pilosa.NewInternalHTTPClient(m.Server.URI.HostPort(), pilosa.GetHTTPClient(nil))
client, err := pilosa.NewInternalHTTPClient(m.Server.URI.HostPort(), server.GetHTTPClient(nil))
if err != nil {
panic(err)
}

7
test/test.go
View file

@ -13,10 +13,3 @@
// limitations under the License.
package test
import "flag"
// Test flags.
var (
Network = flag.String("network", "tcp", "network name")
)