diff --git a/broadcast.go b/broadcast.go index 19e927c18..3f6b960e4 100644 --- a/broadcast.go +++ b/broadcast.go @@ -25,11 +25,15 @@ import ( // broadcaster is an interface for broadcasting messages. type broadcaster interface { - SendSync(pb proto.Message) error - SendAsync(pb proto.Message) error - SendTo(to *Node, pb proto.Message) error + SendSync(Message) error + SendAsync(Message) error + SendTo(*Node, Message) error } +// Message is the interface implemented by all core pilosa types which can be serialized to messages. +// TODO add at least a single "isMessage()" method. +type Message interface{} + func init() { NopBroadcaster = &nopBroadcaster{} } @@ -40,13 +44,13 @@ var NopBroadcaster broadcaster type nopBroadcaster struct{} // SendSync A no-op implementation of Broadcaster SendSync method. -func (n nopBroadcaster) SendSync(pb proto.Message) error { return nil } +func (nopBroadcaster) SendSync(Message) error { return nil } // SendAsync A no-op implementation of Broadcaster SendAsync method. -func (n nopBroadcaster) SendAsync(pb proto.Message) error { return nil } +func (nopBroadcaster) SendAsync(Message) error { return nil } // SendTo is a no-op implementation of Broadcaster SendTo method. -func (c nopBroadcaster) SendTo(to *Node, pb proto.Message) error { return nil } +func (nopBroadcaster) SendTo(*Node, Message) error { return nil } // Broadcast message types. const ( @@ -69,7 +73,8 @@ const ( ) // MarshalMessage encodes the protobuf message into a byte slice. -func MarshalMessage(m proto.Message) ([]byte, error) { +func MarshalMessage(pm Message) ([]byte, error) { + m := encode(pm) var typ uint8 switch obj := m.(type) { case *internal.CreateShardMessage: @@ -114,8 +119,20 @@ func MarshalMessage(m proto.Message) ([]byte, error) { return append([]byte{typ}, buf...), nil } +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 + } + return nil +} + // UnmarshalMessage decodes the byte slice into a protobuf message. -func UnmarshalMessage(buf []byte) (proto.Message, error) { +func UnmarshalMessage(buf []byte) (Message, error) { typ, buf := buf[0], buf[1:] var m proto.Message switch typ { diff --git a/cluster.go b/cluster.go index acf06dae0..fcfb08d73 100644 --- a/cluster.go +++ b/cluster.go @@ -69,52 +69,6 @@ func (n Node) String() string { return fmt.Sprintf("Node: %s", n.ID) } -// EncodeNodes converts a slice of Nodes into its internal representation. -func EncodeNodes(a []*Node) []*internal.Node { - other := make([]*internal.Node, len(a)) - for i := range a { - other[i] = EncodeNode(a[i]) - } - return other -} - -// EncodeNode converts a Node into its internal representation. -func EncodeNode(n *Node) *internal.Node { - return &internal.Node{ - ID: n.ID, - URI: n.URI.Encode(), - IsCoordinator: n.IsCoordinator, - } -} - -// DecodeNodes converts a proto message into a slice of Nodes. -func DecodeNodes(a []*internal.Node) []*Node { - if len(a) == 0 { - return nil - } - other := make([]*Node, len(a)) - for i := range a { - other[i] = DecodeNode(a[i]) - } - return other -} - -// DecodeNode converts a proto message into a Node. -func DecodeNode(node *internal.Node) *Node { - return &Node{ - ID: node.ID, - URI: decodeURI(node.URI), - IsCoordinator: node.IsCoordinator, - } -} - -func DecodeNodeEvent(ne *internal.NodeEventMessage) *nodeEvent { - return &nodeEvent{ - Event: NodeEventType(ne.Event), - Node: DecodeNode(ne.Node), - } -} - // Nodes represents a list of nodes. type Nodes []*Node @@ -1176,7 +1130,7 @@ func (c *cluster) completeCurrentJob(state string) error { } // followResizeInstruction is run by any node that receives a ResizeInstruction. -func (c *cluster) followResizeInstruction(instr *internal.ResizeInstruction) error { +func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error { c.logger.Printf("follow resize instruction on %s", c.Node.ID) // Make sure the cluster status on this node agrees with the Coordinator // before attempting a resize. @@ -1553,32 +1507,6 @@ func (c *cluster) saveTopology() error { return nil } -func encodeTopology(topology *Topology) *internal.Topology { - if topology == nil { - return nil - } - return &internal.Topology{ - ClusterID: topology.ClusterID, - NodeIDs: topology.NodeIDs, - } -} - -func decodeTopology(topology *internal.Topology) (*Topology, error) { - if topology == nil { - return nil, nil - } - - t := NewTopology() - t.ClusterID = topology.ClusterID - t.NodeIDs = topology.NodeIDs - sort.Slice(t.NodeIDs, - func(i, j int) bool { - return t.NodeIDs[i] < t.NodeIDs[j] - }) - - return t, nil -} - func (c *cluster) considerTopology() error { // Create ClusterID if one does not already exist. if c.id == "" { @@ -1752,7 +1680,7 @@ func (c *cluster) nodeLeave(node *Node) error { return nil } -func (c *cluster) mergeClusterStatus(cs *internal.ClusterStatus) error { +func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error { c.mu.Lock() defer c.mu.Unlock() c.logger.Printf("merge cluster status: %v", cs) @@ -1764,7 +1692,7 @@ func (c *cluster) mergeClusterStatus(cs *internal.ClusterStatus) error { // Set ClusterID. c.setID(cs.ClusterID) - officialNodes := DecodeNodes(cs.Nodes) + officialNodes := cs.Nodes // Add all nodes from the coordinator. for _, node := range officialNodes { @@ -1813,3 +1741,231 @@ func (c *cluster) setStatic(hosts []string) error { } return nil } + +type ClusterStatus struct { + ClusterID string + State string + Nodes []*Node +} + +type ResizeInstruction struct { + JobID int64 + Node *Node + Coordinator *Node + Sources []*ResizeSource + Schema *Schema + ClusterStatus *ClusterStatus +} + +func decodeResizeInstruction(ri *internal.ResizeInstruction) ResizeInstruction { + return &ResizeInstruction{ + JobID: ri.JobID, + Node: DecodeNode(ri.Node), + Coordinator: DecodeNode(ri.Coordinator), + Sources: decodeResizeSources(ri.Sources), + Schema: decodeSchema(ri.Schema), + ClusterStatus: decodeClusterStatus(ri.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"` + Field string `protobuf:"bytes,3,opt,name=Field,proto3" json:"Field,omitempty"` + View string `protobuf:"bytes,4,opt,name=View,proto3" json:"View,omitempty"` + Shard uint64 `protobuf:"varint,5,opt,name=Shard,proto3" json:"Shard,omitempty"` +} + +func decodeResizeSources(srcs []*internal.ResizeSource) []*ResizeSource { + new := make([]*ResizeSource, 0, len(srcs)) + for _, src := range srcs { + new = append(new, decodeResizeSource(src)) + } + return new +} + +func decodeResizeSource(rs *internal.ResizeSource) *ResizeSource { + return &ResizeSource{ + Node: DecodeNode(rs.Node), + Index: rs.Index, + Field: rs.Field, + View: rs.View, + Shard: rs.Shard, + } +} + +// Schema is a schema +type Schema struct { + Indexes []*IndexInfo +} + +func decodeSchema(s *internal.Schema) *Schema { + return &Schema{ + Indexes: decodeIndexes(s.Indexes), + } +} + +func decodeIndexes(idxs []*internal.Index) []*IndexInfo { + new := make([]*IndexInfo, 0, len(idxs)) + for _, idx := range idxs { + new = append(new, decodeIndex(idx)) + } + return new +} + +func decodeIndex(idx *internal.Index) *IndexInfo { + return &IndexInfo{ + Name: idx.Name, + Fields: decodeFields(idx.Fields), + } +} + +func decodeFields(fs []*internal.Field) []*FieldInfo { + new := make([]*FieldInfo, 0, len(fs)) + for _, f := range fs { + new = append(new, decodeField(f)) + } + return new +} + +func decodeField(f *internal.Field) *FieldInfo { + fi := &FieldInfo{ + Name: f.Name, + Options: *decodeFieldOptions(f.Meta), + Views: make([]*viewInfo, 0, len(f.Views)), + } + for _, viewname := range f.Views { + fi.Views = append(fi.Views, &viewInfo{Name: viewname}) + } + return fi +} + +// EncodeNodes converts a slice of Nodes into its internal representation. +func EncodeNodes(a []*Node) []*internal.Node { + other := make([]*internal.Node, len(a)) + for i := range a { + other[i] = EncodeNode(a[i]) + } + return other +} + +// EncodeNode converts a Node into its internal representation. +func EncodeNode(n *Node) *internal.Node { + return &internal.Node{ + ID: n.ID, + URI: n.URI.Encode(), + IsCoordinator: n.IsCoordinator, + } +} + +// DecodeNodes converts a proto message into a slice of Nodes. +func DecodeNodes(a []*internal.Node) []*Node { + if len(a) == 0 { + return nil + } + other := make([]*Node, len(a)) + for i := range a { + other[i] = DecodeNode(a[i]) + } + return other +} + +func decodeClusterStatus(cs *internal.ClusterStatus) *ClusterStatus { + return &ClusterStatus{ + State: cs.State, + ClusterID: cs.ClusterID, + Nodes: DecodeNodes(cs.Nodes), + } +} + +// DecodeNode converts a proto message into a Node. +func DecodeNode(node *internal.Node) *Node { + return &Node{ + ID: node.ID, + URI: decodeURI(node.URI), + IsCoordinator: node.IsCoordinator, + } +} + +func DecodeNodeEvent(ne *internal.NodeEventMessage) *nodeEvent { + return &nodeEvent{ + Event: NodeEventType(ne.Event), + Node: DecodeNode(ne.Node), + } +} + +func encodeTopology(topology *Topology) *internal.Topology { + if topology == nil { + return nil + } + return &internal.Topology{ + ClusterID: topology.ClusterID, + NodeIDs: topology.NodeIDs, + } +} + +func decodeTopology(topology *internal.Topology) (*Topology, error) { + if topology == nil { + return nil, nil + } + + t := NewTopology() + t.ClusterID = topology.ClusterID + t.NodeIDs = topology.NodeIDs + sort.Slice(t.NodeIDs, + func(i, j int) bool { + return t.NodeIDs[i] < t.NodeIDs[j] + }) + + return t, nil +} + +type CreateShardMessage struct { + Index string + Shard uint64 +} + +func encodeCreateShardMessage(m *CreateShardMessage) *internal.CreateShardMessage { + return &internal.CreateShardMessage{ + Index: m.Index, + Shard: m.Shard, + } +} + +func decodeCreateShardMessage(pb *internal.CreateShardMessage) *CreateShardMessage { + return &CreateShardMessage{ + Index: pb.Index, + Shard: pb.Shard, + } +} + +type CreateIndexMessage struct { + Index string + Meta *IndexOptions +} + +func encodeCreateIndexMessage(m *CreateIndexMessage) *internal.CreateIndexMessage { + return &internal.CreateIndexMessage{ + Index: m.Index, + Meta: encodeIndexMeta(m.Meta), + } +} + +func decodeCreateIndexMessage(pb *internal.CreateIndexMessage) *CreateIndexMessage { + return &CreateIndexMessage{ + Index: pb.Index, + Meta: decodeIndexMeta(pb.Meta), + } +} + +func encodeIndexMeta(m *IndexOptions) *internal.IndexMeta { + return &internal.IndexMeta{ + Keys: m.Keys, + } +} + +func decodeIndexMeta(pb *internal.IndexMeta) *IndexOptions { + return &IndexOptions{ + Keys: pb.Keys, + } +} diff --git a/server.go b/server.go index 56195b427..dffeb94a0 100644 --- a/server.go +++ b/server.go @@ -483,12 +483,12 @@ func (s *Server) receiveMessage(pb proto.Message) error { return err } case *internal.ClusterStatus: - err := s.cluster.mergeClusterStatus(obj) + err := s.cluster.mergeClusterStatus(decodeClusterStatus(obj)) if err != nil { return err } case *internal.ResizeInstruction: - err := s.cluster.followResizeInstruction(obj) + err := s.cluster.followResizeInstruction(decodeResizeInstruction(obj)) if err != nil { return err } @@ -518,7 +518,7 @@ func (s *Server) receiveMessage(pb proto.Message) error { } // SendSync represents an implementation of Broadcaster. -func (s *Server) SendSync(pb proto.Message) error { +func (s *Server) SendSync(m Message) error { var eg errgroup.Group for _, node := range s.cluster.Nodes { node := node