scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-15 08:41:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1454 from jaffee/core-structs

Browse source 
invert encoding/decoding and remove internal references
This commit is contained in:
Matthew Jaffee 2018-07-05 17:51:52 -05:00 • committed by GitHub
commit b7b46f1d66
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
29 changed files with 1618 additions and 830 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

53
api.go
View file

@ -26,8 +26,6 @@ import (
"strings"
"time"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/pql"
"github.com/pkg/errors"
)
@ -38,6 +36,8 @@ type API struct {
holder *Holder
cluster *cluster
server *Server
Serializer Serializer
}
// APIOption is a functional option type for pilosa.API
@ -48,6 +48,7 @@ func OptAPIServer(s *Server) APIOption {
a.server = s
a.holder = s.holder
a.cluster = s.cluster
a.Serializer = s.serializer
return nil
}
}
@ -149,6 +150,10 @@ func (api *API) Query(ctx context.Context, req *QueryRequest) (QueryResponse, er
return resp, nil
}
func (api *API) Holder() *Holder {
return api.server.Holder()
}
// readColumnAttrSets returns a list of column attribute objects by id.
func (api *API) readColumnAttrSets(index *Index, ids []uint64) ([]*ColumnAttrSet, error) {
if index == nil {
@ -185,12 +190,11 @@ func (api *API) CreateIndex(ctx context.Context, indexName string, options Index
}
// Send the create index message to all nodes.
err = api.server.SendSync(
&internal.CreateIndexMessage{
&CreateIndexMessage{
Index: indexName,
Meta: options.Encode(),
Meta: &options,
})
if err != nil {
api.server.logger.Printf("problem sending CreateIndex message: %s", err)
return nil, errors.Wrap(err, "sending CreateIndex message")
}
api.holder.Stats.Count("createIndex", 1, 1.0)
@ -224,7 +228,7 @@ func (api *API) DeleteIndex(ctx context.Context, indexName string) error {
}
// Send the delete index message to all nodes.
err = api.server.SendSync(
&internal.DeleteIndexMessage{
&DeleteIndexMessage{
Index: indexName,
})
if err != nil {
@ -244,7 +248,7 @@ func (api *API) CreateField(ctx context.Context, indexName string, fieldName str
}
// Apply functional options.
fo := fieldOptions{}
fo := FieldOptions{}
for _, opt := range opts {
err := opt(&fo)
if err != nil {
@ -266,10 +270,10 @@ func (api *API) CreateField(ctx context.Context, indexName string, fieldName str
// Send the create field message to all nodes.
err = api.server.SendSync(
&internal.CreateFieldMessage{
&CreateFieldMessage{
Index: indexName,
Field: fieldName,
Meta: fo.Encode(),
Meta: &fo,
})
if err != nil {
api.server.logger.Printf("problem sending CreateField message: %s", err)
@ -313,7 +317,7 @@ func (api *API) DeleteField(ctx context.Context, indexName string, fieldName str
// Send the delete field message to all nodes.
err := api.server.SendSync(
&internal.DeleteFieldMessage{
&DeleteFieldMessage{
Index: indexName,
Field: fieldName,
})
@ -384,8 +388,8 @@ func (api *API) FragmentBlockData(ctx context.Context, body io.Reader) ([]byte,
if err != nil {
return nil, NewBadRequestError(errors.Wrap(err, "read body error"))
}
var req internal.BlockDataRequest
if err := proto.Unmarshal(reqBytes, &req); err != nil {
var req BlockDataRequest
if err := api.Serializer.Unmarshal(reqBytes, &req); err != nil {
return nil, NewBadRequestError(errors.Wrap(err, "unmarshal body error"))
}
@ -395,11 +399,11 @@ func (api *API) FragmentBlockData(ctx context.Context, body io.Reader) ([]byte,
return nil, ErrFragmentNotFound
}
var resp = internal.BlockDataResponse{}
var resp = BlockDataResponse{}
resp.RowIDs, resp.ColumnIDs = f.blockData(int(req.Block))
// Encode response.
buf, err := proto.Marshal(&resp)
buf, err := api.Serializer.Marshal(&resp)
if err != nil {
return nil, errors.Wrap(err, "merge block response encoding error")
}
@ -442,7 +446,7 @@ func (api *API) RecalculateCaches(ctx context.Context) error {
return errors.Wrap(err, "validating api method")
}
err := api.server.SendSync(&internal.RecalculateCaches{})
err := api.server.SendSync(&RecalculateCaches{})
if err != nil {
return errors.Wrap(err, "broacasting message")
}
@ -463,14 +467,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(pb); err != nil {
if err := api.server.receiveMessage(msg); err != nil {
return errors.Wrap(err, "receiving message")
}
return nil
@ -521,7 +526,7 @@ func (api *API) DeleteView(ctx context.Context, indexName string, fieldName stri
// Send the delete view message to all nodes.
err := api.server.SendSync(
&internal.DeleteViewMessage{
&DeleteViewMessage{
Index: indexName,
Field: fieldName,
View: viewName,
@ -603,7 +608,7 @@ func (api *API) FieldAttrDiff(ctx context.Context, indexName string, fieldName s
}
// Import bulk imports data into a particular index,field,shard.
func (api *API) Import(ctx context.Context, req internal.ImportRequest) error {
func (api *API) Import(ctx context.Context, req *ImportRequest) error {
if err := api.validate(apiImport); err != nil {
return errors.Wrap(err, "validating api method")
}
@ -632,7 +637,7 @@ func (api *API) Import(ctx context.Context, req internal.ImportRequest) error {
}
// ImportValue bulk imports values into a particular field.
func (api *API) ImportValue(ctx context.Context, req internal.ImportValueRequest) error {
func (api *API) ImportValue(ctx context.Context, req *ImportValueRequest) error {
if err := api.validate(apiImportValue); err != nil {
return errors.Wrap(err, "validating api method")
}
@ -716,8 +721,8 @@ func (api *API) SetCoordinator(ctx context.Context, id string) (oldNode, newNode
// Send the set-coordinator message to new node.
err = api.server.SendTo(
newNode,
&internal.SetCoordinatorMessage{
New: EncodeNode(newNode),
&SetCoordinatorMessage{
New: newNode,
})
if err != nil {
return nil, nil, fmt.Errorf("problem sending SetCoordinator message: %s", err)

155
broadcast.go
View file

@ -16,20 +16,27 @@ package pilosa
import (
"fmt"
"reflect"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/internal"
"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(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 +47,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 (
@ -68,95 +75,91 @@ const (
messageTypeNodeStatus
)
// MarshalMessage encodes the protobuf message into a byte slice.
func MarshalMessage(m proto.Message) ([]byte, error) {
var typ uint8
switch obj := m.(type) {
case *internal.CreateShardMessage:
typ = messageTypeCreateShard
case *internal.CreateIndexMessage:
typ = messageTypeCreateIndex
case *internal.DeleteIndexMessage:
typ = messageTypeDeleteIndex
case *internal.CreateFieldMessage:
typ = messageTypeCreateField
case *internal.DeleteFieldMessage:
typ = messageTypeDeleteField
case *internal.CreateViewMessage:
typ = messageTypeCreateView
case *internal.DeleteViewMessage:
typ = messageTypeDeleteView
case *internal.ClusterStatus:
typ = messageTypeClusterStatus
case *internal.ResizeInstruction:
typ = messageTypeResizeInstruction
case *internal.ResizeInstructionComplete:
typ = messageTypeResizeInstructionComplete
case *internal.SetCoordinatorMessage:
typ = messageTypeSetCoordinator
case *internal.UpdateCoordinatorMessage:
typ = messageTypeUpdateCoordinator
case *internal.NodeStateMessage:
typ = messageTypeNodeState
case *internal.RecalculateCaches:
typ = messageTypeRecalculateCaches
case *internal.NodeEventMessage:
typ = messageTypeNodeEvent
case *internal.NodeStatus:
typ = messageTypeNodeStatus
default:
return nil, fmt.Errorf("message type not implemented for marshalling: %s", reflect.TypeOf(obj))
}
buf, err := proto.Marshal(m)
// MarshalInternalMessage serializes the pilosa message and adds pilosa internal
// type info which is used by the internal messaging stuff.
func MarshalInternalMessage(m Message, s Serializer) ([]byte, error) {
typ := getMessageType(m)
buf, err := s.Marshal(m)
if err != nil {
return nil, errors.Wrap(err, "marshalling")
return nil, errors.Wrap(err, "marshaling")
}
return append([]byte{typ}, buf...), nil
}
// UnmarshalMessage decodes the byte slice into a protobuf message.
func UnmarshalMessage(buf []byte) (proto.Message, error) {
typ, buf := buf[0], buf[1:]
var m proto.Message
func getMessage(typ byte) Message {
switch typ {
case messageTypeCreateShard:
m = &internal.CreateShardMessage{}
return &CreateShardMessage{}
case messageTypeCreateIndex:
m = &internal.CreateIndexMessage{}
return &CreateIndexMessage{}
case messageTypeDeleteIndex:
m = &internal.DeleteIndexMessage{}
return &DeleteIndexMessage{}
case messageTypeCreateField:
m = &internal.CreateFieldMessage{}
return &CreateFieldMessage{}
case messageTypeDeleteField:
m = &internal.DeleteFieldMessage{}
return &DeleteFieldMessage{}
case messageTypeCreateView:
m = &internal.CreateViewMessage{}
return &CreateViewMessage{}
case messageTypeDeleteView:
m = &internal.DeleteViewMessage{}
return &DeleteViewMessage{}
case messageTypeClusterStatus:
m = &internal.ClusterStatus{}
return &ClusterStatus{}
case messageTypeResizeInstruction:
m = &internal.ResizeInstruction{}
return &ResizeInstruction{}
case messageTypeResizeInstructionComplete:
m = &internal.ResizeInstructionComplete{}
return &ResizeInstructionComplete{}
case messageTypeSetCoordinator:
m = &internal.SetCoordinatorMessage{}
return &SetCoordinatorMessage{}
case messageTypeUpdateCoordinator:
m = &internal.UpdateCoordinatorMessage{}
return &UpdateCoordinatorMessage{}
case messageTypeNodeState:
m = &internal.NodeStateMessage{}
return &NodeStateMessage{}
case messageTypeRecalculateCaches:
m = &internal.RecalculateCaches{}
return &RecalculateCaches{}
case messageTypeNodeEvent:
m = &internal.NodeEventMessage{}
return &NodeEvent{}
case messageTypeNodeStatus:
m = &internal.NodeStatus{}
return &NodeStatus{}
default:
return nil, fmt.Errorf("invalid message type: %d", typ)
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))
}
if err := proto.Unmarshal(buf, m); err != nil {
return nil, errors.Wrap(err, "unmarshalling")
}
return m, nil
}

51
broadcast_test.go
View file

@ -1,51 +0,0 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package pilosa_test
import (
"reflect"
"testing"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/internal"
)
// Ensure a message can be marshaled and unmarshaled.
func TestMessage_Marshal(t *testing.T) {
testMessageMarshal(t, &internal.CreateShardMessage{
Index: "i",
Shard: 8,
})
testMessageMarshal(t, &internal.DeleteIndexMessage{
Index: "i",
})
}
func testMessageMarshal(t *testing.T, m proto.Message) {
marshalled, err := pilosa.MarshalMessage(m)
if err != nil {
t.Fatal(err)
}
unmarshalled, err := pilosa.UnmarshalMessage(marshalled)
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(unmarshalled, m) {
t.Fatalf("unexpected message marshalling: %s", unmarshalled)
}
}

33
cache.go
View file

@ -22,7 +22,6 @@ import (
"sync"
"time"
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/lru"
)
@ -318,22 +317,6 @@ type Pair struct {
Count uint64 `json:"count"`
}
func encodePair(p Pair) *internal.Pair {
return &internal.Pair{
ID: p.ID,
Key: p.Key,
Count: p.Count,
}
}
func decodePair(pb *internal.Pair) Pair {
return Pair{
ID: pb.ID,
Key: pb.Key,
Count: pb.Count,
}
}
// Pairs is a sortable slice of Pair objects.
type Pairs []Pair
@ -409,22 +392,6 @@ func (p Pairs) String() string {
return buf.String()
}
func EncodePairs(a Pairs) []*internal.Pair {
other := make([]*internal.Pair, len(a))
for i := range a {
other[i] = encodePair(a[i])
}
return other
}
func decodePairs(a []*internal.Pair) []Pair {
other := make([]Pair, len(a))
for i := range a {
other[i] = decodePair(a[i])
}
return other
}
// uint64Slice represents a sortable slice of uint64 numbers.
type uint64Slice []uint64

19
client.go
View file

@ -3,9 +3,6 @@ package pilosa
import (
"context"
"io"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/internal"
)
// Bit represents the intersection of a row and a column. It can be specifed by
@ -36,8 +33,8 @@ type InternalClient interface {
Schema(ctx context.Context) ([]*IndexInfo, error)
CreateIndex(ctx context.Context, index string, opt IndexOptions) error
FragmentNodes(ctx context.Context, index string, shard uint64) ([]*Node, error)
Query(ctx context.Context, index string, queryRequest *internal.QueryRequest) (*internal.QueryResponse, error)
QueryNode(ctx context.Context, uri *URI, index string, queryRequest *internal.QueryRequest) (*internal.QueryResponse, error)
Query(ctx context.Context, index string, queryRequest *QueryRequest) (*QueryResponse, error)
QueryNode(ctx context.Context, uri *URI, index string, queryRequest *QueryRequest) (*QueryResponse, error)
Import(ctx context.Context, index, field string, shard uint64, bits []Bit) error
ImportK(ctx context.Context, index, field string, bits []Bit) error
EnsureIndex(ctx context.Context, name string, options IndexOptions) error
@ -49,19 +46,19 @@ 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)
}
//===============
type InternalQueryClient interface {
QueryNode(ctx context.Context, uri *URI, index string, queryRequest *internal.QueryRequest) (*internal.QueryResponse, error)
QueryNode(ctx context.Context, uri *URI, index string, queryRequest *QueryRequest) (*QueryResponse, error)
}
type NopInternalQueryClient struct{}
func (n *NopInternalQueryClient) QueryNode(ctx context.Context, uri *URI, index string, queryRequest *internal.QueryRequest) (*internal.QueryResponse, error) {
func (n *NopInternalQueryClient) QueryNode(ctx context.Context, uri *URI, index string, queryRequest *QueryRequest) (*QueryResponse, error) {
return nil, nil
}
@ -91,10 +88,10 @@ func (n NopInternalClient) CreateIndex(ctx context.Context, index string, opt In
func (n NopInternalClient) FragmentNodes(ctx context.Context, index string, shard uint64) ([]*Node, error) {
return nil, nil
}
func (n NopInternalClient) Query(ctx context.Context, index string, queryRequest *internal.QueryRequest) (*internal.QueryResponse, error) {
func (n NopInternalClient) Query(ctx context.Context, index string, queryRequest *QueryRequest) (*QueryResponse, error) {
return nil, nil
}
func (n NopInternalClient) QueryNode(ctx context.Context, uri *URI, index string, queryRequest *internal.QueryRequest) (*internal.QueryResponse, error) {
func (n NopInternalClient) QueryNode(ctx context.Context, uri *URI, index string, queryRequest *QueryRequest) (*QueryResponse, error) {
return nil, nil
}
func (n NopInternalClient) Import(ctx context.Context, index, field string, shard uint64, bits []Bit) error {
@ -128,7 +125,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) {

253
cluster.go
View file

@ -68,52 +68,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
@ -313,8 +267,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)
@ -468,7 +422,7 @@ func (c *cluster) setNodeState(state string) error {
}
// Send node state to coordinator.
ns := &internal.NodeStateMessage{
ns := &NodeStateMessage{
NodeID: c.Node.ID,
State: state,
}
@ -505,12 +459,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, its ID, and current state.
func (c *cluster) Status() *ClusterStatus {
return &ClusterStatus{
ClusterID: c.id,
State: c.state,
Nodes: EncodeNodes(c.Nodes),
Nodes: c.Nodes,
}
}
@ -685,8 +639,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)
@ -745,7 +699,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
@ -756,8 +710,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,
@ -901,9 +855,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)
@ -1013,8 +967,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
@ -1120,7 +1074,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
@ -1144,12 +1098,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)
@ -1175,7 +1129,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.
@ -1193,7 +1147,7 @@ func (c *cluster) followResizeInstruction(instr *internal.ResizeInstruction) err
<-c.holder.opened
// Prepare the return message.
complete := &internal.ResizeInstructionComplete{
complete := &ResizeInstructionComplete{
JobID: instr.JobID,
Node: instr.Node,
Error: "",
@ -1212,7 +1166,7 @@ func (c *cluster) followResizeInstruction(instr *internal.ResizeInstruction) err
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)
@ -1264,14 +1218,14 @@ func (c *cluster) followResizeInstruction(instr *internal.ResizeInstruction) err
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)
@ -1308,7 +1262,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
@ -1411,7 +1365,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 {
@ -1552,32 +1506,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 == "" {
@ -1611,7 +1539,7 @@ func (c *cluster) considerTopology() error {
}
// ReceiveEvent represents an implementation of EventHandler.
func (c *cluster) ReceiveEvent(e *nodeEvent) error {
func (c *cluster) ReceiveEvent(e *NodeEvent) error {
// Ignore events sent from this node.
if e.Node.ID == c.Node.ID {
return nil
@ -1751,7 +1679,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)
@ -1763,7 +1691,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 {
@ -1812,3 +1740,120 @@ 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
}
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"`
}
// Schema contains information about indexes and their configuration.
type Schema struct {
Indexes []*IndexInfo
}
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
}
type CreateIndexMessage struct {
Index string
Meta *IndexOptions
}
type DeleteIndexMessage struct {
Index string
}
type CreateFieldMessage struct {
Index string
Field string
Meta *FieldOptions
}
type DeleteFieldMessage struct {
Index string
Field string
}
type CreateViewMessage struct {
Index string
Field string
View string
}
type DeleteViewMessage struct {
Index string
Field string
View string
}
type ResizeInstructionComplete struct {
JobID int64
Node *Node
Error string
}
type SetCoordinatorMessage struct {
New *Node
}
type UpdateCoordinatorMessage struct {
New *Node
}
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"`
}
type NodeStatus struct {
Node *Node
MaxShards map[string]uint64
Schema *Schema
}
type RecalculateCaches struct{}

41
cluster_internal_test.go
View file

@ -24,7 +24,6 @@ import (
"testing/quick"
"github.com/davecgh/go-spew/spew"
"github.com/pilosa/pilosa/internal"
"github.com/pkg/errors"
)
@ -175,19 +174,19 @@ func TestFragSources(t *testing.T) {
from *cluster
to *cluster
idx *Index
expected map[string][]*internal.ResizeSource
expected map[string][]*ResizeSource
err string
}{
{
from: c1,
to: c2,
idx: idx,
expected: map[string][]*internal.ResizeSource{
"node0": []*internal.ResizeSource{},
"node1": []*internal.ResizeSource{},
"node2": []*internal.ResizeSource{
{&internal.Node{"node0", &internal.URI{"http", "host0", 10101}, false}, "i", "f", "standard", uint64(0)},
{&internal.Node{"node1", &internal.URI{"http", "host1", 10101}, false}, "i", "f", "standard", uint64(2)},
expected: map[string][]*ResizeSource{
"node0": []*ResizeSource{},
"node1": []*ResizeSource{},
"node2": []*ResizeSource{
{&Node{"node0", URI{"http", "host0", 10101}, false}, "i", "f", "standard", uint64(0)},
{&Node{"node1", URI{"http", "host1", 10101}, false}, "i", "f", "standard", uint64(2)},
},
},
err: "",
@ -196,13 +195,13 @@ func TestFragSources(t *testing.T) {
from: c4,
to: c3,
idx: idx,
expected: map[string][]*internal.ResizeSource{
"node0": []*internal.ResizeSource{
{&internal.Node{"node1", &internal.URI{"http", "host1", 10101}, false}, "i", "f", "standard", uint64(1)},
expected: map[string][]*ResizeSource{
"node0": []*ResizeSource{
{&Node{"node1", URI{"http", "host1", 10101}, false}, "i", "f", "standard", uint64(1)},
},
"node1": []*internal.ResizeSource{
{&internal.Node{"node0", &internal.URI{"http", "host0", 10101}, false}, "i", "f", "standard", uint64(0)},
{&internal.Node{"node0", &internal.URI{"http", "host0", 10101}, false}, "i", "f", "standard", uint64(2)},
"node1": []*ResizeSource{
{&Node{"node0", URI{"http", "host0", 10101}, false}, "i", "f", "standard", uint64(0)},
{&Node{"node0", URI{"http", "host0", 10101}, false}, "i", "f", "standard", uint64(2)},
},
},
err: "",
@ -211,15 +210,15 @@ func TestFragSources(t *testing.T) {
from: c5,
to: c4,
idx: idx,
expected: map[string][]*internal.ResizeSource{
"node0": []*internal.ResizeSource{
{&internal.Node{"node2", &internal.URI{"http", "host2", 10101}, false}, "i", "f", "standard", uint64(0)},
{&internal.Node{"node2", &internal.URI{"http", "host2", 10101}, false}, "i", "f", "standard", uint64(2)},
expected: map[string][]*ResizeSource{
"node0": []*ResizeSource{
{&Node{"node2", URI{"http", "host2", 10101}, false}, "i", "f", "standard", uint64(0)},
{&Node{"node2", URI{"http", "host2", 10101}, false}, "i", "f", "standard", uint64(2)},
},
"node1": []*internal.ResizeSource{
{&internal.Node{"node0", &internal.URI{"http", "host0", 10101}, false}, "i", "f", "standard", uint64(3)},
"node1": []*ResizeSource{
{&Node{"node0", URI{"http", "host0", 10101}, false}, "i", "f", "standard", uint64(3)},
},
"node2": []*internal.ResizeSource{},
"node2": []*ResizeSource{},
},
err: "",
},

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

File diff suppressed because it is too large Load diff

4
event.go
View file

@ -23,8 +23,8 @@ const (
NodeUpdate
)
// nodeEvent is a single event related to node activity in the cluster.
type nodeEvent struct {
// NodeEvent is a single event related to node activity in the cluster.
type NodeEvent struct {
Event NodeEventType
Node *Node
}

62
executor.go
View file

@ -20,7 +20,6 @@ import (
"sort"
"time"
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/pql"
"github.com/pkg/errors"
)
@ -335,7 +334,7 @@ func (e *executor) executeBitmapCall(ctx context.Context, index string, c *pql.C
other, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn)
if err != nil {
return nil, err
return nil, errors.Wrap(err, "map reduce")
}
// Attach attributes for Row() calls.
@ -1375,7 +1374,7 @@ func (e *executor) executeSetColumnAttrs(ctx context.Context, index string, c *p
// exec executes a PQL query remotely for a set of shards on a node.
func (e *executor) remoteExec(ctx context.Context, node *Node, index string, q *pql.Query, shards []uint64, opt *execOptions) (results []interface{}, err error) {
// Encode request object.
pbreq := &internal.QueryRequest{
pbreq := &QueryRequest{
Query: q.String(),
Shards: shards,
Remote: true,
@ -1386,40 +1385,7 @@ func (e *executor) remoteExec(ctx context.Context, node *Node, index string, q *
return nil, err
}
// Return an error, if specified on response.
if err := decodeError(pb.Err); err != nil {
return nil, err
}
// Return appropriate data for the query.
results = make([]interface{}, len(q.Calls))
for i, call := range q.Calls {
var v interface{}
var err error
switch call.Name {
case "Average", "Sum":
v, err = decodeValCount(pb.Results[i].GetValCount()), nil
case "TopN":
v, err = decodePairs(pb.Results[i].GetPairs()), nil
case "Count":
v, err = pb.Results[i].N, nil
case "Set":
v, err = pb.Results[i].Changed, nil
case "Clear":
v, err = pb.Results[i].Changed, nil
case "SetRowAttrs":
case "SetColumnAttrs":
default:
v, err = DecodeRow(pb.Results[i].GetRow()), nil
}
if err != nil {
return nil, err
}
results[i] = v
}
return results, nil
return pb.Results, pb.Err
}
// shardsByNode returns a mapping of nodes to shards.
@ -1473,7 +1439,7 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
for {
select {
case <-ctx.Done():
return nil, ctx.Err()
return nil, errors.Wrap(ctx.Err(), "context done")
case resp := <-ch:
// On error retry against remaining nodes. If an error returns then
// the context will cancel and cause all open goroutines to return.
@ -1483,10 +1449,10 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
nodes = Nodes(nodes).Filter(resp.node)
// Begin mapper against secondary nodes.
if err := e.mapper(ctx, ch, nodes, index, resp.shards, c, opt, mapFn, reduceFn); err == errShardUnavailable {
if err := e.mapper(ctx, ch, nodes, index, resp.shards, c, opt, mapFn, reduceFn); errors.Cause(err) == errShardUnavailable {
return nil, resp.err
} else if err != nil {
return nil, err
return nil, errors.Wrap(err, "calling mapper")
}
continue
}
@ -1507,7 +1473,7 @@ func (e *executor) mapper(ctx context.Context, ch chan mapResponse, nodes []*Nod
// Group shards together by nodes.
m, err := e.shardsByNode(nodes, index, shards)
if err != nil {
return err
return errors.Wrap(err, "shards by node")
}
// Execute each node in a separate goroutine.
@ -1754,20 +1720,6 @@ func (vc *ValCount) Add(other ValCount) ValCount {
}
}
func EncodeValCount(vc ValCount) *internal.ValCount {
return &internal.ValCount{
Val: vc.Val,
Count: vc.Count,
}
}
func decodeValCount(pb *internal.ValCount) ValCount {
return ValCount{
Val: pb.Val,
Count: pb.Count,
}
}
// Smaller returns the smaller of the two ValCounts.
func (vc *ValCount) Smaller(other ValCount) ValCount {
if vc.Count == 0 || (other.Val < vc.Val && other.Count > 0) {

2
executor_test.go
View file

@ -385,7 +385,7 @@ func TestExecutor_Execute_OldPQL(t *testing.T) {
hldr.SetBit("i", "f", 1, 0)
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `SetBit(frame=f, row=11, col=1)`}); err == nil || errors.Cause(err).Error() != "unknown call: SetBit" {
t.Fatalf("Expected error: 'unknown call: SetBit', got: %v", errors.Cause(err))
t.Fatalf("Expected error: 'unknown call: SetBit', got: %v. Full: %v", errors.Cause(err), err)
}
}

82
field.go
View file

@ -68,7 +68,7 @@ type Field struct {
Stats StatsClient
// Field options.
options fieldOptions
options FieldOptions
bsiGroups []*bsiGroup
@ -76,17 +76,17 @@ type Field struct {
}
// FieldOption is a functional option type for pilosa.fieldOptions.
type FieldOption func(fo *fieldOptions) error
type FieldOption func(fo *FieldOptions) error
func OptFieldKeys() FieldOption {
return func(fo *fieldOptions) error {
return func(fo *FieldOptions) error {
fo.Keys = true
return nil
}
}
func OptFieldTypeDefault() FieldOption {
return func(fo *fieldOptions) error {
return func(fo *FieldOptions) error {
if fo.Type != "" {
return errors.Errorf("field type is already set to: %s", fo.Type)
}
@ -98,7 +98,7 @@ func OptFieldTypeDefault() FieldOption {
}
func OptFieldTypeSet(cacheType string, cacheSize uint32) FieldOption {
return func(fo *fieldOptions) error {
return func(fo *FieldOptions) error {
if fo.Type != "" {
return errors.Errorf("field type is already set to: %s", fo.Type)
}
@ -110,7 +110,7 @@ func OptFieldTypeSet(cacheType string, cacheSize uint32) FieldOption {
}
func OptFieldTypeInt(min, max int64) FieldOption {
return func(fo *fieldOptions) error {
return func(fo *FieldOptions) error {
if fo.Type != "" {
return errors.Errorf("field type is already set to: %s", fo.Type)
}
@ -125,7 +125,7 @@ func OptFieldTypeInt(min, max int64) FieldOption {
}
func OptFieldTypeTime(timeQuantum TimeQuantum) FieldOption {
return func(fo *fieldOptions) error {
return func(fo *FieldOptions) error {
if fo.Type != "" {
return errors.Errorf("field type is already set to: %s", fo.Type)
}
@ -146,7 +146,7 @@ func NewField(path, index, name string, opts FieldOption) (*Field, error) {
}
// Apply functional option.
fo := fieldOptions{}
fo := FieldOptions{}
err = opts(&fo)
if err != nil {
return nil, errors.Wrap(err, "applying option")
@ -233,7 +233,7 @@ func (f *Field) CacheSize() uint32 {
}
// Options returns all options for this field.
func (f *Field) Options() fieldOptions {
func (f *Field) Options() FieldOptions {
f.mu.RLock()
defer f.mu.RUnlock()
return f.options
@ -351,7 +351,7 @@ func (f *Field) saveMeta() error {
}
// applyOptions configures the field based on opt.
func (f *Field) applyOptions(opt fieldOptions) error {
func (f *Field) applyOptions(opt FieldOptions) error {
switch opt.Type {
case FieldTypeSet, "":
f.options.Type = FieldTypeSet
@ -624,7 +624,7 @@ func (f *Field) createViewIfNotExists(name string) (*view, error) {
if created {
// Broadcast view creation to the cluster.
err = f.broadcaster.SendSync(
&internal.CreateViewMessage{
&CreateViewMessage{
Index: f.index,
Field: f.name,
View: name,
@ -1095,37 +1095,18 @@ func (f *Field) ImportValue(columnIDs []uint64, values []int64) error {
func (f *Field) MarshalJSON() ([]byte, error) {
thing := struct {
Name string
Options fieldOptions
Views []*viewInfo
Options FieldOptions
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)
}
// encodeFields converts a into its internal representation.
func encodeFields(a []*Field) []*internal.Field {
other := make([]*internal.Field, len(a))
for i := range a {
other[i] = encodeField(a[i])
}
return other
}
// encodeField converts f into its internal representation.
func encodeField(f *Field) *internal.Field {
fo := f.options
return &internal.Field{
Name: f.name,
Meta: fo.Encode(),
Views: f.viewNames(),
}
}
type fieldSlice []*Field
func (p fieldSlice) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
@ -1135,8 +1116,8 @@ func (p fieldSlice) Less(i, j int) bool { return p[i].Name() < p[j].Name() }
// FieldInfo represents schema information for a field.
type FieldInfo struct {
Name string `json:"name"`
Options fieldOptions `json:"options"`
Views []*viewInfo `json:"views,omitempty"`
Options FieldOptions `json:"options"`
Views []*ViewInfo `json:"views,omitempty"`
}
type fieldInfoSlice []*FieldInfo
@ -1145,8 +1126,8 @@ func (p fieldInfoSlice) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
func (p fieldInfoSlice) Len() int { return len(p) }
func (p fieldInfoSlice) Less(i, j int) bool { return p[i].Name < p[j].Name }
// fieldOptions represents options to set when initializing a field.
type fieldOptions struct {
// FieldOptions represents options to set when initializing a field.
type FieldOptions struct {
Type string `json:"type,omitempty"`
CacheType string `json:"cacheType,omitempty"`
CacheSize uint32 `json:"cacheSize,omitempty"`
@ -1156,11 +1137,11 @@ type fieldOptions struct {
Keys bool `json:"keys,omitempty"`
}
// applyDefaultOptions returns a new fieldOptions object
// applyDefaultOptions returns a new FieldOptions object
// with default values if o does not contain a valid type.
func applyDefaultOptions(o fieldOptions) fieldOptions {
func applyDefaultOptions(o FieldOptions) FieldOptions {
if o.Type == "" {
return fieldOptions{
return FieldOptions{
Type: DefaultFieldType,
CacheType: DefaultCacheType,
CacheSize: DefaultCacheSize,
@ -1170,11 +1151,11 @@ func applyDefaultOptions(o fieldOptions) fieldOptions {
}
// Encode converts o into its internal representation.
func (o *fieldOptions) Encode() *internal.FieldOptions {
func (o *FieldOptions) Encode() *internal.FieldOptions {
return encodeFieldOptions(o)
}
func encodeFieldOptions(o *fieldOptions) *internal.FieldOptions {
func encodeFieldOptions(o *FieldOptions) *internal.FieldOptions {
if o == nil {
return nil
}
@ -1189,22 +1170,7 @@ func encodeFieldOptions(o *fieldOptions) *internal.FieldOptions {
}
}
func decodeFieldOptions(options *internal.FieldOptions) *fieldOptions {
if options == nil {
return nil
}
return &fieldOptions{
Type: options.Type,
CacheType: options.CacheType,
CacheSize: options.CacheSize,
Min: options.Min,
Max: options.Max,
TimeQuantum: TimeQuantum(options.TimeQuantum),
Keys: options.Keys,
}
}
func (o *fieldOptions) MarshalJSON() ([]byte, error) {
func (o *FieldOptions) MarshalJSON() ([]byte, error) {
switch o.Type {
case FieldTypeSet:
return json.Marshal(struct {

2
fragment.go
View file

@ -1889,7 +1889,7 @@ func (s *fragmentSyncer) syncBlock(id int) error {
}
// Execute query.
queryRequest := &internal.QueryRequest{
queryRequest := &QueryRequest{
Query: buffers[k].String(),
Remote: true,
}

32
gossip/gossip.go
View file

@ -26,10 +26,8 @@ import (
"sync"
"time"
"github.com/gogo/protobuf/proto"
"github.com/hashicorp/memberlist"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/toml"
"github.com/pkg/errors"
)
@ -148,7 +146,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 +191,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
@ -222,7 +220,7 @@ func NewGossipMemberSet(cfg Config, api *pilosa.API, options ...GossipMemberSetO
// NodeMeta implementation of the memberlist.Delegate interface.
func (g *GossipMemberSet) NodeMeta(limit int) []byte {
buf, err := proto.Marshal(pilosa.EncodeNode(g.papi.Node()))
buf, err := g.papi.Serializer.Marshal(g.papi.Node())
if err != nil {
g.Logger.Printf("marshal message error: %s", err)
return []byte{}
@ -248,14 +246,14 @@ func (g *GossipMemberSet) GetBroadcasts(overhead, limit int) [][]byte {
// LocalState implementation of the memberlist.Delegate interface
// sends this Node's state data.
func (g *GossipMemberSet) LocalState(join bool) []byte {
pb := &internal.NodeStatus{
Node: pilosa.EncodeNode(g.papi.Node()),
MaxShards: &internal.MaxShards{Standard: g.papi.MaxShards(context.Background())},
Schema: &internal.Schema{Indexes: pilosa.EncodeIndexes(g.papi.Schema(context.Background()))},
m := &pilosa.NodeStatus{
Node: g.papi.Node(),
MaxShards: g.papi.MaxShards(context.Background()),
Schema: &pilosa.Schema{Indexes: g.papi.Holder().Schema()},
}
// Marshal nodestate data to bytes.
buf, err := pilosa.MarshalMessage(pb)
buf, err := pilosa.MarshalInternalMessage(m, g.papi.Serializer)
if err != nil {
g.Logger.Printf("error marshalling nodestate data, err=%s", err)
return []byte{}
@ -323,16 +321,16 @@ func (g *gossipEventReceiver) listen() {
}
// Get the node from the event.Node meta data.
var n internal.Node
if err := proto.Unmarshal(e.Node.Meta, &n); err != nil {
panic("failed to unmarshal event node meta data")
var n pilosa.Node
if err := g.papi.Serializer.Unmarshal(e.Node.Meta, &n); err != nil {
panic("failed to unmarshal event node meta into node")
}
ne := &internal.NodeEventMessage{
Event: uint32(nodeEventType),
ne := &pilosa.NodeEvent{
Event: nodeEventType,
Node: &n,
}
buf, err := pilosa.MarshalMessage(ne)
buf, err := pilosa.MarshalInternalMessage(ne, g.papi.Serializer)
if err != nil {
panic(err)
}

37
handler.go
View file

@ -75,3 +75,40 @@ func (n nopHandler) Close() error {
}
var NopHandler Handler = nopHandler{}
type ImportValueRequest struct {
Index string
Field string
Shard uint64
ColumnIDs []uint64
ColumnKeys []string
Values []int64
}
type ImportRequest struct {
Index string
Field string
Shard uint64
RowIDs []uint64
ColumnIDs []uint64
RowKeys []string
ColumnKeys []string
Timestamps []int64
}
type ImportResponse struct {
Err string
}
type BlockDataRequest struct {
Index string
Field string
View string
Shard uint64
Block uint64
}
type BlockDataResponse struct {
RowIDs []uint64
ColumnIDs []uint64
}

24
holder.go
View file

@ -27,7 +27,6 @@ import (
"syscall"
"time"
"github.com/pilosa/pilosa/internal"
"github.com/pkg/errors"
uuid "github.com/satori/go.uuid"
)
@ -217,7 +216,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)
@ -230,7 +229,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 +239,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")
}
@ -257,20 +255,6 @@ func (h *Holder) applySchema(schema *internal.Schema) error {
return nil
}
// encodeMaxShards creates and internal representation of max shards.
func (h *Holder) encodeMaxShards() *internal.MaxShards {
return &internal.MaxShards{
Standard: h.maxShards(),
}
}
// encodeSchema creates an internal representation of schema.
func (h *Holder) encodeSchema() *internal.Schema {
return &internal.Schema{
Indexes: EncodeIndexes(h.Indexes()),
}
}
// IndexPath returns the path where a given index is stored.
func (h *Holder) IndexPath(name string) string { return filepath.Join(h.Path, name) }

61
http/client.go
View file

@ -29,9 +29,8 @@ import (
"crypto/tls"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/encoding/proto"
"github.com/pkg/errors"
)
@ -43,6 +42,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 +66,7 @@ func NewInternalClient(host string, remoteClient *http.Client) (*InternalClient,
func NewInternalClientFromURI(defaultURI *pilosa.URI, remoteClient *http.Client) *InternalClient {
return &InternalClient{
defaultURI: defaultURI,
serializer: proto.Serializer{},
HTTPClient: remoteClient,
}
}
@ -217,22 +218,21 @@ func (c *InternalClient) FragmentNodes(ctx context.Context, index string, shard
}
// Query executes query against the index.
func (c *InternalClient) Query(ctx context.Context, index string, queryRequest *internal.QueryRequest) (*internal.QueryResponse, error) {
func (c *InternalClient) Query(ctx context.Context, index string, queryRequest *pilosa.QueryRequest) (*pilosa.QueryResponse, error) {
return c.QueryNode(ctx, c.defaultURI, index, queryRequest)
}
// QueryNode executes query against the index, sending the request to the node specified.
func (c *InternalClient) QueryNode(ctx context.Context, uri *pilosa.URI, index string, queryRequest *internal.QueryRequest) (*internal.QueryResponse, error) {
func (c *InternalClient) QueryNode(ctx context.Context, uri *pilosa.URI, index string, queryRequest *pilosa.QueryRequest) (*pilosa.QueryResponse, error) {
if index == "" {
return nil, pilosa.ErrIndexRequired
} else if queryRequest.Query == "" {
return nil, pilosa.ErrQueryRequired
}
// Encode request object.
buf, err := proto.Marshal(queryRequest)
buf, err := c.serializer.Marshal(queryRequest)
if err != nil {
return nil, errors.Wrap(err, "marshaling")
return nil, errors.Wrap(err, "marshaling queryRequest")
}
// Create HTTP request.
@ -262,11 +262,11 @@ func (c *InternalClient) QueryNode(ctx context.Context, uri *pilosa.URI, index s
return nil, errors.New(string(body))
}
qresp := &internal.QueryResponse{}
if err := proto.Unmarshal(body, qresp); err != nil {
qresp := &pilosa.QueryResponse{}
if err := c.serializer.Unmarshal(body, qresp); err != nil {
return nil, fmt.Errorf("unmarshal response: %s", err)
} else if s := qresp.Err; s != "" {
return nil, errors.New(s)
} else if qresp.Err != nil {
return nil, qresp.Err
}
return qresp, nil
@ -280,7 +280,7 @@ func (c *InternalClient) Import(ctx context.Context, index, field string, shard
return pilosa.ErrFieldRequired
}
buf, err := marshalImportPayload(index, field, shard, bits)
buf, err := c.marshalImportPayload(index, field, shard, bits)
if err != nil {
return fmt.Errorf("Error Creating Payload: %s", err)
}
@ -309,7 +309,7 @@ func (c *InternalClient) ImportK(ctx context.Context, index, field string, colum
return pilosa.ErrFieldRequired
}
buf, err := marshalImportPayloadK(index, field, columns)
buf, err := c.marshalImportPayloadK(index, field, columns)
if err != nil {
return fmt.Errorf("Error Creating Payload: %s", err)
}
@ -343,14 +343,14 @@ func (c *InternalClient) EnsureField(ctx context.Context, indexName string, fiel
}
// marshalImportPayload marshalls the import parameters into a protobuf byte slice.
func marshalImportPayload(index, field string, shard uint64, bits []pilosa.Bit) ([]byte, error) {
func (c *InternalClient) marshalImportPayload(index, field string, shard uint64, bits []pilosa.Bit) ([]byte, error) {
// Separate row and column IDs to reduce allocations.
rowIDs := Bits(bits).RowIDs()
columnIDs := Bits(bits).ColumnIDs()
timestamps := Bits(bits).Timestamps()
// Marshal data to protobuf.
buf, err := proto.Marshal(&internal.ImportRequest{
buf, err := c.serializer.Marshal(&pilosa.ImportRequest{
Index: index,
Field: field,
Shard: shard,
@ -365,14 +365,14 @@ func marshalImportPayload(index, field string, shard uint64, bits []pilosa.Bit)
}
// marshalImportPayloadK marshalls the import parameters into a protobuf byte slice.
func marshalImportPayloadK(index, field string, bits []pilosa.Bit) ([]byte, error) {
func (c *InternalClient) marshalImportPayloadK(index, field string, bits []pilosa.Bit) ([]byte, error) {
// Separate row and column IDs to reduce allocations.
rowKeys := Bits(bits).RowKeys()
columnKeys := Bits(bits).ColumnKeys()
timestamps := Bits(bits).Timestamps()
// Marshal data to protobuf.
buf, err := proto.Marshal(&internal.ImportRequest{
buf, err := c.serializer.Marshal(&pilosa.ImportRequest{
Index: index,
Field: field,
RowKeys: rowKeys,
@ -414,8 +414,8 @@ func (c *InternalClient) importNode(ctx context.Context, node *pilosa.Node, inde
return errors.New(string(body))
}
var isresp internal.ImportResponse
if err := proto.Unmarshal(body, &isresp); err != nil {
var isresp pilosa.ImportResponse
if err := c.serializer.Unmarshal(body, &isresp); err != nil {
return fmt.Errorf("unmarshal import response: %s", err)
} else if s := isresp.Err; s != "" {
return errors.New(s)
@ -432,7 +432,7 @@ func (c *InternalClient) ImportValue(ctx context.Context, index, field string, s
return pilosa.ErrFieldRequired
}
buf, err := marshalImportValuePayload(index, field, shard, vals)
buf, err := c.marshalImportValuePayload(index, field, shard, vals)
if err != nil {
return fmt.Errorf("Error Creating Payload: %s", err)
}
@ -454,13 +454,13 @@ func (c *InternalClient) ImportValue(ctx context.Context, index, field string, s
}
// marshalImportValuePayload marshalls the import parameters into a protobuf byte slice.
func marshalImportValuePayload(index, field string, shard uint64, vals []pilosa.FieldValue) ([]byte, error) {
func (c *InternalClient) marshalImportValuePayload(index, field string, shard uint64, vals []pilosa.FieldValue) ([]byte, error) {
// Separate row and column IDs to reduce allocations.
columnIDs := FieldValues(vals).ColumnIDs()
values := FieldValues(vals).Values()
// Marshal data to protobuf.
buf, err := proto.Marshal(&internal.ImportValueRequest{
buf, err := c.serializer.Marshal(&pilosa.ImportValueRequest{
Index: index,
Field: field,
Shard: shard,
@ -683,7 +683,7 @@ func (c *InternalClient) BlockData(ctx context.Context, uri *pilosa.URI, index,
if uri == nil {
panic("need to pass a URI to BlockData")
}
buf, err := proto.Marshal(&internal.BlockDataRequest{
buf, err := c.serializer.Marshal(&pilosa.BlockDataRequest{
Index: index,
Field: field,
Shard: shard,
@ -719,10 +719,10 @@ func (c *InternalClient) BlockData(ctx context.Context, uri *pilosa.URI, index,
}
// Decode response object.
var rsp internal.BlockDataResponse
var rsp pilosa.BlockDataResponse
if body, err := ioutil.ReadAll(resp.Body); err != nil {
return nil, nil, errors.Wrap(err, "reading")
} else if err := proto.Unmarshal(body, &rsp); err != nil {
} else if err := c.serializer.Unmarshal(body, &rsp); err != nil {
return nil, nil, errors.Wrap(err, "unmarshalling")
}
return rsp.RowIDs, rsp.ColumnIDs, nil
@ -819,12 +819,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 +993,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 +1001,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,
}

9
http/client_test.go
View file

@ -24,7 +24,6 @@ import (
"github.com/davecgh/go-spew/spew"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/http"
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/pql"
"github.com/pilosa/pilosa/server"
"github.com/pilosa/pilosa/test"
@ -131,7 +130,7 @@ func TestClient_MultiNode(t *testing.T) {
client[2] = MustNewClient(c[2].URL(), defaultClient)
topN := 4
queryRequest := &internal.QueryRequest{
queryRequest := &pilosa.QueryRequest{
Query: fmt.Sprintf(`TopN(f, n=%d)`, topN),
Remote: false,
}
@ -147,17 +146,17 @@ func TestClient_MultiNode(t *testing.T) {
}
// Test must return exactly N results.
if len(result.Results[0].Pairs) != topN {
if len(result.Results[0].([]pilosa.Pair)) != topN {
t.Fatalf("unexpected number of TopN results: %s", spew.Sdump(result))
}
p := []*internal.Pair{
p := []pilosa.Pair{
{ID: 100, Count: 12},
{ID: 22, Count: 10},
{ID: 98, Count: 8},
{ID: 99, Count: 7}}
// Valdidate the Top 4 result counts.
if !reflect.DeepEqual(result.Results[0].Pairs, p) {
if !reflect.DeepEqual(result.Results[0].([]pilosa.Pair), p) {
t.Fatalf("Invalid TopN result set: %s", spew.Sdump(result))
}

77
http/handler.go
View file

@ -33,11 +33,9 @@ import (
"strings"
"time"
"github.com/gogo/protobuf/proto"
"github.com/gorilla/handlers"
"github.com/gorilla/mux"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/internal"
"github.com/pkg/errors"
)
@ -676,7 +674,7 @@ type postFieldRequest struct {
Options fieldOptions `json:"options"`
}
// fieldOptions tracks pilosa.fieldOptions. It is made up of pointers to values,
// fieldOptions tracks pilosa.FieldOptions. It is made up of pointers to values,
// and used for input validation.
type fieldOptions struct {
Type string `json:"type,omitempty"`
@ -820,13 +818,12 @@ func (h *Handler) readProtobufQueryRequest(r *http.Request) (*pilosa.QueryReques
return nil, errors.Wrap(err, "reading")
}
// Unmarshal into object.
var req internal.QueryRequest
if err := proto.Unmarshal(body, &req); err != nil {
return nil, errors.Wrap(err, "unmarshalling")
qreq := &pilosa.QueryRequest{}
err = h.API.Serializer.Unmarshal(body, qreq)
if err != nil {
return nil, errors.Wrap(err, "unmarshalling query request")
}
return decodeQueryRequest(&req), nil
return qreq, nil
}
// readURLQueryRequest parses query parameters from URL parameters from r.
@ -865,7 +862,7 @@ func (h *Handler) writeQueryResponse(w http.ResponseWriter, r *http.Request, res
// writeProtobufQueryResponse writes the response from the executor to w as protobuf.
func (h *Handler) writeProtobufQueryResponse(w http.ResponseWriter, resp *pilosa.QueryResponse) error {
if buf, err := proto.Marshal(encodeQueryResponse(resp)); err != nil {
if buf, err := h.API.Serializer.Marshal(resp); err != nil {
return errors.Wrap(err, "marshalling")
} else if _, err := w.Write(buf); err != nil {
return errors.Wrap(err, "writing")
@ -917,8 +914,8 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) {
if field.Type() == pilosa.FieldTypeInt {
// Field type: Int
// Marshal into request object.
var req internal.ImportValueRequest
if err := proto.Unmarshal(body, &req); err != nil {
req := &pilosa.ImportValueRequest{}
if err := h.API.Serializer.Unmarshal(body, req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
@ -935,8 +932,8 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) {
} else {
// Field type: Set, Time
// Marshal into request object.
var req internal.ImportRequest
if err := proto.Unmarshal(body, &req); err != nil {
req := &pilosa.ImportRequest{}
if err := h.API.Serializer.Unmarshal(body, req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
@ -953,7 +950,7 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) {
}
// Marshal response object.
buf, e := proto.Marshal(&internal.ImportResponse{Err: ""})
buf, e := h.API.Serializer.Marshal(&pilosa.ImportResponse{Err: ""})
if e != nil {
http.Error(w, fmt.Sprintf("marshal import response"), http.StatusInternalServerError)
return
@ -1108,56 +1105,6 @@ const (
QueryResultTypeBool
)
func decodeQueryRequest(pb *internal.QueryRequest) *pilosa.QueryRequest {
req := &pilosa.QueryRequest{
Query: pb.Query,
Shards: pb.Shards,
ColumnAttrs: pb.ColumnAttrs,
Remote: pb.Remote,
ExcludeRowAttrs: pb.ExcludeRowAttrs,
ExcludeColumns: pb.ExcludeColumns,
}
return req
}
func encodeQueryResponse(resp *pilosa.QueryResponse) *internal.QueryResponse {
pb := &internal.QueryResponse{
Results: make([]*internal.QueryResult, len(resp.Results)),
ColumnAttrSets: pilosa.EncodeColumnAttrSets(resp.ColumnAttrSets),
}
for i := range resp.Results {
pb.Results[i] = &internal.QueryResult{}
switch result := resp.Results[i].(type) {
case *pilosa.Row:
pb.Results[i].Type = QueryResultTypeRow
pb.Results[i].Row = pilosa.EncodeRow(result)
case []pilosa.Pair:
pb.Results[i].Type = QueryResultTypePairs
pb.Results[i].Pairs = pilosa.EncodePairs(result)
case pilosa.ValCount:
pb.Results[i].Type = QueryResultTypeValCount
pb.Results[i].ValCount = pilosa.EncodeValCount(result)
case uint64:
pb.Results[i].Type = QueryResultTypeUint64
pb.Results[i].N = result
case bool:
pb.Results[i].Type = QueryResultTypeBool
pb.Results[i].Changed = result
case nil:
pb.Results[i].Type = QueryResultTypeNil
}
}
if resp.Err != nil {
pb.Err = resp.Err.Error()
}
return pb
}
// parseUint64Slice returns a slice of uint64s from a comma-delimited string.
func parseUint64Slice(s string) ([]uint64, error) {
var a []uint64

32
index.go
View file

@ -297,7 +297,7 @@ func (i *Index) CreateField(name string, opts ...FieldOption) (*Field, error) {
}
// Apply functional options.
fo := fieldOptions{}
fo := FieldOptions{}
for _, opt := range opts {
err := opt(&fo)
if err != nil {
@ -319,7 +319,7 @@ func (i *Index) CreateFieldIfNotExists(name string, opts FieldOption) (*Field, e
}
// Apply functional option.
fo := fieldOptions{}
fo := FieldOptions{}
err := opts(&fo)
if err != nil {
return nil, errors.Wrap(err, "applying option")
@ -328,7 +328,7 @@ func (i *Index) CreateFieldIfNotExists(name string, opts FieldOption) (*Field, e
return i.createField(name, fo)
}
func (i *Index) createFieldIfNotExists(name string, opt fieldOptions) (*Field, error) {
func (i *Index) createFieldIfNotExists(name string, opt FieldOptions) (*Field, error) {
i.mu.Lock()
defer i.mu.Unlock()
@ -340,7 +340,7 @@ func (i *Index) createFieldIfNotExists(name string, opt fieldOptions) (*Field, e
return i.createField(name, opt)
}
func (i *Index) createField(name string, opt fieldOptions) (*Field, error) {
func (i *Index) createField(name string, opt FieldOptions) (*Field, error) {
if name == "" {
return nil, errors.New("field name required")
} else if opt.CacheType != "" && !isValidCacheType(opt.CacheType) {
@ -432,35 +432,11 @@ func (p indexInfoSlice) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
func (p indexInfoSlice) Len() int { return len(p) }
func (p indexInfoSlice) Less(i, j int) bool { return p[i].Name < p[j].Name }
// EncodeIndexes converts a into its internal representation.
func EncodeIndexes(a []*Index) []*internal.Index {
other := make([]*internal.Index, len(a))
for i := range a {
other[i] = encodeIndex(a[i])
}
return other
}
// encodeIndex converts d into its internal representation.
func encodeIndex(d *Index) *internal.Index {
return &internal.Index{
Name: d.name,
Fields: encodeFields(d.Fields()),
}
}
// IndexOptions represents options to set when initializing an index.
type IndexOptions struct {
Keys bool `json:"keys"`
}
// Encode converts i into its internal representation.
func (i *IndexOptions) Encode() *internal.IndexMeta {
return &internal.IndexMeta{
Keys: i.Keys,
}
}
// hasTime returns true if a contains a non-nil time.
func hasTime(a []*time.Time) bool {
for _, t := range a {

19
pilosa.go
View file

@ -17,8 +17,6 @@ package pilosa
import (
"errors"
"regexp"
"github.com/pilosa/pilosa/internal"
)
// System errors.
@ -122,23 +120,6 @@ type ColumnAttrSet struct {
Attrs map[string]interface{} `json:"attrs,omitempty"`
}
// EncodeColumnAttrSets converts a into its internal representation.
func EncodeColumnAttrSets(a []*ColumnAttrSet) []*internal.ColumnAttrSet {
other := make([]*internal.ColumnAttrSet, len(a))
for i := range a {
other[i] = EncodeColumnAttrSet(a[i])
}
return other
}
// EncodeColumnAttrSet converts set into its internal representation.
func EncodeColumnAttrSet(set *ColumnAttrSet) *internal.ColumnAttrSet {
return &internal.ColumnAttrSet{
ID: set.ID,
Attrs: encodeAttrs(set.Attrs),
}
}
// TimeFormat is the go-style time format used to parse string dates.
const TimeFormat = "2006-01-02T15:04"

27
row.go
View file

@ -18,7 +18,6 @@ import (
"encoding/json"
"sort"
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/roaring"
)
@ -252,32 +251,6 @@ func (r *Row) Columns() []uint64 {
return a
}
// EncodeRow converts r into its internal representation.
func EncodeRow(r *Row) *internal.Row {
if r == nil {
return nil
}
return &internal.Row{
Columns: r.Columns(),
Attrs: encodeAttrs(r.Attrs),
}
}
// DecodeRow converts r from its internal representation.
func DecodeRow(pr *internal.Row) *Row {
if pr == nil {
return nil
}
r := NewRow()
r.Attrs = decodeAttrs(pr.Attrs)
for _, v := range pr.Columns {
r.SetBit(v)
}
return r
}
// RowSegment holds a subset of a row.
// This could point to a mmapped roaring bitmap or an in-memory bitmap. The
// width of the segment will always match the shard width.

88
server.go
View file

@ -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"
@ -56,6 +54,7 @@ type Server struct {
executor *executor
hosts []string
clusterDisabled bool
serializer Serializer
// External
systemInfo SystemInfo
@ -202,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
@ -431,40 +437,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 +479,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,44 +488,49 @@ func (s *Server) receiveMessage(pb proto.Message) error {
if err != nil {
return err
}
case *internal.ClusterStatus:
case *ClusterStatus:
err := s.cluster.mergeClusterStatus(obj)
if err != nil {
return err
}
case *internal.ResizeInstruction:
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
}
// 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
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)
@ -529,7 +540,7 @@ func (s *Server) SendSync(pb proto.Message) error {
}
eg.Go(func() error {
return s.defaultClient.SendMessage(context.Background(), &node.URI, pb)
return s.defaultClient.SendMessage(context.Background(), &node.URI, msg)
})
}
@ -537,14 +548,19 @@ func (s *Server) SendSync(pb proto.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 {
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
@ -554,7 +570,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 +580,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 +600,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
@ -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())

80
server/handler_test.go
View file

@ -27,10 +27,8 @@ import (
gohttp "net/http"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/http"
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/server"
"github.com/pilosa/pilosa/test"
)
@ -144,7 +142,7 @@ func TestHandler_Endpoints(t *testing.T) {
t.Run("Shards args protobuf", func(t *testing.T) {
// Generate request body.
reqBody, err := proto.Marshal(&internal.QueryRequest{
reqBody, err := cmd.API.Serializer.Marshal(&pilosa.QueryRequest{
Query: "Count(Row(f0=30))",
Shards: []uint64{0, 1},
})
@ -196,13 +194,11 @@ func TestHandler_Endpoints(t *testing.T) {
t.Fatalf("unexpected status code: %d", w.Code)
}
var resp internal.QueryResponse
if err := proto.Unmarshal(w.Body.Bytes(), &resp); err != nil {
var resp pilosa.QueryResponse
if err := cmd.API.Serializer.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
} else if rt := resp.Results[0].Type; rt != http.QueryResultTypeUint64 {
t.Fatalf("unexpected response type: %d", resp.Results[0].Type)
} else if n := resp.Results[0].N; n != 3 {
t.Fatalf("unexpected n: %d", n)
} else if rt, ok := resp.Results[0].(uint64); !ok || rt != 3 {
t.Fatalf("unexpected response type: %#v", resp.Results[0])
}
})
@ -244,27 +240,25 @@ func TestHandler_Endpoints(t *testing.T) {
t.Fatalf("unexpected status code: %d", w.Code)
}
var resp internal.QueryResponse
if err := proto.Unmarshal(w.Body.Bytes(), &resp); err != nil {
var resp pilosa.QueryResponse
if err := cmd.API.Serializer.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
} else if rt := resp.Results[0].Type; rt != http.QueryResultTypeRow {
t.Fatalf("unexpected response type: %d", resp.Results[0].Type)
} else if columns := resp.Results[0].Row.Columns; !reflect.DeepEqual(columns, []uint64{pilosa.ShardWidth + 1, pilosa.ShardWidth + 2, (3 * pilosa.ShardWidth) + 4}) {
} else if columns := resp.Results[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(columns, []uint64{pilosa.ShardWidth + 1, pilosa.ShardWidth + 2, (3 * pilosa.ShardWidth) + 4}) {
t.Fatalf("unexpected columns: %+v", columns)
} else if attrs := resp.Results[0].Row.Attrs; len(attrs) != 3 {
} else if attrs := resp.Results[0].(*pilosa.Row).Attrs; len(attrs) != 3 {
t.Fatalf("unexpected attr length: %d", len(attrs))
} else if k, v := attrs[0].Key, attrs[0].StringValue; k != "a" || v != "b" {
t.Fatalf("unexpected attr[0]: %s=%v", k, v)
} else if k, v := attrs[1].Key, attrs[1].IntValue; k != "c" || v != int64(1) {
t.Fatalf("unexpected attr[1]: %s=%v", k, v)
} else if k, v := attrs[2].Key, attrs[2].BoolValue; k != "d" || !v {
t.Fatalf("unexpected attr[2]: %s=%v", k, v)
} else if attrs["a"] != "b" {
t.Fatalf("unexpected attr[a]: %v", attrs["a"])
} else if attrs["c"] != int64(1) {
t.Fatalf("unexpected attr[c]: %v", attrs["c"])
} else if !attrs["d"].(bool) {
t.Fatalf("unexpected attr[d]: %v", attrs["d"])
}
})
t.Run("Row columnattrs protobuf", func(t *testing.T) {
// Encode request body.
buf, err := proto.Marshal(&internal.QueryRequest{
buf, err := cmd.API.Serializer.Marshal(&pilosa.QueryRequest{
Query: "Row(f0=30)",
ColumnAttrs: true,
})
@ -281,22 +275,22 @@ func TestHandler_Endpoints(t *testing.T) {
t.Fatalf("unexpected status code: %d", w.Code)
}
var resp internal.QueryResponse
if err := proto.Unmarshal(w.Body.Bytes(), &resp); err != nil {
var resp pilosa.QueryResponse
if err := cmd.API.Serializer.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
}
if columns := resp.Results[0].Row.Columns; !reflect.DeepEqual(columns, []uint64{pilosa.ShardWidth + 1, pilosa.ShardWidth + 2, (3 * pilosa.ShardWidth) + 4}) {
if columns := resp.Results[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(columns, []uint64{pilosa.ShardWidth + 1, pilosa.ShardWidth + 2, (3 * pilosa.ShardWidth) + 4}) {
t.Fatalf("unexpected columns: %+v", columns)
} else if rt := resp.Results[0].Type; rt != http.QueryResultTypeRow {
t.Fatalf("unexpected response type: %d", resp.Results[0].Type)
} else if attrs := resp.Results[0].Row.Attrs; len(attrs) != 3 {
} else if _, ok := resp.Results[0].(*pilosa.Row); !ok {
t.Fatalf("unexpected response type: %#v", resp.Results[0])
} else if attrs := resp.Results[0].(*pilosa.Row).Attrs; len(attrs) != 3 {
t.Fatalf("unexpected attr length: %d", len(attrs))
} else if k, v := attrs[0].Key, attrs[0].StringValue; k != "a" || v != "b" {
t.Fatalf("unexpected attr[0]: %s=%v", k, v)
} else if k, v := attrs[1].Key, attrs[1].IntValue; k != "c" || v != int64(1) {
t.Fatalf("unexpected attr[1]: %s=%v", k, v)
} else if k, v := attrs[2].Key, attrs[2].BoolValue; k != "d" || !v {
t.Fatalf("unexpected attr[2]: %s=%v", k, v)
} else if attrs["a"] != "b" {
t.Fatalf("unexpected attr[a]: %v", attrs["a"])
} else if attrs["c"] != int64(1) {
t.Fatalf("unexpected attr[c]: %v", attrs["c"])
} else if !attrs["d"].(bool) {
t.Fatalf("unexpected attr[d]: %v", attrs["d"])
}
if a := resp.ColumnAttrSets; len(a) != 2 {
@ -305,8 +299,8 @@ func TestHandler_Endpoints(t *testing.T) {
t.Fatalf("unexpected id: %d", a[0].ID)
} else if len(a[0].Attrs) != 1 {
t.Fatalf("unexpected column attr length: %d", len(a))
} else if k, v := a[0].Attrs[0].Key, a[0].Attrs[0].StringValue; k != "x" || v != "y" {
t.Fatalf("unexpected attr[0]: %s=%v", k, v)
} else if a[0].Attrs["x"] != "y" {
t.Fatalf("unexpected attr[x]: %v", a[0].Attrs["x"])
}
})
@ -329,12 +323,10 @@ func TestHandler_Endpoints(t *testing.T) {
t.Fatalf("unexpected status code: %d", w.Code)
}
var resp internal.QueryResponse
if err := proto.Unmarshal(w.Body.Bytes(), &resp); err != nil {
var resp pilosa.QueryResponse
if err := cmd.API.Serializer.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
} else if rt := resp.Results[0].Type; rt != http.QueryResultTypePairs {
t.Fatalf("unexpected response type: %d", resp.Results[0].Type)
} else if a := resp.Results[0].GetPairs(); len(a) != 2 {
} else if a := resp.Results[0].([]pilosa.Pair); len(a) != 2 {
t.Fatalf("unexpected pair length: %d", len(a))
}
})
@ -358,10 +350,10 @@ func TestHandler_Endpoints(t *testing.T) {
t.Fatalf("unexpected status code: %d", w.Code)
}
var resp internal.QueryResponse
if err := proto.Unmarshal(w.Body.Bytes(), &resp); err != nil {
var resp pilosa.QueryResponse
if err := cmd.API.Serializer.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
} else if s := resp.Err; s != `executing: field not found` {
} else if s := resp.Err.Error(); s != `executing: field not found` {
t.Fatalf("unexpected error: %s", s)
}
})

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

91
uri.go
View file

@ -21,7 +21,6 @@ import (
"strconv"
"strings"
"github.com/pilosa/pilosa/internal"
"github.com/pkg/errors"
)
@ -43,17 +42,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 +82,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 +93,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 +108,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 +128,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)
}
// Path returns URI with path
@ -191,41 +190,13 @@ 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
}
// Encode converts o into its internal representation.
func (u URI) Encode() *internal.URI {
return encodeURI(u)
}
func encodeURI(u URI) *internal.URI {
return &internal.URI{
Scheme: u.scheme,
Host: u.host,
Port: uint32(u.port),
}
}
func DecodeURI(i *internal.URI) URI {
return decodeURI(i)
}
func decodeURI(i *internal.URI) URI {
if i == nil {
return URI{}
}
return URI{
scheme: i.Scheme,
host: i.Host,
port: uint16(i.Port),
}
}
// MarshalJSON marshals URI into a JSON-encoded byte slice.
func (u *URI) MarshalJSON() ([]byte, error) {
var output struct {
@ -233,9 +204,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)
}
@ -249,8 +220,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

@ -83,8 +83,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)
}
}
@ -95,8 +95,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)
}
}
@ -104,8 +104,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)
}
}
@ -137,14 +137,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)
}
}

30
utils_internal_test.go
View file

@ -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.
@ -162,7 +161,7 @@ func (t *ClusterCluster) addNode() error {
// Send NodeJoin event to coordinator.
if id > 0 {
coord := t.Clusters[0]
ev := &nodeEvent{
ev := &NodeEvent{
Event: NodeJoin,
Node: c.Node,
}
@ -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,8 +364,7 @@ func (t *ClusterCluster) FollowResizeInstruction(instr *internal.ResizeInstructi
}
for _, src := range instr.Sources {
srcNode := DecodeNode(src.Node)
srcCluster := t.clusterByID(srcNode.ID)
srcCluster := t.clusterByID(src.Node.ID)
srcFragment := srcCluster.holder.fragment(src.Index, src.Field, src.View, src.Shard)
destFragment := destCluster.holder.fragment(src.Index, src.Field, src.View, src.Shard)
@ -405,6 +403,6 @@ func (t *ClusterCluster) FollowResizeInstruction(instr *internal.ResizeInstructi
complete.Error = err.Error()
}
node := DecodeNode(instr.Coordinator)
node := instr.Coordinator
return t.SendTo(node, complete)
}

9
view.go
View file

@ -22,7 +22,6 @@ import (
"strings"
"sync"
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/pql"
"github.com/pkg/errors"
)
@ -232,7 +231,7 @@ func (v *view) createFragmentIfNotExists(shard uint64) (*fragment, error) {
// Send the create shard message to all nodes.
err := v.broadcaster.SendSync(
&internal.CreateShardMessage{
&CreateShardMessage{
Index: v.index,
Shard: shard,
})
@ -422,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) }