From 4515a24e487fada4fac19f877effe05d76206b5e Mon Sep 17 00:00:00 2001 From: Travis Date: Wed, 6 Jan 2021 16:09:24 -0600 Subject: [PATCH] change all references to use subpackages: topology, net --- api.go | 13 +- broadcast.go | 5 +- client.go | 87 +++++++------- cluster.go | 239 ++++++++++--------------------------- cluster_internal_test.go | 124 +++++++++---------- cmd/badloader/badloader.go | 14 ++- cmd/slurp/slurp.go | 7 +- encoding/proto/proto.go | 40 ++++--- event.go | 4 +- executor.go | 37 +++--- fragment.go | 6 +- gossip/gossip.go | 14 ++- holder.go | 13 +- http/client.go | 72 +++++------ http/handler.go | 15 +-- pilosa.go | 7 +- server.go | 18 +-- server/server.go | 7 +- utils_internal_test.go | 26 ++-- 19 files changed, 330 insertions(+), 418 deletions(-) diff --git a/api.go b/api.go index 9140153a1..8f8a831e4 100644 --- a/api.go +++ b/api.go @@ -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, } } diff --git a/broadcast.go b/broadcast.go index 37d2bb39d..f883d421d 100644 --- a/broadcast.go +++ b/broadcast.go @@ -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 ( diff --git a/client.go b/client.go index 4cd410345..ad53cdf4f 100644 --- a/client.go +++ b/client.go @@ -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 } diff --git a/cluster.go b/cluster.go index 3a9976e72..405b65901 100644 --- a/cluster.go +++ b/cluster.go @@ -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 } diff --git a/cluster_internal_test.go b/cluster_internal_test.go index a9ec930be..77fc173b4 100644 --- a/cluster_internal_test.go +++ b/cluster_internal_test.go @@ -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)), }) diff --git a/cmd/badloader/badloader.go b/cmd/badloader/badloader.go index 719a256f5..5a6b0fad8 100644 --- a/cmd/badloader/badloader.go +++ b/cmd/badloader/badloader.go @@ -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 } diff --git a/cmd/slurp/slurp.go b/cmd/slurp/slurp.go index 9d49d20a5..11bbec314 100644 --- a/cmd/slurp/slurp.go +++ b/cmd/slurp/slurp.go @@ -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 } diff --git a/encoding/proto/proto.go b/encoding/proto/proto.go index 10a2f05a2..1444247e3 100644 --- a/encoding/proto/proto.go +++ b/encoding/proto/proto.go @@ -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) diff --git a/event.go b/event.go index b27bd1bf6..39e688f07 100644 --- a/event.go +++ b/event.go @@ -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 } diff --git a/executor.go b/executor.go index 6536c17f6..76eadf161 100644 --- a/executor.go +++ b/executor.go @@ -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{} diff --git a/fragment.go b/fragment.go index aa93dcc9a..20cdcf380 100644 --- a/fragment.go +++ b/fragment.go @@ -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 { diff --git a/gossip/gossip.go b/gossip/gossip.go index d7b6377e7..e37e8e8ca 100644 --- a/gossip/gossip.go +++ b/gossip/gossip.go @@ -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") } diff --git a/holder.go b/holder.go index 256871058..f5145ba31 100644 --- a/holder.go +++ b/holder.go @@ -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 diff --git a/http/client.go b/http/client.go index 4f37d7a36..7eb5d025f 100644 --- a/http/client.go +++ b/http/client.go @@ -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() diff --git a/http/handler.go b/http/handler.go index bc2fbb1ab..efc993f45 100644 --- a/http/handler.go +++ b/http/handler.go @@ -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. diff --git a/pilosa.go b/pilosa.go index a05228087..edb0240d1 100644 --- a/pilosa.go +++ b/pilosa.go @@ -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) } diff --git a/server.go b/server.go index 2fb74e560..fb37f3a0d 100644 --- a/server.go +++ b/server.go @@ -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 } diff --git a/server/server.go b/server/server.go index 291c986e1..b27f7741f 100644 --- a/server/server.go +++ b/server/server.go @@ -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) diff --git a/utils_internal_test.go b/utils_internal_test.go index d59f6aabc..4b5820351 100644 --- a/utils_internal_test.go +++ b/utils_internal_test.go @@ -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)), })