remove holder.Peek, combine with HasData, move server logic

Server initializing happens more in NewServer than Open now - expecting to
continue this trend. goal was to remove remoteClient from Server (since it has a
defaultClient) as well, but we'll have to refactor the client usage in fragment
and frame first.

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

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.

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)
 		}
 	})
 }

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 {

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)