scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-12 07:41:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'develop' into update-getting-started

Browse source
This commit is contained in:
Yuce Tekol 2018-06-27 01:16:05 +03:00 • committed by GitHub
commit 345165e7df
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
14 changed files with 248 additions and 153 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
Gopkg.lock generated
View file

@ -197,12 +197,6 @@
revision = "645ef00459ed84a119197bfb8d8205042c6df63d"
version = "v0.8.0"
[[projects]]
name = "github.com/rakyll/statik"
packages = ["fs"]
revision = "fd36b3595eb2ec8da4b8153b107f7ea08504899d"
version = "v0.1.1"
[[projects]]
name = "github.com/satori/go.uuid"
packages = ["."]

27
broadcast_test.go
View file

@ -56,6 +56,7 @@ func testMessageMarshal(t *testing.T, m proto.Message) {
// Ensure that BroadcastReceiver can register a BroadcastHandler.
func TestBroadcast_BroadcastReceiver(t *testing.T) {
t.Skip("broadcast receiver")
path, err := ioutil.TempDir("", "pilosa-")
if err != nil {
panic(err)
@ -67,24 +68,24 @@ func TestBroadcast_BroadcastReceiver(t *testing.T) {
if err != nil {
t.Fatalf("setting up server: %v", err)
}
s := com.Server
// s := com.Server
sbr := NewSimpleBroadcastReceiver()
sbh := NewSimpleBroadcastHandler()
// sbr := NewSimpleBroadcastReceiver()
// sbh := NewSimpleBroadcastHandler()
s.BroadcastReceiver = sbr
s.BroadcastReceiver.Start(sbh)
// s.BroadcastReceiver = sbr
// s.BroadcastReceiver.Start(sbh)
msg := &internal.DeleteIndexMessage{
Index: "i",
}
// msg := &internal.DeleteIndexMessage{
// Index: "i",
// }
s.BroadcastReceiver.(*SimpleBroadcastReceiver).Receive(msg)
// s.BroadcastReceiver.(*SimpleBroadcastReceiver).Receive(msg)
// Make sure the message received is what was sentd
if !reflect.DeepEqual(sbh.receivedMessage, msg) {
t.Fatalf("unexpected message: %s", sbh.receivedMessage)
}
// // Make sure the message received is what was sentd
// if !reflect.DeepEqual(sbh.receivedMessage, msg) {
// t.Fatalf("unexpected message: %s", sbh.receivedMessage)
// }
}
type SimpleBroadcastReceiver struct {

5
cluster.go
View file

@ -886,11 +886,6 @@ func (c *Cluster) open() error {
return errors.Wrap(err, "adding local node")
}
// Start the EventReceiver.
if err := c.EventReceiver.Start(c); err != nil {
return fmt.Errorf("starting EventReceiver: %v", err)
}
// Open MemberSet communication.
if err := c.MemberSet.Open(c.Node); err != nil {
return fmt.Errorf("opening MemberSet: %v", err)

19
executor.go
View file

@ -1606,6 +1606,9 @@ func (e *Executor) translateCall(index string, idx *Index, c *pql.Call) error {
}
// Translate column key.
if idx.Keys() {
if c.Args[colKey] != nil && !isString(c.Args[colKey]) {
return errors.New("column value must be a string when index 'keys' option enabled")
}
if value := callArgString(c, colKey); value != "" {
ids, err := e.TranslateStore.TranslateColumnsToUint64(index, []string{value})
if err != nil {
@ -1613,6 +1616,10 @@ func (e *Executor) translateCall(index string, idx *Index, c *pql.Call) error {
}
c.Args[colKey] = ids[0]
}
} else {
if isString(c.Args[colKey]) {
return errors.New("string 'col' value not allowed unless index 'keys' option enabled")
}
}
// Translate row key, if field is specified & key exists.
@ -1622,6 +1629,9 @@ func (e *Executor) translateCall(index string, idx *Index, c *pql.Call) error {
return ErrFieldNotFound
}
if field.Keys() {
if c.Args[rowKey] != nil && !isString(c.Args[rowKey]) {
return errors.New("row value must be a string when field 'keys' option enabled")
}
if value := callArgString(c, rowKey); value != "" {
ids, err := e.TranslateStore.TranslateRowsToUint64(index, fieldName, []string{value})
if err != nil {
@ -1629,6 +1639,10 @@ func (e *Executor) translateCall(index string, idx *Index, c *pql.Call) error {
}
c.Args[rowKey] = ids[0]
}
} else {
if isString(c.Args[rowKey]) {
return errors.New("string 'row' value not allowed unless field 'keys' option enabled")
}
}
}
@ -1801,3 +1815,8 @@ func callArgString(call *pql.Call, key string) string {
s, _ := value.(string)
return s
}
func isString(v interface{}) bool {
_, ok := v.(string)
return ok
}

124
executor_test.go
View file

@ -27,6 +27,7 @@ import (
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/pql"
"github.com/pilosa/pilosa/test"
"github.com/pkg/errors"
)
// Ensure a bitmap query can be executed.
@ -264,36 +265,107 @@ func TestExecutor_Execute_Count(t *testing.T) {
}
// Ensure a set query can be executed.
func TestExecutor_Execute_Set(t *testing.T) {
hldr := test.MustOpenHolder()
defer hldr.Close()
func TestExecutor_Execute_SetBit(t *testing.T) {
t.Run("ID", func(t *testing.T) {
cmd := test.MustRunMainWithCluster(t, 1)[0]
holder := cmd.Server.Holder()
hldr := test.Holder{Holder: holder}
hldr.SetBit("i", "f", 1, 0)
// set a bit so the view gets created.
hldr.SetBit("i", "f", 1, 0)
t.Run("OK", func(t *testing.T) {
hldr.ClearBit("i", "f", 11, 1)
if n := hldr.Row("i", "f", 11).Count(); n != 0 {
t.Fatalf("unexpected bitmap count: %d", n)
}
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
if n := hldr.Row("i", "f", 11).Count(); n != 0 {
t.Fatalf("unexpected bitmap count: %d", n)
}
if res, err := cmd.API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set(1, f=11)`}); err != nil {
t.Fatal(err)
} else {
if !res.Results[0].(bool) {
t.Fatalf("expected column changed")
}
}
if res, err := e.Execute(context.Background(), "i", test.MustParse(`Set(1, f=11)`), nil, nil); err != nil {
t.Fatal(err)
} else {
if !res[0].(bool) {
t.Fatalf("expected column changed")
}
}
if n := hldr.Row("i", "f", 11).Count(); n != 1 {
t.Fatalf("unexpected bitmap count: %d", n)
}
if res, err := cmd.API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set(1, f=11)`}); err != nil {
t.Fatal(err)
} else {
if res.Results[0].(bool) {
t.Fatalf("expected column unchanged")
}
}
})
if n := hldr.Row("i", "f", 11).Count(); n != 1 {
t.Fatalf("unexpected bitmap count: %d", n)
}
if res, err := e.Execute(context.Background(), "i", test.MustParse(`Set(1, f=11)`), nil, nil); err != nil {
t.Fatal(err)
} else {
if res[0].(bool) {
t.Fatalf("expected column unchanged")
}
}
t.Run("ErrInvalidColValueType", func(t *testing.T) {
if _, err := cmd.API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set("foo", f=1)`}); err == nil || errors.Cause(err).Error() != `string 'col' value not allowed unless index 'keys' option enabled` {
t.Fatalf("The error is: '%v'", err)
}
})
t.Run("ErrInvalidRowValueType", func(t *testing.T) {
if _, err := cmd.API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set(2, f="bar")`}); err == nil || errors.Cause(err).Error() != `string 'row' value not allowed unless field 'keys' option enabled` {
t.Fatal(err)
}
})
})
t.Run("Keys", func(t *testing.T) {
cmd := test.MustRunMainWithCluster(t, 1)[0]
holder := cmd.Server.Holder()
hldr := test.Holder{Holder: holder}
index := hldr.MustCreateIndexIfNotExists("i", pilosa.IndexOptions{Keys: true})
t.Run("OK", func(t *testing.T) {
hldr.SetBit("i", "f", 1, 0)
if n := hldr.Row("i", "f", 11).Count(); n != 0 {
t.Fatalf("unexpected bitmap count: %d", n)
}
if res, err := cmd.API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set("foo", f=11)`}); err != nil {
t.Fatal(err)
} else {
if !res.Results[0].(bool) {
t.Fatalf("expected column changed")
}
}
if n := hldr.Row("i", "f", 11).Count(); n != 1 {
t.Fatalf("unexpected bitmap count: %d", n)
}
if res, err := cmd.API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set("foo", f=11)`}); err != nil {
t.Fatal(err)
} else {
if res.Results[0].(bool) {
t.Fatalf("expected column unchanged")
}
}
})
t.Run("ErrInvalidColValueType", func(t *testing.T) {
if err := index.DeleteField("f"); err != nil {
t.Fatal(err)
}
if _, err := index.CreateField("f", pilosa.FieldOptions{}); err != nil {
t.Fatal(err)
}
if _, err := cmd.API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set(2, f=1)`}); err == nil || errors.Cause(err).Error() != `column value must be a string when index 'keys' option enabled` {
t.Fatal(err)
}
})
t.Run("ErrInvalidRowValueType", func(t *testing.T) {
index := hldr.MustCreateIndexIfNotExists("inokey", pilosa.IndexOptions{})
if _, err := index.CreateField("f", pilosa.FieldOptions{Keys: true}); err != nil {
t.Fatal(err)
}
if _, err := cmd.API.Query(context.Background(), &pilosa.QueryRequest{Index: "inokey", Query: `Set(2, f=1)`}); err == nil || errors.Cause(err).Error() != `row value must be a string when field 'keys' option enabled` {
t.Fatal(err)
}
})
})
}
// Ensure old PQL syntax doesn't break anything too badly.

32
gossip/gossip.go
View file

@ -33,7 +33,6 @@ import (
)
// Ensure GossipMemberSet implements interfaces.
var _ pilosa.BroadcastReceiver = &GossipMemberSet{}
var _ memberlist.Delegate = &GossipMemberSet{}
// GossipMemberSet represents a gossip implementation of MemberSet using memberlist.
@ -45,19 +44,15 @@ type GossipMemberSet struct {
broadcasts *memberlist.TransmitLimitedQueue
statusHandler pilosa.StatusHandler
config *gossipConfig
pserver *pilosa.Server
config *gossipConfig
Logger pilosa.Logger
logger *log.Logger
transport *Transport
}
// Start implements the BroadcastReceiver interface and sets the BroadcastHandler.
func (g *GossipMemberSet) Start(h pilosa.BroadcastHandler) error {
g.handler = h
return nil
gossipEventReceiver *GossipEventReceiver
}
// GetBindAddr returns the gossip bind address based on config and auto bind port.
@ -69,13 +64,16 @@ func (g *GossipMemberSet) GetBindAddr() string {
// Open implements the MemberSet interface to start network activity.
func (g *GossipMemberSet) Open(n *pilosa.Node) error {
err := g.gossipEventReceiver.Start(g.pserver)
if err != nil {
return errors.Wrap(err, "starting event delegate")
}
if g.handler == nil {
return fmt.Errorf("must call Start(pilosa.BroadcastHandler) before calling Open()")
}
g.node = n
err := error(nil)
g.mu.Lock()
g.memberlist, err = memberlist.Create(g.config.memberlistConfig)
g.mu.Unlock()
@ -166,7 +164,7 @@ func WithLogger(logger *log.Logger) GossipMemberSetOption {
}
// NewGossipMemberSet returns a new instance of GossipMemberSet based on options.
func NewGossipMemberSet(name string, host string, cfg Config, ger *GossipEventReceiver, sh pilosa.StatusHandler, options ...GossipMemberSetOption) (*GossipMemberSet, error) {
func NewGossipMemberSet(name string, host string, cfg Config, s *pilosa.Server, options ...GossipMemberSetOption) (*GossipMemberSet, error) {
g := &GossipMemberSet{
Logger: pilosa.NopLogger,
}
@ -177,6 +175,10 @@ func NewGossipMemberSet(name string, host string, cfg Config, ger *GossipEventRe
return nil, errors.Wrap(err, "executing option")
}
}
ger := NewGossipEventReceiver(g.logger)
g.gossipEventReceiver = ger
g.handler = s
if g.transport == nil {
port, err := strconv.Atoi(cfg.Port)
@ -232,7 +234,7 @@ func NewGossipMemberSet(name string, host string, cfg Config, ger *GossipEventRe
gossipSeeds: cfg.Seeds,
}
g.statusHandler = sh
g.pserver = s
return g, nil
}
@ -270,7 +272,7 @@ func (g *GossipMemberSet) GetBroadcasts(overhead, limit int) [][]byte {
// LocalState implementation of the memberlist.Delegate interface
// sends this Node's state data.
func (g *GossipMemberSet) LocalState(join bool) []byte {
pb, err := g.statusHandler.LocalStatus()
pb, err := g.pserver.LocalStatus()
if err != nil {
g.Logger.Printf("error getting local state, err=%s", err)
return []byte{}
@ -294,7 +296,7 @@ func (g *GossipMemberSet) MergeRemoteState(buf []byte, join bool) {
g.Logger.Printf("error unmarshalling nodestate data, err=%s", err)
return
}
err := g.statusHandler.HandleRemoteStatus(&pb)
err := g.pserver.HandleRemoteStatus(&pb)
if err != nil {
g.Logger.Printf("merge state error: %s", err)
}
@ -309,11 +311,11 @@ type GossipEventReceiver struct {
ch chan memberlist.NodeEvent
eventHandler pilosa.EventHandler
Logger pilosa.Logger
Logger *log.Logger
}
// NewGossipEventReceiver returns a new instance of GossipEventReceiver.
func NewGossipEventReceiver(logger pilosa.Logger) *GossipEventReceiver {
func NewGossipEventReceiver(logger *log.Logger) *GossipEventReceiver {
return &GossipEventReceiver{
ch: make(chan memberlist.NodeEvent, 1),
Logger: logger,

33
http/translator_test.go
View file

@ -4,13 +4,13 @@ import (
"context"
"io"
"io/ioutil"
"net/http/httptest"
"testing"
"time"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/http"
"github.com/pilosa/pilosa/mock"
"github.com/pilosa/pilosa/server"
"github.com/pilosa/pilosa/test"
)
@ -52,13 +52,14 @@ func TestTranslateStore_Reader(t *testing.T) {
}
return &mrc, nil
}
h := test.MustNewHandler()
h.API.TranslateStore = &translateStore
s := httptest.NewServer(h)
defer s.Close()
opts := server.OptCommandServerOptions(pilosa.OptServerPrimaryTranslateStore(translateStore))
main := test.MustRunMainWithCluster(t, 1, []server.CommandOption{opts})[0]
defer main.Close()
// Connect to server and stream all available data.
store := http.NewTranslateStore(s.URL)
store := http.NewTranslateStore(main.Server.URI.String())
rc, err := store.Reader(context.Background(), 100)
if err != nil {
t.Fatal(err)
@ -95,15 +96,16 @@ func TestTranslateStore_Reader(t *testing.T) {
translateStore.ReaderFunc = func(ctx context.Context, off int64) (io.ReadCloser, error) {
return &mrc, nil
}
h := test.MustNewHandler()
h.API.TranslateStore = &translateStore
s := httptest.NewServer(h)
defer s.Close()
opts := server.OptCommandServerOptions(pilosa.OptServerPrimaryTranslateStore(translateStore))
main := test.MustRunMainWithCluster(t, 1, []server.CommandOption{opts})[0]
defer main.Close()
defer close(done)
// Connect to server and begin streaming.
ctx, cancel := context.WithCancel(context.Background())
store := http.NewTranslateStore(s.URL)
store := http.NewTranslateStore(main.Server.URI.String())
if _, err := store.Reader(ctx, 0); err != nil {
t.Fatal(err)
}
@ -123,12 +125,11 @@ func TestTranslateStore_Reader(t *testing.T) {
translateStore.ReaderFunc = func(ctx context.Context, off int64) (io.ReadCloser, error) {
return nil, pilosa.ErrNotImplemented
}
h := test.MustNewHandler()
h.API.TranslateStore = &translateStore
s := httptest.NewServer(h)
defer s.Close()
_, err := http.NewTranslateStore(s.URL).Reader(context.Background(), 0)
opts := server.OptCommandServerOptions(pilosa.OptServerPrimaryTranslateStore(translateStore))
main := test.MustRunMainWithCluster(t, 1, []server.CommandOption{opts})[0]
_, err := http.NewTranslateStore(main.Server.URI.String()).Reader(context.Background(), 0)
if err != pilosa.ErrNotImplemented {
t.Fatalf("unexpected error: %s", err)
}

10
mock/translator.go
View file

@ -17,22 +17,22 @@ type TranslateStore struct {
ReaderFunc func(ctx context.Context, off int64) (io.ReadCloser, error)
}
func (s *TranslateStore) TranslateColumnsToUint64(index string, values []string) ([]uint64, error) {
func (s TranslateStore) TranslateColumnsToUint64(index string, values []string) ([]uint64, error) {
return s.TranslateColumnsToUint64Func(index, values)
}
func (s *TranslateStore) TranslateColumnToString(index string, values uint64) (string, error) {
func (s TranslateStore) TranslateColumnToString(index string, values uint64) (string, error) {
return s.TranslateColumnToStringFunc(index, values)
}
func (s *TranslateStore) TranslateRowsToUint64(index, frame string, values []string) ([]uint64, error) {
func (s TranslateStore) TranslateRowsToUint64(index, frame string, values []string) ([]uint64, error) {
return s.TranslateRowsToUint64Func(index, frame, values)
}
func (s *TranslateStore) TranslateRowToString(index, frame string, value uint64) (string, error) {
func (s TranslateStore) TranslateRowToString(index, frame string, value uint64) (string, error) {
return s.TranslateRowToStringFunc(index, frame, value)
}
func (s *TranslateStore) Reader(ctx context.Context, off int64) (io.ReadCloser, error) {
func (s TranslateStore) Reader(ctx context.Context, off int64) (io.ReadCloser, error) {
return s.ReaderFunc(ctx, off)
}

38
server.go
View file

@ -61,10 +61,9 @@ type Server struct {
clusterDisabled bool
// External
BroadcastReceiver BroadcastReceiver
systemInfo SystemInfo
gcNotifier GCNotifier
logger Logger
systemInfo SystemInfo
gcNotifier GCNotifier
logger Logger
NodeID string
URI URI
@ -207,12 +206,11 @@ func OptServerClusterDisabled(disabled bool, hosts []string) ServerOption {
// NewServer returns a new instance of Server.
func NewServer(opts ...ServerOption) (*Server, error) {
s := &Server{
closing: make(chan struct{}),
Cluster: NewCluster(),
holder: NewHolder(),
BroadcastReceiver: NopBroadcastReceiver,
diagnostics: NewDiagnosticsCollector(DefaultDiagnosticServer),
systemInfo: NewNopSystemInfo(),
closing: make(chan struct{}),
Cluster: NewCluster(),
holder: NewHolder(),
diagnostics: NewDiagnosticsCollector(DefaultDiagnosticServer),
systemInfo: NewNopSystemInfo(),
gcNotifier: NopGCNotifier,
@ -246,11 +244,8 @@ func NewServer(opts ...ServerOption) (*Server, error) {
// Initialize translation database.
s.translateFile = NewTranslateFile()
s.translateFile.Path = filepath.Join(path, "keys")
s.translateFile.Path = filepath.Join(path, ".keys")
s.translateFile.PrimaryTranslateStore = s.primaryTranslateStore
if err := s.translateFile.Open(); err != nil {
return nil, err
}
// Get or create NodeID.
s.NodeID = s.LoadNodeID()
@ -290,6 +285,11 @@ func (s *Server) Open() error {
log.Println(errors.Wrap(err, "logging startup"))
}
// Initialize id-key storage.
if err := s.translateFile.Open(); err != nil {
return err
}
// Cluster settings.
s.Cluster.Broadcaster = s
s.Cluster.MaxWritesPerRequest = s.maxWritesPerRequest
@ -297,11 +297,6 @@ func (s *Server) Open() error {
// Initialize Holder.
s.holder.Broadcaster = s
// Start the BroadcastReceiver.
if err := s.BroadcastReceiver.Start(s); err != nil {
return fmt.Errorf("starting BroadcastReceiver: %v", err)
}
// Open Cluster management.
if err := s.Cluster.open(); err != nil {
return fmt.Errorf("opening Cluster: %v", err)
@ -711,6 +706,11 @@ func (s *Server) monitorRuntime() {
}
}
// ReceiveEvent implements the EventHandler interface.
func (s *Server) ReceiveEvent(e *NodeEvent) error {
return s.Cluster.ReceiveEvent(e)
}
// countOpenFiles on operating systems that support lsof.
func countOpenFiles() (int, error) {
switch runtime.GOOS {

3
server/handler_test.go
View file

@ -31,6 +31,7 @@ import (
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/http"
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/server"
"github.com/pilosa/pilosa/test"
)
@ -565,7 +566,7 @@ func TestHandler_Endpoints(t *testing.T) {
t.Fatalf("CORS preflight status should be 405, but is %v", result.StatusCode)
}
clus := test.MustRunMainWithCluster(t, 1, test.OptAllowedOrigins([]string{"http://test/"}))
clus := test.MustRunMainWithCluster(t, 1, []server.CommandOption{test.OptAllowedOrigins([]string{"http://test/"})})
w = httptest.NewRecorder()
h := clus[0].Handler.(*http.Handler).Handler
h.ServeHTTP(w, req)

44
server/server.go
View file

@ -75,12 +75,24 @@ type Command struct {
logger loggerLogger
Handler pilosa.Handler
API *pilosa.API
ln net.Listener
serverOptions []pilosa.ServerOption
}
type CommandOption func(c *Command) error
func OptCommandServerOptions(opts ...pilosa.ServerOption) CommandOption {
return func(c *Command) error {
c.serverOptions = append(c.serverOptions, opts...)
return nil
}
}
// NewCommand returns a new instance of Main.
func NewCommand(stdin io.Reader, stdout, stderr io.Writer) *Command {
return &Command{
func NewCommand(stdin io.Reader, stdout, stderr io.Writer, opts ...CommandOption) *Command {
c := &Command{
Config: NewConfig(),
CmdIO: pilosa.NewCmdIO(stdin, stdout, stderr),
@ -88,6 +100,16 @@ func NewCommand(stdin io.Reader, stdout, stderr io.Writer) *Command {
Started: make(chan struct{}),
done: make(chan struct{}),
}
for _, opt := range opts {
err := opt(c)
if err != nil {
panic(err)
// TODO: Return error instead of panic?
}
}
return c
}
// Start starts the pilosa server - it returns once the server is running.
@ -225,7 +247,7 @@ func (m *Command) SetupServer() error {
primaryTranslateStore = http.NewTranslateStore(m.Config.Translation.PrimaryURL)
}
m.Server, err = pilosa.NewServer(
serverOptions := []pilosa.ServerOption{
pilosa.OptServerAntiEntropyInterval(time.Duration(m.Config.AntiEntropy.Interval)),
pilosa.OptServerLongQueryTime(time.Duration(m.Config.Cluster.LongQueryTime)),
pilosa.OptServerDataDir(m.Config.DataDir),
@ -243,19 +265,24 @@ func (m *Command) SetupServer() error {
pilosa.OptServerInternalClient(http.NewInternalClientFromURI(uri, c)),
pilosa.OptServerPrimaryTranslateStore(primaryTranslateStore),
pilosa.OptServerClusterDisabled(m.Config.Cluster.Disabled, m.Config.Cluster.Hosts),
)
}
serverOptions = append(serverOptions, m.serverOptions...)
m.Server, err = pilosa.NewServer(serverOptions...)
if err != nil {
return errors.Wrap(err, "new server")
}
api, err := pilosa.NewAPI(pilosa.OptAPIServer(m.Server))
m.API, err = pilosa.NewAPI(pilosa.OptAPIServer(m.Server))
if err != nil {
return errors.Wrap(err, "new api")
}
m.Handler, err = http.NewHandler(
http.OptHandlerAllowedOrigins(m.Config.Handler.AllowedOrigins),
http.OptHandlerAPI(api),
http.OptHandlerAPI(m.API),
http.OptHandlerLogger(m.logger),
http.OptHandlerListener(m.ln),
)
@ -292,13 +319,10 @@ func (m *Command) SetupNetworking() error {
m.Server.Cluster.Node.IsCoordinator = true
}
gossipEventReceiver := gossip.NewGossipEventReceiver(m.logger)
m.Server.Cluster.EventReceiver = gossipEventReceiver
gossipMemberSet, err := gossip.NewGossipMemberSet(
m.Server.NodeID,
m.Server.URI.Host(),
m.Config.Gossip,
gossipEventReceiver,
m.Server,
gossip.WithLogger(m.logger.Logger()),
gossip.WithTransport(transport),
@ -306,9 +330,7 @@ func (m *Command) SetupNetworking() error {
if err != nil {
return errors.Wrap(err, "getting memberset")
}
gossipMemberSet.Logger = m.logger
m.Server.Cluster.MemberSet = gossipMemberSet
m.Server.BroadcastReceiver = gossipMemberSet
return nil
}

3
server_test.go
View file

@ -20,6 +20,7 @@ import (
"time"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/server"
"github.com/pilosa/pilosa/test"
)
@ -27,7 +28,7 @@ import (
// pilosa.Server was not having its remoteClient field set by an option and so
// it was using a nil client in monitorAntiEntropy.
func TestMonitorAntiEntropy(t *testing.T) {
cluster := test.MustRunMainWithCluster(t, 3, test.OptAntiEntropyInterval(time.Millisecond*20))
cluster := test.MustRunMainWithCluster(t, 3, []server.CommandOption{test.OptAntiEntropyInterval(time.Millisecond * 20)})
client := cluster[1].Client()
err := client.CreateIndex(context.Background(), "balh", pilosa.IndexOptions{})
if err != nil {

10
statik/statik.go
View file

File diff suppressed because one or more lines are too long

47
test/pilosa.go
View file

@ -34,52 +34,46 @@ import (
)
////////////////////////////////////////////////////////////////////////////////////
// Main represents a test wrapper for main.Main.
// Main represents a test wrapper for server.Command.
type Main struct {
*server.Command
commandOptions []server.CommandOption
Stdin bytes.Buffer
Stdout bytes.Buffer
Stderr bytes.Buffer
}
type MainOpt func(m *Main) error
func OptAntiEntropyInterval(dur time.Duration) MainOpt {
return func(m *Main) error {
m.Command.Config.AntiEntropy.Interval = toml.Duration(dur)
func OptAntiEntropyInterval(dur time.Duration) server.CommandOption {
return func(m *server.Command) error {
m.Config.AntiEntropy.Interval = toml.Duration(dur)
return nil
}
}
func OptAllowedOrigins(origins []string) MainOpt {
return func(m *Main) error {
func OptAllowedOrigins(origins []string) server.CommandOption {
return func(m *server.Command) error {
m.Config.Handler.AllowedOrigins = origins
return nil
}
}
// NewMain returns a new instance of Main with a temporary data directory and random port.
func NewMain(opts ...MainOpt) *Main {
func NewMain(opts ...server.CommandOption) *Main {
path, err := ioutil.TempDir("", "pilosa-")
if err != nil {
panic(err)
}
m := &Main{Command: server.NewCommand(os.Stdin, os.Stdout, os.Stderr)}
m := &Main{Command: server.NewCommand(os.Stdin, os.Stdout, os.Stderr, opts...), commandOptions: opts}
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
for _, opt := range opts {
err := opt(m)
if err != nil {
panic(err)
}
}
err = m.SetupServer()
if err != nil {
panic(err)
@ -94,7 +88,7 @@ func NewMain(opts ...MainOpt) *Main {
}
// NewMainWithCluster returns a new instance of Main with clustering enabled.
func NewMainWithCluster(isCoordinator bool, opts ...MainOpt) *Main {
func NewMainWithCluster(isCoordinator bool, opts ...server.CommandOption) *Main {
m := NewMain(opts...)
m.Config.Cluster.Disabled = false
m.Config.Cluster.Coordinator = isCoordinator
@ -103,7 +97,7 @@ func NewMainWithCluster(isCoordinator bool, opts ...MainOpt) *Main {
// MustRunMainWithCluster ruturns a running array of *Main where
// all nodes are joined via memberlist (i.e. clustering enabled).
func MustRunMainWithCluster(t *testing.T, size int, opts ...MainOpt) []*Main {
func MustRunMainWithCluster(t *testing.T, size int, opts ...[]server.CommandOption) []*Main {
ma, err := runMainWithCluster(size, opts...)
if err != nil {
t.Fatalf("new main array with cluster: %v", err)
@ -113,10 +107,13 @@ func MustRunMainWithCluster(t *testing.T, size int, opts ...MainOpt) []*Main {
// runMainWithCluster runs an array of *Main where all nodes are
// joined via memberlist (i.e. clustering enabled).
func runMainWithCluster(size int, opts ...MainOpt) ([]*Main, error) {
func runMainWithCluster(size int, opts ...[]server.CommandOption) ([]*Main, error) {
if size == 0 {
return nil, errors.New("cluster must contain at least one node")
}
if len(opts) != size && len(opts) != 0 && len(opts) != 1 {
return nil, errors.New("Slice of CommandOptions must be of length 0, 1, or equal to the number of cluster nodes")
}
mains := make([]*Main, size)
@ -126,7 +123,11 @@ func runMainWithCluster(size int, opts ...MainOpt) ([]*Main, error) {
var gossipSeeds = make([]string, size)
for i := 0; i < size; i++ {
m := NewMainWithCluster(i == 0, opts...)
var commandOpts []server.CommandOption
if len(opts) > 0 {
commandOpts = opts[i%len(opts)]
}
m := NewMainWithCluster(i == 0, commandOpts...)
m.Config.Cluster.Disabled = false
gossipSeeds[i], err = m.RunWithTransport(gossipHost, gossipPort, gossipSeeds[:i])
@ -164,7 +165,7 @@ func (m *Main) Reopen() error {
// Create new main with the same config.
config := m.Command.Config
m.Command = server.NewCommand(os.Stdin, os.Stdout, os.Stderr)
m.Command = server.NewCommand(os.Stdin, os.Stdout, os.Stderr, m.commandOptions...)
m.Command.Config = config
err := m.SetupServer()
if err != nil {
@ -223,10 +224,6 @@ func (m *Main) RunWithTransport(host string, bindPort int, joinSeeds []string) (
return seed, err
}
if err = m.Server.BroadcastReceiver.Start(m.Server); err != nil {
return seed, err
}
m.Server.Cluster.Static = false
go func() {