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

Merge pull request #1397 from jaffee/server-cleanup

Browse source 
Server cleanup
This commit is contained in:
Matthew Jaffee 2018-06-20 12:32:09 -05:00 • committed by GitHub
commit 57da20728e
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 60 additions and 68 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

110
server.go
View file

@ -52,9 +52,9 @@ type Server struct {
closing chan struct{}
// Internal
Holder *Holder
holder *Holder
Cluster *Cluster
TranslateFile *TranslateFile
translateFile *TranslateFile
diagnostics *DiagnosticsCollector
executor *Executor
hosts []string
@ -66,7 +66,6 @@ type Server struct {
BroadcastReceiver BroadcastReceiver
systemInfo SystemInfo
gcNotifier GCNotifier
NewAttrStore func(string) AttrStore
logger Logger
ln net.Listener
@ -109,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
}
}
@ -182,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
}
}
@ -224,7 +222,7 @@ 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),
@ -232,8 +230,6 @@ func NewServer(opts ...ServerOption) (*Server, error) {
gcNotifier: NopGCNotifier,
NewAttrStore: NewNopAttrStore,
antiEntropyInterval: time.Minute * 10,
metricInterval: 0,
diagnosticInterval: 0,
@ -254,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
}
@ -292,15 +288,15 @@ func NewServer(opts ...ServerOption) (*Server, error) {
}
// Append the NodeID tag to stats.
s.Holder.Stats = s.Holder.Stats.WithTags(fmt.Sprintf("NodeID:%s", s.NodeID))
s.holder.Stats = s.holder.Stats.WithTags(fmt.Sprintf("NodeID:%s", s.NodeID))
s.executor.Holder = s.Holder
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
}
@ -313,7 +309,7 @@ func (s *Server) Open() error {
}
// Log startup
err := s.Holder.logStartup()
err := s.holder.logStartup()
if err != nil {
log.Println(errors.Wrap(err, "logging startup"))
}
@ -324,14 +320,14 @@ func (s *Server) Open() error {
// 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)
@ -347,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 {
@ -382,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
@ -398,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
@ -426,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 {
@ -448,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)
}
}
@ -456,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)
}
@ -482,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)
}
@ -496,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)
}
@ -529,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))
}
@ -581,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
@ -608,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 {
@ -626,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 {
@ -721,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)
}
}

12
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
}
@ -305,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 {

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