diff --git a/holder.go b/holder.go index 05d21b3c7..0a1af820b 100644 --- a/holder.go +++ b/holder.go @@ -590,7 +590,8 @@ func (h *Holder) setPrimaryTranslateStore(node *Node) { if node != nil { nodeID = node.ID } - h.translateFile.SetPrimaryStore(nodeID, h.NewPrimaryTranslateStore(node)) + ts := h.NewPrimaryTranslateStore(node) + h.translateFile.SetPrimaryStore(nodeID, ts) } // holderSyncer is an active anti-entropy tool that compares the local holder diff --git a/http/translator.go b/http/translator.go index 7ca85a33f..3a2acae04 100644 --- a/http/translator.go +++ b/http/translator.go @@ -6,13 +6,11 @@ import ( "fmt" "io" "io/ioutil" - "log" "net/http" "net/url" "strconv" "github.com/pilosa/pilosa" - "github.com/pkg/errors" ) // Ensure implementation implements inteface. @@ -27,13 +25,13 @@ type translateStore struct { // NewTranslateStore returns a new instance of TranslateStore based on node. // DEPRECATED: Providing a string url to this function is being deprecated. Instead, // provide a *pilosa.Node. +// TODO (2.0) Refactor to avoid panic func NewTranslateStore(node interface{}) pilosa.TranslateStore { var n *pilosa.Node switch v := node.(type) { case string: - log.Printf("WARNING: providing a string url to NewTranslateStore() has been deprecated.") if uri, err := pilosa.NewURIFromAddress(v); err != nil { - log.Println(errors.Wrap(err, "creating uri")) + panic("bad uri for translatestore in deprecated api") } else { n = &pilosa.Node{ ID: v, @@ -43,7 +41,7 @@ func NewTranslateStore(node interface{}) pilosa.TranslateStore { case *pilosa.Node: n = v default: - log.Printf("WARNING: a *pilosa.Node is the only type supported by NewTranslateStore().") + panic("*pilosa.Node is the only type supported by NewTranslateStore().") } return &translateStore{node: n} } diff --git a/http/translator_test.go b/http/translator_test.go index 18ba7c998..ce96b3c2c 100644 --- a/http/translator_test.go +++ b/http/translator_test.go @@ -124,7 +124,8 @@ func TestTranslateStore_Reader(t *testing.T) { primary := test.MustRunCluster(t, 1, []server.CommandOption{opts})[0] defer primary.Close() - _, err := http.NewTranslateStore(primary.URL()).Reader(context.Background(), 0) + ts := http.NewTranslateStore(primary.URL()) + _, err := ts.Reader(context.Background(), 0) if err != pilosa.ErrNotImplemented { t.Fatalf("unexpected error: %s", err) } diff --git a/server.go b/server.go index 70391723e..a4cb8fc8b 100644 --- a/server.go +++ b/server.go @@ -169,6 +169,7 @@ func OptServerPrimaryTranslateStore(store TranslateStore) ServerOption { } func OptServerPrimaryTranslateStoreFunc(tf func(interface{}) TranslateStore) ServerOption { + return func(s *Server) error { s.holder.NewPrimaryTranslateStore = tf return nil @@ -270,6 +271,7 @@ func NewServer(opts ...ServerOption) (*Server, error) { return nil, errors.Wrap(err, "applying option") } } + s.holder.translateFile.logger = s.logger path, err := expandDirName(s.dataDir) if err != nil { diff --git a/translate.go b/translate.go index cb16de929..86ed6c914 100644 --- a/translate.go +++ b/translate.go @@ -8,7 +8,6 @@ import ( "fmt" "io" "io/ioutil" - "log" "os" "path/filepath" "sync" @@ -69,7 +68,7 @@ type TranslateFile struct { Path string mapSize int - + logger Logger // If non-nil, data is streamed from a primary and this is a read-only store. PrimaryTranslateStore TranslateStore primaryID string // unique ID used to identify the primary store @@ -90,6 +89,12 @@ func OptTranslateFileMapSize(mapSize int) TranslateFileOption { return nil } } +func OptTranslateFileLogger(l Logger) TranslateFileOption { + return func(s *TranslateFile) error { + s.logger = l + return nil + } +} // NewTranslateFile returns a new instance of TranslateFile. func NewTranslateFile(opts ...TranslateFileOption) *TranslateFile { @@ -111,6 +116,8 @@ func NewTranslateFile(opts ...TranslateFileOption) *TranslateFile { mapSize: defaultMapSize, + logger: NopLogger, + replicationClosing: make(chan struct{}), primaryStoreEvents: make(chan primaryStoreEvent), @@ -187,12 +194,12 @@ func (s *TranslateFile) handlePrimaryStoreEvent(ev primaryStoreEvent) error { } // Stop translate store replication. - log.Printf("stop monitor replication") + s.logger.Printf("stop monitor replication") close(s.replicationClosing) s.repWG.Wait() // Set the primary node for translate store replication. - log.Printf("set primary translate store to %s", ev.id) + s.logger.Printf("set primary translate store to %s", ev.id) s.primaryID = ev.id if ev.id == "" { s.PrimaryTranslateStore = nil @@ -201,7 +208,7 @@ func (s *TranslateFile) handlePrimaryStoreEvent(ev primaryStoreEvent) error { } // Start translate store replication. Stream from primary, if available. - log.Printf("start monitor replication") + s.logger.Printf("start monitor replication") if s.PrimaryTranslateStore != nil { s.replicationClosing = make(chan struct{}) s.repWG.Add(1) @@ -363,14 +370,14 @@ func (s *TranslateFile) monitorReplication() { // Keep attempting to replicate until the store closes. for { if err := s.replicate(ctx); err != nil { - log.Printf("pilosa: replication error: %s", err) + s.logger.Printf("pilosa: replication error: %s", err) } select { case <-ctx.Done(): return case <-time.After(s.replicationRetryInterval): - log.Printf("pilosa: reconnecting to primary replica") + s.logger.Printf("pilosa: reconnecting to primary replica") } } } @@ -378,7 +385,7 @@ func (s *TranslateFile) monitorReplication() { // monitorPrimaryStoreEvents is executed in a separate goroutine and listens for changes // to the primary store assignment. func (s *TranslateFile) monitorPrimaryStoreEvents() { - log.Printf("monitor primary store events") + s.logger.Printf("monitor primary store events") // Keep handling events until the store closes. for { select { @@ -386,7 +393,7 @@ func (s *TranslateFile) monitorPrimaryStoreEvents() { return case ev := <-s.primaryStoreEvents: if err := s.handlePrimaryStoreEvent(ev); err != nil { - log.Printf("handle primary store event") + s.logger.Printf("handle primary store event") } } } @@ -396,7 +403,7 @@ func (s *TranslateFile) replicate(ctx context.Context) error { off := s.size() // Connect to remote primary. - log.Printf("pilosa: replicating from offset %d", off) + s.logger.Printf("pilosa: replicating from offset %d", off) rc, err := s.PrimaryTranslateStore.Reader(ctx, off) if err != nil { return err