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

change all references to use subpackages: topology, net

Browse source
This commit is contained in:
Travis 2021-01-06 16:09:24 -06:00
parent f8e6115c0e
commit 4515a24e48
No known key found for this signature in database
GPG key ID: 37080CC2042BA34E
19 changed files with 330 additions and 418 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
api.go
View file

@ -34,6 +34,7 @@ import (
"github.com/pilosa/pilosa/v2/pql"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/stats"
"github.com/pilosa/pilosa/v2/topology"
"github.com/pilosa/pilosa/v2/tracing"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
@ -682,7 +683,7 @@ func (api *API) ExportCSV(ctx context.Context, indexName string, fieldName strin
}
// ShardNodes returns the node and all replicas which should contain a shard's data.
func (api *API) ShardNodes(ctx context.Context, indexName string, shard uint64) ([]*Node, error) {
func (api *API) ShardNodes(ctx context.Context, indexName string, shard uint64) ([]*topology.Node, error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.ShardNodes")
defer span.Finish()
@ -796,7 +797,7 @@ func (api *API) TranslateData(ctx context.Context, indexName string, partition i
// Hosts returns a list of the hosts in the cluster including their ID,
// URL, and which is the coordinator.
func (api *API) Hosts(ctx context.Context) []*Node {
func (api *API) Hosts(ctx context.Context) []*topology.Node {
span, _ := tracing.StartSpanFromContext(ctx, "API.Hosts")
defer span.Finish()
return api.cluster.Nodes()
@ -809,7 +810,7 @@ func (api *API) HostStates(ctx context.Context) map[string]string {
}
// Node gets the ID, URI and coordinator status for this particular node.
func (api *API) Node() *Node {
func (api *API) Node() *topology.Node {
node := api.server.node()
return &node
}
@ -1700,7 +1701,7 @@ func (api *API) indexField(indexName string, fieldName string, shard uint64) (*I
}
// SetCoordinator makes a new Node the cluster coordinator.
func (api *API) SetCoordinator(ctx context.Context, id string) (oldNode, newNode *Node, err error) {
func (api *API) SetCoordinator(ctx context.Context, id string) (oldNode, newNode *topology.Node, err error) {
span, _ := tracing.StartSpanFromContext(ctx, "API.SetCoordinator")
defer span.Finish()
@ -1733,7 +1734,7 @@ func (api *API) SetCoordinator(ctx context.Context, id string) (oldNode, newNode
// RemoveNode puts the cluster into the "RESIZING" state and begins the job of
// removing the given node.
func (api *API) RemoveNode(id string) (*Node, error) {
func (api *API) RemoveNode(id string) (*topology.Node, error) {
if err := api.validate(apiRemoveNode); err != nil {
return nil, errors.Wrap(err, "validating api method")
}
@ -1743,7 +1744,7 @@ func (api *API) RemoveNode(id string) (*Node, error) {
if !api.cluster.topologyContainsNode(id) {
return nil, errors.Wrap(ErrNodeIDNotExists, "finding node to remove")
}
removeNode = &Node{
removeNode = &topology.Node{
ID: id,
}
}

5
broadcast.go
View file

@ -17,6 +17,7 @@ package pilosa
import (
"fmt"
"github.com/pilosa/pilosa/v2/topology"
"github.com/pkg/errors"
)
@ -30,7 +31,7 @@ type Serializer interface {
type broadcaster interface {
SendSync(Message) error
SendAsync(Message) error
SendTo(*Node, Message) error
SendTo(*topology.Node, Message) error
}
// Message is the interface implemented by all core pilosa types which can be serialized to messages.
@ -49,7 +50,7 @@ func (nopBroadcaster) SendSync(Message) error { return nil }
func (nopBroadcaster) SendAsync(Message) error { return nil }
// SendTo is a no-op implementation of Broadcaster SendTo method.
func (nopBroadcaster) SendTo(*Node, Message) error { return nil }
func (nopBroadcaster) SendTo(*topology.Node, Message) error { return nil }
// Broadcast message types.
const (

87
client.go
View file

@ -18,6 +18,9 @@ import (
"context"
"io"
"time"
pnet "github.com/pilosa/pilosa/v2/net"
"github.com/pilosa/pilosa/v2/topology"
)
// Bit represents the intersection of a row and a column. It can be specified by
@ -51,10 +54,10 @@ type InternalClient interface {
MaxShardByIndex(ctx context.Context) (map[string]uint64, error)
Schema(ctx context.Context) ([]*IndexInfo, error)
PostSchema(ctx context.Context, uri *URI, s *Schema, remote bool) error
PostSchema(ctx context.Context, uri *pnet.URI, s *Schema, remote bool) error
CreateIndex(ctx context.Context, index string, opt IndexOptions) error
FragmentNodes(ctx context.Context, index string, shard uint64) ([]*Node, error)
Nodes(ctx context.Context) ([]*Node, error)
FragmentNodes(ctx context.Context, index string, shard uint64) ([]*topology.Node, error)
Nodes(ctx context.Context) ([]*topology.Node, error)
Query(ctx context.Context, index string, queryRequest *QueryRequest) (*QueryResponse, error)
Import(ctx context.Context, index, field string, shard uint64, bits []Bit, opts ...ImportOption) error
ImportK(ctx context.Context, index, field string, bits []Bit, opts ...ImportOption) error
@ -67,69 +70,69 @@ type InternalClient interface {
ExportCSV(ctx context.Context, index, field string, shard uint64, w io.Writer) error
CreateField(ctx context.Context, index, field string) error
CreateFieldWithOptions(ctx context.Context, index, field string, opt FieldOptions) error
FragmentBlocks(ctx context.Context, uri *URI, index, field, view string, shard uint64) ([]FragmentBlock, error)
BlockData(ctx context.Context, uri *URI, index, field, view 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, msg []byte) error
RetrieveShardFromURI(ctx context.Context, index, field, view string, shard uint64, uri URI) (io.ReadCloser, error)
RetrieveTranslatePartitionFromURI(ctx context.Context, index string, partition int, uri URI) (io.ReadCloser, error)
ImportRoaring(ctx context.Context, uri *URI, index, field string, shard uint64, remote bool, req *ImportRoaringRequest) error
ImportColumnAttrs(ctx context.Context, uri *URI, index string, req *ImportColumnAttrsRequest) error
FragmentBlocks(ctx context.Context, uri *pnet.URI, index, field, view string, shard uint64) ([]FragmentBlock, error)
BlockData(ctx context.Context, uri *pnet.URI, index, field, view string, shard uint64, block int) ([]uint64, []uint64, error)
ColumnAttrDiff(ctx context.Context, uri *pnet.URI, index string, blks []AttrBlock) (map[uint64]map[string]interface{}, error)
RowAttrDiff(ctx context.Context, uri *pnet.URI, index, field string, blks []AttrBlock) (map[uint64]map[string]interface{}, error)
SendMessage(ctx context.Context, uri *pnet.URI, msg []byte) error
RetrieveShardFromURI(ctx context.Context, index, field, view string, shard uint64, uri pnet.URI) (io.ReadCloser, error)
RetrieveTranslatePartitionFromURI(ctx context.Context, index string, partition int, uri pnet.URI) (io.ReadCloser, error)
ImportRoaring(ctx context.Context, uri *pnet.URI, index, field string, shard uint64, remote bool, req *ImportRoaringRequest) error
ImportColumnAttrs(ctx context.Context, uri *pnet.URI, index string, req *ImportColumnAttrsRequest) error
StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool) (*Transaction, error)
FinishTransaction(ctx context.Context, id string) (*Transaction, error)
Transactions(ctx context.Context) (map[string]*Transaction, error)
GetTransaction(ctx context.Context, id string) (*Transaction, error)
GetNodeUsage(ctx context.Context, uri *URI) (map[string]NodeUsage, error)
GetPastQueries(ctx context.Context, uri *URI) ([]PastQueryStatus, error)
GetNodeUsage(ctx context.Context, uri *pnet.URI) (map[string]NodeUsage, error)
GetPastQueries(ctx context.Context, uri *pnet.URI) ([]PastQueryStatus, error)
}
//===============
// InternalQueryClient is the internal interface for querying a node.
type InternalQueryClient interface {
QueryNode(ctx context.Context, uri *URI, index string, queryRequest *QueryRequest) (*QueryResponse, error)
QueryNode(ctx context.Context, uri *pnet.URI, index string, queryRequest *QueryRequest) (*QueryResponse, error)
// Trasnlate keys on the particular node. The parameter writable informs TranslateStore if we can generate a new ID if any of keys does not exist.
TranslateKeysNode(ctx context.Context, uri *URI, index, field string, keys []string, writable bool) ([]uint64, error)
TranslateIDsNode(ctx context.Context, uri *URI, index, field string, id []uint64) ([]string, error)
TranslateKeysNode(ctx context.Context, uri *pnet.URI, index, field string, keys []string, writable bool) ([]uint64, error)
TranslateIDsNode(ctx context.Context, uri *pnet.URI, index, field string, id []uint64) ([]string, error)
FindIndexKeysNode(ctx context.Context, uri *URI, index string, keys ...string) (map[string]uint64, error)
FindFieldKeysNode(ctx context.Context, uri *URI, index string, field string, keys ...string) (map[string]uint64, error)
FindIndexKeysNode(ctx context.Context, uri *pnet.URI, index string, keys ...string) (map[string]uint64, error)
FindFieldKeysNode(ctx context.Context, uri *pnet.URI, index string, field string, keys ...string) (map[string]uint64, error)
CreateIndexKeysNode(ctx context.Context, uri *URI, index string, keys ...string) (map[string]uint64, error)
CreateFieldKeysNode(ctx context.Context, uri *URI, index string, field string, keys ...string) (map[string]uint64, error)
CreateIndexKeysNode(ctx context.Context, uri *pnet.URI, index string, keys ...string) (map[string]uint64, error)
CreateFieldKeysNode(ctx context.Context, uri *pnet.URI, index string, field string, keys ...string) (map[string]uint64, error)
}
type nopInternalQueryClient struct{}
func (n nopInternalQueryClient) QueryNode(ctx context.Context, uri *URI, index string, queryRequest *QueryRequest) (*QueryResponse, error) {
func (n nopInternalQueryClient) QueryNode(ctx context.Context, uri *pnet.URI, index string, queryRequest *QueryRequest) (*QueryResponse, error) {
return nil, nil
}
func (n nopInternalQueryClient) TranslateKeysNode(ctx context.Context, uri *URI, index, field string, keys []string, writable bool) ([]uint64, error) {
func (n nopInternalQueryClient) TranslateKeysNode(ctx context.Context, uri *pnet.URI, index, field string, keys []string, writable bool) ([]uint64, error) {
return nil, nil
}
func (n nopInternalQueryClient) TranslateIDsNode(ctx context.Context, uri *URI, index, field string, ids []uint64) ([]string, error) {
func (n nopInternalQueryClient) TranslateIDsNode(ctx context.Context, uri *pnet.URI, index, field string, ids []uint64) ([]string, error) {
return nil, nil
}
func (n nopInternalQueryClient) FindIndexKeysNode(ctx context.Context, uri *URI, index string, keys ...string) (map[string]uint64, error) {
func (n nopInternalQueryClient) FindIndexKeysNode(ctx context.Context, uri *pnet.URI, index string, keys ...string) (map[string]uint64, error) {
return nil, nil
}
func (n nopInternalQueryClient) FindFieldKeysNode(ctx context.Context, uri *URI, index string, field string, keys ...string) (map[string]uint64, error) {
func (n nopInternalQueryClient) FindFieldKeysNode(ctx context.Context, uri *pnet.URI, index string, field string, keys ...string) (map[string]uint64, error) {
return nil, nil
}
func (n nopInternalQueryClient) CreateIndexKeysNode(ctx context.Context, uri *URI, index string, keys ...string) (map[string]uint64, error) {
func (n nopInternalQueryClient) CreateIndexKeysNode(ctx context.Context, uri *pnet.URI, index string, keys ...string) (map[string]uint64, error) {
return nil, nil
}
func (n nopInternalQueryClient) CreateFieldKeysNode(ctx context.Context, uri *URI, index string, field string, keys ...string) (map[string]uint64, error) {
func (n nopInternalQueryClient) CreateFieldKeysNode(ctx context.Context, uri *pnet.URI, index string, field string, keys ...string) (map[string]uint64, error) {
return nil, nil
}
@ -153,17 +156,17 @@ func (n nopInternalClient) MaxShardByIndex(context.Context) (map[string]uint64,
return nil, nil
}
func (n nopInternalClient) Schema(ctx context.Context) ([]*IndexInfo, error) { return nil, nil }
func (n nopInternalClient) PostSchema(ctx context.Context, uri *URI, s *Schema, remote bool) error {
func (n nopInternalClient) PostSchema(ctx context.Context, uri *pnet.URI, s *Schema, remote bool) error {
return nil
}
func (n nopInternalClient) CreateIndex(ctx context.Context, index string, opt IndexOptions) error {
return nil
}
func (n nopInternalClient) FragmentNodes(ctx context.Context, index string, shard uint64) ([]*Node, error) {
func (n nopInternalClient) FragmentNodes(ctx context.Context, index string, shard uint64) ([]*topology.Node, error) {
return nil, nil
}
func (n nopInternalClient) Nodes(ctx context.Context) ([]*Node, error) {
func (n nopInternalClient) Nodes(ctx context.Context) ([]*topology.Node, error) {
return nil, nil
}
func (n nopInternalClient) Query(ctx context.Context, index string, queryRequest *QueryRequest) (*QueryResponse, error) {
@ -179,11 +182,11 @@ func (n nopInternalClient) ImportValue2(ctx context.Context, req *ImportValueReq
return nil
}
func (n nopInternalClient) ImportRoaring(ctx context.Context, uri *URI, index, field string, shard uint64, remote bool, req *ImportRoaringRequest) error {
func (n nopInternalClient) ImportRoaring(ctx context.Context, uri *pnet.URI, index, field string, shard uint64, remote bool, req *ImportRoaringRequest) error {
return nil
}
func (n nopInternalClient) ImportColumnAttrs(ctx context.Context, uri *URI, index string, req *ImportColumnAttrsRequest) error {
func (n nopInternalClient) ImportColumnAttrs(ctx context.Context, uri *pnet.URI, index string, req *ImportColumnAttrsRequest) error {
return nil
}
@ -209,25 +212,25 @@ func (n nopInternalClient) CreateField(ctx context.Context, index, field string)
func (n nopInternalClient) CreateFieldWithOptions(ctx context.Context, index, field string, opt FieldOptions) error {
return nil
}
func (n nopInternalClient) FragmentBlocks(ctx context.Context, uri *URI, index, field, view string, shard uint64) ([]FragmentBlock, error) {
func (n nopInternalClient) FragmentBlocks(ctx context.Context, uri *pnet.URI, index, field, view string, shard uint64) ([]FragmentBlock, error) {
return nil, nil
}
func (n nopInternalClient) BlockData(ctx context.Context, uri *URI, index, field, view string, shard uint64, block int) ([]uint64, []uint64, error) {
func (n nopInternalClient) BlockData(ctx context.Context, uri *pnet.URI, index, field, view string, shard uint64, block int) ([]uint64, []uint64, error) {
return nil, nil, nil
}
func (n nopInternalClient) ColumnAttrDiff(ctx context.Context, uri *URI, index string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) {
func (n nopInternalClient) ColumnAttrDiff(ctx context.Context, uri *pnet.URI, index string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) {
return nil, nil
}
func (n nopInternalClient) RowAttrDiff(ctx context.Context, uri *URI, index, field string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) {
func (n nopInternalClient) RowAttrDiff(ctx context.Context, uri *pnet.URI, index, field string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) {
return nil, nil
}
func (n nopInternalClient) SendMessage(ctx context.Context, uri *URI, msg []byte) error {
func (n nopInternalClient) SendMessage(ctx context.Context, uri *pnet.URI, msg []byte) error {
return nil
}
func (n nopInternalClient) RetrieveShardFromURI(ctx context.Context, index, field, view string, shard uint64, uri URI) (io.ReadCloser, error) {
func (n nopInternalClient) RetrieveShardFromURI(ctx context.Context, index, field, view string, shard uint64, uri pnet.URI) (io.ReadCloser, error) {
return nil, nil
}
func (n nopInternalClient) RetrieveTranslatePartitionFromURI(ctx context.Context, index string, partition int, uri URI) (io.ReadCloser, error) {
func (n nopInternalClient) RetrieveTranslatePartitionFromURI(ctx context.Context, index string, partition int, uri pnet.URI) (io.ReadCloser, error) {
return nil, nil
}
@ -244,10 +247,10 @@ func (n nopInternalClient) GetTransaction(ctx context.Context, id string) (*Tran
return nil, nil
}
func (n nopInternalClient) GetNodeUsage(ctx context.Context, uri *URI) (map[string]NodeUsage, error) {
func (n nopInternalClient) GetNodeUsage(ctx context.Context, uri *pnet.URI) (map[string]NodeUsage, error) {
return nil, nil
}
func (n nopInternalClient) GetPastQueries(ctx context.Context, uri *URI) ([]PastQueryStatus, error) {
func (n nopInternalClient) GetPastQueries(ctx context.Context, uri *pnet.URI) ([]PastQueryStatus, error) {
return nil, nil
}

239
cluster.go
View file

@ -32,7 +32,9 @@ import (
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/v2/internal"
"github.com/pilosa/pilosa/v2/logger"
pnet "github.com/pilosa/pilosa/v2/net"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/topology"
"github.com/pilosa/pilosa/v2/tracing"
"github.com/pkg/errors"
uuid "github.com/satori/go.uuid"
@ -66,138 +68,17 @@ const (
defaultConfirmDownSleep = 1 * time.Second
)
// Node represents a node in the cluster.
type Node struct {
ID string `json:"id"`
URI URI `json:"uri"`
GRPCURI URI `json:"grpc-uri"`
IsCoordinator bool `json:"isCoordinator"`
State string `json:"state"`
}
func (n *Node) Clone() *Node {
if n == nil {
return nil
}
other := *n
return &other
}
func (n Node) String() string {
return fmt.Sprintf("Node:%s:%s:%s", n.URI, n.State, n.ID)
}
// Nodes represents a list of nodes.
type Nodes []*Node
// Contains returns true if a node exists in the list.
func (a Nodes) Contains(n *Node) bool {
for i := range a {
if a[i] == n {
return true
}
}
return false
}
// ContainsID returns true if host matches one of the node's id.
func (a Nodes) ContainsID(id string) bool {
for _, n := range a {
if n.ID == id {
return true
}
}
return false
}
// NodeByID returns the node for an ID. If the ID is not found,
// it returns nil.
func (a Nodes) NodeByID(id string) *Node {
for _, n := range a {
if n.ID == id {
return n
}
}
return nil
}
// Filter returns a new list of nodes with node removed.
func (a Nodes) Filter(n *Node) []*Node {
other := make([]*Node, 0, len(a))
for i := range a {
if a[i] != n {
other = append(other, a[i])
}
}
return other
}
// FilterID returns a new list of nodes with ID removed.
func (a Nodes) FilterID(id string) []*Node {
other := make([]*Node, 0, len(a))
for _, node := range a {
if node.ID != id {
other = append(other, node)
}
}
return other
}
// FilterURI returns a new list of nodes with URI removed.
func (a Nodes) FilterURI(uri URI) []*Node {
other := make([]*Node, 0, len(a))
for _, node := range a {
if node.URI != uri {
other = append(other, node)
}
}
return other
}
// IDs returns a list of all node IDs.
func (a Nodes) IDs() []string {
ids := make([]string, len(a))
for i, n := range a {
ids[i] = n.ID
}
return ids
}
// URIs returns a list of all uris.
func (a Nodes) URIs() []URI {
uris := make([]URI, len(a))
for i, n := range a {
uris[i] = n.URI
}
return uris
}
// Clone returns a shallow copy of nodes.
func (a Nodes) Clone() []*Node {
other := make([]*Node, len(a))
copy(other, a)
return other
}
// byID implements sort.Interface for []Node based on
// the ID field.
type byID []*Node
func (h byID) Len() int { return len(h) }
func (h byID) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
func (h byID) Less(i, j int) bool { return h[i].ID < h[j].ID }
// nodeAction represents a node that is joining or leaving the cluster.
type nodeAction struct {
node *Node
node *topology.Node
action string
}
// cluster represents a collection of nodes.
type cluster struct { // nolint: maligned
id string
Node *Node
nodes []*Node
Node *topology.Node
nodes []*topology.Node
// Hashing algorithm used to assign partitions to nodes.
Hasher Hasher
@ -303,14 +184,14 @@ func (c *cluster) abortAntiEntropy() {
}
}
func (c *cluster) coordinatorNode() *Node {
func (c *cluster) coordinatorNode() *topology.Node {
c.mu.RLock()
defer c.mu.RUnlock()
return c.unprotectedCoordinatorNode()
}
// unprotectedCoordinatorNode returns the coordinator node.
func (c *cluster) unprotectedCoordinatorNode() *Node {
func (c *cluster) unprotectedCoordinatorNode() *topology.Node {
return c.unprotectedNodeByID(c.Coordinator)
}
@ -329,7 +210,7 @@ func (c *cluster) unprotectedIsCoordinator() bool {
// Coordinator. In response to this, the current node
// will consider itself coordinator and update the other
// nodes with its version of Cluster.Status.
func (c *cluster) setCoordinator(n *Node) error {
func (c *cluster) setCoordinator(n *topology.Node) error {
c.mu.Lock()
defer c.mu.Unlock()
// Verify that the new Coordinator value matches
@ -376,13 +257,13 @@ func (c *cluster) unprotectedSendSync(m Message) error {
// changing the corresponding node's IsCoordinator value
// to true, and sets all other nodes to false. Returns true if the value
// changed.
func (c *cluster) updateCoordinator(n *Node) bool { // nolint: unparam
func (c *cluster) updateCoordinator(n *topology.Node) bool { // nolint: unparam
c.mu.Lock()
defer c.mu.Unlock()
return c.unprotectedUpdateCoordinator(n)
}
func (c *cluster) unprotectedUpdateCoordinator(n *Node) bool {
func (c *cluster) unprotectedUpdateCoordinator(n *topology.Node) bool {
var changed bool
if c.Coordinator != n.ID {
c.Coordinator = n.ID
@ -400,7 +281,7 @@ func (c *cluster) unprotectedUpdateCoordinator(n *Node) bool {
// addNode adds a node to the Cluster and updates and saves the
// new topology. unprotected.
func (c *cluster) addNode(node *Node) error {
func (c *cluster) addNode(node *topology.Node) error {
// If the node being added is the coordinator, set it for this node.
if node.IsCoordinator {
c.Coordinator = node.ID
@ -444,7 +325,7 @@ func (c *cluster) removeNode(nodeID string) error {
// nodeIDs returns the list of IDs in the cluster.
func (c *cluster) nodeIDs() []string {
return Nodes(c.nodes).IDs()
return topology.Nodes(c.nodes).IDs()
}
func (c *cluster) unprotectedSetID(id string) {
@ -629,14 +510,14 @@ func (c *cluster) unprotectedStatus() *ClusterStatus {
}
}
func (c *cluster) nodeByID(id string) *Node {
func (c *cluster) nodeByID(id string) *topology.Node {
c.mu.RLock()
defer c.mu.RUnlock()
return c.unprotectedNodeByID(id)
}
// unprotectedNodeByID returns a node reference by ID.
func (c *cluster) unprotectedNodeByID(id string) *Node {
func (c *cluster) unprotectedNodeByID(id string) *topology.Node {
for _, n := range c.nodes {
if n.ID == id {
return n
@ -668,7 +549,7 @@ func (c *cluster) nodePositionByID(nodeID string) int {
// addNodeBasicSorted adds a node to the cluster, sorted by id. Returns a
// pointer to the node and true if the node was added. unprotected.
func (c *cluster) addNodeBasicSorted(node *Node) bool {
func (c *cluster) addNodeBasicSorted(node *topology.Node) bool {
n := c.unprotectedNodeByID(node.ID)
if n != nil {
if n.State != node.State || n.IsCoordinator != node.IsCoordinator || n.URI != node.URI {
@ -684,17 +565,17 @@ func (c *cluster) addNodeBasicSorted(node *Node) bool {
c.nodes = append(c.nodes, node)
// All hosts must be merged in the same order on all nodes in the cluster.
sort.Sort(byID(c.nodes))
sort.Sort(topology.ByID(c.nodes))
return true
}
// Nodes returns a copy of the slice of nodes in the cluster. Safe for
// concurrent use, result may be modified.
func (c *cluster) Nodes() []*Node {
func (c *cluster) Nodes() []*topology.Node {
c.mu.RLock()
defer c.mu.RUnlock()
ret := make([]*Node, len(c.nodes))
ret := make([]*topology.Node, len(c.nodes))
copy(ret, c.nodes)
return ret
}
@ -851,7 +732,7 @@ func (c *cluster) fragSources(to *cluster, idx *Index) (map[string][]*ResizeSour
srcCluster := c
if action == resizeJobActionAdd && c.ReplicaN > 1 {
srcCluster = newCluster()
srcCluster.nodes = Nodes(c.nodes).Clone()
srcCluster.nodes = topology.Nodes(c.nodes).Clone()
srcCluster.Hasher = c.Hasher
srcCluster.partitionN = c.partitionN
srcCluster.ReplicaN = 1
@ -1041,26 +922,26 @@ func (c *cluster) idPartition(index string, id uint64) int {
}
// ShardNodes returns a list of nodes that own a fragment. Safe for concurrent use.
func (c *cluster) ShardNodes(index string, shard uint64) []*Node {
func (c *cluster) ShardNodes(index string, shard uint64) []*topology.Node {
c.mu.RLock()
defer c.mu.RUnlock()
return c.shardNodes(index, shard)
}
// shardNodes returns a list of nodes that own a shard. unprotected
func (c *cluster) shardNodes(index string, shard uint64) []*Node {
func (c *cluster) shardNodes(index string, shard uint64) []*topology.Node {
return c.partitionNodes(c.shardToShardPartition(index, shard))
}
// KeyNodes returns a list of nodes that own a fragment. Safe for concurrent use.
func (c *cluster) KeyNodes(index, key string) []*Node {
func (c *cluster) KeyNodes(index, key string) []*topology.Node {
c.mu.RLock()
defer c.mu.RUnlock()
return c.keyNodes(index, key)
}
// keyNodes returns a list of nodes that own a key. unprotected
func (c *cluster) keyNodes(index, key string) []*Node {
func (c *cluster) keyNodes(index, key string) []*topology.Node {
return c.partitionNodes(c.Topology.KeyPartition(index, key))
}
@ -1068,11 +949,11 @@ func (c *cluster) keyNodes(index, key string) []*Node {
func (c *cluster) ownsShard(nodeID string, index string, shard uint64) bool {
c.mu.RLock()
defer c.mu.RUnlock()
return Nodes(c.shardNodes(index, shard)).ContainsID(nodeID)
return topology.Nodes(c.shardNodes(index, shard)).ContainsID(nodeID)
}
// partitionNodes returns a list of nodes that own a partition. unprotected.
func (c *cluster) partitionNodes(partitionID int) []*Node {
func (c *cluster) partitionNodes(partitionID int) []*topology.Node {
// Default replica count to between one and the number of nodes.
// The replica count can be zero if there are no nodes.
@ -1114,11 +995,11 @@ func (c *cluster) partitionNodes(partitionID int) []*Node {
return nil
}
// Collect nodes around the ring.
nodes := make([]*Node, 0, replicaN)
nodes := make([]*topology.Node, 0, replicaN)
for i := 0; i < replicaN; i++ {
if useTopology {
maybeNodeID := c.Topology.nodeIDs[(nodeIndex+i)%nodeN]
if node := Nodes(c.nodes).NodeByID(maybeNodeID); node != nil {
if node := topology.Nodes(c.nodes).NodeByID(maybeNodeID); node != nil {
nodes = append(nodes, node)
}
} else {
@ -1129,14 +1010,14 @@ func (c *cluster) partitionNodes(partitionID int) []*Node {
return nodes
}
func (c *cluster) primaryPartitionNode(partition int) *Node {
func (c *cluster) primaryPartitionNode(partition int) *topology.Node {
c.mu.RLock()
defer c.mu.RUnlock()
return c.unprotectedPrimaryPartitionNode(partition)
}
// unprotectedPrimaryPartition returns tprimary node of partition.
func (c *cluster) unprotectedPrimaryPartitionNode(partition int) *Node {
func (c *cluster) unprotectedPrimaryPartitionNode(partition int) *topology.Node {
if nodes := c.partitionNodes(partition); len(nodes) > 0 {
return nodes[0]
}
@ -1199,7 +1080,7 @@ func (topo *Topology) GetReplicasForPrimary(primary int) (replicaNodeIDs, nonRep
}
// containsShards is like OwnsShards, but it includes replicas.
func (c *cluster) containsShards(index string, availableShards *roaring.Bitmap, node *Node) []uint64 {
func (c *cluster) containsShards(index string, availableShards *roaring.Bitmap, node *topology.Node) []uint64 {
var shards []uint64
_ = availableShards.ForEach(func(i uint64) error {
p := c.shardToShardPartition(index, i)
@ -1417,7 +1298,7 @@ func (c *cluster) unprotectedSetStateAndBroadcast(state string) error {
return c.unprotectedSendSync(status) // TODO fix c.Status
}
func (c *cluster) sendTo(node *Node, m Message) error {
func (c *cluster) sendTo(node *topology.Node, m Message) error {
if err := c.broadcaster.SendTo(node, m); err != nil {
return errors.Wrap(err, "sending")
}
@ -1512,7 +1393,7 @@ func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (*
// toCluster is a clone of Cluster with the new node added/removed for comparison.
toCluster := newCluster()
toCluster.nodes = Nodes(c.nodes).Clone()
toCluster.nodes = topology.Nodes(c.nodes).Clone()
toCluster.Hasher = c.Hasher
toCluster.partitionN = c.partitionN
toCluster.ReplicaN = c.ReplicaN
@ -1830,7 +1711,7 @@ type resizeJob struct {
}
// newResizeJob returns a new instance of resizeJob.
func newResizeJob(existingNodes []*Node, node *Node, action string) *resizeJob {
func newResizeJob(existingNodes []*topology.Node, node *topology.Node, action string) *resizeJob {
// Build a map of uris to track their resize status.
// The value for a node will be set to true after that node
@ -1918,7 +1799,7 @@ func (j *resizeJob) distributeResizeInstructions() error {
for _, instr := range j.Instructions {
// Because the node may not be in the cluster yet, create
// a dummy node object to use in the SendTo() method.
node := &Node{
node := &topology.Node{
ID: instr.Node.ID,
URI: instr.Node.URI,
GRPCURI: instr.Node.GRPCURI,
@ -2147,7 +2028,7 @@ func (c *cluster) considerTopology() error {
// band aid to protect against false nodeLeave events from memberlist
// the test is the lightest weight endpoint of the node in question /version
// TODO provide more robust solution to false nodeLeave events
func (c *cluster) confirmNodeDown(uri URI) bool {
func (c *cluster) confirmNodeDown(uri pnet.URI) bool {
u := url.URL{
Scheme: uri.Scheme,
Host: uri.HostPort(),
@ -2219,7 +2100,7 @@ func (c *cluster) ReceiveEvent(e *NodeEvent) (err error) {
}
// nodeJoin should only be called by the coordinator.
func (c *cluster) nodeJoin(node *Node) error {
func (c *cluster) nodeJoin(node *topology.Node) error {
c.abortAntiEntropy()
// Technically there is a race condition here which could
// allow the anti-entropy process to re-start (and acquire
@ -2343,7 +2224,7 @@ func (c *cluster) nodeLeave(nodeID string) error {
// See if resize job can be generated
if _, err := c.unprotectedGenerateResizeJobByAction(
nodeAction{
node: &Node{ID: nodeID},
node: &topology.Node{ID: nodeID},
action: resizeJobActionRemove},
); err != nil {
return errors.Wrap(err, "generating job")
@ -2364,7 +2245,7 @@ func (c *cluster) nodeLeave(nodeID string) error {
if err := c.unprotectedSetStateAndBroadcast(ClusterStateResizing); err != nil {
return errors.Wrap(err, "broadcasting state")
}
c.joiningLeavingNodes <- nodeAction{node: &Node{ID: nodeID}, action: resizeJobActionRemove}
c.joiningLeavingNodes <- nodeAction{node: &topology.Node{ID: nodeID}, action: resizeJobActionRemove}
return nil
}
@ -2433,7 +2314,7 @@ func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error {
if node.ID == c.Node.ID {
continue
}
if Nodes(officialNodes).ContainsID(node.ID) {
if topology.Nodes(officialNodes).ContainsID(node.ID) {
continue
}
nodeIDsToRemove = append(nodeIDsToRemove, node.ID)
@ -2455,7 +2336,7 @@ func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error {
// unprotectedPreviousNode returns the node listed before the current node in c.Nodes.
// If there is only one node in the cluster, returns nil.
// If the current node is the first node in the list, returns the last node.
func (c *cluster) unprotectedPreviousNode() *Node {
func (c *cluster) unprotectedPreviousNode() *topology.Node {
if len(c.nodes) <= 1 {
return nil
}
@ -2472,13 +2353,13 @@ func (c *cluster) unprotectedPreviousNode() *Node {
// PrimaryReplicaNode returns the node listed before the current node in c.Nodes.
// This is different than "previous node" as the first node always returns nil.
func (c *cluster) PrimaryReplicaNode() *Node {
func (c *cluster) PrimaryReplicaNode() *topology.Node {
c.mu.RLock()
defer c.mu.RUnlock()
return c.unprotectedPrimaryReplicaNode()
}
func (c *cluster) unprotectedPrimaryReplicaNode() *Node {
func (c *cluster) unprotectedPrimaryReplicaNode() *topology.Node {
pos := c.nodePositionByID(c.Node.ID)
if pos <= 0 {
return nil
@ -2492,11 +2373,11 @@ func (c *cluster) setStatic(hosts []string) error {
c.Static = true
c.Coordinator = c.Node.ID
for _, address := range hosts {
uri, err := NewURIFromAddress(address)
uri, err := pnet.NewURIFromAddress(address)
if err != nil {
return errors.Wrap(err, "getting URI")
}
c.nodes = append(c.nodes, &Node{URI: *uri})
c.nodes = append(c.nodes, &topology.Node{URI: *uri})
}
return nil
}
@ -2822,7 +2703,7 @@ func (c *cluster) findIndexKeys(ctx context.Context, indexName string, keys ...s
// TODO: use local replicas to short-circuit network traffic
// Group keys by node.
keysByNode := make(map[*Node][]string)
keysByNode := make(map[*topology.Node][]string)
for partitionID, keys := range keysByPartition {
// Find the primary node for this partition.
primary := c.primaryPartitionNode(partitionID)
@ -2929,7 +2810,7 @@ func (c *cluster) createIndexKeys(ctx context.Context, indexName string, keys ..
// Group keys by node.
// Delete remote keys from the by-partition map so that it can be used for local translation.
keysByNode := make(map[*Node][]string)
keysByNode := make(map[*topology.Node][]string)
for partitionID, keys := range keysByPartition {
// Find the primary node for this partition.
primary := c.primaryPartitionNode(partitionID)
@ -3088,7 +2969,7 @@ func (c *cluster) translateIndexIDSet(ctx context.Context, indexName string, idS
type ClusterStatus struct {
ClusterID string
State string
Nodes []*Node
Nodes []*topology.Node
Schema *Schema
}
@ -3096,8 +2977,8 @@ type ClusterStatus struct {
// during a cluster resize operation.
type ResizeInstruction struct {
JobID int64
Node *Node
Coordinator *Node
Node *topology.Node
Coordinator *topology.Node
Sources []*ResizeSource
TranslationSources []*TranslationResizeSource
NodeStatus *NodeStatus
@ -3107,17 +2988,17 @@ type ResizeInstruction struct {
// ResizeSource is the source of data for a node acting on a
// ResizeInstruction.
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"`
Node *topology.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"`
}
// TranslationResizeSource is the source of translation data for
// a node acting on a ResizeInstruction.
type TranslationResizeSource struct {
Node *Node
Node *topology.Node
Index string
PartitionID int
}
@ -3125,7 +3006,7 @@ type TranslationResizeSource struct {
// translateResizeNode holds the node/partition pairs used
// to create a TranslationResizeSource for each index.
type translationResizeNode struct {
node *Node
node *topology.Node
partitionID int
}
@ -3219,18 +3100,18 @@ type DeleteViewMessage struct {
// that the resize instructions performed on a single node have completed.
type ResizeInstructionComplete struct {
JobID int64
Node *Node
Node *topology.Node
Error string
}
// SetCoordinatorMessage is an internal message instructing nodes to honor a new coordinator.
type SetCoordinatorMessage struct {
New *Node
New *topology.Node
}
// UpdateCoordinatorMessage is an internal message for reassigning the coordinator.
type UpdateCoordinatorMessage struct {
New *Node
New *topology.Node
}
// NodeStateMessage is an internal message for broadcasting a node's state.
@ -3241,7 +3122,7 @@ type NodeStateMessage struct {
// NodeStatus is an internal message representing the contents of a node.
type NodeStatus struct {
Node *Node
Node *topology.Node
Indexes []*IndexStatus
Schema *Schema
}

124
cluster_internal_test.go
View file

@ -33,24 +33,26 @@ import (
"github.com/davecgh/go-spew/spew"
"github.com/gorilla/mux"
"github.com/pilosa/pilosa/v2/logger"
pnet "github.com/pilosa/pilosa/v2/net"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/testhook"
"github.com/pilosa/pilosa/v2/topology"
"github.com/pkg/errors"
)
// Ensure that fragCombos creates the correct fragment mapping.
func TestFragCombos(t *testing.T) {
uri0, err := NewURIFromAddress("host0")
uri0, err := pnet.NewURIFromAddress("host0")
if err != nil {
t.Fatal(err)
}
uri1, err := NewURIFromAddress("host1")
uri1, err := pnet.NewURIFromAddress("host1")
if err != nil {
t.Fatal(err)
}
node0 := &Node{ID: "node0", URI: *uri0}
node1 := &Node{ID: "node1", URI: *uri1}
node0 := &topology.Node{ID: "node0", URI: *uri0}
node1 := &topology.Node{ID: "node1", URI: *uri1}
c := newCluster()
c.addNodeBasicSorted(node0)
@ -110,27 +112,27 @@ func newIndexWithTempPath(tb testing.TB, name string) *Index {
// Ensure that fragSources creates the correct fragment mapping.
func TestFragSources(t *testing.T) {
uri0, err := NewURIFromAddress("host0")
uri0, err := pnet.NewURIFromAddress("host0")
if err != nil {
t.Fatal(err)
}
uri1, err := NewURIFromAddress("host1")
uri1, err := pnet.NewURIFromAddress("host1")
if err != nil {
t.Fatal(err)
}
uri2, err := NewURIFromAddress("host2")
uri2, err := pnet.NewURIFromAddress("host2")
if err != nil {
t.Fatal(err)
}
uri3, err := NewURIFromAddress("host3")
uri3, err := pnet.NewURIFromAddress("host3")
if err != nil {
t.Fatal(err)
}
node0 := &Node{ID: "node0", URI: *uri0}
node1 := &Node{ID: "node1", URI: *uri1}
node2 := &Node{ID: "node2", URI: *uri2}
node3 := &Node{ID: "node3", URI: *uri3}
node0 := &topology.Node{ID: "node0", URI: *uri0}
node1 := &topology.Node{ID: "node1", URI: *uri1}
node2 := &topology.Node{ID: "node2", URI: *uri2}
node3 := &topology.Node{ID: "node3", URI: *uri3}
c1 := newCluster()
c1.ReplicaN = 1
@ -224,8 +226,8 @@ func TestFragSources(t *testing.T) {
"node0": {},
"node1": {},
"node2": {
{&Node{ID: "node0", URI: URI{"http", "host0", 10101}, IsCoordinator: false}, "i", "f", "standard", uint64(0)},
{&Node{ID: "node1", URI: URI{"http", "host1", 10101}, IsCoordinator: false}, "i", "f", "standard", uint64(2)},
{&topology.Node{ID: "node0", URI: pnet.URI{Scheme: "http", Host: "host0", Port: 10101}, IsCoordinator: false}, "i", "f", "standard", uint64(0)},
{&topology.Node{ID: "node1", URI: pnet.URI{Scheme: "http", Host: "host1", Port: 10101}, IsCoordinator: false}, "i", "f", "standard", uint64(2)},
},
},
err: "",
@ -236,11 +238,11 @@ func TestFragSources(t *testing.T) {
idx: idx,
expected: map[string][]*ResizeSource{
"node0": {
{&Node{ID: "node1", URI: URI{"http", "host1", 10101}, IsCoordinator: false}, "i", "f", "standard", uint64(1)},
{&topology.Node{ID: "node1", URI: pnet.URI{Scheme: "http", Host: "host1", Port: 10101}, IsCoordinator: false}, "i", "f", "standard", uint64(1)},
},
"node1": {
{&Node{ID: "node0", URI: URI{"http", "host0", 10101}, IsCoordinator: false}, "i", "f", "standard", uint64(0)},
{&Node{ID: "node0", URI: URI{"http", "host0", 10101}, IsCoordinator: false}, "i", "f", "standard", uint64(2)},
{&topology.Node{ID: "node0", URI: pnet.URI{Scheme: "http", Host: "host0", Port: 10101}, IsCoordinator: false}, "i", "f", "standard", uint64(0)},
{&topology.Node{ID: "node0", URI: pnet.URI{Scheme: "http", Host: "host0", Port: 10101}, IsCoordinator: false}, "i", "f", "standard", uint64(2)},
},
},
err: "",
@ -251,11 +253,11 @@ func TestFragSources(t *testing.T) {
idx: idx,
expected: map[string][]*ResizeSource{
"node0": {
{&Node{ID: "node2", URI: URI{"http", "host2", 10101}, IsCoordinator: false}, "i", "f", "standard", uint64(0)},
{&Node{ID: "node2", URI: URI{"http", "host2", 10101}, IsCoordinator: false}, "i", "f", "standard", uint64(2)},
{&topology.Node{ID: "node2", URI: pnet.URI{Scheme: "http", Host: "host2", Port: 10101}, IsCoordinator: false}, "i", "f", "standard", uint64(0)},
{&topology.Node{ID: "node2", URI: pnet.URI{Scheme: "http", Host: "host2", Port: 10101}, IsCoordinator: false}, "i", "f", "standard", uint64(2)},
},
"node1": {
{&Node{ID: "node0", URI: URI{"http", "host0", 10101}, IsCoordinator: false}, "i", "f", "standard", uint64(3)},
{&topology.Node{ID: "node0", URI: pnet.URI{Scheme: "http", Host: "host0", Port: 10101}, IsCoordinator: false}, "i", "f", "standard", uint64(3)},
},
"node2": {},
},
@ -304,37 +306,37 @@ func TestFragSources(t *testing.T) {
// Ensure that fragSources creates the correct fragment mapping.
func TestResizeJob(t *testing.T) {
uri0, err := NewURIFromAddress("host0")
uri0, err := pnet.NewURIFromAddress("host0")
if err != nil {
t.Fatal(err)
}
uri1, err := NewURIFromAddress("host1")
uri1, err := pnet.NewURIFromAddress("host1")
if err != nil {
t.Fatal(err)
}
uri2, err := NewURIFromAddress("host2")
uri2, err := pnet.NewURIFromAddress("host2")
if err != nil {
t.Fatal(err)
}
node0 := &Node{ID: "node0", URI: *uri0}
node1 := &Node{ID: "node1", URI: *uri1}
node2 := &Node{ID: "node2", URI: *uri2}
node0 := &topology.Node{ID: "node0", URI: *uri0}
node1 := &topology.Node{ID: "node1", URI: *uri1}
node2 := &topology.Node{ID: "node2", URI: *uri2}
tests := []struct {
existingNodes []*Node
node *Node
existingNodes []*topology.Node
node *topology.Node
action string
expectedIDs map[string]bool
}{
{
existingNodes: []*Node{node0, node1},
existingNodes: []*topology.Node{node0, node1},
node: node2,
action: resizeJobActionAdd,
expectedIDs: map[string]bool{node0.ID: false, node1.ID: false, node2.ID: false},
},
{
existingNodes: []*Node{node0, node1, node2},
existingNodes: []*topology.Node{node0, node1, node2},
node: node2,
action: resizeJobActionRemove,
expectedIDs: map[string]bool{node0.ID: false, node1.ID: false},
@ -355,7 +357,7 @@ func TestResizeJob(t *testing.T) {
// Ensure the cluster can fairly distribute partitions across the nodes.
func TestCluster_Owners(t *testing.T) {
c := cluster{
nodes: []*Node{
nodes: []*topology.Node{
{URI: NewTestURIFromHostPort("serverA", 1000)},
{URI: NewTestURIFromHostPort("serverB", 1000)},
{URI: NewTestURIFromHostPort("serverC", 1000)},
@ -365,12 +367,12 @@ func TestCluster_Owners(t *testing.T) {
}
// Verify nodes are distributed.
if a := c.partitionNodes(0); !reflect.DeepEqual(a, []*Node{c.nodes[0], c.nodes[1]}) {
if a := c.partitionNodes(0); !reflect.DeepEqual(a, []*topology.Node{c.nodes[0], c.nodes[1]}) {
t.Fatalf("unexpected owners: %s", spew.Sdump(a))
}
// Verify nodes go around the ring.
if a := c.partitionNodes(2); !reflect.DeepEqual(a, []*Node{c.nodes[2], c.nodes[0]}) {
if a := c.partitionNodes(2); !reflect.DeepEqual(a, []*topology.Node{c.nodes[2], c.nodes[0]}) {
t.Fatalf("unexpected owners: %s", spew.Sdump(a))
}
}
@ -436,15 +438,15 @@ func TestCluster_Nodes(t *testing.T) {
uri2 := NewTestURIFromHostPort("node2", 0)
uri3 := NewTestURIFromHostPort("node3", 0)
node0 := &Node{ID: "node0", URI: uri0}
node1 := &Node{ID: "node1", URI: uri1}
node2 := &Node{ID: "node2", URI: uri2}
node3 := &Node{ID: "node3", URI: uri3}
node0 := &topology.Node{ID: "node0", URI: uri0}
node1 := &topology.Node{ID: "node1", URI: uri1}
node2 := &topology.Node{ID: "node2", URI: uri2}
node3 := &topology.Node{ID: "node3", URI: uri3}
nodes := []*Node{node0, node1, node2}
nodes := []*topology.Node{node0, node1, node2}
t.Run("NodeIDs", func(t *testing.T) {
actual := Nodes(nodes).IDs()
actual := topology.Nodes(nodes).IDs()
expected := []string{node0.ID, node1.ID, node2.ID}
if !reflect.DeepEqual(actual, expected) {
t.Errorf("expected: %v, but got: %v", expected, actual)
@ -452,24 +454,24 @@ func TestCluster_Nodes(t *testing.T) {
})
t.Run("Filter", func(t *testing.T) {
actual := Nodes(Nodes(nodes).Filter(nodes[1])).URIs()
expected := []URI{uri0, uri2}
actual := topology.Nodes(topology.Nodes(nodes).Filter(nodes[1])).URIs()
expected := []pnet.URI{uri0, uri2}
if !reflect.DeepEqual(actual, expected) {
t.Errorf("expected: %v, but got: %v", expected, actual)
}
})
t.Run("FilterURI", func(t *testing.T) {
actual := Nodes(Nodes(nodes).FilterURI(uri1)).URIs()
expected := []URI{uri0, uri2}
actual := topology.Nodes(topology.Nodes(nodes).FilterURI(uri1)).URIs()
expected := []pnet.URI{uri0, uri2}
if !reflect.DeepEqual(actual, expected) {
t.Errorf("expected: %v, but got: %v", expected, actual)
}
})
t.Run("Contains", func(t *testing.T) {
actualTrue := Nodes(nodes).Contains(node1)
actualFalse := Nodes(nodes).Contains(node3)
actualTrue := topology.Nodes(nodes).Contains(node1)
actualFalse := topology.Nodes(nodes).Contains(node3)
if !reflect.DeepEqual(actualTrue, true) {
t.Errorf("expected: %v, but got: %v", true, actualTrue)
}
@ -479,9 +481,9 @@ func TestCluster_Nodes(t *testing.T) {
})
t.Run("Clone", func(t *testing.T) {
clone := Nodes(nodes).Clone()
actual := Nodes(clone).URIs()
expected := []URI{uri0, uri1, uri2}
clone := topology.Nodes(nodes).Clone()
actual := topology.Nodes(clone).URIs()
expected := []pnet.URI{uri0, uri1, uri2}
if !reflect.DeepEqual(actual, expected) {
t.Errorf("expected: %v, but got: %v", expected, actual)
}
@ -489,9 +491,9 @@ func TestCluster_Nodes(t *testing.T) {
}
func TestCluster_PreviousNode(t *testing.T) {
node0 := &Node{ID: "node0"}
node1 := &Node{ID: "node1"}
node2 := &Node{ID: "node2"}
node0 := &topology.Node{ID: "node0"}
node1 := &topology.Node{ID: "node1"}
node2 := &topology.Node{ID: "node2"}
t.Run("OneNode", func(t *testing.T) {
c := newCluster()
@ -547,8 +549,8 @@ func TestCluster_Coordinator(t *testing.T) {
uri1 := NewTestURIFromHostPort("node1", 0)
uri2 := NewTestURIFromHostPort("node2", 0)
node1 := &Node{ID: "node1", URI: uri1}
node2 := &Node{ID: "node2", URI: uri2}
node1 := &topology.Node{ID: "node1", URI: uri1}
node2 := &topology.Node{ID: "node2", URI: uri2}
c1 := *newCluster()
c1.Node = node1
@ -574,10 +576,10 @@ func TestCluster_Topology(t *testing.T) {
uri2 := NewTestURIFromHostPort("host2", 0)
invalid := NewTestURIFromHostPort("invalid", 0)
node0 := &Node{ID: "node0", URI: uri0}
node1 := &Node{ID: "node1", URI: uri1}
node2 := &Node{ID: "node2", URI: uri2}
nodeinvalid := &Node{ID: "nodeinvalid", URI: invalid}
node0 := &topology.Node{ID: "node0", URI: uri0}
node1 := &topology.Node{ID: "node1", URI: uri1}
node2 := &topology.Node{ID: "node2", URI: uri2}
nodeinvalid := &topology.Node{ID: "nodeinvalid", URI: invalid}
t.Run("AddNode", func(t *testing.T) {
err := c1.addNode(node1)
@ -984,7 +986,7 @@ func TestCluster_confirmNodeDownUp(t *testing.T) {
if err != nil {
t.Error("bad test setup")
}
uri := URI{}
uri := pnet.URI{}
host, port, _ := net.SplitHostPort(u.Host)
uri.Scheme = u.Scheme
uri.Host = host
@ -1018,7 +1020,7 @@ func TestCluster_confirmNodeDownTimeout(t *testing.T) {
if err != nil {
t.Error("bad test setup")
}
uri := URI{}
uri := pnet.URI{}
host, port, _ := net.SplitHostPort(u.Host)
uri.Scheme = u.Scheme
uri.Host = host
@ -1040,7 +1042,7 @@ func TestCluster_confirmNodeDownDown(t *testing.T) {
if testing.Short() {
t.Skip()
}
uri := URI{}
uri := pnet.URI{}
uri.Scheme = "http"
uri.Host = "DoesntMatter"
uri.Port = 6666
@ -1063,7 +1065,7 @@ func TestCluster_GetNonPrimaryReplicas(t *testing.T) {
nNodes := 4
for i := 0; i < nNodes; i++ {
nodeID := fmt.Sprintf("node%d", i)
c.nodes = append(c.nodes, &Node{
c.nodes = append(c.nodes, &topology.Node{
ID: nodeID,
URI: NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0)),
})

14
cmd/badloader/badloader.go
View file

@ -19,13 +19,17 @@ import (
"compress/gzip"
"context"
"time"
//"fmt"
"fmt"
"github.com/pilosa/pilosa/v2"
"github.com/pilosa/pilosa/v2/http"
"io"
"io/ioutil"
gohttp "net/http"
"github.com/pilosa/pilosa/v2"
"github.com/pilosa/pilosa/v2/http"
pnet "github.com/pilosa/pilosa/v2/net"
//"log"
"os"
//"path/filepath"
@ -140,15 +144,15 @@ func main() {
vv("total elapsed '%v'", time.Since(t0))
}
var globURI *pilosa.URI
var globURI *pnet.URI
func init() {
var err error
globURI, err = pilosa.NewURIFromHostPort("127.0.0.1", 10101)
globURI, err = pnet.NewURIFromHostPort("127.0.0.1", 10101)
panicOn(err)
}
// get correct node to go to.
func GetImportRoaringURI(index string, shard uint64) *pilosa.URI {
func GetImportRoaringURI(index string, shard uint64) *pnet.URI {
return globURI
}

7
cmd/slurp/slurp.go
View file

@ -32,6 +32,7 @@ import (
"github.com/pilosa/pilosa/v2"
"github.com/pilosa/pilosa/v2/http"
pnet "github.com/pilosa/pilosa/v2/net"
)
// slurp: slurp is a load-tester for importing bulk data.
@ -191,7 +192,7 @@ func main() {
flag.StringVar(&tarSrcPath, "src", "q2.tar.gz", "data to import")
flag.Parse()
uri, err := pilosa.NewURIFromAddress(host)
uri, err := pnet.NewURIFromAddress(host)
panicOn(err)
globURI = uri
@ -253,9 +254,9 @@ func stopProfile(host, outfile string) {
}
var globURI *pilosa.URI
var globURI *pnet.URI
// get correct node to go to.
func GetImportRoaringURI(index string, shard uint64) *pilosa.URI {
func GetImportRoaringURI(index string, shard uint64) *pnet.URI {
return globURI
}

40
encoding/proto/proto.go
View file

@ -22,8 +22,10 @@ import (
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/v2"
"github.com/pilosa/pilosa/v2/internal"
pnet "github.com/pilosa/pilosa/v2/net"
"github.com/pilosa/pilosa/v2/pql"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/topology"
"github.com/pkg/errors"
)
@ -184,7 +186,7 @@ func (s Serializer) Unmarshal(buf []byte, m pilosa.Message) error {
}
s.decodeNodeStatus(msg, mt)
return nil
case *pilosa.Node:
case *topology.Node:
msg := &internal.Node{}
err := proto.Unmarshal(buf, msg)
if err != nil {
@ -361,7 +363,7 @@ func (s Serializer) encodeToProto(m pilosa.Message) proto.Message {
return s.encodeNodeEventMessage(mt)
case *pilosa.NodeStatus:
return s.encodeNodeStatus(mt)
case *pilosa.Node:
case *topology.Node:
return s.encodeNode(mt)
case *pilosa.QueryRequest:
return s.encodeQueryRequest(mt)
@ -679,7 +681,7 @@ func (s Serializer) encodeFieldOptions(o *pilosa.FieldOptions) *internal.FieldOp
}
// s.encodeNodes converts a slice of Nodes into its internal representation.
func (s Serializer) encodeNodes(a []*pilosa.Node) []*internal.Node {
func (s Serializer) encodeNodes(a []*topology.Node) []*internal.Node {
other := make([]*internal.Node, len(a))
for i := range a {
other[i] = s.encodeNode(a[i])
@ -688,7 +690,7 @@ func (s Serializer) encodeNodes(a []*pilosa.Node) []*internal.Node {
}
// s.encodeNode converts a Node into its internal representation.
func (s Serializer) encodeNode(n *pilosa.Node) *internal.Node {
func (s Serializer) encodeNode(n *topology.Node) *internal.Node {
return &internal.Node{
ID: n.ID,
URI: s.encodeURI(n.URI),
@ -698,7 +700,7 @@ func (s Serializer) encodeNode(n *pilosa.Node) *internal.Node {
}
}
func (s Serializer) encodeURI(u pilosa.URI) *internal.URI {
func (s Serializer) encodeURI(u pnet.URI) *internal.URI {
return &internal.URI{
Scheme: u.Scheme,
Host: u.Host,
@ -948,9 +950,9 @@ func (s Serializer) encodeTransactionStats(stats pilosa.TransactionStats) *inter
func (s Serializer) decodeResizeInstruction(ri *internal.ResizeInstruction, m *pilosa.ResizeInstruction) {
m.JobID = ri.JobID
m.Node = &pilosa.Node{}
m.Node = &topology.Node{}
s.decodeNode(ri.Node, m.Node)
m.Coordinator = &pilosa.Node{}
m.Coordinator = &topology.Node{}
s.decodeNode(ri.Coordinator, m.Coordinator)
m.Sources = make([]*pilosa.ResizeSource, len(ri.Sources))
s.decodeResizeSources(ri.Sources, m.Sources)
@ -970,7 +972,7 @@ func (s Serializer) decodeResizeSources(srcs []*internal.ResizeSource, m []*pilo
}
func (s Serializer) decodeResizeSource(rs *internal.ResizeSource, m *pilosa.ResizeSource) {
m.Node = &pilosa.Node{}
m.Node = &topology.Node{}
s.decodeNode(rs.Node, m.Node)
m.Index = rs.Index
m.Field = rs.Field
@ -986,7 +988,7 @@ func (s Serializer) decodeTranslationResizeSources(srcs []*internal.TranslationR
}
func (s Serializer) decodeTranslationResizeSource(rs *internal.TranslationResizeSource, m *pilosa.TranslationResizeSource) {
m.Node = &pilosa.Node{}
m.Node = &topology.Node{}
s.decodeNode(rs.Node, m.Node)
m.Index = rs.Index
m.PartitionID = int(rs.PartitionID)
@ -1050,9 +1052,9 @@ func (s Serializer) decodeDecimal(d *internal.Decimal, m *pql.Decimal) {
m.Scale = d.Scale
}
func (s Serializer) decodeNodes(a []*internal.Node, m []*pilosa.Node) {
func (s Serializer) decodeNodes(a []*internal.Node, m []*topology.Node) {
for i := range a {
m[i] = &pilosa.Node{}
m[i] = &topology.Node{}
s.decodeNode(a[i], m[i])
}
}
@ -1060,13 +1062,13 @@ func (s Serializer) decodeNodes(a []*internal.Node, m []*pilosa.Node) {
func (s Serializer) decodeClusterStatus(cs *internal.ClusterStatus, m *pilosa.ClusterStatus) {
m.State = cs.State
m.ClusterID = cs.ClusterID
m.Nodes = make([]*pilosa.Node, len(cs.Nodes))
m.Nodes = make([]*topology.Node, len(cs.Nodes))
s.decodeNodes(cs.Nodes, m.Nodes)
m.Schema = &pilosa.Schema{}
s.decodeSchema(cs.Schema, m.Schema)
}
func (s Serializer) decodeNode(node *internal.Node, m *pilosa.Node) {
func (s Serializer) decodeNode(node *internal.Node, m *topology.Node) {
m.ID = node.ID
s.decodeURI(node.URI, &m.URI)
s.decodeURI(node.GRPCURI, &m.GRPCURI)
@ -1074,7 +1076,7 @@ func (s Serializer) decodeNode(node *internal.Node, m *pilosa.Node) {
m.State = node.State
}
func (s Serializer) decodeURI(i *internal.URI, m *pilosa.URI) {
func (s Serializer) decodeURI(i *internal.URI, m *pnet.URI) {
m.Scheme = i.Scheme
m.Host = i.Host
m.Port = uint16(i.Port)
@ -1137,18 +1139,18 @@ func (s Serializer) decodeDeleteViewMessage(pb *internal.DeleteViewMessage, m *p
func (s Serializer) decodeResizeInstructionComplete(pb *internal.ResizeInstructionComplete, m *pilosa.ResizeInstructionComplete) {
m.JobID = pb.JobID
m.Node = &pilosa.Node{}
m.Node = &topology.Node{}
s.decodeNode(pb.Node, m.Node)
m.Error = pb.Error
}
func (s Serializer) decodeSetCoordinatorMessage(pb *internal.SetCoordinatorMessage, m *pilosa.SetCoordinatorMessage) {
m.New = &pilosa.Node{}
m.New = &topology.Node{}
s.decodeNode(pb.New, m.New)
}
func (s Serializer) decodeUpdateCoordinatorMessage(pb *internal.UpdateCoordinatorMessage, m *pilosa.UpdateCoordinatorMessage) {
m.New = &pilosa.Node{}
m.New = &topology.Node{}
s.decodeNode(pb.New, m.New)
}
@ -1159,12 +1161,12 @@ func (s Serializer) decodeNodeStateMessage(pb *internal.NodeStateMessage, m *pil
func (s Serializer) decodeNodeEventMessage(pb *internal.NodeEventMessage, m *pilosa.NodeEvent) {
m.Event = pilosa.NodeEventType(pb.Event)
m.Node = &pilosa.Node{}
m.Node = &topology.Node{}
s.decodeNode(pb.Node, m.Node)
}
func (s Serializer) decodeNodeStatus(pb *internal.NodeStatus, m *pilosa.NodeStatus) {
m.Node = &pilosa.Node{}
m.Node = &topology.Node{}
m.Indexes = s.decodeIndexStatuses(pb.Indexes)
m.Schema = &pilosa.Schema{}
s.decodeSchema(pb.Schema, m.Schema)

4
event.go
View file

@ -14,6 +14,8 @@
package pilosa
import "github.com/pilosa/pilosa/v2/topology"
// NodeEventType are the types of node events.
type NodeEventType int
@ -27,5 +29,5 @@ const (
// NodeEvent is a single event related to node activity in the cluster.
type NodeEvent struct {
Event NodeEventType
Node *Node
Node *topology.Node
}

37
executor.go
View file

@ -30,6 +30,7 @@ import (
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/shardwidth"
"github.com/pilosa/pilosa/v2/testhook"
"github.com/pilosa/pilosa/v2/topology"
"github.com/pilosa/pilosa/v2/tracing"
"github.com/pkg/errors"
)
@ -51,7 +52,7 @@ type executor struct {
Holder *Holder
// Local hostname & cluster configuration.
Node *Node
Node *topology.Node
Cluster *cluster
// Client used for remote requests.
@ -5102,10 +5103,10 @@ func (e *executor) executeSetRowAttrs(ctx context.Context, qcx *Qcx, index strin
}
// Execute on remote nodes in parallel.
nodes := Nodes(e.Cluster.nodes).FilterID(e.Node.ID)
nodes := topology.Nodes(e.Cluster.nodes).FilterID(e.Node.ID)
resp := make(chan error, len(nodes))
for _, node := range nodes {
go func(node *Node) {
go func(node *topology.Node) {
_, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil, nil)
resp <- err
}(node)
@ -5214,10 +5215,10 @@ func (e *executor) executeBulkSetRowAttrs(ctx context.Context, qcx *Qcx, index s
}
// Execute on remote nodes in parallel.
nodes := Nodes(e.Cluster.nodes).FilterID(e.Node.ID)
nodes := topology.Nodes(e.Cluster.nodes).FilterID(e.Node.ID)
resp := make(chan error, len(nodes))
for _, node := range nodes {
go func(node *Node) {
go func(node *topology.Node) {
_, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: calls}, nil, nil)
resp <- err
}(node)
@ -5266,10 +5267,10 @@ func (e *executor) executeSetColumnAttrs(ctx context.Context, qcx *Qcx, index st
}
// Execute on remote nodes in parallel.
nodes := Nodes(e.Cluster.nodes).FilterID(e.Node.ID)
nodes := topology.Nodes(e.Cluster.nodes).FilterID(e.Node.ID)
resp := make(chan error, len(nodes))
for _, node := range nodes {
go func(node *Node) {
go func(node *topology.Node) {
_, err := e.remoteExec(ctx, node, index, &pql.Query{Calls: []*pql.Call{c}}, nil, nil)
resp <- err
}(node)
@ -5286,7 +5287,7 @@ func (e *executor) executeSetColumnAttrs(ctx context.Context, qcx *Qcx, index st
}
// remoteExec 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, embed []*Row) (results []interface{}, err error) { // nolint: interfacer
func (e *executor) remoteExec(ctx context.Context, node *topology.Node, index string, q *pql.Query, shards []uint64, embed []*Row) (results []interface{}, err error) { // nolint: interfacer
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeExec")
defer span.Finish()
@ -5308,13 +5309,13 @@ func (e *executor) remoteExec(ctx context.Context, node *Node, index string, q *
// shardsByNode returns a mapping of nodes to shards.
// Returns errShardUnavailable if a shard cannot be allocated to a node.
func (e *executor) shardsByNode(nodes []*Node, index string, shards []uint64) (map[*Node][]uint64, error) {
m := make(map[*Node][]uint64)
func (e *executor) shardsByNode(nodes []*topology.Node, index string, shards []uint64) (map[*topology.Node][]uint64, error) {
m := make(map[*topology.Node][]uint64)
loop:
for _, shard := range shards {
for _, node := range e.Cluster.ShardNodes(index, shard) {
if Nodes(nodes).Contains(node) {
if topology.Nodes(nodes).Contains(node) {
m[node] = append(m[node], shard)
continue loop
}
@ -5342,11 +5343,11 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
//
// However, if this request is being sent from the coordinator then all
// processing should be done locally so we start with just the local node.
var nodes []*Node
var nodes []*topology.Node
if !opt.Remote {
nodes = Nodes(e.Cluster.nodes).Clone()
nodes = topology.Nodes(e.Cluster.nodes).Clone()
} else {
nodes = []*Node{e.Cluster.nodeByID(e.Node.ID)}
nodes = []*topology.Node{e.Cluster.nodeByID(e.Node.ID)}
}
// Start mapping across all primary owners.
@ -5367,7 +5368,7 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
if resp.err != nil {
// Filter out unavailable nodes.
nodes = Nodes(nodes).Filter(resp.node)
nodes = topology.Nodes(nodes).Filter(resp.node)
// Begin mapper against secondary nodes.
if err := e.mapper(ctx, cancel, ch, nodes, index, resp.shards, c, opt, mapFn, reduceFn); errors.Cause(err) == errShardUnavailable {
@ -5434,7 +5435,7 @@ func makeEmbeddedDataForShards(allRows []*Row, shards []uint64) []*Row {
return newRows
}
func (e *executor) mapper(ctx context.Context, cancel context.CancelFunc, ch chan mapResponse, nodes []*Node, index string, shards []uint64, c *pql.Call, opt *execOptions, mapFn mapFunc, reduceFn reduceFunc) error {
func (e *executor) mapper(ctx context.Context, cancel context.CancelFunc, ch chan mapResponse, nodes []*topology.Node, index string, shards []uint64, c *pql.Call, opt *execOptions, mapFn mapFunc, reduceFn reduceFunc) error {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.mapper")
defer span.Finish()
done := ctx.Done()
@ -5447,7 +5448,7 @@ func (e *executor) mapper(ctx context.Context, cancel context.CancelFunc, ch cha
// Execute each node in a separate goroutine.
for n, nodeShards := range m {
go func(n *Node, nodeShards []uint64) {
go func(n *topology.Node, nodeShards []uint64) {
resp := mapResponse{node: n, shards: nodeShards}
// Send local shards to mapper, otherwise remote exec.
@ -6831,7 +6832,7 @@ type mapFunc func(ctx context.Context, shard uint64) (_ interface{}, err error)
type reduceFunc func(ctx context.Context, prev, v interface{}) interface{}
type mapResponse struct {
node *Node
node *topology.Node
shards []uint64
result interface{}

6
fragment.go
View file

@ -42,11 +42,13 @@ import (
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/v2/internal"
"github.com/pilosa/pilosa/v2/logger"
pnet "github.com/pilosa/pilosa/v2/net"
"github.com/pilosa/pilosa/v2/pql"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/shardwidth"
"github.com/pilosa/pilosa/v2/stats"
"github.com/pilosa/pilosa/v2/testhook"
"github.com/pilosa/pilosa/v2/topology"
"github.com/pilosa/pilosa/v2/tracing"
"github.com/pkg/errors"
)
@ -3528,7 +3530,7 @@ func (h *blockHasher) WriteValue(v uint64) {
type fragmentSyncer struct {
Fragment *fragment
Node *Node
Node *topology.Node
Cluster *cluster
// FieldType helps determine which method of syncing to use.
@ -3720,7 +3722,7 @@ func (s *fragmentSyncer) syncBlock(id int) error {
f := s.Fragment
// Read pairs from each remote block.
var uris []*URI
var uris []*pnet.URI
var pairSets []pairSet
for _, node := range s.Cluster.shardNodes(f.index(), f.shard) {
if s.Node.ID == node.ID {

14
gossip/gossip.go
View file

@ -31,8 +31,10 @@ import (
"github.com/hashicorp/memberlist"
"github.com/pilosa/pilosa/v2"
"github.com/pilosa/pilosa/v2/logger"
pnet "github.com/pilosa/pilosa/v2/net"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/toml"
"github.com/pilosa/pilosa/v2/topology"
"github.com/pkg/errors"
)
@ -79,21 +81,21 @@ func (g *memberSet) Open() (err error) {
RetransmitMult: 3,
}
var uris = make([]*pilosa.URI, len(g.config.gossipSeeds))
var uris = make([]*pnet.URI, len(g.config.gossipSeeds))
for i, addr := range g.config.gossipSeeds {
uris[i], err = pilosa.NewURIFromAddress(addr)
uris[i], err = pnet.NewURIFromAddress(addr)
if err != nil {
return fmt.Errorf("new uri from address: %s", err)
}
}
var nodes = make([]*pilosa.Node, len(uris))
var nodes = make([]*topology.Node, len(uris))
for i, uri := range uris {
nodes[i] = &pilosa.Node{URI: *uri}
nodes[i] = &topology.Node{URI: *uri}
}
g.mu.RLock()
err = g.joinWithRetry(pilosa.URIs(pilosa.Nodes(nodes).URIs()).HostPortStrings())
err = g.joinWithRetry(pnet.URIs(topology.Nodes(nodes).URIs()).HostPortStrings())
g.mu.RUnlock()
if err != nil {
return errors.Wrap(err, "joinWithRetry")
@ -447,7 +449,7 @@ func (g *eventReceiver) listen() {
}
// Get the node from the event.Node meta data.
var n pilosa.Node
var n topology.Node
if err := g.papi.Serializer.Unmarshal(e.Node.Meta, &n); err != nil {
panic("failed to unmarshal event node meta into node")
}

13
holder.go
View file

@ -36,6 +36,7 @@ import (
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/stats"
"github.com/pilosa/pilosa/v2/testhook"
"github.com/pilosa/pilosa/v2/topology"
"github.com/pilosa/pilosa/v2/tracing"
"github.com/pkg/errors"
uuid "github.com/satori/go.uuid"
@ -1304,7 +1305,7 @@ type holderSyncer struct {
Holder *Holder
Node *Node
Node *topology.Node
Cluster *cluster
// Translation sync handling.
@ -1416,7 +1417,7 @@ func (s *holderSyncer) syncIndex(index string) error {
s.Stats.CountWithCustomTags(MetricColumnAttrStoreBlocks, int64(len(blks)), 1.0, []string{indexTag})
// Sync with every other host.
for _, node := range Nodes(s.Cluster.nodes).FilterID(s.Node.ID) {
for _, node := range topology.Nodes(s.Cluster.nodes).FilterID(s.Node.ID) {
// Retrieve attributes from differing blocks.
// Skip update and recomputation if no attributes have changed.
m, err := s.Cluster.InternalClient.ColumnAttrDiff(ctx, &node.URI, index, blks)
@ -1463,7 +1464,7 @@ func (s *holderSyncer) syncField(index, name string) error {
s.Stats.CountWithCustomTags(MetricRowAttrStoreBlocks, int64(len(blks)), 1.0, []string{indexTag, fieldTag})
// Sync with every other host.
for _, node := range Nodes(s.Cluster.nodes).FilterID(s.Node.ID) {
for _, node := range topology.Nodes(s.Cluster.nodes).FilterID(s.Node.ID) {
// Retrieve attributes from differing blocks.
// Skip update and recomputation if no attributes have changed.
m, err := s.Cluster.InternalClient.RowAttrDiff(ctx, &node.URI, index, name, blks)
@ -1669,8 +1670,8 @@ func (s *holderSyncer) initializeIndexTranslateReplication() error {
}
for partitionID := 0; partitionID < s.Cluster.partitionN; partitionID++ {
partitionNodes := s.Cluster.partitionNodes(partitionID)
isPrimary := partitionNodes[0].ID == node.ID // remote is primary?
isReplica := Nodes(partitionNodes[1:]).ContainsID(s.Node.ID) // local is replica?
isPrimary := partitionNodes[0].ID == node.ID // remote is primary?
isReplica := topology.Nodes(partitionNodes[1:]).ContainsID(s.Node.ID) // local is replica?
if !isPrimary || !isReplica {
continue
}
@ -1797,7 +1798,7 @@ func (s *holderSyncer) readFieldTranslateReader(rd TranslateEntryReader) {
// holderCleaner removes fragments and data files that are no longer used.
type holderCleaner struct {
Node *Node
Node *topology.Node
Holder *Holder
Cluster *cluster

72
http/client.go
View file

@ -30,13 +30,15 @@ import (
"github.com/pilosa/pilosa/v2"
"github.com/pilosa/pilosa/v2/encoding/proto"
pnet "github.com/pilosa/pilosa/v2/net"
"github.com/pilosa/pilosa/v2/topology"
"github.com/pilosa/pilosa/v2/tracing"
"github.com/pkg/errors"
)
// InternalClient represents a client to the Pilosa cluster.
type InternalClient struct {
defaultURI *pilosa.URI
defaultURI *pnet.URI
serializer pilosa.Serializer
// The client to use for HTTP communication.
@ -49,7 +51,7 @@ func NewInternalClient(host string, remoteClient *http.Client) (*InternalClient,
return nil, pilosa.ErrHostRequired
}
uri, err := pilosa.NewURIFromAddress(host)
uri, err := pnet.NewURIFromAddress(host)
if err != nil {
return nil, errors.Wrap(err, "getting URI")
}
@ -58,7 +60,7 @@ func NewInternalClient(host string, remoteClient *http.Client) (*InternalClient,
return client, nil
}
func NewInternalClientFromURI(defaultURI *pilosa.URI, remoteClient *http.Client) *InternalClient {
func NewInternalClientFromURI(defaultURI *pnet.URI, remoteClient *http.Client) *InternalClient {
return &InternalClient{
defaultURI: defaultURI,
serializer: proto.Serializer{},
@ -133,7 +135,7 @@ func (c *InternalClient) Schema(ctx context.Context) ([]*pilosa.IndexInfo, error
return rsp.Indexes, nil
}
func (c *InternalClient) PostSchema(ctx context.Context, uri *pilosa.URI, s *pilosa.Schema, remote bool) error {
func (c *InternalClient) PostSchema(ctx context.Context, uri *pnet.URI, s *pilosa.Schema, remote bool) error {
u := uri.Path(fmt.Sprintf("/schema?remote=%v", remote))
buf, err := json.Marshal(s)
if err != nil {
@ -207,7 +209,7 @@ func (c *InternalClient) CreateIndex(ctx context.Context, index string, opt pilo
}
// FragmentNodes returns a list of nodes that own a shard.
func (c *InternalClient) FragmentNodes(ctx context.Context, index string, shard uint64) ([]*pilosa.Node, error) {
func (c *InternalClient) FragmentNodes(ctx context.Context, index string, shard uint64) ([]*topology.Node, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.FragmentNodes")
defer span.Finish()
@ -231,7 +233,7 @@ func (c *InternalClient) FragmentNodes(ctx context.Context, index string, shard
}
defer resp.Body.Close()
var a []*pilosa.Node
var a []*topology.Node
if err := json.NewDecoder(resp.Body).Decode(&a); err != nil {
return nil, fmt.Errorf("json decode: %s", err)
}
@ -239,7 +241,7 @@ func (c *InternalClient) FragmentNodes(ctx context.Context, index string, shard
}
// Nodes returns a list of all nodes.
func (c *InternalClient) Nodes(ctx context.Context) ([]*pilosa.Node, error) {
func (c *InternalClient) Nodes(ctx context.Context) ([]*topology.Node, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Nodes")
defer span.Finish()
@ -262,7 +264,7 @@ func (c *InternalClient) Nodes(ctx context.Context) ([]*pilosa.Node, error) {
}
defer resp.Body.Close()
var a []*pilosa.Node
var a []*topology.Node
if err := json.NewDecoder(resp.Body).Decode(&a); err != nil {
return nil, fmt.Errorf("json decode: %s", err)
}
@ -277,7 +279,7 @@ func (c *InternalClient) Query(ctx context.Context, index string, 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 *pilosa.QueryRequest) (*pilosa.QueryResponse, error) {
func (c *InternalClient) QueryNode(ctx context.Context, uri *pnet.URI, index string, queryRequest *pilosa.QueryRequest) (*pilosa.QueryResponse, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "QueryNode")
defer span.Finish()
@ -368,7 +370,7 @@ func (c *InternalClient) Import(ctx context.Context, index, field string, shard
return nil
}
func getCoordinatorNode(nodes []*pilosa.Node) *pilosa.Node {
func getCoordinatorNode(nodes []*topology.Node) *topology.Node {
for _, node := range nodes {
if node.IsCoordinator {
return node
@ -482,7 +484,7 @@ func (c *InternalClient) marshalImportPayload(index, field string, shard uint64,
}
// importNode sends a pre-marshaled import request to a node.
func (c *InternalClient) importNode(ctx context.Context, node *pilosa.Node, index, field string, buf []byte, opts *pilosa.ImportOptions) error {
func (c *InternalClient) importNode(ctx context.Context, node *topology.Node, index, field string, buf []byte, opts *pilosa.ImportOptions) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.importNode")
defer span.Finish()
@ -664,7 +666,7 @@ func (c *InternalClient) marshalImportValuePayload(index, field string, shard ui
// ImportRoaring does fast import of raw bits in roaring format (pilosa or
// official format, see API.ImportRoaring).
func (c *InternalClient) ImportRoaring(ctx context.Context, uri *pilosa.URI, index, field string, shard uint64, remote bool, req *pilosa.ImportRoaringRequest) error {
func (c *InternalClient) ImportRoaring(ctx context.Context, uri *pnet.URI, index, field string, shard uint64, remote bool, req *pilosa.ImportRoaringRequest) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportRoaring")
defer span.Finish()
@ -718,7 +720,7 @@ func (c *InternalClient) ImportRoaring(ctx context.Context, uri *pilosa.URI, ind
}
// ImportColumnAttrs does bulk import of column attrs
func (c *InternalClient) ImportColumnAttrs(ctx context.Context, uri *pilosa.URI, index string, req *pilosa.ImportColumnAttrsRequest) error {
func (c *InternalClient) ImportColumnAttrs(ctx context.Context, uri *pnet.URI, index string, req *pilosa.ImportColumnAttrsRequest) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportRoaring")
defer span.Finish()
@ -802,7 +804,7 @@ func (c *InternalClient) ExportCSV(ctx context.Context, index, field string, sha
}
// exportNode copies a CSV export from a node to w.
func (c *InternalClient) exportNodeCSV(ctx context.Context, node *pilosa.Node, index, field string, shard uint64, w io.Writer) error {
func (c *InternalClient) exportNodeCSV(ctx context.Context, node *topology.Node, index, field string, shard uint64, w io.Writer) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.exportNodeCSV")
defer span.Finish()
@ -840,11 +842,11 @@ func (c *InternalClient) exportNodeCSV(ctx context.Context, node *pilosa.Node, i
// RetrieveShardFromURI returns a ReadCloser which contains the data of the
// specified shard from the specified node. Caller *must* close the returned
// ReadCloser or risk leaking goroutines/tcp connections.
func (c *InternalClient) RetrieveShardFromURI(ctx context.Context, index, field, view string, shard uint64, uri pilosa.URI) (io.ReadCloser, error) {
func (c *InternalClient) RetrieveShardFromURI(ctx context.Context, index, field, view string, shard uint64, uri pnet.URI) (io.ReadCloser, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.RetrieveShardFromURI")
defer span.Finish()
node := &pilosa.Node{
node := &topology.Node{
URI: uri,
}
@ -961,7 +963,7 @@ func (c *InternalClient) CreateFieldWithOptions(ctx context.Context, index, fiel
// FragmentBlocks returns a list of block checksums for a fragment on a host.
// Only returns blocks which contain data.
func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pilosa.URI, index, field, view string, shard uint64) ([]pilosa.FragmentBlock, error) {
func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pnet.URI, index, field, view string, shard uint64) ([]pilosa.FragmentBlock, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.FragmentBlocks")
defer span.Finish()
@ -1005,7 +1007,7 @@ func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pilosa.URI, in
}
// BlockData returns row/column id pairs for a block.
func (c *InternalClient) BlockData(ctx context.Context, uri *pilosa.URI, index, field, view string, shard uint64, block int) ([]uint64, []uint64, error) {
func (c *InternalClient) BlockData(ctx context.Context, uri *pnet.URI, index, field, view string, shard uint64, block int) ([]uint64, []uint64, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.BlockData")
defer span.Finish()
@ -1054,7 +1056,7 @@ func (c *InternalClient) BlockData(ctx context.Context, uri *pilosa.URI, index,
}
// ColumnAttrDiff returns data from differing blocks on a remote host.
func (c *InternalClient) ColumnAttrDiff(ctx context.Context, uri *pilosa.URI, index string, blks []pilosa.AttrBlock) (map[uint64]map[string]interface{}, error) {
func (c *InternalClient) ColumnAttrDiff(ctx context.Context, uri *pnet.URI, index string, blks []pilosa.AttrBlock) (map[uint64]map[string]interface{}, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ColumnAttrDiff")
defer span.Finish()
@ -1094,7 +1096,7 @@ func (c *InternalClient) ColumnAttrDiff(ctx context.Context, uri *pilosa.URI, in
}
// RowAttrDiff returns data from differing blocks on a remote host.
func (c *InternalClient) RowAttrDiff(ctx context.Context, uri *pilosa.URI, index, field string, blks []pilosa.AttrBlock) (map[uint64]map[string]interface{}, error) {
func (c *InternalClient) RowAttrDiff(ctx context.Context, uri *pnet.URI, index, field string, blks []pilosa.AttrBlock) (map[uint64]map[string]interface{}, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.RowAttrDiff")
defer span.Finish()
@ -1137,7 +1139,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, msg []byte) error {
func (c *InternalClient) SendMessage(ctx context.Context, uri *pnet.URI, msg []byte) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.SendMessage")
defer span.Finish()
@ -1163,7 +1165,7 @@ func (c *InternalClient) SendMessage(ctx context.Context, uri *pilosa.URI, msg [
// TranslateKeysNode function is mainly called to translate keys from coordinator node.
// If coordinator node returns 404 error the function wraps it with pilosa.ErrTranslatingKeyNotFound.
func (c *InternalClient) TranslateKeysNode(ctx context.Context, uri *pilosa.URI, index, field string, keys []string, writable bool) ([]uint64, error) {
func (c *InternalClient) TranslateKeysNode(ctx context.Context, uri *pnet.URI, index, field string, keys []string, writable bool) ([]uint64, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "TranslateKeysNode")
defer span.Finish()
@ -1218,7 +1220,7 @@ func (c *InternalClient) TranslateKeysNode(ctx context.Context, uri *pilosa.URI,
}
// TranslateIDsNode sends an id translation request to a specific node.
func (c *InternalClient) TranslateIDsNode(ctx context.Context, uri *pilosa.URI, index, field string, ids []uint64) ([]string, error) {
func (c *InternalClient) TranslateIDsNode(ctx context.Context, uri *pnet.URI, index, field string, ids []uint64) ([]string, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "TranslateIDsNode")
defer span.Finish()
@ -1269,7 +1271,7 @@ func (c *InternalClient) TranslateIDsNode(ctx context.Context, uri *pilosa.URI,
}
// GetNodeUsage retrieves the size-on-disk information for the specified node.
func (c *InternalClient) GetNodeUsage(ctx context.Context, uri *pilosa.URI) (map[string]pilosa.NodeUsage, error) {
func (c *InternalClient) GetNodeUsage(ctx context.Context, uri *pnet.URI) (map[string]pilosa.NodeUsage, error) {
u := uri.Path("/ui/usage?remote=true")
req, err := http.NewRequest("GET", u, nil)
if err != nil {
@ -1300,7 +1302,7 @@ func (c *InternalClient) GetNodeUsage(ctx context.Context, uri *pilosa.URI) (map
}
// GetPastQueries retrieves the query history log for the specified node.
func (c *InternalClient) GetPastQueries(ctx context.Context, uri *pilosa.URI) ([]pilosa.PastQueryStatus, error) {
func (c *InternalClient) GetPastQueries(ctx context.Context, uri *pnet.URI) ([]pilosa.PastQueryStatus, error) {
u := uri.Path("/query-history?remote=true")
req, err := http.NewRequest("GET", u, nil)
if err != nil {
@ -1330,7 +1332,7 @@ func (c *InternalClient) GetPastQueries(ctx context.Context, uri *pilosa.URI) ([
return queries, nil
}
func (c *InternalClient) FindIndexKeysNode(ctx context.Context, uri *pilosa.URI, index string, keys ...string) (transMap map[string]uint64, err error) {
func (c *InternalClient) FindIndexKeysNode(ctx context.Context, uri *pnet.URI, index string, keys ...string) (transMap map[string]uint64, err error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.FindIndexKeysNode")
defer span.Finish()
@ -1379,7 +1381,7 @@ func (c *InternalClient) FindIndexKeysNode(ctx context.Context, uri *pilosa.URI,
return transMap, nil
}
func (c *InternalClient) FindFieldKeysNode(ctx context.Context, uri *pilosa.URI, index string, field string, keys ...string) (transMap map[string]uint64, err error) {
func (c *InternalClient) FindFieldKeysNode(ctx context.Context, uri *pnet.URI, index string, field string, keys ...string) (transMap map[string]uint64, err error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.FindFieldKeysNode")
defer span.Finish()
@ -1427,7 +1429,7 @@ func (c *InternalClient) FindFieldKeysNode(ctx context.Context, uri *pilosa.URI,
return transMap, nil
}
func (c *InternalClient) CreateIndexKeysNode(ctx context.Context, uri *pilosa.URI, index string, keys ...string) (transMap map[string]uint64, err error) {
func (c *InternalClient) CreateIndexKeysNode(ctx context.Context, uri *pnet.URI, index string, keys ...string) (transMap map[string]uint64, err error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.CreateIndexKeysNode")
defer span.Finish()
@ -1476,7 +1478,7 @@ func (c *InternalClient) CreateIndexKeysNode(ctx context.Context, uri *pilosa.UR
return transMap, nil
}
func (c *InternalClient) CreateFieldKeysNode(ctx context.Context, uri *pilosa.URI, index string, field string, keys ...string) (transMap map[string]uint64, err error) {
func (c *InternalClient) CreateFieldKeysNode(ctx context.Context, uri *pnet.URI, index string, field string, keys ...string) (transMap map[string]uint64, err error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.CreateFieldKeysNode")
defer span.Finish()
@ -1922,7 +1924,7 @@ func pos(rowID, columnID uint64) uint64 {
return (rowID * pilosa.ShardWidth) + (columnID % pilosa.ShardWidth)
}
func uriPathToURL(uri *pilosa.URI, path string) url.URL {
func uriPathToURL(uri *pnet.URI, path string) url.URL {
return url.URL{
Scheme: uri.Scheme,
Host: uri.HostPort(),
@ -1930,7 +1932,7 @@ func uriPathToURL(uri *pilosa.URI, path string) url.URL {
}
}
func nodePathToURL(node *pilosa.Node, path string) url.URL {
func nodePathToURL(node *topology.Node, path string) url.URL {
return url.URL{
Scheme: node.URI.Scheme,
Host: node.URI.HostPort(),
@ -1941,11 +1943,11 @@ func nodePathToURL(node *pilosa.Node, path string) url.URL {
// RetrieveTranslatePartitionFromURI returns a ReadCloser which contains the data of the
// specified translate partition from the specified node. Caller *must* close the returned
// ReadCloser or risk leaking goroutines/tcp connections.
func (c *InternalClient) RetrieveTranslatePartitionFromURI(ctx context.Context, index string, partition int, uri pilosa.URI) (io.ReadCloser, error) {
func (c *InternalClient) RetrieveTranslatePartitionFromURI(ctx context.Context, index string, partition int, uri pnet.URI) (io.ReadCloser, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.RetrieveTranslatePartitionFromURI")
defer span.Finish()
node := &pilosa.Node{
node := &topology.Node{
URI: uri,
}
@ -1974,7 +1976,7 @@ func (c *InternalClient) RetrieveTranslatePartitionFromURI(ctx context.Context,
return resp.Body, nil
}
func (c *InternalClient) ImportIndexKeys(ctx context.Context, uri *pilosa.URI, index string, partitionID int, remote bool, rddbdata io.Reader) error {
func (c *InternalClient) ImportIndexKeys(ctx context.Context, uri *pnet.URI, index string, partitionID int, remote bool, rddbdata io.Reader) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportIndexKeys")
defer span.Finish()
@ -2006,7 +2008,7 @@ func (c *InternalClient) ImportIndexKeys(ctx context.Context, uri *pilosa.URI, i
return nil
}
func (c *InternalClient) ImportFieldKeys(ctx context.Context, uri *pilosa.URI, index, field string, remote bool, rddbdata io.Reader) error {
func (c *InternalClient) ImportFieldKeys(ctx context.Context, uri *pnet.URI, index, field string, remote bool, rddbdata io.Reader) error {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.ImportFieldKeys")
defer span.Finish()

15
http/handler.go
View file

@ -44,6 +44,7 @@ import (
"github.com/pilosa/pilosa/v2/encoding/proto"
"github.com/pilosa/pilosa/v2/logger"
"github.com/pilosa/pilosa/v2/pql"
"github.com/pilosa/pilosa/v2/topology"
"github.com/pilosa/pilosa/v2/tracing"
"github.com/pkg/errors"
"github.com/prometheus/client_golang/prometheus/promhttp"
@ -793,10 +794,10 @@ type getSchemaResponse struct {
}
type getStatusResponse struct {
State string `json:"state"`
Nodes []*pilosa.Node `json:"nodes"`
LocalID string `json:"localID"`
ClusterName string `json:"clusterName"`
State string `json:"state"`
Nodes []*topology.Node `json:"nodes"`
LocalID string `json:"localID"`
ClusterName string `json:"clusterName"`
}
func hash(s string) string {
@ -2053,8 +2054,8 @@ type setCoordinatorRequest struct {
}
type setCoordinatorResponse struct {
Old *pilosa.Node `json:"old"`
New *pilosa.Node `json:"new"`
Old *topology.Node `json:"old"`
New *topology.Node `json:"new"`
}
// handlePostClusterResizeRemoveNode handles POST /cluster/resize/remove-node request.
@ -2095,7 +2096,7 @@ type removeNodeRequest struct {
}
type removeNodeResponse struct {
Remove *pilosa.Node `json:"remove"`
Remove *topology.Node `json:"remove"`
}
// handlePostClusterResizeAbort handles POST /cluster/resize/abort request.

7
pilosa.go
View file

@ -19,6 +19,7 @@ import (
"regexp"
"time"
pnet "github.com/pilosa/pilosa/v2/net"
"github.com/pkg/errors"
)
@ -208,9 +209,9 @@ func timestamp() int64 {
// AddressWithDefaults converts addr into a valid address,
// using defaults when necessary.
func AddressWithDefaults(addr string) (*URI, error) {
func AddressWithDefaults(addr string) (*pnet.URI, error) {
if addr == "" {
return defaultURI(), nil
return pnet.DefaultURI(), nil
}
return NewURIFromAddress(addr)
return pnet.NewURIFromAddress(addr)
}

18
server.go
View file

@ -30,9 +30,11 @@ import (
uuid "github.com/satori/go.uuid"
"github.com/pilosa/pilosa/v2/logger"
pnet "github.com/pilosa/pilosa/v2/net"
rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/stats"
"github.com/pilosa/pilosa/v2/topology"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
)
@ -68,8 +70,8 @@ type Server struct { // nolint: maligned
snapshotQueue SnapshotQueue
nodeID string
uri URI
grpcURI URI
uri pnet.URI
grpcURI pnet.URI
antiEntropyInterval time.Duration
metricInterval time.Duration
diagnosticInterval time.Duration
@ -248,7 +250,7 @@ func OptServerNodeDownRetries(retries int, sleep time.Duration) ServerOption {
// OptServerURI is a functional option on Server
// used to set the server URI.
func OptServerURI(uri *URI) ServerOption {
func OptServerURI(uri *pnet.URI) ServerOption {
return func(s *Server) error {
s.uri = *uri
return nil
@ -257,7 +259,7 @@ func OptServerURI(uri *URI) ServerOption {
// OptServerGRPCURI is a functional option on Server
// used to set the server gRPC URI.
func OptServerGRPCURI(uri *URI) ServerOption {
func OptServerGRPCURI(uri *pnet.URI) ServerOption {
return func(s *Server) error {
s.grpcURI = *uri
return nil
@ -459,7 +461,7 @@ func NewServer(opts ...ServerOption) (*Server, error) {
}
// Set Cluster Node.
node := &Node{
node := &topology.Node{
ID: s.nodeID,
URI: s.uri,
GRPCURI: s.grpcURI,
@ -499,7 +501,7 @@ func (s *Server) InternalClient() InternalClient {
return s.defaultClient
}
func (s *Server) GRPCURI() URI {
func (s *Server) GRPCURI() pnet.URI {
return s.grpcURI
}
@ -905,7 +907,7 @@ func (s *Server) SendAsync(m Message) error {
}
// SendTo represents an implementation of Broadcaster.
func (s *Server) SendTo(to *Node, m Message) error {
func (s *Server) SendTo(to *topology.Node, m Message) error {
msg, err := s.serializer.Marshal(m)
if err != nil {
return fmt.Errorf("marshaling message: %v", err)
@ -916,7 +918,7 @@ func (s *Server) SendTo(to *Node, m Message) error {
// node returns the pilosa.node object. It is used by membership protocols to
// get this node's name(ID), location(URI), and coordinator status.
func (s *Server) node() Node {
func (s *Server) node() topology.Node {
return *s.cluster.Node
}

7
server/server.go
View file

@ -46,6 +46,7 @@ import (
"github.com/pilosa/pilosa/v2/gossip"
"github.com/pilosa/pilosa/v2/http"
"github.com/pilosa/pilosa/v2/logger"
pnet "github.com/pilosa/pilosa/v2/net"
"github.com/pilosa/pilosa/v2/prometheus"
"github.com/pilosa/pilosa/v2/statik"
"github.com/pilosa/pilosa/v2/stats"
@ -88,7 +89,7 @@ type Command struct {
grpcLn net.Listener
API *pilosa.API
ln net.Listener
listenURI *pilosa.URI
listenURI *pnet.URI
tlsConfig *tls.Config
closeTimeout time.Duration
pgserver *PostgresServer
@ -368,7 +369,7 @@ func (m *Command) SetupServer() error {
}
// Get grpc advertise address as uri.
advertiseGRPCURI, err := pilosa.NewURIFromAddress(m.Config.AdvertiseGRPC)
advertiseGRPCURI, err := pnet.NewURIFromAddress(m.Config.AdvertiseGRPC)
if err != nil {
return errors.Wrap(err, "processing grpc advertise address")
}
@ -595,7 +596,7 @@ func newStatsClient(name string, host string) (stats.StatsClient, error) {
}
// getListener gets a net.Listener based on the config.
func getListener(uri pilosa.URI, tlsconf *tls.Config) (ln net.Listener, err error) {
func getListener(uri pnet.URI, tlsconf *tls.Config) (ln net.Listener, err error) {
// If bind URI has the https scheme, enable TLS
if uri.Scheme == "https" && tlsconf != nil {
ln, err = tls.Listen("tcp", uri.HostPort(), tlsconf)

26
utils_internal_test.go
View file

@ -24,8 +24,10 @@ import (
"time"
"github.com/gogo/protobuf/proto"
pnet "github.com/pilosa/pilosa/v2/net"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/testhook"
"github.com/pilosa/pilosa/v2/topology"
"github.com/pkg/errors"
)
@ -73,7 +75,7 @@ func NewTestCluster(tb testing.TB, n int) *cluster {
c.Topology = NewTopology(c.Hasher, c.partitionN, c.ReplicaN, c)
for i := 0; i < n; i++ {
c.nodes = append(c.nodes, &Node{
c.nodes = append(c.nodes, &topology.Node{
ID: fmt.Sprintf("node%d", i),
URI: NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0)),
})
@ -87,17 +89,17 @@ func NewTestCluster(tb testing.TB, n int) *cluster {
}
// NewTestURI is a test URI creator that intentionally swallows errors.
func NewTestURI(scheme, host string, port uint16) URI {
uri := defaultURI()
_ = uri.setScheme(scheme)
_ = uri.setHost(host)
func NewTestURI(scheme, host string, port uint16) pnet.URI {
uri := pnet.DefaultURI()
_ = uri.SetScheme(scheme)
_ = uri.SetHost(host)
uri.SetPort(port)
return *uri
}
func NewTestURIFromHostPort(host string, port uint16) URI {
uri := defaultURI()
_ = uri.setHost(host)
func NewTestURIFromHostPort(host string, port uint16) pnet.URI {
uri := pnet.DefaultURI()
_ = uri.SetHost(host)
uri.SetPort(port)
return *uri
}
@ -127,7 +129,7 @@ type ClusterCluster struct {
}
type commonClusterSettings struct {
Nodes []*Node
Nodes []*topology.Node
}
func (t *ClusterCluster) CreateIndex(name string) error {
@ -257,7 +259,7 @@ func (t *ClusterCluster) addCluster(i int, saveTopology bool) (*cluster, error)
id := fmt.Sprintf("node%d", i)
uri := NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0))
node := &Node{
node := &topology.Node{
ID: id,
URI: uri,
}
@ -406,7 +408,7 @@ func (bcast) SendAsync(Message) error {
}
// SendTo is a test implementation of Broadcaster SendTo method.
func (b bcast) SendTo(to *Node, m Message) error {
func (b bcast) SendTo(to *topology.Node, m Message) error {
switch obj := m.(type) {
case *ResizeInstruction:
err := b.t.FollowResizeInstruction(obj)
@ -551,7 +553,7 @@ func NewTestClusterWithReplication(tb testing.TB, nNodes, nReplicas, partitionN
for i := 0; i < nNodes; i++ {
nodeID := fmt.Sprintf("node%d", i)
c.nodes = append(c.nodes, &Node{
c.nodes = append(c.nodes, &topology.Node{
ID: nodeID,
URI: NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0)),
})