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

Merge remote-tracking branch 'upstream/develop' into accept-json

Browse source
This commit is contained in:
Todd Gruben 2018-06-20 13:14:37 -05:00
commit ff81322d69
13 changed files with 146 additions and 161 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
cluster.go
View file

@ -1801,3 +1801,17 @@ func (c *Cluster) mergeClusterStatus(cs *internal.ClusterStatus) error {
return nil
}
func (c *Cluster) setStatic(hosts []string) error {
c.Static = true
c.Coordinator = c.Node.ID
for _, address := range hosts {
uri, err := NewURIFromAddress(address)
if err != nil {
return errors.Wrap(err, "getting URI")
}
c.Nodes = append(c.Nodes, &Node{URI: *uri})
}
c.MemberSet = NewStaticMemberSet(c.Nodes)
return nil
}

32
cluster_internal_test.go
View file

@ -153,19 +153,19 @@ func TestFragSources(t *testing.T) {
if err != nil {
t.Fatal(err)
}
_, err = field.SetBit("standard", 1, 101, nil)
_, err = field.SetBit(1, 101, nil)
if err != nil {
t.Fatal(err)
}
_, err = field.SetBit("standard", 1, 1300000, nil)
_, err = field.SetBit(1, 1300000, nil)
if err != nil {
t.Fatal(err)
}
_, err = field.SetBit("standard", 1, 2600000, nil)
_, err = field.SetBit(1, 2600000, nil)
if err != nil {
t.Fatal(err)
}
_, err = field.SetBit("standard", 1, 3900000, nil)
_, err = field.SetBit(1, 3900000, nil)
if err != nil {
t.Fatal(err)
}
@ -552,7 +552,7 @@ func TestCluster_ResizeStates(t *testing.T) {
t.Run("Single node, in topology", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.addNode(false)
tc.addNode()
node := tc.Clusters[0]
@ -580,7 +580,7 @@ func TestCluster_ResizeStates(t *testing.T) {
t.Run("Single node, not in topology", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.addNode(false)
tc.addNode()
node := tc.Clusters[0]
@ -605,14 +605,14 @@ func TestCluster_ResizeStates(t *testing.T) {
t.Run("Multiple nodes, no data", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.addNode(false)
tc.addNode()
// Open TestCluster.
if err := tc.Open(); err != nil {
t.Fatal(err)
}
tc.addNode(false)
tc.addNode()
node0 := tc.Clusters[0]
node1 := tc.Clusters[1]
@ -643,7 +643,7 @@ func TestCluster_ResizeStates(t *testing.T) {
t.Run("Multiple nodes, in/not in topology", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.addNode(false)
tc.addNode()
node0 := tc.Clusters[0]
// write topology to data file
@ -664,12 +664,12 @@ func TestCluster_ResizeStates(t *testing.T) {
// Expect an error by adding a node not in the topology.
expectedError := "host is not in topology: node1"
err := tc.addNode(false)
err := tc.addNode()
if err == nil || err.Error() != expectedError {
t.Errorf("did not receive expected error: %s", expectedError)
}
tc.addNode(false)
tc.addNode()
node2 := tc.Clusters[2]
// Ensure that node comes up in state NORMAL.
@ -687,7 +687,7 @@ func TestCluster_ResizeStates(t *testing.T) {
t.Run("Multiple nodes, with data", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.addNode(false)
tc.addNode()
node0 := tc.Clusters[0]
// Open TestCluster.
@ -699,8 +699,8 @@ func TestCluster_ResizeStates(t *testing.T) {
if err := tc.CreateField("i", "f", FieldOptions{}); err != nil {
t.Fatal(err)
}
tc.SetBit("i", "f", "standard", 1, 101, nil)
tc.SetBit("i", "f", "standard", 1, 1300000, nil)
tc.SetBit("i", "f", 1, 101, nil)
tc.SetBit("i", "f", 1, 1300000, nil)
// Before starting the resize, get the CheckSum to use for
// comparison later.
@ -709,8 +709,8 @@ func TestCluster_ResizeStates(t *testing.T) {
node0Fragment := node0View.Fragment(1)
node0Checksum := node0Fragment.Checksum()
// AddNode needs to block until the resize process has completed.
tc.addNode(false)
// addNode needs to block until the resize process has completed.
tc.addNode()
node1 := tc.Clusters[1]
// Ensure that nodes come up in state NORMAL.

2
diagnostics.go
View file

@ -222,7 +222,7 @@ func (d *DiagnosticsCollector) EnrichWithSchemaProperties() {
bsiFieldCount := 0
timeQuantumEnabled := false
for _, index := range d.server.Holder.Indexes() {
for _, index := range d.server.holder.Indexes() {
numSlices += index.MaxSlice() + 1
numIndexes += 1
for _, field := range index.Fields() {

16
executor.go
View file

@ -1026,17 +1026,17 @@ func (e *Executor) executeClearBit(ctx context.Context, index string, c *pql.Cal
return false, fmt.Errorf("ClearBit col field '%v' required", columnLabel)
}
return e.executeClearBitView(ctx, index, c, f, ViewStandard, colID, rowID, opt)
return e.executeClearBitField(ctx, index, c, f, colID, rowID, opt)
}
// executeClearBitView executes a ClearBit() call for a single view.
func (e *Executor) executeClearBitView(ctx context.Context, index string, c *pql.Call, f *Field, view string, colID, rowID uint64, opt *ExecOptions) (bool, error) {
// executeClearBitField executes a ClearBit() call for a single view.
func (e *Executor) executeClearBitField(ctx context.Context, index string, c *pql.Call, f *Field, colID, rowID uint64, opt *ExecOptions) (bool, error) {
slice := colID / SliceWidth
ret := false
for _, node := range e.Cluster.sliceNodes(index, slice) {
// Update locally if host matches.
if node.ID == e.Node.ID {
val, err := f.ClearBit(view, rowID, colID, nil)
val, err := f.ClearBit(rowID, colID, nil)
if err != nil {
return false, err
} else if val {
@ -1101,18 +1101,18 @@ func (e *Executor) executeSetBit(ctx context.Context, index string, c *pql.Call,
timestamp = &t
}
return e.executeSetBitView(ctx, index, c, f, ViewStandard, colID, rowID, timestamp, opt)
return e.executeSetBitField(ctx, index, c, f, colID, rowID, timestamp, opt)
}
// executeSetBitView executes a SetBit() call for a specific view.
func (e *Executor) executeSetBitView(ctx context.Context, index string, c *pql.Call, f *Field, view string, colID, rowID uint64, timestamp *time.Time, opt *ExecOptions) (bool, error) {
// executeSetBitField executes a SetBit() call for a specific view.
func (e *Executor) executeSetBitField(ctx context.Context, index string, c *pql.Call, f *Field, colID, rowID uint64, timestamp *time.Time, opt *ExecOptions) (bool, error) {
slice := colID / SliceWidth
ret := false
for _, node := range e.Cluster.sliceNodes(index, slice) {
// Update locally if host matches.
if node.ID == e.Node.ID {
val, err := f.SetBit(view, rowID, colID, timestamp)
val, err := f.SetBit(rowID, colID, timestamp)
if err != nil {
return false, err
} else if val {

22
field.go
View file

@ -654,14 +654,11 @@ func (f *Field) ViewRow(viewName string, rowID uint64) (*Row, error) {
}
// SetBit sets a bit on a view within the field.
func (f *Field) SetBit(name string, rowID, colID uint64, t *time.Time) (changed bool, err error) {
// Validate view name.
if !isValidView(name) {
return false, ErrInvalidView
}
func (f *Field) SetBit(rowID, colID uint64, t *time.Time) (changed bool, err error) {
viewName := ViewStandard
// Retrieve view. Exit if it doesn't exist.
view, err := f.CreateViewIfNotExists(name)
view, err := f.CreateViewIfNotExists(viewName)
if err != nil {
return changed, errors.Wrap(err, "creating view")
}
@ -679,7 +676,7 @@ func (f *Field) SetBit(name string, rowID, colID uint64, t *time.Time) (changed
}
// If a timestamp is specified then set bits across all views for the quantum.
for _, subname := range viewsByTime(name, *t, f.TimeQuantum()) {
for _, subname := range viewsByTime(viewName, *t, f.TimeQuantum()) {
view, err := f.CreateViewIfNotExists(subname)
if err != nil {
return changed, errors.Wrapf(err, "creating view %s", subname)
@ -696,14 +693,11 @@ func (f *Field) SetBit(name string, rowID, colID uint64, t *time.Time) (changed
}
// ClearBit clears a bit within the field.
func (f *Field) ClearBit(name string, rowID, colID uint64, t *time.Time) (changed bool, err error) {
// Validate view name.
if !isValidView(name) {
return false, ErrInvalidView
}
func (f *Field) ClearBit(rowID, colID uint64, t *time.Time) (changed bool, err error) {
viewName := ViewStandard
// Retrieve view. Exit if it doesn't exist.
view, err := f.CreateViewIfNotExists(name)
view, err := f.CreateViewIfNotExists(viewName)
if err != nil {
return changed, errors.Wrap(err, "creating view")
}
@ -721,7 +715,7 @@ func (f *Field) ClearBit(name string, rowID, colID uint64, t *time.Time) (change
}
// If a timestamp is specified then clear bits across all views for the quantum.
for _, subname := range viewsByTime(name, *t, f.TimeQuantum()) {
for _, subname := range viewsByTime(viewName, *t, f.TimeQuantum()) {
view, err := f.CreateViewIfNotExists(subname)
if err != nil {
return changed, errors.Wrapf(err, "creating view %s", subname)

6
holder_test.go
View file

@ -210,7 +210,7 @@ func TestHolder_Open(t *testing.T) {
t.Fatal(err)
} else if field, err := idx.CreateField("bar", pilosa.FieldOptions{}); err != nil {
t.Fatal(err)
} else if _, err := field.SetBit(pilosa.ViewStandard, 0, 0, nil); err != nil {
} else if _, err := field.SetBit(0, 0, nil); err != nil {
t.Fatal(err)
} else if err := h.Holder.Close(); err != nil {
t.Fatal(err)
@ -231,7 +231,7 @@ func TestHolder_Open(t *testing.T) {
t.Fatal(err)
} else if field, err := idx.CreateField("bar", pilosa.FieldOptions{}); err != nil {
t.Fatal(err)
} else if _, err := field.SetBit(pilosa.ViewStandard, 0, 0, nil); err != nil {
} else if _, err := field.SetBit(0, 0, nil); err != nil {
t.Fatal(err)
} else if err := h.Holder.Close(); err != nil {
t.Fatal(err)
@ -257,7 +257,7 @@ func TestHolder_Open(t *testing.T) {
t.Fatal(err)
} else if view, err := field.CreateViewIfNotExists(pilosa.ViewStandard); err != nil {
t.Fatal(err)
} else if _, err := field.SetBit(pilosa.ViewStandard, 0, 0, nil); err != nil {
} else if _, err := field.SetBit(0, 0, nil); err != nil {
t.Fatal(err)
} else if err := view.Fragment(0).FlushCache(); err != nil {
t.Fatal(err)

8
http/handler_test.go
View file

@ -85,12 +85,12 @@ func TestHandler_Schema(t *testing.T) {
if f, err := i0.CreateFieldIfNotExists("f1", pilosa.FieldOptions{}); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(pilosa.ViewStandard, 0, 0, nil); err != nil {
} else if _, err := f.SetBit(0, 0, nil); err != nil {
t.Fatal(err)
}
if f, err := i1.CreateFieldIfNotExists("f0", pilosa.FieldOptions{}); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(pilosa.ViewStandard, 0, 0, nil); err != nil {
} else if _, err := f.SetBit(0, 0, nil); err != nil {
t.Fatal(err)
}
if _, err := i0.CreateFieldIfNotExists("f0", pilosa.FieldOptions{}); err != nil {
@ -122,12 +122,12 @@ func TestHandler_Status(t *testing.T) {
if f, err := i0.CreateFieldIfNotExists("f1", pilosa.FieldOptions{}); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(pilosa.ViewStandard, 0, 0, nil); err != nil {
} else if _, err := f.SetBit(0, 0, nil); err != nil {
t.Fatal(err)
}
if f, err := i1.CreateFieldIfNotExists("f0", pilosa.FieldOptions{}); err != nil {
t.Fatal(err)
} else if _, err := f.SetBit(pilosa.ViewStandard, 0, 0, nil); err != nil {
} else if _, err := f.SetBit(0, 0, nil); err != nil {
t.Fatal(err)
}
if _, err := i0.CreateFieldIfNotExists("f0", pilosa.FieldOptions{}); err != nil {

148
server.go
View file

@ -52,11 +52,13 @@ type Server struct {
closing chan struct{}
// Internal
Holder *Holder
Cluster *Cluster
TranslateFile *TranslateFile
diagnostics *DiagnosticsCollector
executor *Executor
holder *Holder
Cluster *Cluster
translateFile *TranslateFile
diagnostics *DiagnosticsCollector
executor *Executor
hosts []string
clusterDisabled bool
// External
handler Handler
@ -64,7 +66,6 @@ type Server struct {
BroadcastReceiver BroadcastReceiver
systemInfo SystemInfo
gcNotifier GCNotifier
NewAttrStore func(string) AttrStore
logger Logger
ln net.Listener
@ -107,8 +108,7 @@ func OptServerDataDir(dir string) ServerOption {
func OptServerAttrStoreFunc(af func(string) AttrStore) ServerOption {
return func(s *Server) error {
s.NewAttrStore = af
s.Holder.NewAttrStore = af
s.holder.NewAttrStore = af
return nil
}
}
@ -180,7 +180,7 @@ func OptServerPrimaryTranslateStore(store TranslateStore) ServerOption {
func OptServerStatsClient(sc StatsClient) ServerOption {
return func(s *Server) error {
s.Holder.Stats = sc
s.holder.Stats = sc
return nil
}
}
@ -207,12 +207,22 @@ func OptServerURI(uri *URI) ServerOption {
}
}
// OptClusterDisabled tells the server whether to use a static cluster with the
// defined hosts. Mostly used for testing.
func OptServerClusterDisabled(disabled bool, hosts []string) ServerOption {
return func(s *Server) error {
s.hosts = hosts
s.clusterDisabled = disabled
return nil
}
}
// NewServer returns a new instance of Server.
func NewServer(opts ...ServerOption) (*Server, error) {
s := &Server{
closing: make(chan struct{}),
Cluster: NewCluster(),
Holder: NewHolder(),
holder: NewHolder(),
Broadcaster: NopBroadcaster,
BroadcastReceiver: NopBroadcastReceiver,
diagnostics: NewDiagnosticsCollector(DefaultDiagnosticServer),
@ -220,8 +230,6 @@ func NewServer(opts ...ServerOption) (*Server, error) {
gcNotifier: NopGCNotifier,
NewAttrStore: NewNopAttrStore,
antiEntropyInterval: time.Minute * 10,
metricInterval: 0,
diagnosticInterval: 0,
@ -242,19 +250,19 @@ func NewServer(opts ...ServerOption) (*Server, error) {
return nil, err
}
s.Holder.Path = path
s.Holder.Logger = s.logger
s.Holder.Stats.SetLogger(s.logger)
s.holder.Path = path
s.holder.Logger = s.logger
s.holder.Stats.SetLogger(s.logger)
s.Cluster.Path = path
s.Cluster.Logger = s.logger
s.Cluster.Holder = s.Holder
s.Cluster.Holder = s.holder
// Initialize translation database.
s.TranslateFile = NewTranslateFile()
s.TranslateFile.Path = filepath.Join(path, "keys")
s.TranslateFile.PrimaryTranslateStore = s.primaryTranslateStore
if err := s.TranslateFile.Open(); err != nil {
s.translateFile = NewTranslateFile()
s.translateFile.Path = filepath.Join(path, "keys")
s.translateFile.PrimaryTranslateStore = s.primaryTranslateStore
if err := s.translateFile.Open(); err != nil {
return nil, err
}
@ -263,6 +271,7 @@ func NewServer(opts ...ServerOption) (*Server, error) {
s.URI.SetPort(uint16(s.ln.Addr().(*net.TCPAddr).Port))
}
// Get or create NodeID.
s.NodeID = s.LoadNodeID()
// Set Cluster Node.
node := &Node{
@ -271,15 +280,23 @@ func NewServer(opts ...ServerOption) (*Server, error) {
IsCoordinator: s.Cluster.Coordinator == s.NodeID,
}
s.Cluster.Node = node
s.Holder.Stats = s.Holder.Stats.WithTags(fmt.Sprintf("NodeID:%s", s.NodeID))
if s.clusterDisabled {
err := s.Cluster.setStatic(s.hosts)
if err != nil {
return nil, errors.Wrap(err, "setting cluster static")
}
}
s.executor.Holder = s.Holder
// Append the NodeID tag to stats.
s.holder.Stats = s.holder.Stats.WithTags(fmt.Sprintf("NodeID:%s", s.NodeID))
s.executor.Holder = s.holder
s.executor.Node = node
s.executor.Cluster = s.Cluster
s.executor.TranslateStore = s.TranslateFile
s.executor.TranslateStore = s.translateFile
s.executor.MaxWritesPerRequest = s.maxWritesPerRequest
s.handler.GetAPI().Executor = s.executor
s.handler.GetAPI().TranslateStore = s.TranslateFile
s.handler.GetAPI().TranslateStore = s.translateFile
return s, nil
}
@ -287,39 +304,30 @@ func NewServer(opts ...ServerOption) (*Server, error) {
// Open opens and initializes the server.
func (s *Server) Open() error {
s.logger.Printf("open server")
// s.ln can be configured prior to Open() via s.OpenListener().
if s.ln == nil {
return errors.New("Must pass a listener option to NewServer")
return errors.New("must pass a listener option to NewServer")
}
// Log startup
err := s.Holder.logStartup()
err := s.holder.logStartup()
if err != nil {
log.Println(errors.Wrap(err, "logging startup"))
}
// Get or create NodeID.
// Append the NodeID tag to stats.
// Create default HTTP client
// Create executor for executing queries.
// Cluster settings.
s.Cluster.Broadcaster = s.Broadcaster
s.Cluster.MaxWritesPerRequest = s.maxWritesPerRequest
// Initialize HTTP handler.
api := s.handler.GetAPI()
api.Holder = s.Holder
api.Holder = s.holder
api.Broadcaster = s.Broadcaster
api.BroadcastHandler = s
api.StatusHandler = s
api.Cluster = s.Cluster
// Initialize Holder.
s.Holder.Broadcaster = s.Broadcaster
s.holder.Broadcaster = s.Broadcaster
// Serve handler.
go s.handler.Serve(s.ln, s.closing)
@ -335,7 +343,7 @@ func (s *Server) Open() error {
}
// Open holder.
if err := s.Holder.Open(); err != nil {
if err := s.holder.Open(); err != nil {
return fmt.Errorf("opening Holder: %v", err)
}
if err := s.Cluster.setNodeState(NodeStateReady); err != nil {
@ -370,11 +378,11 @@ func (s *Server) Close() error {
if s.Cluster != nil {
s.Cluster.close()
}
if s.Holder != nil {
s.Holder.Close()
if s.holder != nil {
s.holder.Close()
}
if s.TranslateFile != nil {
s.TranslateFile.Close()
if s.translateFile != nil {
s.translateFile.Close()
}
return nil
@ -386,7 +394,7 @@ func (s *Server) LoadNodeID() string {
if s.NodeID != "" {
return s.NodeID
}
nodeID, err := s.Holder.loadNodeID()
nodeID, err := s.holder.loadNodeID()
if err != nil {
s.logger.Printf("loading NodeID: %v", err)
return s.NodeID
@ -414,18 +422,18 @@ func (s *Server) monitorAntiEntropy() {
case <-s.closing:
return
case <-ticker.C:
s.Holder.Stats.Count("AntiEntropy", 1, 1.0)
s.holder.Stats.Count("AntiEntropy", 1, 1.0)
}
t := time.Now()
s.logger.Printf("holder sync beginning")
// Initialize syncer with local holder and remote client.
var syncer HolderSyncer
syncer.Holder = s.Holder
syncer.Holder = s.holder
syncer.Node = s.Cluster.Node
syncer.Cluster = s.Cluster
syncer.Closing = s.closing
syncer.Stats = s.Holder.Stats.WithTags("HolderSyncer")
syncer.Stats = s.holder.Stats.WithTags("HolderSyncer")
// Sync holders.
if err := syncer.SyncHolder(); err != nil {
@ -436,7 +444,7 @@ func (s *Server) monitorAntiEntropy() {
// Record successful sync in log.
s.logger.Printf("holder sync complete")
dif := time.Since(t)
s.Holder.Stats.Histogram("AntiEntropyDuration", float64(dif), 1.0)
s.holder.Stats.Histogram("AntiEntropyDuration", float64(dif), 1.0)
}
}
@ -444,23 +452,23 @@ func (s *Server) monitorAntiEntropy() {
func (s *Server) ReceiveMessage(pb proto.Message) error {
switch obj := pb.(type) {
case *internal.CreateSliceMessage:
idx := s.Holder.Index(obj.Index)
idx := s.holder.Index(obj.Index)
if idx == nil {
return fmt.Errorf("Local Index not found: %s", obj.Index)
}
idx.SetRemoteMaxSlice(obj.Slice)
case *internal.CreateIndexMessage:
opt := IndexOptions{}
_, err := s.Holder.CreateIndex(obj.Index, opt)
_, err := s.holder.CreateIndex(obj.Index, opt)
if err != nil {
return err
}
case *internal.DeleteIndexMessage:
if err := s.Holder.DeleteIndex(obj.Index); err != nil {
if err := s.holder.DeleteIndex(obj.Index); err != nil {
return err
}
case *internal.CreateFieldMessage:
idx := s.Holder.Index(obj.Index)
idx := s.holder.Index(obj.Index)
if idx == nil {
return fmt.Errorf("Local Index not found: %s", obj.Index)
}
@ -470,12 +478,12 @@ func (s *Server) ReceiveMessage(pb proto.Message) error {
return err
}
case *internal.DeleteFieldMessage:
idx := s.Holder.Index(obj.Index)
idx := s.holder.Index(obj.Index)
if err := idx.DeleteField(obj.Field); err != nil {
return err
}
case *internal.CreateViewMessage:
f := s.Holder.Field(obj.Index, obj.Field)
f := s.holder.Field(obj.Index, obj.Field)
if f == nil {
return fmt.Errorf("Local Field not found: %s", obj.Field)
}
@ -484,7 +492,7 @@ func (s *Server) ReceiveMessage(pb proto.Message) error {
return err
}
case *internal.DeleteViewMessage:
f := s.Holder.Field(obj.Index, obj.Field)
f := s.holder.Field(obj.Index, obj.Field)
if f == nil {
return fmt.Errorf("Local Field not found: %s", obj.Field)
}
@ -517,7 +525,7 @@ func (s *Server) ReceiveMessage(pb proto.Message) error {
return err
}
case *internal.RecalculateCaches:
s.Holder.RecalculateCaches()
s.holder.RecalculateCaches()
case *internal.NodeEventMessage:
s.Cluster.ReceiveEvent(DecodeNodeEvent(obj))
}
@ -569,14 +577,14 @@ func (s *Server) LocalStatus() (proto.Message, error) {
if s.Cluster == nil {
return nil, errors.New("Server.Cluster is nil")
}
if s.Holder == nil {
if s.holder == nil {
return nil, errors.New("Server.Holder is nil")
}
ns := internal.NodeStatus{
Node: EncodeNode(s.Cluster.Node),
MaxSlices: s.Holder.EncodeMaxSlices(),
Schema: s.Holder.EncodeSchema(),
MaxSlices: s.holder.EncodeMaxSlices(),
Schema: s.holder.EncodeSchema(),
}
return &ns, nil
@ -596,7 +604,7 @@ func (s *Server) HandleRemoteStatus(pb proto.Message) error {
go func() {
// Make sure the holder has opened.
<-s.Holder.opened
<-s.holder.opened
err := s.mergeRemoteStatus(pb.(*internal.NodeStatus))
if err != nil {
@ -614,14 +622,14 @@ func (s *Server) mergeRemoteStatus(ns *internal.NodeStatus) error {
}
// Sync schema.
if err := s.Holder.ApplySchema(ns.Schema); err != nil {
if err := s.holder.ApplySchema(ns.Schema); err != nil {
return errors.Wrap(err, "applying schema")
}
// Sync maxSlices.
oldmaxslices := s.Holder.MaxSlices()
oldmaxslices := s.holder.MaxSlices()
for index, newMax := range ns.MaxSlices.Standard {
localIndex := s.Holder.Index(index)
localIndex := s.holder.Index(index)
// if we don't know about an index locally, log an error because
// indexes should be created and synced prior to slice creation
if localIndex == nil {
@ -709,26 +717,26 @@ func (s *Server) monitorRuntime() {
return
case <-s.gcNotifier.AfterGC():
// GC just ran.
s.Holder.Stats.Count("garbage_collection", 1, 1.0)
s.holder.Stats.Count("garbage_collection", 1, 1.0)
case <-ticker.C:
}
// Record the number of go routines.
s.Holder.Stats.Gauge("goroutines", float64(runtime.NumGoroutine()), 1.0)
s.holder.Stats.Gauge("goroutines", float64(runtime.NumGoroutine()), 1.0)
openFiles, err := countOpenFiles()
// Open File handles.
if err == nil {
s.Holder.Stats.Gauge("OpenFiles", float64(openFiles), 1.0)
s.holder.Stats.Gauge("OpenFiles", float64(openFiles), 1.0)
}
// Runtime memory metrics.
runtime.ReadMemStats(&m)
s.Holder.Stats.Gauge("HeapAlloc", float64(m.HeapAlloc), 1.0)
s.Holder.Stats.Gauge("HeapInuse", float64(m.HeapInuse), 1.0)
s.Holder.Stats.Gauge("StackInuse", float64(m.StackInuse), 1.0)
s.Holder.Stats.Gauge("Mallocs", float64(m.Mallocs), 1.0)
s.Holder.Stats.Gauge("Frees", float64(m.Frees), 1.0)
s.holder.Stats.Gauge("HeapAlloc", float64(m.HeapAlloc), 1.0)
s.holder.Stats.Gauge("HeapInuse", float64(m.HeapInuse), 1.0)
s.holder.Stats.Gauge("StackInuse", float64(m.StackInuse), 1.0)
s.holder.Stats.Gauge("Mallocs", float64(m.Mallocs), 1.0)
s.holder.Stats.Gauge("Frees", float64(m.Frees), 1.0)
}
}

31
server/server.go
View file

@ -65,10 +65,10 @@ type Command struct {
// Standard input/output
*pilosa.CmdIO
// Started will be closed once Command.Run is finished.
// Started will be closed once Command.Start is finished.
Started chan struct{}
// Done will be closed when Command.Close() is called
Done chan struct{}
// done will be closed when Command.Close() is called
done chan struct{}
// Passed to the Gossip implementation.
logOutput io.Writer
@ -83,7 +83,7 @@ func NewCommand(stdin io.Reader, stdout, stderr io.Writer) *Command {
CmdIO: pilosa.NewCmdIO(stdin, stdout, stderr),
Started: make(chan struct{}),
Done: make(chan struct{}),
done: make(chan struct{}),
}
}
@ -125,7 +125,7 @@ func (m *Command) Wait() error {
// Second signal causes a hard shutdown.
go func() { <-c; os.Exit(1) }()
return errors.Wrap(m.Close(), "closing command")
case <-m.Done:
case <-m.done:
m.logger.Printf("Server closed externally")
return nil
}
@ -242,6 +242,7 @@ func (m *Command) SetupServer() error {
pilosa.OptServerURI(uri),
pilosa.OptServerInternalClient(http.NewInternalClientFromURI(uri, c)),
pilosa.OptServerPrimaryTranslateStore(primaryTranslateStore),
pilosa.OptServerClusterDisabled(m.Config.Cluster.Disabled, m.Config.Cluster.Hosts),
)
return errors.Wrap(err, "new server")
@ -249,25 +250,7 @@ func (m *Command) SetupServer() error {
// SetupNetworking sets up internode communication based on the configuration.
func (m *Command) SetupNetworking() error {
m.Server.NodeID = m.Server.LoadNodeID()
if m.Config.Cluster.Disabled {
m.Server.Cluster.Static = true
m.Server.Cluster.Coordinator = m.Server.NodeID
for _, address := range m.Config.Cluster.Hosts {
uri, err := pilosa.NewURIFromAddress(address)
if err != nil {
return errors.Wrap(err, "getting URI")
}
m.Server.Cluster.Nodes = append(m.Server.Cluster.Nodes, &pilosa.Node{
URI: *uri,
})
}
m.Server.Broadcaster = pilosa.NopBroadcaster
m.Server.Cluster.MemberSet = pilosa.NewStaticMemberSet(m.Server.Cluster.Nodes)
m.Server.BroadcastReceiver = pilosa.NopBroadcastReceiver
return nil
}
@ -322,7 +305,7 @@ func (m *Command) Close() error {
if closer, ok := m.logOutput.(io.Closer); ok {
logErr = closer.Close()
}
close(m.Done)
close(m.done)
if serveErr != nil && logErr != nil {
return fmt.Errorf("closing server: '%v', closing logs: '%v'", serveErr, logErr)
} else if logErr != nil {

10
test/frame.go → test/field.go
View file

@ -18,7 +18,6 @@ import (
"io/ioutil"
"os"
"testing"
"time"
"github.com/pilosa/pilosa"
)
@ -75,15 +74,6 @@ func (f *Field) Reopen() error {
return nil
}
// MustSetBit sets a bit on the field. Panic on error.
func (f *Field) MustSetBit(view string, rowID, columnID uint64, t *time.Time) (changed bool) {
changed, err := f.SetBit(view, rowID, columnID, t)
if err != nil {
panic(err)
}
return changed
}
// Ensure field can set its cache
func TestField_SetCacheSize(t *testing.T) {
f := MustOpenField()

4
test/holder.go
View file

@ -142,7 +142,7 @@ func (h *Holder) SetBit(index, field string, rowID, columnID uint64) {
if err != nil {
panic(err)
}
f.SetBit(pilosa.ViewStandard, rowID, columnID, nil)
f.SetBit(rowID, columnID, nil)
}
// ClearBit clears a bit on the given field.
@ -152,7 +152,7 @@ func (h *Holder) ClearBit(index, field string, rowID, columnID uint64) {
if err != nil {
panic(err)
}
f.ClearBit(pilosa.ViewStandard, rowID, columnID, nil)
f.ClearBit(rowID, columnID, nil)
}
// MustSetBits sets columns on a row. Panic on error.

4
test/pilosa.go
View file

@ -25,7 +25,6 @@ import (
"testing"
"time"
"github.com/pilosa/pilosa/boltdb"
"github.com/pilosa/pilosa/gossip"
"github.com/pilosa/pilosa/http"
"github.com/pilosa/pilosa/server"
@ -164,9 +163,6 @@ func (m *Main) Reopen() error {
return errors.Wrap(err, "setting up server")
}
m.Server.NewAttrStore = boltdb.NewAttrStore
m.Server.Holder.NewAttrStore = m.Server.NewAttrStore
// Run new program.
if err := m.Start(); err != nil {
return err

10
utils_internal_test.go
View file

@ -117,7 +117,7 @@ func (t *ClusterCluster) CreateField(index, field string, opt FieldOptions) erro
return nil
}
func (t *ClusterCluster) SetBit(index, field, view string, rowID, colID uint64, x *time.Time) error {
func (t *ClusterCluster) SetBit(index, field string, rowID, colID uint64, x *time.Time) error {
// Determine which node should receive the SetBit.
c0 := t.Clusters[0] // use the first node's cluster to determine slice location.
slice := colID / SliceWidth
@ -132,7 +132,7 @@ func (t *ClusterCluster) SetBit(index, field, view string, rowID, colID uint64,
if f == nil {
return fmt.Errorf("index/field does not exist: %s/%s", index, field)
}
_, err := f.SetBit(view, rowID, colID, x)
_, err := f.SetBit(rowID, colID, x)
if err != nil {
return err
}
@ -150,11 +150,11 @@ func (t *ClusterCluster) clusterByID(id string) *Cluster {
return nil
}
// AddNode adds a node to the cluster and (potentially) starts a resize job.
func (t *ClusterCluster) addNode(saveTopology bool) error {
// addNode adds a node to the cluster and (potentially) starts a resize job.
func (t *ClusterCluster) addNode() error {
id := len(t.Clusters)
c, err := t.addCluster(id, saveTopology)
c, err := t.addCluster(id, false)
if err != nil {
return err
}