scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-10-08 11:57:51 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

add proto encoding subpackage and use for send and receive message

Browse source
This commit is contained in:
Matt Jaffee 2018-07-05 12:00:42 -05:00
parent 1b2aaa26bf
commit 2cec75e399
No known key found for this signature in database
GPG key ID: 08A3DFFF987B11BF
14 changed files with 778 additions and 122 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
api.go
View file

@ -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

116
broadcast.go
View file

@ -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.

5
client.go
View file

@ -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) {

4
cluster.go
View file

@ -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
}

594
encoding/proto/proto.go Normal file
View file

@ -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) {}

6
field.go
View file

@ -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

6
gossip/gossip.go
View file

@ -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

2
holder.go
View file

@ -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)

14
http/client.go
View file

@ -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,
}

26
server.go
View file

@ -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())

14
server/server.go
View file

@ -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

74
uri.go
View file

@ -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
}

24
uri_internal_test.go
View file

@ -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)
}
}

6
view.go
View file

@ -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) }