diff --git a/cluster.go b/cluster.go index 39bf5afff..d66883f5e 100644 --- a/cluster.go +++ b/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 } diff --git a/server.go b/server.go index b677147bd..7328eac80 100644 --- a/server.go +++ b/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 {