diff --git a/api.go b/api.go index 73bdd6665..09f0531d5 100644 --- a/api.go +++ b/api.go @@ -509,14 +509,15 @@ func (api *API) ClusterMessage(ctx context.Context, reqBody io.Reader) error { return errors.Wrap(err, "reading body") } - // Marshal into request object. - pb, err := UnmarshalMessage(body) + typ := body[0] + msg := getMessage(typ) + err = api.server.serializer.Unmarshal(body[1:], msg) if err != nil { - return errors.Wrap(err, "unmarshaling message") + return errors.Wrap(err, "deserializing cluster message") } // Forward the error message. - if err := api.server.receiveMessage(decode(pb)); err != nil { + if err := api.server.receiveMessage(msg); err != nil { return errors.Wrap(err, "receiving message") } return nil diff --git a/broadcast.go b/broadcast.go index d3c4a2562..1da6ca57f 100644 --- a/broadcast.go +++ b/broadcast.go @@ -23,6 +23,12 @@ import ( "github.com/pkg/errors" ) +// Serializer is an interface for serializing pilosa types to bytes and back. +type Serializer interface { + Marshal(Message) ([]byte, error) + Unmarshal([]byte, Message) error +} + // broadcaster is an interface for broadcasting messages. type broadcaster interface { SendSync(Message) error @@ -118,42 +124,82 @@ func MarshalMessage(m proto.Message) ([]byte, error) { return append([]byte{typ}, buf...), nil } -func encode(m Message) proto.Message { - switch mt := m.(type) { - case *CreateShardMessage: - return encodeCreateShardMessage(mt) - case *CreateIndexMessage: - return encodeCreateIndexMessage(mt) - 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) +func getMessage(typ byte) Message { + switch typ { + case messageTypeCreateShard: + return &CreateShardMessage{} + case messageTypeCreateIndex: + return &CreateIndexMessage{} + case messageTypeDeleteIndex: + return &DeleteIndexMessage{} + case messageTypeCreateField: + return &CreateFieldMessage{} + case messageTypeDeleteField: + return &DeleteFieldMessage{} + case messageTypeCreateView: + return &CreateViewMessage{} + case messageTypeDeleteView: + return &DeleteViewMessage{} + case messageTypeClusterStatus: + return &ClusterStatus{} + case messageTypeResizeInstruction: + return &ResizeInstruction{} + case messageTypeResizeInstructionComplete: + return &ResizeInstructionComplete{} + case messageTypeSetCoordinator: + return &SetCoordinatorMessage{} + case messageTypeUpdateCoordinator: + return &UpdateCoordinatorMessage{} + case messageTypeNodeState: + return &NodeStateMessage{} + case messageTypeRecalculateCaches: + return &RecalculateCaches{} + case messageTypeNodeEvent: + return &NodeEvent{} + case messageTypeNodeStatus: + return &NodeStatus{} + default: + panic(fmt.Sprintf("unknown message type %d", typ)) + } +} + +func getMessageType(m Message) byte { + switch m.(type) { + case *CreateShardMessage: + return messageTypeCreateShard + case *CreateIndexMessage: + return messageTypeCreateIndex + case *DeleteIndexMessage: + return messageTypeDeleteIndex + case *CreateFieldMessage: + return messageTypeCreateField + case *DeleteFieldMessage: + return messageTypeDeleteField + case *CreateViewMessage: + return messageTypeCreateView + case *DeleteViewMessage: + return messageTypeDeleteView + case *ClusterStatus: + return messageTypeClusterStatus + case *ResizeInstruction: + return messageTypeResizeInstruction + case *ResizeInstructionComplete: + return messageTypeResizeInstructionComplete + case *SetCoordinatorMessage: + return messageTypeSetCoordinator + case *UpdateCoordinatorMessage: + return messageTypeUpdateCoordinator + case *NodeStateMessage: + return messageTypeNodeState + case *RecalculateCaches: + return messageTypeRecalculateCaches + case *NodeEvent: + return messageTypeNodeEvent + case *NodeStatus: + return messageTypeNodeStatus + default: + panic(fmt.Sprintf("don't have type for message %#v", m)) } - return nil } // UnmarshalMessage decodes the byte slice into a protobuf message. diff --git a/client.go b/client.go index 59e3ad59b..01e0b847a 100644 --- a/client.go +++ b/client.go @@ -4,7 +4,6 @@ import ( "context" "io" - "github.com/gogo/protobuf/proto" "github.com/pilosa/pilosa/internal" ) @@ -49,7 +48,7 @@ type InternalClient interface { BlockData(ctx context.Context, uri *URI, index, field string, shard uint64, block int) ([]uint64, []uint64, error) ColumnAttrDiff(ctx context.Context, uri *URI, index string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) RowAttrDiff(ctx context.Context, uri *URI, index, field string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) - SendMessage(ctx context.Context, uri *URI, pb proto.Message) error + SendMessage(ctx context.Context, uri *URI, msg []byte) error RetrieveShardFromURI(ctx context.Context, index, field string, shard uint64, uri URI) (io.ReadCloser, error) } @@ -128,7 +127,7 @@ func (n NopInternalClient) ColumnAttrDiff(ctx context.Context, uri *URI, index s func (n NopInternalClient) RowAttrDiff(ctx context.Context, uri *URI, index, field string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) { return nil, nil } -func (n NopInternalClient) SendMessage(ctx context.Context, uri *URI, pb proto.Message) error { +func (n NopInternalClient) SendMessage(ctx context.Context, uri *URI, msg []byte) error { return nil } func (n NopInternalClient) RetrieveShardFromURI(ctx context.Context, index, field string, shard uint64, uri URI) (io.ReadCloser, error) { diff --git a/cluster.go b/cluster.go index 2d25ebeaa..e2f29fde5 100644 --- a/cluster.go +++ b/cluster.go @@ -1890,10 +1890,10 @@ func decodeField(f *internal.Field) *FieldInfo { fi := &FieldInfo{ Name: f.Name, Options: *decodeFieldOptions(f.Meta), - Views: make([]*viewInfo, 0, len(f.Views)), + Views: make([]*ViewInfo, 0, len(f.Views)), } for _, viewname := range f.Views { - fi.Views = append(fi.Views, &viewInfo{Name: viewname}) + fi.Views = append(fi.Views, &ViewInfo{Name: viewname}) } return fi } diff --git a/encoding/proto/proto.go b/encoding/proto/proto.go new file mode 100644 index 000000000..1c83ba4ea --- /dev/null +++ b/encoding/proto/proto.go @@ -0,0 +1,594 @@ +package proto + +import ( + "fmt" + + "github.com/gogo/protobuf/proto" + "github.com/pilosa/pilosa" + "github.com/pilosa/pilosa/internal" + "github.com/pkg/errors" +) + +// Serializer implements pilosa.Serializer for protobufs. +type Serializer struct{} + +// Marshal turns pilosa messages into protobuf serialized bytes. +func (Serializer) Marshal(m pilosa.Message) ([]byte, error) { + pm := encodeToProto(m) + if pm == nil { + return nil, errors.New("passed invalid pilosa.Message") + } + buf, err := proto.Marshal(pm) + return buf, errors.Wrap(err, "marshalling") +} + +// Unmarshal takes byte slices and protobuf deserializes them into a pilosa Message. +func (Serializer) Unmarshal(buf []byte, m pilosa.Message) error { + switch mt := m.(type) { + case *pilosa.CreateShardMessage: + msg := &internal.CreateShardMessage{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling CreateShardMessage") + } + decodeCreateShardMessage(msg, mt) + return nil + case *pilosa.CreateIndexMessage: + msg := &internal.CreateIndexMessage{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling CreateIndexMessage") + } + decodeCreateIndexMessage(msg, mt) + return nil + case *pilosa.DeleteIndexMessage: + msg := &internal.DeleteIndexMessage{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling DeleteIndexMessage") + } + decodeDeleteIndexMessage(msg, mt) + return nil + case *pilosa.CreateFieldMessage: + msg := &internal.CreateFieldMessage{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling CreateFieldMessage") + } + decodeCreateFieldMessage(msg, mt) + return nil + case *pilosa.DeleteFieldMessage: + msg := &internal.DeleteFieldMessage{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling DeleteFieldMessage") + } + decodeDeleteFieldMessage(msg, mt) + return nil + case *pilosa.CreateViewMessage: + msg := &internal.CreateViewMessage{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling CreateViewMessage") + } + decodeCreateViewMessage(msg, mt) + return nil + case *pilosa.DeleteViewMessage: + msg := &internal.DeleteViewMessage{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling DeleteViewMessage") + } + decodeDeleteViewMessage(msg, mt) + return nil + case *pilosa.ClusterStatus: + msg := &internal.ClusterStatus{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling ClusterStatus") + } + decodeClusterStatus(msg, mt) + return nil + case *pilosa.ResizeInstruction: + msg := &internal.ResizeInstruction{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling ResizeInstruction") + } + decodeResizeInstruction(msg, mt) + return nil + case *pilosa.ResizeInstructionComplete: + msg := &internal.ResizeInstructionComplete{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling ResizeInstructionComplete") + } + decodeResizeInstructionComplete(msg, mt) + return nil + case *pilosa.SetCoordinatorMessage: + msg := &internal.SetCoordinatorMessage{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling SetCoordinatorMessage") + } + decodeSetCoordinatorMessage(msg, mt) + return nil + case *pilosa.UpdateCoordinatorMessage: + msg := &internal.UpdateCoordinatorMessage{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling UpdateCoordinatorMessage") + } + decodeUpdateCoordinatorMessage(msg, mt) + return nil + case *pilosa.NodeStateMessage: + msg := &internal.NodeStateMessage{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling NodeStateMessage") + } + decodeNodeStateMessage(msg, mt) + return nil + case *pilosa.RecalculateCaches: + msg := &internal.RecalculateCaches{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling RecalculateCaches") + } + decodeRecalculateCaches(msg, mt) + return nil + case *pilosa.NodeEvent: + msg := &internal.NodeEventMessage{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling NodeEvent") + } + decodeNodeEventMessage(msg, mt) + return nil + case *pilosa.NodeStatus: + msg := &internal.NodeStatus{} + err := proto.Unmarshal(buf, msg) + if err != nil { + return errors.Wrap(err, "unmarshaling NodeStatus") + } + decodeNodeStatus(msg, mt) + return nil + default: + panic(fmt.Sprintf("unhandled pilosa.Message of type %T: %#v", mt, m)) + } +} + +func encodeToProto(m pilosa.Message) proto.Message { + switch mt := m.(type) { + case *pilosa.CreateShardMessage: + return encodeCreateShardMessage(mt) + case *pilosa.CreateIndexMessage: + return encodeCreateIndexMessage(mt) + case *pilosa.DeleteIndexMessage: + return encodeDeleteIndexMessage(mt) + case *pilosa.CreateFieldMessage: + return encodeCreateFieldMessage(mt) + case *pilosa.DeleteFieldMessage: + return encodeDeleteFieldMessage(mt) + case *pilosa.CreateViewMessage: + return encodeCreateViewMessage(mt) + case *pilosa.DeleteViewMessage: + return encodeDeleteViewMessage(mt) + case *pilosa.ClusterStatus: + return encodeClusterStatus(mt) + case *pilosa.ResizeInstruction: + return encodeResizeInstruction(mt) + case *pilosa.ResizeInstructionComplete: + return encodeResizeInstructionComplete(mt) + case *pilosa.SetCoordinatorMessage: + return encodeSetCoordinatorMessage(mt) + case *pilosa.UpdateCoordinatorMessage: + return encodeUpdateCoordinatorMessage(mt) + case *pilosa.NodeStateMessage: + return encodeNodeStateMessage(mt) + case *pilosa.RecalculateCaches: + return encodeRecalculateCaches(mt) + case *pilosa.NodeEvent: + return encodeNodeEventMessage(mt) + case *pilosa.NodeStatus: + return encodeNodeStatus(mt) + } + return nil +} + +func encodeResizeInstruction(m *pilosa.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), + } +} + +func encodeResizeSources(srcs []*pilosa.ResizeSource) []*internal.ResizeSource { + new := make([]*internal.ResizeSource, 0, len(srcs)) + for _, src := range srcs { + new = append(new, encodeResizeSource(src)) + } + return new +} + +func encodeResizeSource(m *pilosa.ResizeSource) *internal.ResizeSource { + return &internal.ResizeSource{ + Node: EncodeNode(m.Node), + Index: m.Index, + Field: m.Field, + View: m.View, + Shard: m.Shard, + } +} + +func encodeSchema(m *pilosa.Schema) *internal.Schema { + return &internal.Schema{ + Indexes: encodeIndexInfos(m.Indexes), + } +} + +func encodeIndexInfos(idxs []*pilosa.IndexInfo) []*internal.Index { + new := make([]*internal.Index, 0, len(idxs)) + for _, idx := range idxs { + new = append(new, encodeIndexInfo(idx)) + } + return new +} + +func encodeIndexInfo(idx *pilosa.IndexInfo) *internal.Index { + return &internal.Index{ + Name: idx.Name, + Fields: encodeFieldInfos(idx.Fields), + } +} + +func encodeFieldInfos(fs []*pilosa.FieldInfo) []*internal.Field { + new := make([]*internal.Field, 0, len(fs)) + for _, f := range fs { + new = append(new, encodeFieldInfo(f)) + } + return new +} + +func encodeFieldInfo(f *pilosa.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 +} + +func encodeFieldOptions(o *pilosa.FieldOptions) *internal.FieldOptions { + if o == nil { + return nil + } + return &internal.FieldOptions{ + Type: o.Type, + CacheType: o.CacheType, + CacheSize: o.CacheSize, + Min: o.Min, + Max: o.Max, + TimeQuantum: string(o.TimeQuantum), + Keys: o.Keys, + } +} + +// EncodeNodes converts a slice of Nodes into its internal representation. +func EncodeNodes(a []*pilosa.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 *pilosa.Node) *internal.Node { + return &internal.Node{ + ID: n.ID, + URI: n.URI.Encode(), + IsCoordinator: n.IsCoordinator, + } +} + +func encodeClusterStatus(m *pilosa.ClusterStatus) *internal.ClusterStatus { + return &internal.ClusterStatus{ + State: m.State, + ClusterID: m.ClusterID, + Nodes: EncodeNodes(m.Nodes), + } +} + +func encodeCreateShardMessage(m *pilosa.CreateShardMessage) *internal.CreateShardMessage { + return &internal.CreateShardMessage{ + Index: m.Index, + Shard: m.Shard, + } +} + +func encodeCreateIndexMessage(m *pilosa.CreateIndexMessage) *internal.CreateIndexMessage { + return &internal.CreateIndexMessage{ + Index: m.Index, + Meta: encodeIndexMeta(m.Meta), + } +} + +func encodeIndexMeta(m *pilosa.IndexOptions) *internal.IndexMeta { + return &internal.IndexMeta{ + Keys: m.Keys, + } +} + +func encodeDeleteIndexMessage(m *pilosa.DeleteIndexMessage) *internal.DeleteIndexMessage { + return &internal.DeleteIndexMessage{ + Index: m.Index, + } +} + +func encodeCreateFieldMessage(m *pilosa.CreateFieldMessage) *internal.CreateFieldMessage { + return &internal.CreateFieldMessage{ + Index: m.Index, + Field: m.Field, + Meta: encodeFieldOptions(m.Meta), + } +} + +func encodeDeleteFieldMessage(m *pilosa.DeleteFieldMessage) *internal.DeleteFieldMessage { + return &internal.DeleteFieldMessage{ + Index: m.Index, + Field: m.Field, + } +} + +func encodeCreateViewMessage(m *pilosa.CreateViewMessage) *internal.CreateViewMessage { + return &internal.CreateViewMessage{ + Index: m.Index, + Field: m.Field, + View: m.View, + } +} + +func encodeDeleteViewMessage(m *pilosa.DeleteViewMessage) *internal.DeleteViewMessage { + return &internal.DeleteViewMessage{ + Index: m.Index, + Field: m.Field, + View: m.View, + } +} + +func encodeResizeInstructionComplete(m *pilosa.ResizeInstructionComplete) *internal.ResizeInstructionComplete { + return &internal.ResizeInstructionComplete{ + JobID: m.JobID, + Node: EncodeNode(m.Node), + Error: m.Error, + } +} + +func encodeSetCoordinatorMessage(m *pilosa.SetCoordinatorMessage) *internal.SetCoordinatorMessage { + return &internal.SetCoordinatorMessage{ + New: EncodeNode(m.New), + } +} + +func encodeUpdateCoordinatorMessage(m *pilosa.UpdateCoordinatorMessage) *internal.UpdateCoordinatorMessage { + return &internal.UpdateCoordinatorMessage{ + New: EncodeNode(m.New), + } +} + +func encodeNodeStateMessage(m *pilosa.NodeStateMessage) *internal.NodeStateMessage { + return &internal.NodeStateMessage{ + NodeID: m.NodeID, + State: m.State, + } +} + +func encodeNodeEventMessage(m *pilosa.NodeEvent) *internal.NodeEventMessage { + return &internal.NodeEventMessage{ + Event: uint32(m.Event), + Node: EncodeNode(m.Node), + } +} + +func encodeNodeStatus(m *pilosa.NodeStatus) *internal.NodeStatus { + return &internal.NodeStatus{ + Node: EncodeNode(m.Node), + MaxShards: &internal.MaxShards{Standard: m.MaxShards}, + Schema: encodeSchema(m.Schema), + } +} + +func encodeRecalculateCaches(*pilosa.RecalculateCaches) *internal.RecalculateCaches { + return &internal.RecalculateCaches{} +} + +func decodeResizeInstruction(ri *internal.ResizeInstruction, m *pilosa.ResizeInstruction) { + m.JobID = ri.JobID + m.Node = &pilosa.Node{} + decodeNode(ri.Node, m.Node) + m.Coordinator = &pilosa.Node{} + decodeNode(ri.Coordinator, m.Coordinator) + m.Sources = make([]*pilosa.ResizeSource, len(ri.Sources)) + decodeResizeSources(ri.Sources, m.Sources) + m.Schema = &pilosa.Schema{} + decodeSchema(ri.Schema, m.Schema) + m.ClusterStatus = &pilosa.ClusterStatus{} + decodeClusterStatus(ri.ClusterStatus, m.ClusterStatus) +} + +func decodeResizeSources(srcs []*internal.ResizeSource, m []*pilosa.ResizeSource) { + for i := range srcs { + m[i] = &pilosa.ResizeSource{} + decodeResizeSource(srcs[i], m[i]) + } +} + +func decodeResizeSource(rs *internal.ResizeSource, m *pilosa.ResizeSource) { + m.Node = &pilosa.Node{} + decodeNode(rs.Node, m.Node) + m.Index = rs.Index + m.Field = rs.Field + m.View = rs.View + m.Shard = rs.Shard +} + +func decodeSchema(s *internal.Schema, m *pilosa.Schema) { + m.Indexes = make([]*pilosa.IndexInfo, len(s.Indexes)) + decodeIndexes(s.Indexes, m.Indexes) +} + +func decodeIndexes(idxs []*internal.Index, m []*pilosa.IndexInfo) { + for i := range idxs { + m[i] = &pilosa.IndexInfo{} + decodeIndex(idxs[i], m[i]) + } +} + +func decodeIndex(idx *internal.Index, m *pilosa.IndexInfo) { + m.Name = idx.Name + m.Fields = make([]*pilosa.FieldInfo, len(idx.Fields)) + decodeFields(idx.Fields, m.Fields) +} + +func decodeFields(fs []*internal.Field, m []*pilosa.FieldInfo) { + for i := range fs { + m[i] = &pilosa.FieldInfo{} + decodeField(fs[i], m[i]) + } +} + +func decodeField(f *internal.Field, m *pilosa.FieldInfo) { + m.Name = f.Name + m.Options = pilosa.FieldOptions{} + decodeFieldOptions(f.Meta, &m.Options) + m.Views = make([]*pilosa.ViewInfo, 0, len(f.Views)) + for _, viewname := range f.Views { + m.Views = append(m.Views, &pilosa.ViewInfo{Name: viewname}) + } +} + +func decodeFieldOptions(options *internal.FieldOptions, m *pilosa.FieldOptions) { + m.Type = options.Type + m.CacheType = options.CacheType + m.CacheSize = options.CacheSize + m.Min = options.Min + m.Max = options.Max + m.TimeQuantum = pilosa.TimeQuantum(options.TimeQuantum) + m.Keys = options.Keys +} + +func decodeNodes(a []*internal.Node, m []*pilosa.Node) { + for i := range a { + m[i] = &pilosa.Node{} + decodeNode(a[i], m[i]) + } +} + +func decodeClusterStatus(cs *internal.ClusterStatus, m *pilosa.ClusterStatus) { + m.State = cs.State + m.ClusterID = cs.ClusterID + m.Nodes = make([]*pilosa.Node, len(cs.Nodes)) + decodeNodes(cs.Nodes, m.Nodes) +} + +func decodeNode(node *internal.Node, m *pilosa.Node) { + m.ID = node.ID + decodeURI(node.URI, &m.URI) + m.IsCoordinator = node.IsCoordinator +} + +func decodeURI(i *internal.URI, m *pilosa.URI) { + m.Scheme = i.Scheme + m.Host = i.Host + m.Port = uint16(i.Port) +} + +func decodeCreateShardMessage(pb *internal.CreateShardMessage, m *pilosa.CreateShardMessage) { + m.Index = pb.Index + m.Shard = pb.Shard +} + +func decodeCreateIndexMessage(pb *internal.CreateIndexMessage, m *pilosa.CreateIndexMessage) { + m.Index = pb.Index + m.Meta = &pilosa.IndexOptions{} + decodeIndexMeta(pb.Meta, m.Meta) +} + +func decodeIndexMeta(pb *internal.IndexMeta, m *pilosa.IndexOptions) { + m.Keys = pb.Keys +} + +func decodeDeleteIndexMessage(pb *internal.DeleteIndexMessage, m *pilosa.DeleteIndexMessage) { + m.Index = pb.Index +} + +func decodeCreateFieldMessage(pb *internal.CreateFieldMessage, m *pilosa.CreateFieldMessage) { + m.Index = pb.Index + m.Field = pb.Field + m.Meta = &pilosa.FieldOptions{} + decodeFieldOptions(pb.Meta, m.Meta) +} + +func decodeDeleteFieldMessage(pb *internal.DeleteFieldMessage, m *pilosa.DeleteFieldMessage) { + m.Index = pb.Index + m.Field = pb.Field +} + +func decodeCreateViewMessage(pb *internal.CreateViewMessage, m *pilosa.CreateViewMessage) { + m.Index = pb.Index + m.Field = pb.Field + m.View = pb.View +} + +func decodeDeleteViewMessage(pb *internal.DeleteViewMessage, m *pilosa.DeleteViewMessage) { + m.Index = pb.Index + m.Field = pb.Field + m.View = pb.View +} + +func decodeResizeInstructionComplete(pb *internal.ResizeInstructionComplete, m *pilosa.ResizeInstructionComplete) { + m.JobID = pb.JobID + m.Node = &pilosa.Node{} + decodeNode(pb.Node, m.Node) + m.Error = pb.Error +} + +func decodeSetCoordinatorMessage(pb *internal.SetCoordinatorMessage, m *pilosa.SetCoordinatorMessage) { + m.New = &pilosa.Node{} + decodeNode(pb.New, m.New) +} + +func decodeUpdateCoordinatorMessage(pb *internal.UpdateCoordinatorMessage, m *pilosa.UpdateCoordinatorMessage) { + m.New = &pilosa.Node{} + decodeNode(pb.New, m.New) +} + +func decodeNodeStateMessage(pb *internal.NodeStateMessage, m *pilosa.NodeStateMessage) { + m.NodeID = pb.NodeID + m.State = pb.State +} + +func decodeNodeEventMessage(pb *internal.NodeEventMessage, m *pilosa.NodeEvent) { + m.Event = pilosa.NodeEventType(pb.Event) + m.Node = &pilosa.Node{} + decodeNode(pb.Node, m.Node) +} + +func decodeNodeStatus(pb *internal.NodeStatus, m *pilosa.NodeStatus) { + m.Node = &pilosa.Node{} + decodeNode(pb.Node, m.Node) + m.MaxShards = pb.MaxShards.Standard + m.Schema = &pilosa.Schema{} + decodeSchema(pb.Schema, m.Schema) +} + +func decodeRecalculateCaches(pb *internal.RecalculateCaches, m *pilosa.RecalculateCaches) {} diff --git a/field.go b/field.go index eea6bb10e..fb7e9af4f 100644 --- a/field.go +++ b/field.go @@ -1077,13 +1077,13 @@ func (f *Field) MarshalJSON() ([]byte, error) { thing := struct { Name string Options FieldOptions - Views []*viewInfo + Views []*ViewInfo }{ Name: f.Name(), Options: f.Options(), } for _, viewname := range f.viewNames() { - thing.Views = append(thing.Views, &viewInfo{Name: viewname}) + thing.Views = append(thing.Views, &ViewInfo{Name: viewname}) } return json.Marshal(thing) } @@ -1117,7 +1117,7 @@ func (p fieldSlice) Less(i, j int) bool { return p[i].Name() < p[j].Name() } type FieldInfo struct { Name string `json:"name"` Options FieldOptions `json:"options"` - Views []*viewInfo `json:"views,omitempty"` + Views []*ViewInfo `json:"views,omitempty"` } type fieldInfoSlice []*FieldInfo diff --git a/gossip/gossip.go b/gossip/gossip.go index 3dc7c202b..cd1d9ad0b 100644 --- a/gossip/gossip.go +++ b/gossip/gossip.go @@ -148,7 +148,7 @@ func WithLogger(logger *log.Logger) GossipMemberSetOption { // NewGossipMemberSet returns a new instance of GossipMemberSet based on options. func NewGossipMemberSet(cfg Config, api *pilosa.API, options ...GossipMemberSetOption) (*GossipMemberSet, error) { - host := api.Node().URI.Host() + host := api.Node().URI.GetHost() g := &GossipMemberSet{ papi: api, Logger: pilosa.NopLogger, @@ -193,10 +193,10 @@ func NewGossipMemberSet(cfg Config, api *pilosa.API, options ...GossipMemberSetO conf := memberlist.DefaultWANConfig() conf.Transport = g.transport.Net conf.Name = api.Node().ID - conf.BindAddr = api.Node().URI.Host() + conf.BindAddr = api.Node().URI.GetHost() conf.BindPort = port conf.AdvertisePort = port - conf.AdvertiseAddr = hostToIP(api.Node().URI.Host()) + conf.AdvertiseAddr = hostToIP(api.Node().URI.GetHost()) // conf.TCPTimeout = time.Duration(cfg.StreamTimeout) conf.SuspicionMult = cfg.SuspicionMult diff --git a/holder.go b/holder.go index 83ffd6967..10e147098 100644 --- a/holder.go +++ b/holder.go @@ -217,7 +217,7 @@ func (h *Holder) Schema() []*IndexInfo { for _, field := range index.Fields() { fi := &FieldInfo{Name: field.Name(), Options: field.Options()} for _, view := range field.views() { - fi.Views = append(fi.Views, &viewInfo{Name: view.name}) + fi.Views = append(fi.Views, &ViewInfo{Name: view.name}) } sort.Sort(viewInfoSlice(fi.Views)) di.Fields = append(di.Fields, fi) diff --git a/http/client.go b/http/client.go index 27597a5cb..41602b778 100644 --- a/http/client.go +++ b/http/client.go @@ -31,6 +31,7 @@ import ( "github.com/gogo/protobuf/proto" "github.com/pilosa/pilosa" + pilosaproto "github.com/pilosa/pilosa/encoding/proto" "github.com/pilosa/pilosa/internal" "github.com/pkg/errors" ) @@ -43,6 +44,7 @@ type ClientOptions struct { // InternalClient represents a client to the Pilosa cluster. type InternalClient struct { defaultURI *pilosa.URI + serializer pilosa.Serializer // The client to use for HTTP communication. HTTPClient *http.Client @@ -66,6 +68,7 @@ func NewInternalClient(host string, remoteClient *http.Client) (*InternalClient, func NewInternalClientFromURI(defaultURI *pilosa.URI, remoteClient *http.Client) *InternalClient { return &InternalClient{ defaultURI: defaultURI, + serializer: pilosaproto.Serializer{}, HTTPClient: remoteClient, } } @@ -819,12 +822,7 @@ func (c *InternalClient) RowAttrDiff(ctx context.Context, uri *pilosa.URI, index } // SendMessage posts a message synchronously. -func (c *InternalClient) SendMessage(ctx context.Context, uri *pilosa.URI, pb proto.Message) error { - msg, err := pilosa.MarshalMessage(pb) - if err != nil { - return fmt.Errorf("marshaling message: %v", err) - } - +func (c *InternalClient) SendMessage(ctx context.Context, uri *pilosa.URI, msg []byte) error { u := uriPathToURL(uri, "/internal/cluster/message") req, err := http.NewRequest("POST", u.String(), bytes.NewReader(msg)) if err != nil { @@ -998,7 +996,7 @@ func pos(rowID, columnID uint64) uint64 { func uriPathToURL(uri *pilosa.URI, path string) url.URL { return url.URL{ - Scheme: uri.Scheme(), + Scheme: uri.GetScheme(), Host: uri.HostPort(), Path: path, } @@ -1006,7 +1004,7 @@ func uriPathToURL(uri *pilosa.URI, path string) url.URL { func nodePathToURL(node *pilosa.Node, path string) url.URL { return url.URL{ - Scheme: node.URI.Scheme(), + Scheme: node.URI.GetScheme(), Host: node.URI.HostPort(), Path: path, } diff --git a/server.go b/server.go index b57935acb..b3bc6e224 100644 --- a/server.go +++ b/server.go @@ -54,6 +54,7 @@ type Server struct { executor *executor hosts []string clusterDisabled bool + serializer Serializer // External systemInfo SystemInfo @@ -200,6 +201,13 @@ func OptServerClusterDisabled(disabled bool, hosts []string) ServerOption { } } +func OptServerSerializer(ser Serializer) ServerOption { + return func(s *Server) error { + s.serializer = ser + return nil + } +} + func OptServerIsCoordinator(is bool) ServerOption { return func(s *Server) error { s.isCoordinator = is @@ -517,8 +525,12 @@ func (s *Server) receiveMessage(m Message) error { // SendSync represents an implementation of Broadcaster. func (s *Server) SendSync(m Message) error { - pb := encode(m) var eg errgroup.Group + msg, err := s.serializer.Marshal(m) + if err != nil { + return fmt.Errorf("marshaling message: %v", err) + } + msg = append([]byte{getMessageType(m)}, msg...) for _, node := range s.cluster.Nodes { node := node s.logger.Printf("SendSync to: %s", node.URI) @@ -528,7 +540,7 @@ func (s *Server) SendSync(m Message) error { } eg.Go(func() error { - return s.defaultClient.SendMessage(context.Background(), &node.URI, pb) + return s.defaultClient.SendMessage(context.Background(), &node.URI, msg) }) } @@ -542,9 +554,13 @@ func (s *Server) SendAsync(m Message) error { // SendTo represents an implementation of Broadcaster. 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) + msg, err := s.serializer.Marshal(m) + if err != nil { + return fmt.Errorf("marshaling message: %v", err) + } + msg = append([]byte{getMessageType(m)}, msg...) + return s.defaultClient.SendMessage(context.Background(), &to.URI, msg) } // node returns the pilosa.node object. It is used by membership protocols to @@ -613,7 +629,7 @@ func (s *Server) monitorDiagnostics() { s.diagnostics.Logger = s.logger s.diagnostics.SetVersion(Version) - s.diagnostics.Set("Host", s.uri.host) + s.diagnostics.Set("Host", s.uri.Host) s.diagnostics.Set("Cluster", strings.Join(s.cluster.nodeIDs(), ",")) s.diagnostics.Set("NumNodes", len(s.cluster.Nodes)) s.diagnostics.Set("NumCPU", runtime.NumCPU()) diff --git a/server/server.go b/server/server.go index c8de0664c..f46f6a64a 100644 --- a/server/server.go +++ b/server/server.go @@ -35,6 +35,7 @@ import ( "github.com/pilosa/pilosa" "github.com/pilosa/pilosa/boltdb" + "github.com/pilosa/pilosa/encoding/proto" "github.com/pilosa/pilosa/gcnotify" "github.com/pilosa/pilosa/gopsutil" "github.com/pilosa/pilosa/gossip" @@ -202,7 +203,7 @@ func (m *Command) SetupServer() error { // Setup TLS var TLSConfig *tls.Config - if uri.Scheme() == "https" { + if uri.GetScheme() == "https" { if m.Config.TLS.CertificatePath == "" { return errors.New("certificate path is required for TLS sockets") } @@ -235,7 +236,7 @@ func (m *Command) SetupServer() error { } // If port is 0, get auto-allocated port from listener - if uri.Port() == 0 { + if uri.GetPort() == 0 { uri.SetPort(uint16(m.ln.Addr().(*net.TCPAddr).Port)) } @@ -271,6 +272,7 @@ func (m *Command) SetupServer() error { pilosa.OptServerInternalClient(http.NewInternalClientFromURI(uri, c)), pilosa.OptServerPrimaryTranslateStore(primaryTranslateStore), pilosa.OptServerClusterDisabled(m.Config.Cluster.Disabled, m.Config.Cluster.Hosts), + pilosa.OptServerSerializer(proto.Serializer{}), coordinatorOpt, } @@ -309,7 +311,7 @@ func (m *Command) SetupNetworking() error { } // get the host portion of addr to use for binding - gossipHost := m.API.Node().URI.Host() + gossipHost := m.API.Node().URI.GetHost() m.gossipTransport, err = gossip.NewTransport(gossipHost, gossipPort, m.logger.Logger()) if err != nil { return errors.Wrap(err, "getting transport") @@ -366,19 +368,19 @@ func NewStatsClient(name string, host string) (pilosa.StatsClient, error) { // getListener gets a net.Listener based on the config. func getListener(uri pilosa.URI, tlsconf *tls.Config) (ln net.Listener, err error) { // If bind URI has the https scheme, enable TLS - if uri.Scheme() == "https" && tlsconf != nil { + if uri.GetScheme() == "https" && tlsconf != nil { ln, err = tls.Listen("tcp", uri.HostPort(), tlsconf) if err != nil { return nil, errors.Wrap(err, "tls.Listener") } - } else if uri.Scheme() == "http" { + } else if uri.GetScheme() == "http" { // Open HTTP listener to determine port (if specified as :0). ln, err = net.Listen("tcp", uri.HostPort()) if err != nil { return nil, errors.Wrap(err, "net.Listen") } } else { - return nil, errors.Errorf("unsupported scheme: %s", uri.Scheme()) + return nil, errors.Errorf("unsupported scheme: %s", uri.GetScheme()) } return ln, nil diff --git a/uri.go b/uri.go index 5d823d8b6..2058f70fa 100644 --- a/uri.go +++ b/uri.go @@ -43,17 +43,17 @@ var addressRegexp = regexp.MustCompile(`^(([+a-z]+):\/\/)?([0-9a-z.-]+|\[[:0-9a- // localhost // :10101 type URI struct { - scheme string `json:"scheme"` - host string `json:"host"` - port uint16 `json:"port"` + Scheme string `json:"scheme"` + Host string `json:"host"` + Port uint16 `json:"port"` } // DefaultURI creates and returns the default URI. func DefaultURI() *URI { return &URI{ - scheme: "http", - host: "localhost", - port: 10101, + Scheme: "http", + Host: "localhost", + Port: 10101, } } @@ -83,9 +83,9 @@ func NewURIFromAddress(address string) (*URI, error) { return parseAddress(address) } -// Scheme returns the scheme of this URI. -func (u *URI) Scheme() string { - return u.scheme +// GetScheme returns the scheme of this URI. +func (u *URI) GetScheme() string { + return u.Scheme } // SetScheme sets the scheme of this URI. @@ -94,13 +94,13 @@ func (u *URI) SetScheme(scheme string) error { if m == nil { return errors.New("invalid scheme") } - u.scheme = scheme + u.Scheme = scheme return nil } -// Host returns the host of this URI. -func (u *URI) Host() string { - return u.host +// GetHost returns the host of this URI. +func (u *URI) GetHost() string { + return u.Host } // SetHost sets the host of this URI. @@ -109,18 +109,18 @@ func (u *URI) SetHost(host string) error { if m == nil { return errors.New("invalid host") } - u.host = host + u.Host = host return nil } -// Port returns the port of this URI. -func (u *URI) Port() uint16 { - return u.port +// GetPort returns the port of this URI. +func (u *URI) GetPort() uint16 { + return u.Port } // SetPort sets the port of this URI. func (u *URI) SetPort(port uint16) { - u.port = port + u.Port = port } // HostPort returns `Host:Port` @@ -129,23 +129,23 @@ func (u *URI) HostPort() string { if u == nil { return "" } - s := fmt.Sprintf("%s:%d", u.host, u.port) + s := fmt.Sprintf("%s:%d", u.Host, u.Port) return s } // Normalize returns the address in a form usable by a HTTP client. func (u *URI) Normalize() string { - scheme := u.scheme + scheme := u.Scheme index := strings.Index(scheme, "+") if index >= 0 { scheme = scheme[:index] } - return fmt.Sprintf("%s://%s:%d", scheme, u.host, u.port) + return fmt.Sprintf("%s://%s:%d", scheme, u.Host, u.Port) } // String returns the address as a string. func (u URI) String() string { - return fmt.Sprintf("%s://%s:%d", u.scheme, u.host, u.port) + return fmt.Sprintf("%s://%s:%d", u.Scheme, u.Host, u.Port) } // Equals returns true if the checked URI is equivalent to this URI. @@ -199,9 +199,9 @@ func parseAddress(address string) (uri *URI, err error) { } } uri = &URI{ - scheme: scheme, - host: host, - port: uint16(port), + Scheme: scheme, + Host: host, + Port: uint16(port), } return uri, nil } @@ -213,9 +213,9 @@ func (u URI) Encode() *internal.URI { func encodeURI(u URI) *internal.URI { return &internal.URI{ - Scheme: u.scheme, - Host: u.host, - Port: uint32(u.port), + Scheme: u.Scheme, + Host: u.Host, + Port: uint32(u.Port), } } @@ -228,9 +228,9 @@ func decodeURI(i *internal.URI) URI { return URI{} } return URI{ - scheme: i.Scheme, - host: i.Host, - port: uint16(i.Port), + Scheme: i.Scheme, + Host: i.Host, + Port: uint16(i.Port), } } @@ -241,9 +241,9 @@ func (u *URI) MarshalJSON() ([]byte, error) { Host string `json:"host,omitempty"` Port uint16 `json:"port,omitempty"` } - output.Scheme = u.scheme - output.Host = u.host - output.Port = u.port + output.Scheme = u.Scheme + output.Host = u.Host + output.Port = u.Port return json.Marshal(output) } @@ -257,8 +257,8 @@ func (u *URI) UnmarshalJSON(b []byte) error { if err := json.Unmarshal(b, &input); err != nil { return err } - u.scheme = input.Scheme - u.host = input.Host - u.port = input.Port + u.Scheme = input.Scheme + u.Host = input.Host + u.Port = input.Port return nil } diff --git a/uri_internal_test.go b/uri_internal_test.go index dbcbfa04d..2aac9651f 100644 --- a/uri_internal_test.go +++ b/uri_internal_test.go @@ -93,8 +93,8 @@ func TestSetScheme(t *testing.T) { if err != nil { t.Fatal(err) } - if uri.Scheme() != target { - t.Fatalf("%s != %s", uri.Scheme(), target) + if uri.GetScheme() != target { + t.Fatalf("%s != %s", uri.GetScheme(), target) } } @@ -105,8 +105,8 @@ func TestSetHost(t *testing.T) { if err != nil { t.Fatal(err) } - if uri.Host() != target { - t.Fatalf("%s != %s", uri.host, target) + if uri.GetHost() != target { + t.Fatalf("%s != %s", uri.Host, target) } } @@ -114,8 +114,8 @@ func TestSetPort(t *testing.T) { uri := DefaultURI() target := uint16(9999) uri.SetPort(target) - if uri.Port() != target { - t.Fatalf("%d != %d", uri.port, target) + if uri.GetPort() != target { + t.Fatalf("%d != %d", uri.Port, target) } } @@ -147,14 +147,14 @@ func TestHostPort(t *testing.T) { } func compare(t *testing.T, uri *URI, scheme string, host string, port uint16) { - if uri.Scheme() != scheme { - t.Fatalf("Scheme does not match: %s != %s", uri.scheme, scheme) + if uri.GetScheme() != scheme { + t.Fatalf("Scheme does not match: %s != %s", uri.Scheme, scheme) } - if uri.Host() != host { - t.Fatalf("Host does not match: %s != %s", uri.host, host) + if uri.GetHost() != host { + t.Fatalf("Host does not match: %s != %s", uri.Host, host) } - if uri.Port() != port { - t.Fatalf("Port does not match: %d != %d", uri.port, port) + if uri.GetPort() != port { + t.Fatalf("Port does not match: %d != %d", uri.Port, port) } } diff --git a/view.go b/view.go index 0f2e189cd..609664304 100644 --- a/view.go +++ b/view.go @@ -421,12 +421,12 @@ func (v *view) rangeBetween(bitDepth uint, predicateMin, predicateMax uint64) (* return r, nil } -// viewInfo represents schema information for a view. -type viewInfo struct { +// ViewInfo represents schema information for a view. +type ViewInfo struct { Name string `json:"name"` } -type viewInfoSlice []*viewInfo +type viewInfoSlice []*ViewInfo func (p viewInfoSlice) Swap(i, j int) { p[i], p[j] = p[j], p[i] } func (p viewInfoSlice) Len() int { return len(p) }