diff --git a/api.go b/api.go index abf2a84ea..65ff90870 100644 --- a/api.go +++ b/api.go @@ -517,7 +517,7 @@ func (api *API) ClusterMessage(ctx context.Context, reqBody io.Reader) error { } // Forward the error message. - if err := api.server.receiveMessage(pb); err != nil { + if err := api.server.receiveMessage(decode(pb)); err != nil { return errors.Wrap(err, "receiving message") } return nil diff --git a/broadcast.go b/broadcast.go index 3f6b960e4..0a33d4ca7 100644 --- a/broadcast.go +++ b/broadcast.go @@ -73,8 +73,7 @@ const ( ) // MarshalMessage encodes the protobuf message into a byte slice. -func MarshalMessage(pm Message) ([]byte, error) { - m := encode(pm) +func MarshalMessage(m proto.Message) ([]byte, error) { var typ uint8 switch obj := m.(type) { case *internal.CreateShardMessage: @@ -120,19 +119,45 @@ func MarshalMessage(pm Message) ([]byte, error) { } func encode(m Message) proto.Message { - var pm proto.Message switch mt := m.(type) { case *CreateShardMessage: return encodeCreateShardMessage(mt) case *CreateIndexMessage: return encodeCreateIndexMessage(mt) - // TODO, the rest + case *DeleteIndexMessage: + return encodeDeleteIndexMessage(mt) + case *CreateFieldMessage: + return encodeCreateFieldMessage(mt) + case *DeleteFieldMessage: + return encodeDeleteFieldMessage(mt) + case *CreateViewMessage: + return encodeCreateViewMessage(mt) + case *DeleteViewMessage: + return encodeDeleteViewMessage(mt) + case *ClusterStatus: + return encodeClusterStatus(mt) + case *ResizeInstruction: + return encodeResizeInstruction(mt) + case *ResizeInstructionComplete: + return encodeResizeInstructionComplete(mt) + case *SetCoordinatorMessage: + return encodeSetCoordinatorMessage(mt) + case *UpdateCoordinatorMessage: + return encodeUpdateCoordinatorMessage(mt) + case *NodeStateMessage: + return encodeNodeStateMessage(mt) + case *RecalculateCaches: + return encodeRecalculateCaches(mt) + case *nodeEvent: + return encodeNodeEventMessage(mt) + case *NodeStatus: + return encodeNodeStatus(mt) } return nil } // UnmarshalMessage decodes the byte slice into a protobuf message. -func UnmarshalMessage(buf []byte) (Message, error) { +func UnmarshalMessage(buf []byte) (proto.Message, error) { typ, buf := buf[0], buf[1:] var m proto.Message switch typ { @@ -177,3 +202,41 @@ func UnmarshalMessage(buf []byte) (Message, error) { } return m, nil } + +func decode(m proto.Message) Message { + switch mt := m.(type) { + case *internal.CreateShardMessage: + return decodeCreateShardMessage(mt) + case *internal.CreateIndexMessage: + return decodeCreateIndexMessage(mt) + case *internal.DeleteIndexMessage: + return decodeDeleteIndexMessage(mt) + case *internal.CreateFieldMessage: + return decodeCreateFieldMessage(mt) + case *internal.DeleteFieldMessage: + return decodeDeleteFieldMessage(mt) + case *internal.CreateViewMessage: + return decodeCreateViewMessage(mt) + case *internal.DeleteViewMessage: + return decodeDeleteViewMessage(mt) + case *internal.ClusterStatus: + return decodeClusterStatus(mt) + case *internal.ResizeInstruction: + return decodeResizeInstruction(mt) + case *internal.ResizeInstructionComplete: + return decodeResizeInstructionComplete(mt) + case *internal.SetCoordinatorMessage: + return decodeSetCoordinatorMessage(mt) + case *internal.UpdateCoordinatorMessage: + return decodeUpdateCoordinatorMessage(mt) + case *internal.NodeStateMessage: + return decodeNodeStateMessage(mt) + case *internal.RecalculateCaches: + return decodeRecalculateCaches(mt) + case *internal.NodeEventMessage: + return decodeNodeEventMessage(mt) + case *internal.NodeStatus: + return decodeNodeStatus(mt) + } + return nil +} diff --git a/cluster.go b/cluster.go index fcfb08d73..e97ba46e4 100644 --- a/cluster.go +++ b/cluster.go @@ -268,8 +268,8 @@ func (c *cluster) setCoordinator(n *Node) error { c.mu.Unlock() // Send the update coordinator message to all nodes. err := c.broadcaster.SendSync( - &internal.UpdateCoordinatorMessage{ - New: EncodeNode(n), + &UpdateCoordinatorMessage{ + New: n, }) if err != nil { return fmt.Errorf("problem sending UpdateCoordinator message: %v", err) @@ -423,7 +423,7 @@ func (c *cluster) setNodeState(state string) error { } // Send node state to coordinator. - ns := &internal.NodeStateMessage{ + ns := &NodeStateMessage{ NodeID: c.Node.ID, State: state, } @@ -460,12 +460,12 @@ func (c *cluster) receiveNodeState(nodeID string, state string) error { return nil } -// Status returns the internal ClusterStatus representation. -func (c *cluster) Status() *internal.ClusterStatus { - return &internal.ClusterStatus{ +// Status returns the the cluster's status including what nodes it contains, it's ID, and current state. +func (c *cluster) Status() *ClusterStatus { + return &ClusterStatus{ ClusterID: c.id, State: c.state, - Nodes: EncodeNodes(c.Nodes), + Nodes: c.Nodes, } } @@ -640,8 +640,8 @@ func (c *cluster) diff(other *cluster) (action string, nodeID string, err error) // fragSources returns a list of ResizeSources - for each node in the `to` cluster - // required to move from cluster `c` to cluster `to`. -func (c *cluster) fragSources(to *cluster, idx *Index) (map[string][]*internal.ResizeSource, error) { - m := make(map[string][]*internal.ResizeSource) +func (c *cluster) fragSources(to *cluster, idx *Index) (map[string][]*ResizeSource, error) { + m := make(map[string][]*ResizeSource) // Determine if a node is being added or removed. action, diffNodeID, err := c.diff(to) @@ -700,7 +700,7 @@ func (c *cluster) fragSources(to *cluster, idx *Index) (map[string][]*internal.R // Get the ResizeSource for each diff. for nodeID, diff := range diffs { - m[nodeID] = []*internal.ResizeSource{} + m[nodeID] = []*ResizeSource{} for _, frag := range diff { // If there is no valid source node ID for a fragment, // it likely means that the replica factor was not @@ -711,8 +711,8 @@ func (c *cluster) fragSources(to *cluster, idx *Index) (map[string][]*internal.R return nil, errors.New("not enough data to perform resize (replica factor may need to be increased)") } - src := &internal.ResizeSource{ - Node: EncodeNode(c.unprotectedNodeByID(srcNodeID)), + src := &ResizeSource{ + Node: c.unprotectedNodeByID(srcNodeID), Index: idx.Name(), Field: frag.field, View: frag.view, @@ -856,9 +856,9 @@ func (c *cluster) waitForStarted() error { // TODO: Because the normal code path already sends a NodeJoin event (via // memberlist), this it a bit redundant in most cases. Perhaps determine // that the node has been restarted and don't do this step. - msg := &internal.NodeEventMessage{ - Event: uint32(NodeJoin), - Node: EncodeNode(c.Node), + msg := &nodeEvent{ + Event: NodeJoin, + Node: c.Node, } if err := c.broadcaster.SendSync(msg); err != nil { return fmt.Errorf("sending restart NodeJoin: %v", err) @@ -968,8 +968,8 @@ func (c *cluster) setStateAndBroadcast(state string) error { return c.broadcaster.SendSync(c.Status()) } -func (c *cluster) sendTo(node *Node, msg proto.Message) error { - if err := c.broadcaster.SendTo(node, msg); err != nil { +func (c *cluster) sendTo(node *Node, m Message) error { + if err := c.broadcaster.SendTo(node, m); err != nil { return errors.Wrap(err, "sending") } return nil @@ -1075,7 +1075,7 @@ func (c *cluster) generateResizeJobByAction(nodeAction nodeAction) (*resizeJob, } // multiIndex is a map of sources initialized with all the nodes in toCluster. - multiIndex := make(map[string][]*internal.ResizeSource) + multiIndex := make(map[string][]*ResizeSource) for _, n := range toCluster.Nodes { multiIndex[n.ID] = nil @@ -1099,12 +1099,12 @@ func (c *cluster) generateResizeJobByAction(nodeAction nodeAction) (*resizeJob, j.IDs[id] = true continue } - instr := &internal.ResizeInstruction{ + instr := &ResizeInstruction{ JobID: j.ID, - Node: EncodeNode(toCluster.unprotectedNodeByID(id)), - Coordinator: EncodeNode(c.coordinatorNode()), + Node: toCluster.unprotectedNodeByID(id), + Coordinator: c.coordinatorNode(), Sources: sources, - Schema: c.holder.encodeSchema(), // Include the schema to ensure it's in sync on the receiving node. + Schema: &Schema{Indexes: c.holder.Schema()}, // Include the schema to ensure it's in sync on the receiving node. ClusterStatus: c.Status(), } j.Instructions = append(j.Instructions, instr) @@ -1148,7 +1148,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error { <-c.holder.opened // Prepare the return message. - complete := &internal.ResizeInstructionComplete{ + complete := &ResizeInstructionComplete{ JobID: instr.JobID, Node: instr.Node, Error: "", @@ -1167,7 +1167,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error { for _, src := range instr.Sources { c.logger.Printf("get shard %d for index %s from host %s", src.Shard, src.Index, src.Node.URI) - srcURI := decodeURI(src.Node.URI) + srcURI := src.Node.URI // Retrieve field. f := c.holder.Field(src.Index, src.Field) @@ -1219,14 +1219,14 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error { complete.Error = err.Error() } - if err := c.sendTo(DecodeNode(instr.Coordinator), complete); err != nil { + if err := c.sendTo(instr.Coordinator, complete); err != nil { c.logger.Printf("sending resizeInstructionComplete error: err=%s", err) } }() return nil } -func (c *cluster) markResizeInstructionComplete(complete *internal.ResizeInstructionComplete) error { +func (c *cluster) markResizeInstructionComplete(complete *ResizeInstructionComplete) error { j := c.job(complete.JobID) @@ -1263,7 +1263,7 @@ func (c *cluster) job(id int64) *resizeJob { type resizeJob struct { ID int64 IDs map[string]bool - Instructions []*internal.ResizeInstruction + Instructions []*ResizeInstruction Broadcaster broadcaster action string @@ -1366,7 +1366,7 @@ func (j *resizeJob) distributeResizeInstructions() error { // a dummy node object to use in the SendTo() method. node := &Node{ ID: instr.Node.ID, - URI: decodeURI(instr.Node.URI), + URI: instr.Node.URI, } j.Logger.Printf("send resize instructions: %v", instr) if err := j.Broadcaster.SendTo(node, instr); err != nil { @@ -1757,7 +1757,7 @@ type ResizeInstruction struct { ClusterStatus *ClusterStatus } -func decodeResizeInstruction(ri *internal.ResizeInstruction) ResizeInstruction { +func decodeResizeInstruction(ri *internal.ResizeInstruction) *ResizeInstruction { return &ResizeInstruction{ JobID: ri.JobID, Node: DecodeNode(ri.Node), @@ -1768,6 +1768,17 @@ func decodeResizeInstruction(ri *internal.ResizeInstruction) ResizeInstruction { } } +func encodeResizeInstruction(m *ResizeInstruction) *internal.ResizeInstruction { + return &internal.ResizeInstruction{ + JobID: m.JobID, + Node: EncodeNode(m.Node), + Coordinator: EncodeNode(m.Coordinator), + Sources: encodeResizeSources(m.Sources), + Schema: encodeSchema(m.Schema), + ClusterStatus: encodeClusterStatus(m.ClusterStatus), + } +} + type ResizeSource struct { Node *Node `protobuf:"bytes,1,opt,name=Node" json:"Node,omitempty"` Index string `protobuf:"bytes,2,opt,name=Index,proto3" json:"Index,omitempty"` @@ -1784,6 +1795,14 @@ func decodeResizeSources(srcs []*internal.ResizeSource) []*ResizeSource { return new } +func encodeResizeSources(srcs []*ResizeSource) []*internal.ResizeSource { + new := make([]*internal.ResizeSource, 0, len(srcs)) + for _, src := range srcs { + new = append(new, encodeResizeSource(src)) + } + return new +} + func decodeResizeSource(rs *internal.ResizeSource) *ResizeSource { return &ResizeSource{ Node: DecodeNode(rs.Node), @@ -1794,6 +1813,16 @@ func decodeResizeSource(rs *internal.ResizeSource) *ResizeSource { } } +func encodeResizeSource(m *ResizeSource) *internal.ResizeSource { + return &internal.ResizeSource{ + Node: EncodeNode(m.Node), + Index: m.Index, + Field: m.Field, + View: m.View, + Shard: m.Shard, + } +} + // Schema is a schema type Schema struct { Indexes []*IndexInfo @@ -1805,6 +1834,12 @@ func decodeSchema(s *internal.Schema) *Schema { } } +func encodeSchema(m *Schema) *internal.Schema { + return &internal.Schema{ + Indexes: encodeIndexInfos(m.Indexes), + } +} + func decodeIndexes(idxs []*internal.Index) []*IndexInfo { new := make([]*IndexInfo, 0, len(idxs)) for _, idx := range idxs { @@ -1813,6 +1848,14 @@ func decodeIndexes(idxs []*internal.Index) []*IndexInfo { return new } +func encodeIndexInfos(idxs []*IndexInfo) []*internal.Index { + new := make([]*internal.Index, 0, len(idxs)) + for _, idx := range idxs { + new = append(new, encodeIndexInfo(idx)) + } + return new +} + func decodeIndex(idx *internal.Index) *IndexInfo { return &IndexInfo{ Name: idx.Name, @@ -1820,6 +1863,13 @@ func decodeIndex(idx *internal.Index) *IndexInfo { } } +func encodeIndexInfo(idx *IndexInfo) *internal.Index { + return &internal.Index{ + Name: idx.Name, + Fields: encodeFieldInfos(idx.Fields), + } +} + func decodeFields(fs []*internal.Field) []*FieldInfo { new := make([]*FieldInfo, 0, len(fs)) for _, f := range fs { @@ -1828,6 +1878,14 @@ func decodeFields(fs []*internal.Field) []*FieldInfo { return new } +func encodeFieldInfos(fs []*FieldInfo) []*internal.Field { + new := make([]*internal.Field, 0, len(fs)) + for _, f := range fs { + new = append(new, encodeFieldInfo(f)) + } + return new +} + func decodeField(f *internal.Field) *FieldInfo { fi := &FieldInfo{ Name: f.Name, @@ -1840,6 +1898,19 @@ func decodeField(f *internal.Field) *FieldInfo { return fi } +func encodeFieldInfo(f *FieldInfo) *internal.Field { + ifield := &internal.Field{ + Name: f.Name, + Meta: encodeFieldOptions(&f.Options), + Views: make([]string, 0, len(f.Views)), + } + + for _, viewinfo := range f.Views { + ifield.Views = append(ifield.Views, viewinfo.Name) + } + return ifield +} + // EncodeNodes converts a slice of Nodes into its internal representation. func EncodeNodes(a []*Node) []*internal.Node { other := make([]*internal.Node, len(a)) @@ -1878,6 +1949,14 @@ func decodeClusterStatus(cs *internal.ClusterStatus) *ClusterStatus { } } +func encodeClusterStatus(m *ClusterStatus) *internal.ClusterStatus { + return &internal.ClusterStatus{ + State: m.State, + ClusterID: m.ClusterID, + Nodes: EncodeNodes(m.Nodes), + } +} + // DecodeNode converts a proto message into a Node. func DecodeNode(node *internal.Node) *Node { return &Node{ @@ -1969,3 +2048,223 @@ func decodeIndexMeta(pb *internal.IndexMeta) *IndexOptions { Keys: pb.Keys, } } + +type DeleteIndexMessage struct { + Index string +} + +func encodeDeleteIndexMessage(m *DeleteIndexMessage) *internal.DeleteIndexMessage { + return &internal.DeleteIndexMessage{ + Index: m.Index, + } +} + +func decodeDeleteIndexMessage(pb *internal.DeleteIndexMessage) *DeleteIndexMessage { + return &DeleteIndexMessage{ + Index: pb.Index, + } +} + +type CreateFieldMessage struct { + Index string + Field string + Meta *FieldOptions +} + +func encodeCreateFieldMessage(m *CreateFieldMessage) *internal.CreateFieldMessage { + return &internal.CreateFieldMessage{ + Index: m.Index, + Field: m.Field, + Meta: encodeFieldOptions(m.Meta), + } +} + +func decodeCreateFieldMessage(pb *internal.CreateFieldMessage) *CreateFieldMessage { + return &CreateFieldMessage{ + Index: pb.Index, + Field: pb.Field, + Meta: decodeFieldOptions(pb.Meta), + } +} + +type DeleteFieldMessage struct { + Index string + Field string +} + +func encodeDeleteFieldMessage(m *DeleteFieldMessage) *internal.DeleteFieldMessage { + return &internal.DeleteFieldMessage{ + Index: m.Index, + Field: m.Field, + } +} + +func decodeDeleteFieldMessage(pb *internal.DeleteFieldMessage) *DeleteFieldMessage { + return &DeleteFieldMessage{ + Index: pb.Index, + Field: pb.Field, + } +} + +type CreateViewMessage struct { + Index string + Field string + View string +} + +func encodeCreateViewMessage(m *CreateViewMessage) *internal.CreateViewMessage { + return &internal.CreateViewMessage{ + Index: m.Index, + Field: m.Field, + View: m.View, + } +} + +func decodeCreateViewMessage(pb *internal.CreateViewMessage) *CreateViewMessage { + return &CreateViewMessage{ + Index: pb.Index, + Field: pb.Field, + View: pb.View, + } +} + +type DeleteViewMessage struct { + Index string + Field string + View string +} + +func encodeDeleteViewMessage(m *DeleteViewMessage) *internal.DeleteViewMessage { + return &internal.DeleteViewMessage{ + Index: m.Index, + Field: m.Field, + View: m.View, + } +} + +func decodeDeleteViewMessage(pb *internal.DeleteViewMessage) *DeleteViewMessage { + return &DeleteViewMessage{ + Index: pb.Index, + Field: pb.Field, + View: pb.View, + } +} + +type ResizeInstructionComplete struct { + JobID int64 + Node *Node + Error string +} + +func encodeResizeInstructionComplete(m *ResizeInstructionComplete) *internal.ResizeInstructionComplete { + return &internal.ResizeInstructionComplete{ + JobID: m.JobID, + Node: EncodeNode(m.Node), + Error: m.Error, + } +} + +func decodeResizeInstructionComplete(pb *internal.ResizeInstructionComplete) *ResizeInstructionComplete { + return &ResizeInstructionComplete{ + JobID: pb.JobID, + Node: DecodeNode(pb.Node), + Error: pb.Error, + } +} + +type SetCoordinatorMessage struct { + New *Node +} + +func encodeSetCoordinatorMessage(m *SetCoordinatorMessage) *internal.SetCoordinatorMessage { + return &internal.SetCoordinatorMessage{ + New: EncodeNode(m.New), + } +} + +func decodeSetCoordinatorMessage(pb *internal.SetCoordinatorMessage) *SetCoordinatorMessage { + return &SetCoordinatorMessage{ + New: DecodeNode(pb.New), + } +} + +type UpdateCoordinatorMessage struct { + New *Node +} + +func encodeUpdateCoordinatorMessage(m *UpdateCoordinatorMessage) *internal.UpdateCoordinatorMessage { + return &internal.UpdateCoordinatorMessage{ + New: EncodeNode(m.New), + } +} + +func decodeUpdateCoordinatorMessage(pb *internal.UpdateCoordinatorMessage) *UpdateCoordinatorMessage { + return &UpdateCoordinatorMessage{ + New: DecodeNode(pb.New), + } +} + +type NodeStateMessage struct { + NodeID string `protobuf:"bytes,1,opt,name=NodeID,proto3" json:"NodeID,omitempty"` + State string `protobuf:"bytes,2,opt,name=State,proto3" json:"State,omitempty"` +} + +func encodeNodeStateMessage(m *NodeStateMessage) *internal.NodeStateMessage { + return &internal.NodeStateMessage{ + NodeID: m.NodeID, + State: m.State, + } +} + +func decodeNodeStateMessage(pb *internal.NodeStateMessage) *NodeStateMessage { + return &NodeStateMessage{ + NodeID: pb.NodeID, + State: pb.State, + } +} + +func encodeNodeEventMessage(m *nodeEvent) *internal.NodeEventMessage { + return &internal.NodeEventMessage{ + Event: uint32(m.Event), + Node: EncodeNode(m.Node), + } +} + +func decodeNodeEventMessage(pb *internal.NodeEventMessage) *nodeEvent { + return &nodeEvent{ + Event: NodeEventType(pb.Event), + Node: DecodeNode(pb.Node), + } +} + +type NodeStatus struct { + Node *Node + MaxShards map[string]uint64 + Schema *Schema +} + +func encodeNodeStatus(m *NodeStatus) *internal.NodeStatus { + return &internal.NodeStatus{ + Node: EncodeNode(m.Node), + MaxShards: &internal.MaxShards{Standard: m.MaxShards}, + Schema: encodeSchema(m.Schema), + } +} + +func decodeNodeStatus(pb *internal.NodeStatus) *NodeStatus { + return &NodeStatus{ + Node: DecodeNode(pb.Node), + MaxShards: pb.MaxShards.Standard, + Schema: decodeSchema(pb.Schema), + } +} + +type RecalculateCaches struct{} + +func decodeRecalculateCaches(pb *internal.RecalculateCaches) *RecalculateCaches { + return &RecalculateCaches{} +} + +func encodeRecalculateCaches(*RecalculateCaches) *internal.RecalculateCaches { + return &internal.RecalculateCaches{} +} diff --git a/holder.go b/holder.go index e599d21f2..83ffd6967 100644 --- a/holder.go +++ b/holder.go @@ -230,7 +230,7 @@ func (h *Holder) Schema() []*IndexInfo { } // applySchema applies an internal Schema to Holder. -func (h *Holder) applySchema(schema *internal.Schema) error { +func (h *Holder) applySchema(schema *Schema) error { // Create indexes that don't exist. for _, index := range schema.Indexes { opt := IndexOptions{} @@ -240,14 +240,13 @@ func (h *Holder) applySchema(schema *internal.Schema) error { } // Create fields that don't exist. for _, f := range index.Fields { - opt := decodeFieldOptions(f.Meta) - field, err := idx.CreateFieldIfNotExists(f.Name, *opt) + field, err := idx.CreateFieldIfNotExists(f.Name, f.Options) if err != nil { return errors.Wrap(err, "creating field") } // Create views that don't exist. for _, v := range f.Views { - _, err := field.createViewIfNotExists(v) + _, err := field.createViewIfNotExists(v.Name) if err != nil { return errors.Wrap(err, "creating view") } diff --git a/server.go b/server.go index dffeb94a0..7199b7178 100644 --- a/server.go +++ b/server.go @@ -27,8 +27,6 @@ import ( "sync" "time" - "github.com/gogo/protobuf/proto" - "github.com/pilosa/pilosa/internal" "github.com/pkg/errors" "golang.org/x/sync/errgroup" @@ -431,40 +429,40 @@ func (s *Server) monitorAntiEntropy() { } // receiveMessage represents an implementation of BroadcastHandler. -func (s *Server) receiveMessage(pb proto.Message) error { - switch obj := pb.(type) { - case *internal.CreateShardMessage: +func (s *Server) receiveMessage(m Message) error { + switch obj := m.(type) { + case *CreateShardMessage: idx := s.holder.Index(obj.Index) if idx == nil { return fmt.Errorf("Local Index not found: %s", obj.Index) } idx.setRemoteMaxShard(obj.Shard) - case *internal.CreateIndexMessage: + case *CreateIndexMessage: opt := IndexOptions{} _, err := s.holder.CreateIndex(obj.Index, opt) if err != nil { return err } - case *internal.DeleteIndexMessage: + case *DeleteIndexMessage: if err := s.holder.DeleteIndex(obj.Index); err != nil { return err } - case *internal.CreateFieldMessage: + case *CreateFieldMessage: idx := s.holder.Index(obj.Index) if idx == nil { return fmt.Errorf("Local Index not found: %s", obj.Index) } - opt := decodeFieldOptions(obj.Meta) + opt := obj.Meta _, err := idx.CreateField(obj.Field, *opt) if err != nil { return err } - case *internal.DeleteFieldMessage: + case *DeleteFieldMessage: idx := s.holder.Index(obj.Index) if err := idx.DeleteField(obj.Field); err != nil { return err } - case *internal.CreateViewMessage: + case *CreateViewMessage: f := s.holder.Field(obj.Index, obj.Field) if f == nil { return fmt.Errorf("Local Field not found: %s", obj.Field) @@ -473,7 +471,7 @@ func (s *Server) receiveMessage(pb proto.Message) error { if err != nil { return err } - case *internal.DeleteViewMessage: + case *DeleteViewMessage: f := s.holder.Field(obj.Index, obj.Field) if f == nil { return fmt.Errorf("Local Field not found: %s", obj.Field) @@ -482,36 +480,36 @@ func (s *Server) receiveMessage(pb proto.Message) error { if err != nil { return err } - case *internal.ClusterStatus: - err := s.cluster.mergeClusterStatus(decodeClusterStatus(obj)) + case *ClusterStatus: + err := s.cluster.mergeClusterStatus(obj) if err != nil { return err } - case *internal.ResizeInstruction: - err := s.cluster.followResizeInstruction(decodeResizeInstruction(obj)) + case *ResizeInstruction: + err := s.cluster.followResizeInstruction(obj) if err != nil { return err } - case *internal.ResizeInstructionComplete: + case *ResizeInstructionComplete: err := s.cluster.markResizeInstructionComplete(obj) if err != nil { return err } - case *internal.SetCoordinatorMessage: - s.cluster.setCoordinator(DecodeNode(obj.New)) - case *internal.UpdateCoordinatorMessage: - s.cluster.updateCoordinator(DecodeNode(obj.New)) - case *internal.NodeStateMessage: + case *SetCoordinatorMessage: + s.cluster.setCoordinator(obj.New) + case *UpdateCoordinatorMessage: + s.cluster.updateCoordinator(obj.New) + case *NodeStateMessage: err := s.cluster.receiveNodeState(obj.NodeID, obj.State) if err != nil { return err } - case *internal.RecalculateCaches: + case *RecalculateCaches: s.holder.RecalculateCaches() - case *internal.NodeEventMessage: - s.cluster.ReceiveEvent(DecodeNodeEvent(obj)) - case *internal.NodeStatus: - s.handleRemoteStatus(pb) + case *nodeEvent: + s.cluster.ReceiveEvent(obj) + case *NodeStatus: + s.handleRemoteStatus(obj) } return nil @@ -519,6 +517,7 @@ func (s *Server) receiveMessage(pb proto.Message) error { // SendSync represents an implementation of Broadcaster. func (s *Server) SendSync(m Message) error { + pb := encode(m) var eg errgroup.Group for _, node := range s.cluster.Nodes { node := node @@ -537,12 +536,13 @@ func (s *Server) SendSync(m Message) error { } // SendAsync represents an implementation of Broadcaster. -func (s *Server) SendAsync(pb proto.Message) error { +func (s *Server) SendAsync(m Message) error { return ErrNotImplemented } // SendTo represents an implementation of Broadcaster. -func (s *Server) SendTo(to *Node, pb proto.Message) error { +func (s *Server) SendTo(to *Node, m Message) error { + pb := encode(m) s.logger.Printf("SendTo: %s", to.URI) return s.defaultClient.SendMessage(context.Background(), &to.URI, pb) } @@ -554,7 +554,7 @@ func (s *Server) node() Node { } // handleRemoteStatus receives incoming NodeStatus from remote nodes. -func (s *Server) handleRemoteStatus(pb proto.Message) { +func (s *Server) handleRemoteStatus(pb Message) { // Ignore NodeStatus messages until the cluster is in a Normal state. if s.cluster.State() != ClusterStateNormal { return @@ -564,16 +564,16 @@ func (s *Server) handleRemoteStatus(pb proto.Message) { // Make sure the holder has opened. <-s.holder.opened - err := s.mergeRemoteStatus(pb.(*internal.NodeStatus)) + err := s.mergeRemoteStatus(pb.(*NodeStatus)) if err != nil { s.logger.Printf("merge remote status: %s", err) } }() } -func (s *Server) mergeRemoteStatus(ns *internal.NodeStatus) error { +func (s *Server) mergeRemoteStatus(ns *NodeStatus) error { // Ignore status updates from self. - if s.nodeID == DecodeNode(ns.Node).ID { + if s.nodeID == ns.Node.ID { return nil } @@ -584,7 +584,7 @@ func (s *Server) mergeRemoteStatus(ns *internal.NodeStatus) error { // Sync maxShards. oldmaxshards := s.holder.maxShards() - for index, newMax := range ns.MaxShards.Standard { + for index, newMax := range ns.MaxShards { localIndex := s.holder.Index(index) // if we don't know about an index locally, log an error because // indexes should be created and synced prior to shard creation diff --git a/utils_internal_test.go b/utils_internal_test.go index ca7bd1fa7..f7c11d6a7 100644 --- a/utils_internal_test.go +++ b/utils_internal_test.go @@ -24,7 +24,6 @@ import ( "time" "github.com/gogo/protobuf/proto" - "github.com/pilosa/pilosa/internal" ) // NewTestCluster returns a cluster with n nodes and uses a mod-based hasher. @@ -304,9 +303,9 @@ func (t *ClusterCluster) Close() error { } // SendSync is a test implemenetation of Broadcaster SendSync method. -func (t *ClusterCluster) SendSync(pb proto.Message) error { - switch obj := pb.(type) { - case *internal.ClusterStatus: +func (t *ClusterCluster) SendSync(m Message) error { + switch obj := m.(type) { + case *ClusterStatus: // Apply the send message to all nodes (except the coordinator). for _, c := range t.Clusters { c.mergeClusterStatus(obj) @@ -322,19 +321,19 @@ func (t *ClusterCluster) SendSync(pb proto.Message) error { } // SendAsync is a test implemenetation of Broadcaster SendAsync method. -func (t *ClusterCluster) SendAsync(pb proto.Message) error { +func (t *ClusterCluster) SendAsync(Message) error { return nil } // SendTo is a test implemenetation of Broadcaster SendTo method. -func (t *ClusterCluster) SendTo(to *Node, pb proto.Message) error { - switch obj := pb.(type) { - case *internal.ResizeInstruction: +func (t *ClusterCluster) SendTo(to *Node, m Message) error { + switch obj := m.(type) { + case *ResizeInstruction: err := t.FollowResizeInstruction(obj) if err != nil { return err } - case *internal.ResizeInstructionComplete: + case *ResizeInstructionComplete: coord := t.clusterByID(to.ID) go coord.markResizeInstructionComplete(obj) } @@ -342,10 +341,10 @@ func (t *ClusterCluster) SendTo(to *Node, pb proto.Message) error { } // FollowResizeInstruction is a version of cluster.FollowResizeInstruction used for testing. -func (t *ClusterCluster) FollowResizeInstruction(instr *internal.ResizeInstruction) error { +func (t *ClusterCluster) FollowResizeInstruction(instr *ResizeInstruction) error { // Prepare the return message. - complete := &internal.ResizeInstructionComplete{ + complete := &ResizeInstructionComplete{ JobID: instr.JobID, Node: instr.Node, Error: "", @@ -356,7 +355,7 @@ func (t *ClusterCluster) FollowResizeInstruction(instr *internal.ResizeInstructi // figure out which node it was meant for, then call the operation on that cluster // basically need to mimic this: client.RetrieveShardFromURI(context.Background(), src.Index, src.Field, src.View, src.Shard, srcURI) - instrNode := DecodeNode(instr.Node) + instrNode := instr.Node destCluster := t.clusterByID(instrNode.ID) // Sync the schema received in the resize instruction. @@ -365,7 +364,7 @@ func (t *ClusterCluster) FollowResizeInstruction(instr *internal.ResizeInstructi } for _, src := range instr.Sources { - srcNode := DecodeNode(src.Node) + srcNode := src.Node srcCluster := t.clusterByID(srcNode.ID) srcFragment := srcCluster.holder.fragment(src.Index, src.Field, src.View, src.Shard) @@ -405,6 +404,6 @@ func (t *ClusterCluster) FollowResizeInstruction(instr *internal.ResizeInstructi complete.Error = err.Error() } - node := DecodeNode(instr.Coordinator) + node := instr.Coordinator return t.SendTo(node, complete) }