diff --git a/cluster.go b/cluster.go index be668a6bc..bbf6f799e 100644 --- a/cluster.go +++ b/cluster.go @@ -17,7 +17,6 @@ package pilosa import ( "context" "encoding/binary" - "errors" "fmt" "hash/fnv" "io/ioutil" @@ -33,6 +32,7 @@ import ( "github.com/gogo/protobuf/proto" "github.com/pilosa/pilosa/internal" + "github.com/pkg/errors" uuid "github.com/satori/go.uuid" ) @@ -899,7 +899,10 @@ func (c *Cluster) Open() error { } // Add the local node to the cluster. - c.AddNode(c.Node) + err := c.AddNode(c.Node) + if err != nil { + return errors.Wrap(err, "adding local node") + } // Start the EventReceiver. if err := c.EventReceiver.Start(c); err != nil { @@ -1684,13 +1687,15 @@ func (c *Cluster) nodeJoin(node *Node) error { // Only change to normal if there is no existing data. Otherwise, // the coordinator needs to wait to receive READY messages (nodeStates) // from remote nodes before setting the cluster to state NORMAL. - if !c.Holder.HasData() { + if ok, err := c.Holder.HasData(); !ok && err == nil { // If the result of the previous AddNode completed the joining of nodes // in the topology, then change the state to NORMAL. if c.haveTopologyAgreement() { return c.setStateAndBroadcast(ClusterStateNormal) } return nil + } else if err != nil { + return errors.Wrap(err, "checking if holder has data") } if c.haveTopologyAgreement() && c.allNodesReady() { @@ -1712,11 +1717,13 @@ func (c *Cluster) nodeJoin(node *Node) error { } // If the holder does not yet contain data, go ahead and add the node. - if !c.Holder.HasData() { + if ok, err := c.Holder.HasData(); !ok && err == nil { if err := c.AddNode(node); err != nil { return err } return c.setStateAndBroadcast(ClusterStateNormal) + } else if err != nil { + return errors.Wrap(err, "checking if holder has data2") } // If the cluster has data, we need to change to RESIZING and @@ -1770,11 +1777,13 @@ func (c *Cluster) nodeLeave(node *Node) error { } // If the holder does not yet contain data, go ahead and remove the node. - if !c.Holder.HasData() { + if ok, err := c.Holder.HasData(); !ok && err == nil { if err := c.RemoveNode(n); err != nil { return err } return c.setStateAndBroadcast(ClusterStateNormal) + } else if err != nil { + return errors.Wrap(err, "checking if holder has data") } // If the cluster has data then change state to RESIZING and diff --git a/holder.go b/holder.go index 2279a3e19..7649caeba 100644 --- a/holder.go +++ b/holder.go @@ -16,7 +16,6 @@ package pilosa import ( "context" - "errors" "fmt" "io/ioutil" "net/http" @@ -30,6 +29,7 @@ import ( "time" "github.com/pilosa/pilosa/internal" + "github.com/pkg/errors" uuid "github.com/satori/go.uuid" ) @@ -47,7 +47,6 @@ type Holder struct { // Indexes by name. indexes map[string]*Index - hasData bool // opened channel is closed once Open() completes. opened chan struct{} @@ -91,36 +90,6 @@ func NewHolder() *Holder { } } -// Peek reads the root data directory for the holder -// without actually loading any data into memory. -// HasData is returned, and h.hasData is set. -func (h *Holder) Peek() bool { - h.Logger.Printf("peek at holder path: %s", h.Path) - h.hasData = false - - // Open path to read all index directories. - f, err := os.Open(h.Path) - if err != nil { - return false - } - defer f.Close() - - fis, err := f.Readdir(0) - if err != nil { - return false - } - - for _, fi := range fis { - if !fi.IsDir() { - continue - } - h.hasData = true - break - } - - return h.hasData -} - // Open initializes the root data directory for the holder. func (h *Holder) Open() error { h.setFileLimit() @@ -198,10 +167,37 @@ func (h *Holder) Close() error { // HasData returns true if Holder contains at least one index. // This is used to determine if the rebalancing of data is necessary // when a node joins the cluster. -func (h *Holder) HasData() bool { - h.mu.RLock() - defer h.mu.RUnlock() - return h.hasData || len(h.indexes) > 0 +func (h *Holder) HasData() (bool, error) { + h.mu.Lock() + defer h.mu.Unlock() + if len(h.indexes) > 0 { + return true, nil + } + // Open path to read all index directories. + if _, err := os.Stat(h.Path); os.IsNotExist(err) { + return false, nil + } else if err != nil { + return false, errors.Wrap(err, "statting data dir") + } + + f, err := os.Open(h.Path) + if err != nil { + return false, errors.Wrap(err, "opening data dir") + } + defer f.Close() + + fis, err := f.Readdir(0) + if err != nil { + return false, errors.Wrap(err, "reading data dir") + } + + for _, fi := range fis { + if !fi.IsDir() { + continue + } + return true, nil + } + return false, nil } // MaxSlices returns MaxSlice map for all indexes. diff --git a/holder_test.go b/holder_test.go index 7053fa1f2..e819c6e94 100644 --- a/holder_test.go +++ b/holder_test.go @@ -283,26 +283,24 @@ func TestHolder_HasData(t *testing.T) { h := test.MustOpenHolder() defer h.Close() - if h.HasData() { - t.Fatal("expected HasData to return false") + if ok, err := h.HasData(); ok || err != nil { + t.Fatal("expected HasData to return false, no err, but", ok, err) } if _, err := h.CreateIndex("test", pilosa.IndexOptions{}); err != nil { t.Fatal(err) } - if !h.HasData() { - t.Fatal("expected HasData to return true") + if ok, err := h.HasData(); !ok || err != nil { + t.Fatal("expected HasData to return true, but ", ok, err) } }) t.Run("Peek", func(t *testing.T) { h := test.NewHolder() - if hasData := h.Peek(); hasData != false { - t.Fatal("expected Peek to return false") - } else if h.HasData() { - t.Fatal("expected HasData to return false") + if ok, err := h.HasData(); ok || err != nil { + t.Fatal("expected HasData to return false, no err, but", ok, err) } // Create an index directory to indicate data exists. @@ -310,24 +308,19 @@ func TestHolder_HasData(t *testing.T) { t.Fatal(err) } - if hasData := h.Peek(); hasData != true { - t.Fatal("expected Peek to return true") - } else if !h.HasData() { - t.Fatal("expected HasData to return true") + if ok, err := h.HasData(); !ok || err != nil { + t.Fatal("expected HasData to return true, no err, but", ok, err) } }) t.Run("Peek at missing directory", func(t *testing.T) { h := test.NewHolder() - // Ensure that hasData is false when trying to peek into - // a directory that doesn't exist. + // Ensure that hasData is false when dir doesn't exist. h.Path = "bad-path" - if hasData := h.Peek(); hasData != false { - t.Fatal("expected Peek to return false") - } else if h.HasData() { - t.Fatal("expected HasData to return false") + if ok, err := h.HasData(); ok || err != nil { + t.Fatal("expected HasData to return false, no err, but", ok, err) } }) } diff --git a/server.go b/server.go index d302a7d97..2b7903fb6 100644 --- a/server.go +++ b/server.go @@ -54,6 +54,7 @@ type Server struct { Holder *Holder Cluster *Cluster diagnostics *DiagnosticsCollector + executor *Executor // External handler *Handler @@ -161,7 +162,8 @@ func OptServerGCNotifier(gcn GCNotifier) ServerOption { func OptServerRemoteClient(c *http.Client) ServerOption { return func(s *Server) error { - s.remoteClient = c + s.executor = NewExecutor(c) + s.defaultClient = NewInternalHTTPClientFromURI(nil, c) s.Cluster.RemoteClient = c return nil } @@ -231,11 +233,28 @@ func NewServer(opts ...ServerOption) (*Server, error) { s.Cluster.Logger = s.logger s.Cluster.Holder = s.Holder - s.Cluster.RemoteClient = s.remoteClient + // update URI port with actual listener port. TODO this should probably be done outside of here. if s.URI.Port() == 0 { s.URI.SetPort(uint16(s.ln.Addr().(*net.TCPAddr).Port)) } + + s.NodeID = s.LoadNodeID() + // Set Cluster Node. + node := &Node{ + ID: s.NodeID, + URI: s.URI, + IsCoordinator: s.Cluster.Coordinator == s.NodeID, + } + s.Cluster.Node = node + 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.MaxWritesPerRequest = s.maxWritesPerRequest + s.handler.API.Executor = s.executor + return s, nil } @@ -248,33 +267,12 @@ func (s *Server) Open() error { } // Get or create NodeID. - s.NodeID = s.LoadNodeID() - - // Set Cluster Node. - node := &Node{ - ID: s.NodeID, - URI: s.URI, - IsCoordinator: s.Cluster.Coordinator == s.NodeID, - } - s.Cluster.Node = node // Append the NodeID tag to stats. - s.Holder.Stats = s.Holder.Stats.WithTags(fmt.Sprintf("NodeID:%s", s.NodeID)) - - // Peek at the holder to determine if there is data on disk. - // Don't actually load the data until after the Cluster - // management starts. - s.Holder.Peek() // Create default HTTP client - s.createDefaultClient(s.remoteClient) // Create executor for executing queries. - e := NewExecutor(s.remoteClient) - e.Holder = s.Holder - e.Node = node - e.Cluster = s.Cluster - e.MaxWritesPerRequest = s.maxWritesPerRequest // Cluster settings. s.Cluster.Broadcaster = s.Broadcaster @@ -287,7 +285,6 @@ func (s *Server) Open() error { s.handler.API.StatusHandler = s s.handler.API.URI = s.URI s.handler.API.Cluster = s.Cluster - s.handler.API.Executor = e // Initialize Holder. s.Holder.Broadcaster = s.Broadcaster @@ -750,10 +747,6 @@ func (s *Server) monitorRuntime() { } } -func (s *Server) createDefaultClient(remoteClient *http.Client) { - s.defaultClient = NewInternalHTTPClientFromURI(nil, remoteClient) -} - // CountOpenFiles on operating systems that support lsof. func CountOpenFiles() (int, error) { switch runtime.GOOS { diff --git a/server/server.go b/server/server.go index 7f5782211..8fd699ce5 100644 --- a/server/server.go +++ b/server/server.go @@ -305,6 +305,7 @@ func (m *Command) SetupNetworking() error { // Set Coordinator. if m.Config.Cluster.Coordinator || len(m.Config.Gossip.Seeds) == 0 { m.Server.Cluster.Coordinator = m.Server.NodeID + m.Server.Cluster.Node.IsCoordinator = true } gossipEventReceiver := gossip.NewGossipEventReceiver(m.logger)