mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-08-28 10:54:59 +00:00
remove remaining external ref to Server.Holder and unexport holder
This commit is contained in:
Matt Jaffee
parent
7a05f32a17
commit
44d3e87d4e
3 changed files with 45 additions and 48 deletions
|
|
@ -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() {
|
||||
|
|
|
|||
88
server.go
88
server.go
|
|
@ -52,7 +52,7 @@ type Server struct {
|
|||
closing chan struct{}
|
||||
|
||||
// Internal
|
||||
Holder *Holder
|
||||
holder *Holder
|
||||
Cluster *Cluster
|
||||
TranslateFile *TranslateFile
|
||||
diagnostics *DiagnosticsCollector
|
||||
|
|
@ -108,7 +108,7 @@ func OptServerDataDir(dir string) ServerOption {
|
|||
|
||||
func OptServerAttrStoreFunc(af func(string) AttrStore) ServerOption {
|
||||
return func(s *Server) error {
|
||||
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
|
||||
}
|
||||
}
|
||||
|
|
@ -222,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),
|
||||
|
|
@ -250,13 +250,13 @@ 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()
|
||||
|
|
@ -288,9 +288,9 @@ 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
|
||||
|
|
@ -309,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"))
|
||||
}
|
||||
|
|
@ -320,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)
|
||||
|
|
@ -343,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 {
|
||||
|
|
@ -378,8 +378,8 @@ 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()
|
||||
|
|
@ -394,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
|
||||
|
|
@ -422,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 {
|
||||
|
|
@ -444,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)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -452,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)
|
||||
}
|
||||
|
|
@ -478,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)
|
||||
}
|
||||
|
|
@ -492,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)
|
||||
}
|
||||
|
|
@ -525,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))
|
||||
}
|
||||
|
|
@ -577,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
|
||||
|
|
@ -604,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 {
|
||||
|
|
@ -622,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 {
|
||||
|
|
@ -717,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)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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,8 +163,6 @@ func (m *Main) Reopen() error {
|
|||
return errors.Wrap(err, "setting up server")
|
||||
}
|
||||
|
||||
m.Server.Holder.NewAttrStore = boltdb.NewAttrStore
|
||||
|
||||
// Run new program.
|
||||
if err := m.Start(); err != nil {
|
||||
return err
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue