diff --git a/diagnostics.go b/diagnostics.go index e4067c6e3..b36335b38 100644 --- a/diagnostics.go +++ b/diagnostics.go @@ -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() { diff --git a/server.go b/server.go index c544ae6fc..1d301df35 100644 --- a/server.go +++ b/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) } } diff --git a/test/pilosa.go b/test/pilosa.go index a65207c91..bd1628aa6 100644 --- a/test/pilosa.go +++ b/test/pilosa.go @@ -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