From 8228ef238203eaebc0a41ad62ad38d390bc44deb Mon Sep 17 00:00:00 2001
From: Travis Turner <travis@pilosa.com>
Date: Tue, 28 Nov 2017 16:40:05 -0600
Subject: [PATCH] Add a Cluster.Static override for tests to treat static nodes
 as Coordinator. Refactor the Server.joining channel to be in Cluster instead.

---
 cluster.go       | 32 +++++++++++++++++++++++++++++++-
 server.go        | 21 +--------------------
 server/server.go | 14 +++++++++++++-
 3 files changed, 45 insertions(+), 22 deletions(-)

diff --git a/cluster.go b/cluster.go
index 6d02d352b..50996d6f1 100644
--- a/cluster.go
+++ b/cluster.go
@@ -166,6 +166,7 @@ type Cluster struct {
 	Topology *Topology
 
 	// Required for cluster Resize.
+	Static      bool // Static is primarily used for testing in a non-gossip environment.
 	State       string
 	Coordinator URI
 	Holder      *Holder
@@ -173,6 +174,11 @@ type Cluster struct {
 
 	joiningLeavingNodes chan nodeAction
 
+	// joining is held open until this node
+	// receives ClusterStatus from the coordinator.
+	joining chan struct{}
+	joined  bool
+
 	mu         sync.RWMutex
 	jobs       map[int64]*ResizeJob
 	currentJob *ResizeJob
@@ -197,6 +203,7 @@ func NewCluster() *Cluster {
 		joiningLeavingNodes: make(chan nodeAction, 10), // buffered channel
 		jobs:                make(map[int64]*ResizeJob),
 		closing:             make(chan struct{}),
+		joining:             make(chan struct{}),
 
 		LogOutput: os.Stderr,
 		prefect:   &NopSecurityManager{},
@@ -210,7 +217,7 @@ func (c *Cluster) logger() *log.Logger {
 
 // IsCoordinator is true if this node is the coordinator.
 func (c *Cluster) IsCoordinator() bool {
-	return c.Coordinator == c.URI
+	return c.Static || c.Coordinator == c.URI
 }
 
 // SetCoordinator updates the Coordinator to new if it is
@@ -709,6 +716,12 @@ func (c *Cluster) Open() error {
 		return fmt.Errorf("opening MemberSet: %v", err)
 	}
 
+	// If not coordinator then wait for ClusterStatus from coordinator.
+	if !c.IsCoordinator() {
+		c.logger().Printf("wait for joining to complete")
+		<-c.joining
+	}
+
 	return nil
 }
 
@@ -720,15 +733,28 @@ func (c *Cluster) Close() error {
 	return nil
 }
 
+func (c *Cluster) MarkAsJoined() {
+	if !c.joined {
+		c.joined = true
+		close(c.joining)
+	}
+}
+
 func (c *Cluster) needTopologyAgreement() bool {
 	return c.State == ClusterStateStarting && !URISlicesAreEqual(c.Topology.NodeSet, c.NodeSet())
 }
 
 func (c *Cluster) haveTopologyAgreement() bool {
+	if c.Static {
+		return true
+	}
 	return URISlicesAreEqual(c.Topology.NodeSet, c.NodeSet())
 }
 
 func (c *Cluster) allNodesReady() bool {
+	if c.Static {
+		return true
+	}
 	for _, uri := range c.Topology.NodeSet {
 		if c.Topology.nodeStates[uri] != NodeStateReady {
 			return false
@@ -1335,6 +1361,10 @@ func decodeTopology(topology *internal.Topology) (*Topology, error) {
 }
 
 func (c *Cluster) considerTopology() error {
+	if c.Static {
+		return nil
+	}
+
 	// If there is no .topology file, it's safe to proceed.
 	if len(c.Topology.NodeSet) == 0 {
 		return nil
diff --git a/server.go b/server.go
index 8a4ad38d5..37270b0f1 100644
--- a/server.go
+++ b/server.go
@@ -51,11 +51,6 @@ type Server struct {
 	wg      sync.WaitGroup
 	closing chan struct{}
 
-	// joining is held open until this node
-	// receives ClusterStatus from the coordinator.
-	joining chan struct{}
-	joined  bool
-
 	// Data storage and HTTP interface.
 	Holder            *Holder
 	Handler           *Handler
@@ -89,7 +84,6 @@ type Server struct {
 func NewServer() *Server {
 	s := &Server{
 		closing: make(chan struct{}),
-		joining: make(chan struct{}),
 
 		Holder:            NewHolder(),
 		Handler:           NewHandler(),
@@ -198,12 +192,6 @@ func (s *Server) Open() error {
 		return fmt.Errorf("opening Cluster: %v", err)
 	}
 
-	// If not coordinator then wait for ClusterStatus from coordinator.
-	if !s.Cluster.IsCoordinator() {
-		s.Logger().Printf("wait for joining to complete")
-		<-s.joining
-	}
-
 	// Open holder.
 	if err := s.Holder.Open(); err != nil {
 		return fmt.Errorf("opening Holder: %v", err)
@@ -228,13 +216,6 @@ func (s *Server) Open() error {
 	return nil
 }
 
-func (s *Server) markAsJoined() {
-	if !s.joined {
-		s.joined = true
-		close(s.joining)
-	}
-}
-
 // Close closes the server and waits for it to shutdown.
 func (s *Server) Close() error {
 	// Notify goroutines to stop.
@@ -370,7 +351,7 @@ func (s *Server) ReceiveMessage(pb proto.Message) error {
 		if err != nil {
 			return err
 		}
-		s.markAsJoined()
+		s.Cluster.MarkAsJoined()
 	case *internal.ResizeInstruction:
 		err := s.Cluster.FollowResizeInstruction(obj)
 		if err != nil {
diff --git a/server/server.go b/server/server.go
index b4d067683..5237df085 100644
--- a/server/server.go
+++ b/server/server.go
@@ -63,7 +63,7 @@ type Command struct {
 	// Standard input/output
 	*pilosa.CmdIO
 
-	// running will be closed once Command.Run is finished.
+	// Started will be closed once Command.Run is finished.
 	Started chan struct{}
 	// Done will be closed when Command.Close() is called
 	Done chan struct{}
@@ -220,6 +220,18 @@ func (m *Command) SetupServer() error {
 		m.Server.Broadcaster = gossipMemberSet
 		m.Server.BroadcastReceiver = gossipMemberSet
 	case pilosa.ClusterStatic, pilosa.ClusterNone:
+
+		m.Server.Cluster.Static = true
+		for _, address := range m.Config.Cluster.Hosts {
+			uri, err := pilosa.NewURIFromAddress(address)
+			if err != nil {
+				return err
+			}
+			cluster.Nodes = append(cluster.Nodes, &pilosa.Node{
+				URI: *uri,
+			})
+		}
+
 		m.Server.Broadcaster = pilosa.NopBroadcaster
 		m.Server.Cluster.MemberSet = pilosa.NewStaticMemberSet()
 		m.Server.BroadcastReceiver = pilosa.NopBroadcastReceiver