refactor resize instruction logic to support multi-index. wait to open holder on non-coordinator nodes.

cluster.go

@@ -876,33 +876,43 @@ func (c *Cluster) generateResizeJobByAction(nodeAction nodeAction) (*ResizeJob,
 
 	pbSchema := c.Holder.EncodeSchema()
 
-	// Add to the ResizeJob the instructions for each index.
+	// multiIndex is a map of sources for each node in toCluster.
+	// Initialize the map with all the nodes in toCluster.
+	multiIndex := make(map[URI][]*internal.ResizeSource)
+	for _, n := range toCluster.Nodes {
+		multiIndex[n.URI] = nil
+	}
+
+	// Add to m the instructions for each index.
 	for _, idx := range c.Holder.Indexes() {
-		// fragSources is map[URI][]*internal.ResizeSource.
 		fragSources, err := c.fragSources(toCluster, idx)
 		if err != nil {
 			return nil, err
 		}
 
 		for u, sources := range fragSources {
-			// If a host doesn't need to request data, mark it as complete.
-			if len(sources) == 0 {
-				j.URIs[u] = true
-				continue
+			for _, src := range sources {
+				multiIndex[u] = append(multiIndex[u], src)
 			}
-			// TODO: we can probably consolidate the instructions that go to the same
-			// node but apply to different indexes. (i.e. don't nest this in the Indexes() loop)
-			instr := &internal.ResizeInstruction{
-				JobID:       j.ID,
-				URI:         u.Encode(),
-				Coordinator: encodeURI(c.Coordinator),
-				Sources:     sources,
-				Schema:      pbSchema, // Include the schema to ensure it's in sync on the receiving node.
-			}
-			j.Instructions = append(j.Instructions, instr)
 		}
 	}
 
+	for u, sources := range multiIndex {
+		// If a host doesn't need to request data, mark it as complete.
+		if len(sources) == 0 {
+			j.URIs[u] = true
+			continue
+		}
+		instr := &internal.ResizeInstruction{
+			JobID:       j.ID,
+			URI:         u.Encode(),
+			Coordinator: encodeURI(c.Coordinator),
+			Sources:     sources,
+			Schema:      pbSchema, // Include the schema to ensure it's in sync on the receiving node.
+		}
+		j.Instructions = append(j.Instructions, instr)
+	}
+
 	return j, nil
 }
 

server.go

@@ -51,6 +51,11 @@ 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
@@ -84,6 +89,7 @@ type Server struct {
 func NewServer() *Server {
 	s := &Server{
 		closing: make(chan struct{}),
+		joining: make(chan struct{}),
 
 		Holder:            NewHolder(),
 		Handler:           NewHandler(),
@@ -192,6 +198,12 @@ 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)
@@ -216,6 +228,13 @@ 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.
@@ -351,6 +370,7 @@ func (s *Server) ReceiveMessage(pb proto.Message) error {
 		if err != nil {
 			return err
 		}
+		s.markAsJoined()
 	case *internal.ResizeInstruction:
 		err := s.Cluster.FollowResizeInstruction(obj)
 		if err != nil {