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

un-export (some) Cluster methods

Browse source
This commit is contained in:
Travis Turner 2018-06-08 15:48:23 -05:00
parent ab652ebe7d
commit 8021fc389b
No known key found for this signature in database
GPG key ID: 7F08008DFD9314C9
14 changed files with 753 additions and 1174 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
api.go
View file

@ -291,7 +291,7 @@ func (api *API) ExportCSV(ctx context.Context, indexName string, fieldName strin
}
// Validate that this handler owns the slice.
if !api.Cluster.OwnsSlice(api.LocalID(), indexName, slice) {
if !api.Cluster.ownsSlice(api.LocalID(), indexName, slice) {
api.Logger.Printf("node %s does not own slice %d of index %s", api.LocalID(), slice, indexName)
return ErrClusterDoesNotOwnSlice
}
@ -327,7 +327,7 @@ func (api *API) SliceNodes(ctx context.Context, indexName string, slice uint64)
return nil, errors.Wrap(err, "validating api method")
}
return api.Cluster.SliceNodes(indexName, slice), nil
return api.Cluster.sliceNodes(indexName, slice), nil
}
// MarshalFragment returns an object which can write the specified fragment's data
@ -681,7 +681,7 @@ func (api *API) LongQueryTime() time.Duration {
func (api *API) indexField(indexName string, fieldName string, slice uint64) (*Index, *Field, error) {
// Validate that this handler owns the slice.
if !api.Cluster.OwnsSlice(api.LocalID(), indexName, slice) {
if !api.Cluster.ownsSlice(api.LocalID(), indexName, slice) {
api.Logger.Printf("node %s does not own slice %d of index %s", api.LocalID(), slice, indexName)
return nil, nil, ErrClusterDoesNotOwnSlice
}
@ -709,15 +709,15 @@ func (api *API) SetCoordinator(ctx context.Context, id string) (oldNode, newNode
return nil, nil, errors.Wrap(err, "validating api method")
}
oldNode = api.Cluster.NodeByID(api.Cluster.Coordinator)
newNode = api.Cluster.NodeByID(id)
oldNode = api.Cluster.nodeByID(api.Cluster.Coordinator)
newNode = api.Cluster.nodeByID(id)
if newNode == nil {
return nil, nil, errors.Wrap(ErrNodeIDNotExists, "getting new node")
}
// If the new coordinator is this node, do the SetCoordinator directly.
if newNode.ID == api.LocalID() {
return oldNode, newNode, api.Cluster.SetCoordinator(newNode)
return oldNode, newNode, api.Cluster.setCoordinator(newNode)
}
// Send the set-coordinator message to new node.
@ -739,13 +739,13 @@ func (api *API) RemoveNode(id string) (*Node, error) {
return nil, errors.Wrap(err, "validating api method")
}
removeNode := api.Cluster.nodeByID(id)
removeNode := api.Cluster.unprotectedNodeByID(id)
if removeNode == nil {
return nil, errors.Wrap(ErrNodeIDNotExists, "finding node to remove")
}
// Start the resize process (similar to NodeJoin)
err := api.Cluster.NodeLeave(removeNode)
err := api.Cluster.nodeLeave(removeNode)
if err != nil {
return removeNode, errors.Wrap(err, "calling node leave")
}
@ -758,7 +758,7 @@ func (api *API) ResizeAbort() error {
return errors.Wrap(err, "validating api method")
}
err := api.Cluster.CompleteCurrentJob(ResizeJobStateAborted)
err := api.Cluster.completeCurrentJob(resizeJobStateAborted)
return errors.Wrap(err, "complete current job")
}

422
cluster.go
View file

@ -49,14 +49,14 @@ const (
NodeStateLoading = "LOADING"
NodeStateReady = "READY"
// ResizeJob states.
ResizeJobStateRunning = "RUNNING"
// resizeJob states.
resizeJobStateRunning = "RUNNING"
// Final states.
ResizeJobStateDone = "DONE"
ResizeJobStateAborted = "ABORTED"
resizeJobStateDone = "DONE"
resizeJobStateAborted = "ABORTED"
ResizeJobActionAdd = "ADD"
ResizeJobActionRemove = "REMOVE"
resizeJobActionAdd = "ADD"
resizeJobActionRemove = "REMOVE"
)
// Node represents a node in the cluster.
@ -255,8 +255,8 @@ type Cluster struct {
joined bool
mu sync.RWMutex
jobs map[int64]*ResizeJob
currentJob *ResizeJob
jobs map[int64]*resizeJob
currentJob *resizeJob
// Close management
wg sync.WaitGroup
@ -279,7 +279,7 @@ func NewCluster() *Cluster {
EventReceiver: NopEventReceiver,
joiningLeavingNodes: make(chan nodeAction, 10), // buffered channel
jobs: make(map[int64]*ResizeJob),
jobs: make(map[int64]*resizeJob),
closing: make(chan struct{}),
joining: make(chan struct{}),
@ -289,27 +289,27 @@ func NewCluster() *Cluster {
}
}
// Coordinator returns the coordinator node.
func (c *Cluster) CoordinatorNode() *Node {
return c.nodeByID(c.Coordinator)
// coordinatorNode returns the coordinator node.
func (c *Cluster) coordinatorNode() *Node {
return c.unprotectedNodeByID(c.Coordinator)
}
// IsCoordinator is true if this node is the coordinator.
func (c *Cluster) IsCoordinator() bool {
// isCoordinator is true if this node is the coordinator.
func (c *Cluster) isCoordinator() bool {
c.mu.RLock()
defer c.mu.RUnlock()
return c.isCoordinator()
return c.unprotectedIsCoordinator()
}
func (c *Cluster) isCoordinator() bool {
func (c *Cluster) unprotectedIsCoordinator() bool {
return c.Coordinator == c.Node.ID
}
// SetCoordinator tells the current node to become the
// setCoordinator tells the current node to become the
// 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 *Node) error {
c.mu.Lock()
// Verify that the new Coordinator value matches
// this node.
@ -319,7 +319,7 @@ func (c *Cluster) SetCoordinator(n *Node) error {
}
// Update IsCoordinator on all nodes (locally).
_ = c.updateCoordinator(n)
_ = c.unprotectedUpdateCoordinator(n)
c.mu.Unlock()
// Send the update coordinator message to all nodes.
err := c.Broadcaster.SendSync(
@ -334,17 +334,17 @@ func (c *Cluster) SetCoordinator(n *Node) error {
return c.Broadcaster.SendSync(c.Status())
}
// UpdateCoordinator updates this nodes Coordinator value as well as
// updateCoordinator updates this nodes Coordinator value as well as
// 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 {
func (c *Cluster) updateCoordinator(n *Node) bool {
c.mu.Lock()
defer c.mu.Unlock()
return c.updateCoordinator(n)
return c.unprotectedUpdateCoordinator(n)
}
func (c *Cluster) updateCoordinator(n *Node) bool {
func (c *Cluster) unprotectedUpdateCoordinator(n *Node) bool {
var changed bool
if c.Coordinator != n.ID {
c.Coordinator = n.ID
@ -360,9 +360,9 @@ func (c *Cluster) updateCoordinator(n *Node) bool {
return changed
}
// AddNode adds a node to the Cluster and updates and saves the
// addNode adds a node to the Cluster and updates and saves the
// new topology.
func (c *Cluster) AddNode(node *Node) error {
func (c *Cluster) addNode(node *Node) error {
c.Logger.Printf("add node %s to cluster on %s", node, c.Node)
// If the node being added is the coordinator, set it for this node.
@ -387,9 +387,9 @@ func (c *Cluster) AddNode(node *Node) error {
return c.saveTopology()
}
// RemoveNode removes a node from the Cluster and updates and saves the
// removeNode removes a node from the Cluster and updates and saves the
// new topology.
func (c *Cluster) RemoveNode(node *Node) error {
func (c *Cluster) removeNode(node *Node) error {
// remove from cluster
if !c.removeNodeBasicSorted(node) {
return nil
@ -407,8 +407,8 @@ func (c *Cluster) RemoveNode(node *Node) error {
return c.saveTopology()
}
// NodeIDs returns the list of IDs in the cluster.
func (c *Cluster) NodeIDs() []string {
// nodeIDs returns the list of IDs in the cluster.
func (c *Cluster) nodeIDs() []string {
return Nodes(c.Nodes).IDs()
}
@ -472,9 +472,9 @@ func (c *Cluster) setState(state string) {
}
}
func (c *Cluster) SetNodeState(state string) error {
if c.IsCoordinator() {
return c.ReceiveNodeState(c.Node.ID, state)
func (c *Cluster) setNodeState(state string) error {
if c.isCoordinator() {
return c.receiveNodeState(c.Node.ID, state)
}
// Send node state to coordinator.
@ -484,18 +484,18 @@ func (c *Cluster) SetNodeState(state string) error {
}
c.Logger.Printf("Sending State %s (%s)", state, c.Coordinator)
if err := c.sendTo(c.CoordinatorNode(), ns); err != nil {
if err := c.sendTo(c.coordinatorNode(), ns); err != nil {
return fmt.Errorf("sending node state error: err=%s", err)
}
return nil
}
// ReceiveNodeState sets node state in Topology in order for the
// receiveNodeState sets node state in Topology in order for the
// Coordinator to keep track of, during startup, which nodes have
// finished opening their Holder.
func (c *Cluster) ReceiveNodeState(nodeID string, state string) error {
if !c.IsCoordinator() {
func (c *Cluster) receiveNodeState(nodeID string, state string) error {
if !c.isCoordinator() {
return nil
}
@ -515,11 +515,6 @@ func (c *Cluster) ReceiveNodeState(nodeID string, state string) error {
return nil
}
// localNode is not being used.
//func (c *Cluster) localNode() *Node {
// return c.NodeByURI(c.URI)
//}
// Status returns the internal ClusterStatus representation.
func (c *Cluster) Status() *internal.ClusterStatus {
return &internal.ClusterStatus{
@ -529,14 +524,14 @@ func (c *Cluster) Status() *internal.ClusterStatus {
}
}
func (c *Cluster) NodeByID(id string) *Node {
func (c *Cluster) nodeByID(id string) *Node {
c.mu.RLock()
defer c.mu.RUnlock()
return c.nodeByID(id)
return c.unprotectedNodeByID(id)
}
// nodeByID returns a node reference by ID.
func (c *Cluster) nodeByID(id string) *Node {
// unprotectedNodeByID returns a node reference by ID.
func (c *Cluster) unprotectedNodeByID(id string) *Node {
for _, n := range c.Nodes {
if n.ID == id {
return n
@ -558,7 +553,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.
func (c *Cluster) addNodeBasicSorted(node *Node) bool {
n := c.nodeByID(node.ID)
n := c.unprotectedNodeByID(node.ID)
if n != nil {
return false
}
@ -645,7 +640,7 @@ func (c *Cluster) fragsByHost(idx *Index) fragsByHost {
func (c *Cluster) fragCombos(idx string, maxSlice uint64, fieldViews viewsByField) fragsByHost {
t := make(fragsByHost)
for i := uint64(0); i <= maxSlice; i++ {
nodes := c.SliceNodes(idx, i)
nodes := c.sliceNodes(idx, i)
for _, n := range nodes {
// for each field/view combination:
for field, views := range fieldViews {
@ -673,10 +668,10 @@ func (c *Cluster) diff(other *Cluster) (action string, nodeID string, err error)
if lenTo-lenFrom > 1 {
return "", "", errors.New("adding more than one node at a time is not supported")
}
action = ResizeJobActionAdd
action = resizeJobActionAdd
// Determine the node ID that is being added.
for _, n := range other.Nodes {
if c.nodeByID(n.ID) == nil {
if c.unprotectedNodeByID(n.ID) == nil {
nodeID = n.ID
break
}
@ -686,10 +681,10 @@ func (c *Cluster) diff(other *Cluster) (action string, nodeID string, err error)
if lenFrom-lenTo > 1 {
return "", "", errors.New("removing more than one node at a time is not supported")
}
action = ResizeJobActionRemove
action = resizeJobActionRemove
// Determine the node ID that is being removed.
for _, n := range c.Nodes {
if other.nodeByID(n.ID) == nil {
if other.unprotectedNodeByID(n.ID) == nil {
nodeID = n.ID
break
}
@ -721,7 +716,7 @@ func (c *Cluster) fragSources(to *Cluster, idx *Index) (map[string][]*internal.R
// If a node is being removed, however, then it will most likely
// require that a replica fragment be the source data.
srcCluster := c
if action == ResizeJobActionAdd && c.ReplicaN > 1 {
if action == resizeJobActionAdd && c.ReplicaN > 1 {
srcCluster = NewCluster()
srcCluster.Nodes = Nodes(c.Nodes).Clone()
srcCluster.Hasher = c.Hasher
@ -740,7 +735,7 @@ func (c *Cluster) fragSources(to *Cluster, idx *Index) (map[string][]*internal.R
srcNodesByFrag := make(map[frag]string)
for nodeID, frags := range srcFrags {
// If a node is being removed, don't consider it as a source.
if action == ResizeJobActionRemove && nodeID == diffNodeID {
if action == resizeJobActionRemove && nodeID == diffNodeID {
continue
}
for _, frag := range frags {
@ -772,7 +767,7 @@ func (c *Cluster) fragSources(to *Cluster, idx *Index) (map[string][]*internal.R
}
src := &internal.ResizeSource{
Node: EncodeNode(c.nodeByID(srcNodeID)),
Node: EncodeNode(c.unprotectedNodeByID(srcNodeID)),
Index: idx.Name(),
Field: frag.field,
View: frag.view,
@ -786,8 +781,8 @@ func (c *Cluster) fragSources(to *Cluster, idx *Index) (map[string][]*internal.R
return m, nil
}
// Partition returns the partition that a slice belongs to.
func (c *Cluster) Partition(index string, slice uint64) int {
// partition returns the partition that a slice belongs to.
func (c *Cluster) partition(index string, slice uint64) int {
var buf [8]byte
binary.BigEndian.PutUint64(buf[:], slice)
@ -798,18 +793,18 @@ func (c *Cluster) Partition(index string, slice uint64) int {
return int(h.Sum64() % uint64(c.PartitionN))
}
// SliceNodes returns a list of nodes that own a fragment.
func (c *Cluster) SliceNodes(index string, slice uint64) []*Node {
return c.PartitionNodes(c.Partition(index, slice))
// sliceNodes returns a list of nodes that own a fragment.
func (c *Cluster) sliceNodes(index string, slice uint64) []*Node {
return c.partitionNodes(c.partition(index, slice))
}
// OwnsSlice returns true if a host owns a fragment.
func (c *Cluster) OwnsSlice(nodeID string, index string, slice uint64) bool {
return Nodes(c.SliceNodes(index, slice)).ContainsID(nodeID)
// ownsSlice returns true if a host owns a fragment.
func (c *Cluster) ownsSlice(nodeID string, index string, slice uint64) bool {
return Nodes(c.sliceNodes(index, slice)).ContainsID(nodeID)
}
// PartitionNodes returns a list of nodes that own a partition.
func (c *Cluster) PartitionNodes(partitionID int) []*Node {
// partitionNodes returns a list of nodes that own a partition.
func (c *Cluster) partitionNodes(partitionID int) []*Node {
// Default replica count to between one and the number of nodes.
// The replica count can be zero if there are no nodes.
replicaN := c.ReplicaN
@ -831,27 +826,13 @@ func (c *Cluster) PartitionNodes(partitionID int) []*Node {
return nodes
}
// OwnsSlices finds the set of slices owned by the node per Index
func (c *Cluster) OwnsSlices(index string, maxSlice uint64, uri URI) []uint64 {
// containsSlices is like OwnsSlices, but it includes replicas.
func (c *Cluster) containsSlices(index string, maxSlice uint64, node *Node) []uint64 {
var slices []uint64
for i := uint64(0); i <= maxSlice; i++ {
p := c.Partition(index, i)
// Determine primary owner node.
nodeIndex := c.Hasher.Hash(uint64(p), len(c.Nodes))
if c.Nodes[nodeIndex].URI == uri {
slices = append(slices, i)
}
}
return slices
}
// ContainsSlices is like OwnsSlices, but it includes replicas.
func (c *Cluster) ContainsSlices(index string, maxSlice uint64, node *Node) []uint64 {
var slices []uint64
for i := uint64(0); i <= maxSlice; i++ {
p := c.Partition(index, i)
p := c.partition(index, i)
// Determine the nodes for partition.
nodes := c.PartitionNodes(p)
nodes := c.partitionNodes(p)
for _, n := range nodes {
if n.ID == node.ID {
slices = append(slices, i)
@ -884,7 +865,7 @@ func (h *jmphasher) Hash(key uint64, n int) int {
return int(b)
}
func (c *Cluster) Open() error {
func (c *Cluster) open() error {
// Cluster always comes up in state STARTING until cluster membership is determined.
c.state = ClusterStateStarting
@ -896,7 +877,7 @@ func (c *Cluster) Open() error {
c.ID = c.Topology.ClusterID
// Only the coordinator needs to consider the .topology file.
if c.IsCoordinator() {
if c.isCoordinator() {
err := c.considerTopology()
if err != nil {
return fmt.Errorf("considerTopology: %v", err)
@ -904,7 +885,7 @@ func (c *Cluster) Open() error {
}
// Add the local node to the cluster.
err := c.AddNode(c.Node)
err := c.addNode(c.Node)
if err != nil {
return errors.Wrap(err, "adding local node")
}
@ -920,7 +901,7 @@ func (c *Cluster) Open() error {
}
// If not coordinator then wait for ClusterStatus from coordinator.
if !c.IsCoordinator() {
if !c.isCoordinator() {
// In the case where a node has been restarted and memberlist has
// not had enough time to determine the node went down/up, then
// the coorninator needs to be alerted that this node is back up
@ -945,7 +926,7 @@ func (c *Cluster) Open() error {
return nil
}
func (c *Cluster) Close() error {
func (c *Cluster) close() error {
// Notify goroutines of closing and wait for completion.
close(c.closing)
c.wg.Wait()
@ -962,14 +943,14 @@ func (c *Cluster) markAsJoined() {
}
func (c *Cluster) needTopologyAgreement() bool {
return c.State() == ClusterStateStarting && !StringSlicesAreEqual(c.Topology.NodeIDs, c.NodeIDs())
return c.State() == ClusterStateStarting && !StringSlicesAreEqual(c.Topology.NodeIDs, c.nodeIDs())
}
func (c *Cluster) haveTopologyAgreement() bool {
if c.Static {
return true
}
return StringSlicesAreEqual(c.Topology.NodeIDs, c.NodeIDs())
return StringSlicesAreEqual(c.Topology.NodeIDs, c.nodeIDs())
}
func (c *Cluster) allNodesReady() bool {
@ -999,10 +980,10 @@ func (c *Cluster) handleNodeAction(nodeAction nodeAction) error {
// channel, which is not consumed until the code below.
var eg errgroup.Group
eg.Go(func() error {
return j.Run()
return j.run()
})
// Wait for the ResizeJob to finish or be aborted.
// Wait for the resizeJob to finish or be aborted.
c.Logger.Printf("wait for jobResult")
jobResult := <-j.result
@ -1013,18 +994,18 @@ func (c *Cluster) handleNodeAction(nodeAction nodeAction) error {
c.Logger.Printf("received jobResult: %s", jobResult)
switch jobResult {
case ResizeJobStateDone:
if err := c.CompleteCurrentJob(ResizeJobStateDone); err != nil {
case resizeJobStateDone:
if err := c.completeCurrentJob(resizeJobStateDone); err != nil {
return errors.Wrap(err, "completing finished job")
}
// Add/remove uri to/from the cluster.
if j.action == ResizeJobActionRemove {
return c.RemoveNode(nodeAction.node)
} else if j.action == ResizeJobActionAdd {
return c.AddNode(nodeAction.node)
if j.action == resizeJobActionRemove {
return c.removeNode(nodeAction.node)
} else if j.action == resizeJobActionAdd {
return c.addNode(nodeAction.node)
}
case ResizeJobStateAborted:
if err := c.CompleteCurrentJob(ResizeJobStateAborted); err != nil {
case resizeJobStateAborted:
if err := c.completeCurrentJob(resizeJobStateAborted); err != nil {
return errors.Wrap(err, "completing aborted job")
}
}
@ -1045,64 +1026,63 @@ func (c *Cluster) sendTo(node *Node, msg proto.Message) error {
return nil
}
// ListenForJoins handles cluster-resize events.
func (c *Cluster) ListenForJoins() {
c.wg.Add(1)
go func() { defer c.wg.Done(); c.listenForJoins() }()
}
// listenForJoins handles cluster-resize events.
func (c *Cluster) listenForJoins() {
// When a cluster starts, the state is STARTING.
// We first want to wait for at least one node to join.
// Then we want to clear out the joiningLeavingNodes queue (buffered channel).
// Then we want to set the cluster state to NORMAL and resume processing of joiningLeavingNodes events.
// We use a bool `setNormal` to indicate when at least one node has joined.
c.wg.Add(1)
go func() {
defer c.wg.Done()
var setNormal bool
// When a cluster starts, the state is STARTING.
// We first want to wait for at least one node to join.
// Then we want to clear out the joiningLeavingNodes queue (buffered channel).
// Then we want to set the cluster state to NORMAL and resume processing of joiningLeavingNodes events.
// We use a bool `setNormal` to indicate when at least one node has joined.
var setNormal bool
for {
for {
// Handle all pending joins before changing state back to NORMAL.
select {
case nodeAction := <-c.joiningLeavingNodes:
err := c.handleNodeAction(nodeAction)
if err != nil {
c.Logger.Printf("handleNodeAction error: err=%s", err)
// Handle all pending joins before changing state back to NORMAL.
select {
case nodeAction := <-c.joiningLeavingNodes:
err := c.handleNodeAction(nodeAction)
if err != nil {
c.Logger.Printf("handleNodeAction error: err=%s", err)
continue
}
setNormal = true
continue
default:
}
// Only change state to NORMAL if we have successfully added at least one host.
if setNormal {
// Put the cluster back to state NORMAL and broadcast.
if err := c.setStateAndBroadcast(ClusterStateNormal); err != nil {
c.Logger.Printf("setStateAndBroadcast error: err=%s", err)
}
}
// Wait for a joining host or a close.
select {
case <-c.closing:
return
case nodeAction := <-c.joiningLeavingNodes:
err := c.handleNodeAction(nodeAction)
if err != nil {
c.Logger.Printf("handleNodeAction error: err=%s", err)
continue
}
setNormal = true
continue
}
setNormal = true
continue
default:
}
// Only change state to NORMAL if we have successfully added at least one host.
if setNormal {
// Put the cluster back to state NORMAL and broadcast.
if err := c.setStateAndBroadcast(ClusterStateNormal); err != nil {
c.Logger.Printf("setStateAndBroadcast error: err=%s", err)
}
}
// Wait for a joining host or a close.
select {
case <-c.closing:
return
case nodeAction := <-c.joiningLeavingNodes:
err := c.handleNodeAction(nodeAction)
if err != nil {
c.Logger.Printf("handleNodeAction error: err=%s", err)
continue
}
setNormal = true
continue
}
}
}()
}
// generateResizeJob creates a new ResizeJob based on the new node being
// added/removed. It also saves a reference to the ResizeJob in the `jobs` map
// generateResizeJob creates a new resizeJob based on the new node being
// added/removed. It also saves a reference to the resizeJob in the `jobs` map
// for future lookup by JobID.
func (c *Cluster) generateResizeJob(nodeAction nodeAction) (*ResizeJob, error) {
func (c *Cluster) generateResizeJob(nodeAction nodeAction) (*resizeJob, error) {
c.Logger.Printf("generateResizeJob: %v", nodeAction)
c.mu.Lock()
defer c.mu.Unlock()
@ -1111,7 +1091,7 @@ func (c *Cluster) generateResizeJob(nodeAction nodeAction) (*ResizeJob, error) {
if err != nil {
return nil, errors.Wrap(err, "generating job")
}
c.Logger.Printf("generated ResizeJob: %d", j.ID)
c.Logger.Printf("generated resizeJob: %d", j.ID)
// Save job in jobs map for future reference.
c.jobs[j.ID] = j
@ -1125,12 +1105,12 @@ func (c *Cluster) generateResizeJob(nodeAction nodeAction) (*ResizeJob, error) {
return j, nil
}
// generateResizeJobByAction returns a ResizeJob with instructions based on
// generateResizeJobByAction returns a resizeJob with instructions based on
// the difference between Cluster and a new Cluster with/without uri.
// Broadcaster is associated to the ResizeJob here for use in broadcasting
// Broadcaster is associated to the resizeJob here for use in broadcasting
// the resize instructions to other nodes in the cluster.
func (c *Cluster) generateResizeJobByAction(nodeAction nodeAction) (*ResizeJob, error) {
j := NewResizeJob(c.Nodes, nodeAction.node, nodeAction.action)
func (c *Cluster) generateResizeJobByAction(nodeAction nodeAction) (*resizeJob, error) {
j := newResizeJob(c.Nodes, nodeAction.node, nodeAction.action)
j.Broadcaster = c.Broadcaster
// toCluster is a clone of Cluster with the new node added/removed for comparison.
@ -1139,9 +1119,9 @@ func (c *Cluster) generateResizeJobByAction(nodeAction nodeAction) (*ResizeJob,
toCluster.Hasher = c.Hasher
toCluster.PartitionN = c.PartitionN
toCluster.ReplicaN = c.ReplicaN
if nodeAction.action == ResizeJobActionRemove {
if nodeAction.action == resizeJobActionRemove {
toCluster.removeNodeBasicSorted(nodeAction.node)
} else if nodeAction.action == ResizeJobActionAdd {
} else if nodeAction.action == resizeJobActionAdd {
toCluster.addNodeBasicSorted(nodeAction.node)
}
@ -1172,8 +1152,8 @@ func (c *Cluster) generateResizeJobByAction(nodeAction nodeAction) (*ResizeJob,
}
instr := &internal.ResizeInstruction{
JobID: j.ID,
Node: EncodeNode(toCluster.nodeByID(id)),
Coordinator: EncodeNode(c.CoordinatorNode()),
Node: EncodeNode(toCluster.unprotectedNodeByID(id)),
Coordinator: EncodeNode(c.coordinatorNode()),
Sources: sources,
Schema: c.Holder.EncodeSchema(), // Include the schema to ensure it's in sync on the receiving node.
ClusterStatus: c.Status(),
@ -1184,28 +1164,28 @@ func (c *Cluster) generateResizeJobByAction(nodeAction nodeAction) (*ResizeJob,
return j, nil
}
// CompleteCurrentJob sets the state of the current ResizeJob
// completeCurrentJob sets the state of the current resizeJob
// then removes the pointer to currentJob.
func (c *Cluster) CompleteCurrentJob(state string) error {
func (c *Cluster) completeCurrentJob(state string) error {
c.mu.Lock()
defer c.mu.Unlock()
if !c.isCoordinator() {
if !c.unprotectedIsCoordinator() {
return ErrNodeNotCoordinator
}
if c.currentJob == nil {
return ErrResizeNotRunning
}
c.currentJob.SetState(state)
c.currentJob.setState(state)
c.currentJob = nil
return nil
}
// FollowResizeInstruction is run by any node that receives a ResizeInstruction.
func (c *Cluster) FollowResizeInstruction(instr *internal.ResizeInstruction) error {
// followResizeInstruction is run by any node that receives a ResizeInstruction.
func (c *Cluster) followResizeInstruction(instr *internal.ResizeInstruction) error {
c.Logger.Printf("follow resize instruction on %s", c.Node.ID)
// Make sure the cluster status on this node agrees with the Coordinator
// before attempting a resize.
if err := c.MergeClusterStatus(instr.ClusterStatus); err != nil {
if err := c.mergeClusterStatus(instr.ClusterStatus); err != nil {
return errors.Wrap(err, "merging cluster status")
}
@ -1297,13 +1277,13 @@ func (c *Cluster) FollowResizeInstruction(instr *internal.ResizeInstruction) err
return nil
}
func (c *Cluster) MarkResizeInstructionComplete(complete *internal.ResizeInstructionComplete) error {
func (c *Cluster) markResizeInstructionComplete(complete *internal.ResizeInstructionComplete) error {
j := c.Job(complete.JobID)
j := c.job(complete.JobID)
// Abort the job if an error exists in the complete object.
if complete.Error != "" {
j.result <- ResizeJobStateAborted
j.result <- resizeJobStateAborted
return errors.New(complete.Error)
}
@ -1311,29 +1291,27 @@ func (c *Cluster) MarkResizeInstructionComplete(complete *internal.ResizeInstruc
defer j.mu.Unlock()
if j.isComplete() {
return fmt.Errorf("ResizeJob %d is no longer running", j.ID)
return fmt.Errorf("resize job %d is no longer running", j.ID)
}
// Mark host complete.
j.IDs[complete.Node.ID] = true
if !j.nodesArePending() {
j.result <- ResizeJobStateDone
j.result <- resizeJobStateDone
}
return nil
}
// Job returns a ResizeJob by id.
func (c *Cluster) Job(id int64) *ResizeJob {
// job returns a resizeJob by id.
func (c *Cluster) job(id int64) *resizeJob {
c.mu.RLock()
defer c.mu.RUnlock()
return c.job(id)
return c.jobs[id]
}
func (c *Cluster) job(id int64) *ResizeJob { return c.jobs[id] }
type ResizeJob struct {
type resizeJob struct {
ID int64
IDs map[string]bool
Instructions []*internal.ResizeInstruction
@ -1348,15 +1326,15 @@ type ResizeJob struct {
Logger Logger
}
// NewResizeJob returns a new instance of ResizeJob.
func NewResizeJob(existingNodes []*Node, node *Node, action string) *ResizeJob {
// newResizeJob returns a new instance of resizeJob.
func newResizeJob(existingNodes []*Node, node *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
// has indicated that it has completed all resize instructions.
ids := make(map[string]bool)
if action == ResizeJobActionRemove {
if action == resizeJobActionRemove {
for _, n := range existingNodes {
// Exclude the removed node from the map.
if n.ID == node.ID {
@ -1364,7 +1342,7 @@ func NewResizeJob(existingNodes []*Node, node *Node, action string) *ResizeJob {
}
ids[n.ID] = false
}
} else if action == ResizeJobActionAdd {
} else if action == resizeJobActionAdd {
for _, n := range existingNodes {
ids[n.ID] = false
}
@ -1372,7 +1350,7 @@ func NewResizeJob(existingNodes []*Node, node *Node, action string) *ResizeJob {
ids[node.ID] = false
}
return &ResizeJob{
return &resizeJob{
ID: rand.Int63(),
IDs: ids,
action: action,
@ -1381,50 +1359,40 @@ func NewResizeJob(existingNodes []*Node, node *Node, action string) *ResizeJob {
}
}
func (j *ResizeJob) State() string {
j.mu.RLock()
defer j.mu.RUnlock()
return j.state
}
func (j *ResizeJob) SetState(state string) {
func (j *resizeJob) setState(state string) {
j.mu.Lock()
j.setState(state)
if j.state == "" || j.state == resizeJobStateRunning {
j.state = state
}
j.mu.Unlock()
}
func (j *ResizeJob) setState(state string) {
if j.state == "" || j.state == ResizeJobStateRunning {
j.state = state
}
}
// Run distributes ResizeInstructions.
func (j *ResizeJob) Run() error {
j.Logger.Printf("run ResizeJob")
// run distributes ResizeInstructions.
func (j *resizeJob) run() error {
j.Logger.Printf("run resizeJob")
// Set job state to RUNNING.
j.SetState(ResizeJobStateRunning)
j.setState(resizeJobStateRunning)
// Job can be considered done in the case where it doesn't require any action.
if !j.nodesArePending() {
j.Logger.Printf("ResizeJob contains no pending tasks; mark as done")
j.result <- ResizeJobStateDone
j.Logger.Printf("resizeJob contains no pending tasks; mark as done")
j.result <- resizeJobStateDone
return nil
}
j.Logger.Printf("distribute tasks for ResizeJob")
j.Logger.Printf("distribute tasks for resizeJob")
err := j.distributeResizeInstructions()
if err != nil {
j.result <- ResizeJobStateAborted
j.result <- resizeJobStateAborted
return errors.Wrap(err, "distributing instructions")
}
return nil
}
// isComplete return true if the job is any one of several completion states.
func (j *ResizeJob) isComplete() bool {
func (j *resizeJob) isComplete() bool {
switch j.state {
case ResizeJobStateDone, ResizeJobStateAborted:
case resizeJobStateDone, resizeJobStateAborted:
return true
default:
return false
@ -1432,7 +1400,7 @@ func (j *ResizeJob) isComplete() bool {
}
// nodesArePending returns true if any node is still working on the resize.
func (j *ResizeJob) nodesArePending() bool {
func (j *resizeJob) nodesArePending() bool {
for _, complete := range j.IDs {
if !complete {
return true
@ -1441,9 +1409,9 @@ func (j *ResizeJob) nodesArePending() bool {
return false
}
func (j *ResizeJob) distributeResizeInstructions() error {
func (j *resizeJob) distributeResizeInstructions() error {
j.Logger.Printf("distributeResizeInstructions for job %d", j.ID)
// Loop through the ResizeInstructions in ResizeJob and send to each host.
// Loop through the ResizeInstructions in resizeJob and send to each host.
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.
@ -1659,7 +1627,7 @@ func (c *Cluster) ReceiveEvent(e *NodeEvent) error {
case NodeJoin:
c.Logger.Printf("received NodeJoin event: %v", e)
// Ignore the event if this is not the coordinator.
if !c.IsCoordinator() {
if !c.isCoordinator() {
return nil
}
return c.nodeJoin(e.Node)
@ -1681,7 +1649,7 @@ func (c *Cluster) nodeJoin(node *Node) error {
return errors.New(err)
}
if err := c.AddNode(node); err != nil {
if err := c.addNode(node); err != nil {
return errors.Wrap(err, "adding node for agreement")
}
@ -1711,13 +1679,13 @@ func (c *Cluster) nodeJoin(node *Node) error {
// If the cluster already contains the node, just send it the cluster status.
// This is useful in the case where a node is restarted or temporarily leaves
// the cluster.
if node := c.nodeByID(node.ID); node != nil {
if node := c.unprotectedNodeByID(node.ID); node != nil {
return c.sendTo(node, c.Status())
}
// If the holder does not yet contain data, go ahead and add the node.
if ok, err := c.Holder.HasData(); !ok && err == nil {
if err := c.AddNode(node); err != nil {
if err := c.addNode(node); err != nil {
return errors.Wrap(err, "adding node")
}
return c.setStateAndBroadcast(ClusterStateNormal)
@ -1730,16 +1698,16 @@ func (c *Cluster) nodeJoin(node *Node) error {
if err := c.setStateAndBroadcast(ClusterStateResizing); err != nil {
return errors.Wrap(err, "broadcasting state")
}
c.joiningLeavingNodes <- nodeAction{node, ResizeJobActionAdd}
c.joiningLeavingNodes <- nodeAction{node, resizeJobActionAdd}
return nil
}
// NodeLeave initiates the removal of a node from the cluster.
func (c *Cluster) NodeLeave(node *Node) error {
// nodeLeave initiates the removal of a node from the cluster.
func (c *Cluster) nodeLeave(node *Node) error {
// Refuse the request if this is not the coordinator.
if !c.IsCoordinator() {
return fmt.Errorf("node removal requests are only valid on the coordinator node: %s", c.CoordinatorNode().ID)
if !c.isCoordinator() {
return fmt.Errorf("node removal requests are only valid on the coordinator node: %s", c.coordinatorNode().ID)
}
if c.State() != ClusterStateNormal {
@ -1747,7 +1715,7 @@ func (c *Cluster) NodeLeave(node *Node) error {
}
// Ensure that node is in the cluster.
if c.nodeByID(node.ID) == nil {
if c.unprotectedNodeByID(node.ID) == nil {
return fmt.Errorf("Node is not a member of the cluster: %s", node.ID)
}
@ -1757,18 +1725,12 @@ func (c *Cluster) NodeLeave(node *Node) error {
}
// See if resize job can be generated
_, err := c.generateResizeJobByAction(nodeAction{c.nodeByID(node.ID), ResizeJobActionRemove})
if err != nil {
if _, err := c.generateResizeJobByAction(nodeAction{c.unprotectedNodeByID(node.ID), resizeJobActionRemove}); err != nil {
return errors.Wrap(err, "generating job")
}
return c.nodeLeave(node)
}
func (c *Cluster) nodeLeave(node *Node) error {
// Get the actual node in the local cluster.
n := c.nodeByID(node.ID)
n := c.unprotectedNodeByID(node.ID)
// Don't do anything else if the cluster doesn't contain the node.
if n == nil {
@ -1777,7 +1739,7 @@ func (c *Cluster) nodeLeave(node *Node) error {
// If the holder does not yet contain data, go ahead and remove the node.
if ok, err := c.Holder.HasData(); !ok && err == nil {
if err := c.RemoveNode(n); err != nil {
if err := c.removeNode(n); err != nil {
return errors.Wrap(err, "removing node")
}
return c.setStateAndBroadcast(ClusterStateNormal)
@ -1790,17 +1752,17 @@ func (c *Cluster) nodeLeave(node *Node) error {
if err := c.setStateAndBroadcast(ClusterStateResizing); err != nil {
return errors.Wrap(err, "broadcasting state")
}
c.joiningLeavingNodes <- nodeAction{n, ResizeJobActionRemove}
c.joiningLeavingNodes <- nodeAction{n, resizeJobActionRemove}
return nil
}
func (c *Cluster) MergeClusterStatus(cs *internal.ClusterStatus) error {
func (c *Cluster) mergeClusterStatus(cs *internal.ClusterStatus) error {
c.mu.Lock()
defer c.mu.Unlock()
c.Logger.Printf("merge cluster status: %v", cs)
// Ignore status updates from self (coordinator).
if c.isCoordinator() {
if c.unprotectedIsCoordinator() {
return nil
}
@ -1811,7 +1773,7 @@ func (c *Cluster) MergeClusterStatus(cs *internal.ClusterStatus) error {
// Add all nodes from the coordinator.
for _, node := range officialNodes {
if err := c.AddNode(node); err != nil {
if err := c.addNode(node); err != nil {
return errors.Wrap(err, "adding node")
}
}
@ -1832,7 +1794,7 @@ func (c *Cluster) MergeClusterStatus(cs *internal.ClusterStatus) error {
}
for _, nodeID := range nodeIDsToRemove {
if err := c.RemoveNode(c.nodeByID(nodeID)); err != nil {
if err := c.removeNode(c.unprotectedNodeByID(nodeID)); err != nil {
return errors.Wrap(err, "removing node")
}
}

470
cluster_internal_test.go
View file

@ -15,11 +15,15 @@
package pilosa
import (
"bytes"
"io/ioutil"
"math/rand"
"reflect"
"strings"
"testing"
"testing/quick"
"github.com/davecgh/go-spew/spew"
"github.com/pilosa/pilosa/internal"
)
@ -287,19 +291,19 @@ func TestResizeJob(t *testing.T) {
{
existingNodes: []*Node{node0, node1},
node: node2,
action: ResizeJobActionAdd,
action: resizeJobActionAdd,
expectedIDs: map[string]bool{node0.ID: false, node1.ID: false, node2.ID: false},
},
{
existingNodes: []*Node{node0, node1, node2},
node: node2,
action: ResizeJobActionRemove,
action: resizeJobActionRemove,
expectedIDs: map[string]bool{node0.ID: false, node1.ID: false},
},
}
for _, test := range tests {
actual := NewResizeJob(test.existingNodes, test.node, test.action)
actual := newResizeJob(test.existingNodes, test.node, test.action)
if err != nil {
t.Fatal(err)
}
@ -308,3 +312,463 @@ 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{
{URI: NewTestURIFromHostPort("serverA", 1000)},
{URI: NewTestURIFromHostPort("serverB", 1000)},
{URI: NewTestURIFromHostPort("serverC", 1000)},
},
Hasher: NewTestModHasher(),
ReplicaN: 2,
}
// Verify nodes are distributed.
if a := c.partitionNodes(0); !reflect.DeepEqual(a, []*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]}) {
t.Fatalf("unexpected owners: %s", spew.Sdump(a))
}
}
// Ensure the partitioner can assign a fragment to a partition.
func TestCluster_Partition(t *testing.T) {
if err := quick.Check(func(index string, slice uint64, partitionN int) bool {
c := NewCluster()
c.PartitionN = partitionN
partitionID := c.partition(index, slice)
if partitionID < 0 || partitionID >= partitionN {
t.Errorf("partition out of range: slice=%d, p=%d, n=%d", slice, partitionID, partitionN)
}
return true
}, &quick.Config{
Values: func(values []reflect.Value, rand *rand.Rand) {
values[0], _ = quick.Value(reflect.TypeOf(""), rand)
values[1] = reflect.ValueOf(uint64(rand.Uint32()))
values[2] = reflect.ValueOf(rand.Intn(1000) + 1)
},
}); err != nil {
t.Fatal(err)
}
}
// Ensure the hasher can hash correctly.
func TestHasher(t *testing.T) {
for _, tt := range []struct {
key uint64
bucket []int
}{
// Generated from the reference C++ code
{0, []int{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}},
{1, []int{0, 0, 0, 0, 0, 0, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 17, 17}},
{0xdeadbeef, []int{0, 1, 2, 3, 3, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 16, 16, 16}},
{0x0ddc0ffeebadf00d, []int{0, 1, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 15, 15, 15, 15}},
} {
for i, v := range tt.bucket {
if got := NewHasher().Hash(tt.key, i+1); got != v {
t.Errorf("hash(%v,%v)=%v, want %v", tt.key, i+1, got, v)
}
}
}
}
// Ensure ContainsSlices can find the actual slice list for node and index.
func TestCluster_ContainsSlices(t *testing.T) {
c := NewTestCluster(5)
c.ReplicaN = 3
slices := c.containsSlices("test", 10, c.Nodes[2])
if !reflect.DeepEqual(slices, []uint64{0, 2, 3, 5, 6, 9, 10}) {
t.Fatalf("unexpected slices for node's index: %v", slices)
}
}
func TestCluster_Nodes(t *testing.T) {
uri0 := NewTestURIFromHostPort("node0", 0)
uri1 := NewTestURIFromHostPort("node1", 0)
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}
nodes := []*Node{node0, node1, node2}
t.Run("NodeIDs", func(t *testing.T) {
actual := 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)
}
})
t.Run("Filter", func(t *testing.T) {
actual := Nodes(Nodes(nodes).Filter(nodes[1])).URIs()
expected := []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}
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)
if !reflect.DeepEqual(actualTrue, true) {
t.Errorf("expected: %v, but got: %v", true, actualTrue)
}
if !reflect.DeepEqual(actualFalse, false) {
t.Errorf("expected: %v, but got: %v", false, actualTrue)
}
})
t.Run("Clone", func(t *testing.T) {
clone := Nodes(nodes).Clone()
actual := Nodes(clone).URIs()
expected := []URI{uri0, uri1, uri2}
if !reflect.DeepEqual(actual, expected) {
t.Errorf("expected: %v, but got: %v", expected, actual)
}
})
}
// NEXT: move this test to internal and unexport IsCoordinator
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}
c1 := *NewCluster()
c1.Node = node1
c1.Coordinator = node1.ID
c2 := *NewCluster()
c2.Node = node2
c2.Coordinator = node1.ID
t.Run("IsCoordinator", func(t *testing.T) {
if !c1.isCoordinator() {
t.Errorf("!IsCoordinator error: %v", c1.Node)
} else if c2.isCoordinator() {
t.Errorf("IsCoordinator error: %v", c2.Node)
}
})
}
func TestCluster_Topology(t *testing.T) {
c1 := NewTestCluster(1) // automatically creates Node{ID: "node0"}
uri0 := NewTestURIFromHostPort("host0", 0)
uri1 := NewTestURIFromHostPort("host1", 0)
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}
t.Run("AddNode", func(t *testing.T) {
err := c1.addNode(node1)
if err != nil {
t.Fatal(err)
}
// add the same host.
err = c1.addNode(node1)
if err != nil {
t.Fatal(err)
}
err = c1.addNode(node2)
if err != nil {
t.Fatal(err)
}
actual := c1.nodeIDs()
expected := []string{node0.ID, node1.ID, node2.ID}
if !reflect.DeepEqual(actual, expected) {
t.Errorf("expected: %v, but got: %v", expected, actual)
}
})
t.Run("ContainsID", func(t *testing.T) {
if !c1.Topology.ContainsID(node1.ID) {
t.Errorf("!ContainsHost error: %v", node1.ID)
} else if c1.Topology.ContainsID(nodeinvalid.ID) {
t.Errorf("ContainsHost error: %v", nodeinvalid.ID)
}
})
}
// Ensure that general cluster functionality works as expected.
func TestCluster_ResizeStates(t *testing.T) {
t.Run("Single node, no data", func(t *testing.T) {
tc := NewClusterCluster(1)
// Open TestCluster.
if err := tc.Open(); err != nil {
t.Fatal(err)
}
node := tc.Clusters[0]
// Ensure that node comes up in state NORMAL.
if node.State() != ClusterStateNormal {
t.Errorf("expected state: %v, but got: %v", ClusterStateNormal, node.State())
}
expectedTop := &Topology{
NodeIDs: []string{node.Node.ID},
}
// Verify topology file.
if !reflect.DeepEqual(node.Topology.NodeIDs, expectedTop.NodeIDs) {
t.Errorf("expected topology: %v, but got: %v", expectedTop.NodeIDs, node.Topology.NodeIDs)
}
// Close TestCluster.
if err := tc.Close(); err != nil {
t.Fatal(err)
}
})
t.Run("Single node, in topology", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.AddNode(false)
node := tc.Clusters[0]
// write topology to data file
top := &Topology{
NodeIDs: []string{node.Node.ID},
}
tc.WriteTopology(node.Path, top)
// Open TestCluster.
if err := tc.Open(); err != nil {
t.Fatal(err)
}
// Ensure that node comes up in state NORMAL.
if node.State() != ClusterStateNormal {
t.Errorf("expected state: %v, but got: %v", ClusterStateNormal, node.State())
}
// Close TestCluster.
if err := tc.Close(); err != nil {
t.Fatal(err)
}
})
t.Run("Single node, not in topology", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.AddNode(false)
node := tc.Clusters[0]
// write topology to data file
top := &Topology{
NodeIDs: []string{"some-other-host"},
}
tc.WriteTopology(node.Path, top)
// Open TestCluster.
expected := "considerTopology: coordinator node0 is not in topology: [some-other-host]"
err := tc.Open()
if err == nil || err.Error() != expected {
t.Errorf("did not receive expected error: %s", expected)
}
// Close TestCluster.
if err := tc.Close(); err != nil {
t.Fatal(err)
}
})
t.Run("Multiple nodes, no data", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.AddNode(false)
// Open TestCluster.
if err := tc.Open(); err != nil {
t.Fatal(err)
}
tc.AddNode(false)
node0 := tc.Clusters[0]
node1 := tc.Clusters[1]
// Ensure that nodes comes up in state NORMAL.
if node0.State() != ClusterStateNormal {
t.Errorf("expected node0 state: %v, but got: %v", ClusterStateNormal, node0.State())
} else if node1.State() != ClusterStateNormal {
t.Errorf("expected node1 state: %v, but got: %v", ClusterStateNormal, node1.State())
}
expectedTop := &Topology{
NodeIDs: []string{node0.Node.ID, node1.Node.ID},
}
// Verify topology file.
if !reflect.DeepEqual(node0.Topology.NodeIDs, expectedTop.NodeIDs) {
t.Errorf("expected node0 topology: %v, but got: %v", expectedTop.NodeIDs, node0.Topology.NodeIDs)
} else if !reflect.DeepEqual(node1.Topology.NodeIDs, expectedTop.NodeIDs) {
t.Errorf("expected node1 topology: %v, but got: %v", expectedTop.NodeIDs, node1.Topology.NodeIDs)
}
// Close TestCluster.
if err := tc.Close(); err != nil {
t.Fatal(err)
}
})
t.Run("Multiple nodes, in/not in topology", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.AddNode(false)
node0 := tc.Clusters[0]
// write topology to data file
top := &Topology{
NodeIDs: []string{"node0", "node2"},
}
tc.WriteTopology(node0.Path, top)
// Open TestCluster.
if err := tc.Open(); err != nil {
t.Fatal(err)
}
// Ensure that node is in state STARTING before the other node joins.
if node0.State() != ClusterStateStarting {
t.Errorf("expected node0 state: %v, but got: %v", ClusterStateStarting, node0.State())
}
// Expect an error by adding a node not in the topology.
expectedError := "host is not in topology: node1"
err := tc.AddNode(false)
if err == nil || err.Error() != expectedError {
t.Errorf("did not receive expected error: %s", expectedError)
}
tc.AddNode(false)
node2 := tc.Clusters[2]
// Ensure that node comes up in state NORMAL.
if node0.State() != ClusterStateNormal {
t.Errorf("expected node0 state: %v, but got: %v", ClusterStateNormal, node0.State())
} else if node2.State() != ClusterStateNormal {
t.Errorf("expected node1 state: %v, but got: %v", ClusterStateNormal, node2.State())
}
// Close TestCluster.
if err := tc.Close(); err != nil {
t.Fatal(err)
}
})
t.Run("Multiple nodes, with data", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.AddNode(false)
node0 := tc.Clusters[0]
// Open TestCluster.
if err := tc.Open(); err != nil {
t.Fatal(err)
}
// Add Bit Data to node0.
if err := tc.CreateField("i", "f", FieldOptions{}); err != nil {
t.Fatal(err)
}
tc.SetBit("i", "f", "standard", 1, 101, nil)
tc.SetBit("i", "f", "standard", 1, 1300000, nil)
// Before starting the resize, get the CheckSum to use for
// comparison later.
node0Field := node0.Holder.Field("i", "f")
node0View := node0Field.View("standard")
node0Fragment := node0View.Fragment(1)
node0Checksum := node0Fragment.Checksum()
// AddNode needs to block until the resize process has completed.
tc.AddNode(false)
node1 := tc.Clusters[1]
// Ensure that nodes come up in state NORMAL.
if node0.State() != ClusterStateNormal {
t.Errorf("expected node0 state: %v, but got: %v", ClusterStateNormal, node0.State())
} else if node1.State() != ClusterStateNormal {
t.Errorf("expected node1 state: %v, but got: %v", ClusterStateNormal, node1.State())
}
expectedTop := &Topology{
NodeIDs: []string{node0.Node.ID, node1.Node.ID},
}
// Verify topology file.
if !reflect.DeepEqual(node0.Topology.NodeIDs, expectedTop.NodeIDs) {
t.Errorf("expected node0 topology: %v, but got: %v", expectedTop.NodeIDs, node0.Topology.NodeIDs)
} else if !reflect.DeepEqual(node1.Topology.NodeIDs, expectedTop.NodeIDs) {
t.Errorf("expected node1 topology: %v, but got: %v", expectedTop.NodeIDs, node1.Topology.NodeIDs)
}
// Bits
// Verify that node-1 contains the fragment (i/f/standard/1) transferred from node-0.
node1Field := node1.Holder.Field("i", "f")
node1View := node1Field.View("standard")
node1Fragment := node1View.Fragment(1)
// Ensure checksums are the same.
if chksum := node1Fragment.Checksum(); !bytes.Equal(chksum, node0Checksum) {
t.Fatalf("expected standard view checksum to match: %x - %x", chksum, node0Checksum)
}
// Close TestCluster.
if err := tc.Close(); err != nil {
t.Fatal(err)
}
})
}
// Ensures that coordinator can be changed.
func TestCluster_UpdateCoordinator(t *testing.T) {
t.Run("UpdateCoordinator", func(t *testing.T) {
c := NewTestCluster(2)
oldNode := c.Nodes[0]
newNode := c.Nodes[1]
// Update coordinator to the same value.
if c.updateCoordinator(oldNode) {
t.Errorf("did not expect coordinator to change")
} else if c.Coordinator != oldNode.ID {
t.Errorf("expected coordinator: %s, but got: %s", c.Coordinator, oldNode.URI)
}
// Update coordinator to a new value.
if !c.updateCoordinator(newNode) {
t.Errorf("expected coordinator to change")
} else if c.Coordinator != newNode.ID {
t.Errorf("expected coordinator: %s, but got: %s", c.Coordinator, newNode.URI)
}
})
}

494
cluster_test.go
View file

@ -1,494 +0,0 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package pilosa
import (
"bytes"
"math/rand"
"reflect"
"testing"
"testing/quick"
"github.com/davecgh/go-spew/spew"
)
// Ensure the cluster can fairly distribute partitions across the nodes.
func TestCluster_Owners(t *testing.T) {
c := Cluster{
Nodes: []*Node{
{URI: NewTestURIFromHostPort("serverA", 1000)},
{URI: NewTestURIFromHostPort("serverB", 1000)},
{URI: NewTestURIFromHostPort("serverC", 1000)},
},
Hasher: NewTestModHasher(),
ReplicaN: 2,
}
// Verify nodes are distributed.
if a := c.PartitionNodes(0); !reflect.DeepEqual(a, []*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]}) {
t.Fatalf("unexpected owners: %s", spew.Sdump(a))
}
}
// Ensure the partitioner can assign a fragment to a partition.
func TestCluster_Partition(t *testing.T) {
if err := quick.Check(func(index string, slice uint64, partitionN int) bool {
c := NewCluster()
c.PartitionN = partitionN
partitionID := c.Partition(index, slice)
if partitionID < 0 || partitionID >= partitionN {
t.Errorf("partition out of range: slice=%d, p=%d, n=%d", slice, partitionID, partitionN)
}
return true
}, &quick.Config{
Values: func(values []reflect.Value, rand *rand.Rand) {
values[0], _ = quick.Value(reflect.TypeOf(""), rand)
values[1] = reflect.ValueOf(uint64(rand.Uint32()))
values[2] = reflect.ValueOf(rand.Intn(1000) + 1)
},
}); err != nil {
t.Fatal(err)
}
}
// Ensure the hasher can hash correctly.
func TestHasher(t *testing.T) {
for _, tt := range []struct {
key uint64
bucket []int
}{
// Generated from the reference C++ code
{0, []int{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}},
{1, []int{0, 0, 0, 0, 0, 0, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 17, 17}},
{0xdeadbeef, []int{0, 1, 2, 3, 3, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 16, 16, 16}},
{0x0ddc0ffeebadf00d, []int{0, 1, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 15, 15, 15, 15}},
} {
for i, v := range tt.bucket {
if got := NewHasher().Hash(tt.key, i+1); got != v {
t.Errorf("hash(%v,%v)=%v, want %v", tt.key, i+1, got, v)
}
}
}
}
// Ensure OwnsSlices can find the actual slice list for node and index.
func TestCluster_OwnsSlices(t *testing.T) {
c := NewTestCluster(5)
slices := c.OwnsSlices("test", 10, NewTestURIFromHostPort("host2", 0))
if !reflect.DeepEqual(slices, []uint64{0, 3, 6, 10}) {
t.Fatalf("unexpected slices for node's index: %v", slices)
}
}
// Ensure ContainsSlices can find the actual slice list for node and index.
func TestCluster_ContainsSlices(t *testing.T) {
c := NewTestCluster(5)
c.ReplicaN = 3
slices := c.ContainsSlices("test", 10, c.Nodes[2])
if !reflect.DeepEqual(slices, []uint64{0, 2, 3, 5, 6, 9, 10}) {
t.Fatalf("unexpected slices for node's index: %v", slices)
}
}
func TestCluster_Nodes(t *testing.T) {
uri0 := NewTestURIFromHostPort("node0", 0)
uri1 := NewTestURIFromHostPort("node1", 0)
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}
nodes := []*Node{node0, node1, node2}
t.Run("NodeIDs", func(t *testing.T) {
actual := 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)
}
})
t.Run("Filter", func(t *testing.T) {
actual := Nodes(Nodes(nodes).Filter(nodes[1])).URIs()
expected := []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}
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)
if !reflect.DeepEqual(actualTrue, true) {
t.Errorf("expected: %v, but got: %v", true, actualTrue)
}
if !reflect.DeepEqual(actualFalse, false) {
t.Errorf("expected: %v, but got: %v", false, actualTrue)
}
})
t.Run("Clone", func(t *testing.T) {
clone := Nodes(nodes).Clone()
actual := Nodes(clone).URIs()
expected := []URI{uri0, uri1, uri2}
if !reflect.DeepEqual(actual, expected) {
t.Errorf("expected: %v, but got: %v", expected, actual)
}
})
}
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}
c1 := *NewCluster()
c1.Node = node1
c1.Coordinator = node1.ID
c2 := *NewCluster()
c2.Node = node2
c2.Coordinator = node1.ID
t.Run("IsCoordinator", func(t *testing.T) {
if !c1.IsCoordinator() {
t.Errorf("!IsCoordinator error: %v", c1.Node)
} else if c2.IsCoordinator() {
t.Errorf("IsCoordinator error: %v", c2.Node)
}
})
}
func TestCluster_Topology(t *testing.T) {
c1 := NewTestCluster(1) // automatically creates Node{ID: "node0"}
uri0 := NewTestURIFromHostPort("host0", 0)
uri1 := NewTestURIFromHostPort("host1", 0)
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}
t.Run("AddNode", func(t *testing.T) {
err := c1.AddNode(node1)
if err != nil {
t.Fatal(err)
}
// add the same host.
err = c1.AddNode(node1)
if err != nil {
t.Fatal(err)
}
err = c1.AddNode(node2)
if err != nil {
t.Fatal(err)
}
actual := c1.NodeIDs()
expected := []string{node0.ID, node1.ID, node2.ID}
if !reflect.DeepEqual(actual, expected) {
t.Errorf("expected: %v, but got: %v", expected, actual)
}
})
t.Run("ContainsID", func(t *testing.T) {
if !c1.Topology.ContainsID(node1.ID) {
t.Errorf("!ContainsHost error: %v", node1.ID)
} else if c1.Topology.ContainsID(nodeinvalid.ID) {
t.Errorf("ContainsHost error: %v", nodeinvalid.ID)
}
})
}
// Ensure that general cluster functionality works as expected.
func TestCluster_ResizeStates(t *testing.T) {
t.Run("Single node, no data", func(t *testing.T) {
tc := NewClusterCluster(1)
// Open TestCluster.
if err := tc.Open(); err != nil {
t.Fatal(err)
}
node := tc.Clusters[0]
// Ensure that node comes up in state NORMAL.
if node.State() != ClusterStateNormal {
t.Errorf("expected state: %v, but got: %v", ClusterStateNormal, node.State())
}
expectedTop := &Topology{
NodeIDs: []string{node.Node.ID},
}
// Verify topology file.
if !reflect.DeepEqual(node.Topology.NodeIDs, expectedTop.NodeIDs) {
t.Errorf("expected topology: %v, but got: %v", expectedTop.NodeIDs, node.Topology.NodeIDs)
}
// Close TestCluster.
if err := tc.Close(); err != nil {
t.Fatal(err)
}
})
t.Run("Single node, in topology", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.AddNode(false)
node := tc.Clusters[0]
// write topology to data file
top := &Topology{
NodeIDs: []string{node.Node.ID},
}
tc.WriteTopology(node.Path, top)
// Open TestCluster.
if err := tc.Open(); err != nil {
t.Fatal(err)
}
// Ensure that node comes up in state NORMAL.
if node.State() != ClusterStateNormal {
t.Errorf("expected state: %v, but got: %v", ClusterStateNormal, node.State())
}
// Close TestCluster.
if err := tc.Close(); err != nil {
t.Fatal(err)
}
})
t.Run("Single node, not in topology", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.AddNode(false)
node := tc.Clusters[0]
// write topology to data file
top := &Topology{
NodeIDs: []string{"some-other-host"},
}
tc.WriteTopology(node.Path, top)
// Open TestCluster.
expected := "considerTopology: coordinator node0 is not in topology: [some-other-host]"
err := tc.Open()
if err == nil || err.Error() != expected {
t.Errorf("did not receive expected error: %s", expected)
}
// Close TestCluster.
if err := tc.Close(); err != nil {
t.Fatal(err)
}
})
t.Run("Multiple nodes, no data", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.AddNode(false)
// Open TestCluster.
if err := tc.Open(); err != nil {
t.Fatal(err)
}
tc.AddNode(false)
node0 := tc.Clusters[0]
node1 := tc.Clusters[1]
// Ensure that nodes comes up in state NORMAL.
if node0.State() != ClusterStateNormal {
t.Errorf("expected node0 state: %v, but got: %v", ClusterStateNormal, node0.State())
} else if node1.State() != ClusterStateNormal {
t.Errorf("expected node1 state: %v, but got: %v", ClusterStateNormal, node1.State())
}
expectedTop := &Topology{
NodeIDs: []string{node0.Node.ID, node1.Node.ID},
}
// Verify topology file.
if !reflect.DeepEqual(node0.Topology.NodeIDs, expectedTop.NodeIDs) {
t.Errorf("expected node0 topology: %v, but got: %v", expectedTop.NodeIDs, node0.Topology.NodeIDs)
} else if !reflect.DeepEqual(node1.Topology.NodeIDs, expectedTop.NodeIDs) {
t.Errorf("expected node1 topology: %v, but got: %v", expectedTop.NodeIDs, node1.Topology.NodeIDs)
}
// Close TestCluster.
if err := tc.Close(); err != nil {
t.Fatal(err)
}
})
t.Run("Multiple nodes, in/not in topology", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.AddNode(false)
node0 := tc.Clusters[0]
// write topology to data file
top := &Topology{
NodeIDs: []string{"node0", "node2"},
}
tc.WriteTopology(node0.Path, top)
// Open TestCluster.
if err := tc.Open(); err != nil {
t.Fatal(err)
}
// Ensure that node is in state STARTING before the other node joins.
if node0.State() != ClusterStateStarting {
t.Errorf("expected node0 state: %v, but got: %v", ClusterStateStarting, node0.State())
}
// Expect an error by adding a node not in the topology.
expectedError := "host is not in topology: node1"
err := tc.AddNode(false)
if err == nil || err.Error() != expectedError {
t.Errorf("did not receive expected error: %s", expectedError)
}
tc.AddNode(false)
node2 := tc.Clusters[2]
// Ensure that node comes up in state NORMAL.
if node0.State() != ClusterStateNormal {
t.Errorf("expected node0 state: %v, but got: %v", ClusterStateNormal, node0.State())
} else if node2.State() != ClusterStateNormal {
t.Errorf("expected node1 state: %v, but got: %v", ClusterStateNormal, node2.State())
}
// Close TestCluster.
if err := tc.Close(); err != nil {
t.Fatal(err)
}
})
t.Run("Multiple nodes, with data", func(t *testing.T) {
tc := NewClusterCluster(0)
tc.AddNode(false)
node0 := tc.Clusters[0]
// Open TestCluster.
if err := tc.Open(); err != nil {
t.Fatal(err)
}
// Add Bit Data to node0.
if err := tc.CreateField("i", "f", FieldOptions{}); err != nil {
t.Fatal(err)
}
tc.SetBit("i", "f", "standard", 1, 101, nil)
tc.SetBit("i", "f", "standard", 1, 1300000, nil)
// Before starting the resize, get the CheckSum to use for
// comparison later.
node0Field := node0.Holder.Field("i", "f")
node0View := node0Field.View("standard")
node0Fragment := node0View.Fragment(1)
node0Checksum := node0Fragment.Checksum()
// AddNode needs to block until the resize process has completed.
tc.AddNode(false)
node1 := tc.Clusters[1]
// Ensure that nodes come up in state NORMAL.
if node0.State() != ClusterStateNormal {
t.Errorf("expected node0 state: %v, but got: %v", ClusterStateNormal, node0.State())
} else if node1.State() != ClusterStateNormal {
t.Errorf("expected node1 state: %v, but got: %v", ClusterStateNormal, node1.State())
}
expectedTop := &Topology{
NodeIDs: []string{node0.Node.ID, node1.Node.ID},
}
// Verify topology file.
if !reflect.DeepEqual(node0.Topology.NodeIDs, expectedTop.NodeIDs) {
t.Errorf("expected node0 topology: %v, but got: %v", expectedTop.NodeIDs, node0.Topology.NodeIDs)
} else if !reflect.DeepEqual(node1.Topology.NodeIDs, expectedTop.NodeIDs) {
t.Errorf("expected node1 topology: %v, but got: %v", expectedTop.NodeIDs, node1.Topology.NodeIDs)
}
// Bits
// Verify that node-1 contains the fragment (i/f/standard/1) transferred from node-0.
node1Field := node1.Holder.Field("i", "f")
node1View := node1Field.View("standard")
node1Fragment := node1View.Fragment(1)
// Ensure checksums are the same.
if chksum := node1Fragment.Checksum(); !bytes.Equal(chksum, node0Checksum) {
t.Fatalf("expected standard view checksum to match: %x - %x", chksum, node0Checksum)
}
// Close TestCluster.
if err := tc.Close(); err != nil {
t.Fatal(err)
}
})
}
// Ensures that coordinator can be changed.
func TestCluster_UpdateCoordinator(t *testing.T) {
t.Run("UpdateCoordinator", func(t *testing.T) {
c := NewTestCluster(2)
oldNode := c.Nodes[0]
newNode := c.Nodes[1]
// Update coordinator to the same value.
if c.UpdateCoordinator(oldNode) {
t.Errorf("did not expect coordinator to change")
} else if c.Coordinator != oldNode.ID {
t.Errorf("expected coordinator: %s, but got: %s", c.Coordinator, oldNode.URI)
}
// Update coordinator to a new value.
if !c.UpdateCoordinator(newNode) {
t.Errorf("expected coordinator to change")
} else if c.Coordinator != newNode.ID {
t.Errorf("expected coordinator: %s, but got: %s", c.Coordinator, newNode.URI)
}
})
}

8
executor.go
View file

@ -1002,7 +1002,7 @@ func (e *Executor) executeClearBit(ctx context.Context, index string, c *pql.Cal
func (e *Executor) executeClearBitView(ctx context.Context, index string, c *pql.Call, f *Field, view string, colID, rowID uint64, opt *ExecOptions) (bool, error) {
slice := colID / SliceWidth
ret := false
for _, node := range e.Cluster.SliceNodes(index, slice) {
for _, node := range e.Cluster.sliceNodes(index, slice) {
// Update locally if host matches.
if node.ID == e.Node.ID {
val, err := f.ClearBit(view, rowID, colID, nil)
@ -1078,7 +1078,7 @@ func (e *Executor) executeSetBitView(ctx context.Context, index string, c *pql.C
slice := colID / SliceWidth
ret := false
for _, node := range e.Cluster.SliceNodes(index, slice) {
for _, node := range e.Cluster.sliceNodes(index, slice) {
// Update locally if host matches.
if node.ID == e.Node.ID {
val, err := f.SetBit(view, rowID, colID, timestamp)
@ -1414,7 +1414,7 @@ func (e *Executor) slicesByNode(nodes []*Node, index string, slices []uint64) (m
loop:
for _, slice := range slices {
for _, node := range e.Cluster.SliceNodes(index, slice) {
for _, node := range e.Cluster.sliceNodes(index, slice) {
if Nodes(nodes).Contains(node) {
m[node] = append(m[node], slice)
continue loop
@ -1444,7 +1444,7 @@ func (e *Executor) mapReduce(ctx context.Context, index string, slices []uint64,
if !opt.Remote {
nodes = Nodes(e.Cluster.Nodes).Clone()
} else {
nodes = []*Node{e.Cluster.nodeByID(e.Node.ID)}
nodes = []*Node{e.Cluster.unprotectedNodeByID(e.Node.ID)}
}
// Start mapping across all primary owners.

66
executor_test.go
View file

@ -38,7 +38,7 @@ func TestExecutor_Execute_Bitmap(t *testing.T) {
t.Fatal(err)
}
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
// Set bits.
if _, err := e.Execute(context.Background(), "i", test.MustParse(``+
@ -87,7 +87,7 @@ func TestExecutor_Execute_Bitmap(t *testing.T) {
t.Fatal(err)
}
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
// Set bits.
if _, err := e.Execute(context.Background(), "i", test.MustParse(``+
@ -113,7 +113,7 @@ func TestExecutor_Execute_Difference(t *testing.T) {
hldr.SetBit("i", "general", 11, 2)
hldr.SetBit("i", "general", 11, 4)
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
if res, err := e.Execute(context.Background(), "i", test.MustParse(`Difference(Bitmap(row=10), Bitmap(row=11))`), nil, nil); err != nil {
t.Fatal(err)
} else if columns := res[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(columns, []uint64{1, 3}) {
@ -127,7 +127,7 @@ func TestExecutor_Execute_Empty_Difference(t *testing.T) {
defer hldr.Close()
hldr.SetBit("i", "general", 10, 1)
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
if res, err := e.Execute(context.Background(), "i", test.MustParse(`Difference()`), nil, nil); err == nil {
t.Fatalf("Empty Difference query should give error, but got %v", res)
}
@ -145,7 +145,7 @@ func TestExecutor_Execute_Intersect(t *testing.T) {
hldr.SetBit("i", "general", 11, 2)
hldr.SetBit("i", "general", 11, SliceWidth+2)
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
if res, err := e.Execute(context.Background(), "i", test.MustParse(`Intersect(Bitmap(row=10), Bitmap(row=11))`), nil, nil); err != nil {
t.Fatal(err)
} else if columns := res[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(columns, []uint64{1, SliceWidth + 2}) {
@ -158,7 +158,7 @@ func TestExecutor_Execute_Empty_Intersect(t *testing.T) {
hldr := test.MustOpenHolder()
defer hldr.Close()
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
if res, err := e.Execute(context.Background(), "i", test.MustParse(`Intersect()`), nil, nil); err == nil {
t.Fatalf("Empty Intersect query should give error, but got %v", res)
}
@ -175,7 +175,7 @@ func TestExecutor_Execute_Union(t *testing.T) {
hldr.SetBit("i", "general", 11, 2)
hldr.SetBit("i", "general", 11, SliceWidth+2)
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
if res, err := e.Execute(context.Background(), "i", test.MustParse(`Union(Bitmap(row=10), Bitmap(row=11))`), nil, nil); err != nil {
t.Fatal(err)
} else if columns := res[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(columns, []uint64{0, 2, SliceWidth + 1, SliceWidth + 2}) {
@ -189,7 +189,7 @@ func TestExecutor_Execute_Empty_Union(t *testing.T) {
defer hldr.Close()
hldr.SetBit("i", "general", 10, 0)
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
if res, err := e.Execute(context.Background(), "i", test.MustParse(`Union()`), nil, nil); err != nil {
t.Fatal(err)
} else if columns := res[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(columns, []uint64{}) {
@ -208,7 +208,7 @@ func TestExecutor_Execute_Xor(t *testing.T) {
hldr.SetBit("i", "general", 11, 2)
hldr.SetBit("i", "general", 11, SliceWidth+2)
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
if res, err := e.Execute(context.Background(), "i", test.MustParse(`Xor(Bitmap(row=10), Bitmap(row=11))`), nil, nil); err != nil {
t.Fatal(err)
} else if columns := res[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(columns, []uint64{0, 2, SliceWidth + 1}) {
@ -224,7 +224,7 @@ func TestExecutor_Execute_Count(t *testing.T) {
hldr.SetBit("i", "f", 10, SliceWidth+1)
hldr.SetBit("i", "f", 10, SliceWidth+2)
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
if res, err := e.Execute(context.Background(), "i", test.MustParse(`Count(Bitmap(row=10, field=f))`), nil, nil); err != nil {
t.Fatal(err)
} else if res[0] != uint64(3) {
@ -240,7 +240,7 @@ func TestExecutor_Execute_SetBit(t *testing.T) {
// set a bit so the view gets created.
hldr.SetBit("i", "f", 1, 0)
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
if n := hldr.Row("i", "f", 11).Count(); n != 0 {
t.Fatalf("unexpected bitmap count: %d", n)
}
@ -284,7 +284,7 @@ func TestExecutor_Execute_SetValue(t *testing.T) {
}
// Set bsiGroup values.
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
if _, err := e.Execute(context.Background(), "i", test.MustParse(`SetValue(col=10, f=25)`), nil, nil); err != nil {
t.Fatal(err)
} else if _, err := e.Execute(context.Background(), "i", test.MustParse(`SetValue(col=100, f=10)`), nil, nil); err != nil {
@ -322,21 +322,21 @@ func TestExecutor_Execute_SetValue(t *testing.T) {
}
t.Run("ErrColumnBSIGroupRequired", func(t *testing.T) {
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
if _, err := e.Execute(context.Background(), "i", test.MustParse(`SetValue(invalid_column_name=10, f=100)`), nil, nil); err == nil || err.Error() != `SetValue() column field 'col' required` {
t.Fatalf("unexpected error: %s", err)
}
})
t.Run("ErrColumnBSIGroupValue", func(t *testing.T) {
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
if _, err := e.Execute(context.Background(), "i", test.MustParse(`SetValue(invalid_column_name="bad_column", f=100)`), nil, nil); err == nil || err.Error() != `SetValue() column field 'col' required` {
t.Fatalf("unexpected error: %s", err)
}
})
t.Run("ErrInvalidBSIGroupValueType", func(t *testing.T) {
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
if _, err := e.Execute(context.Background(), "i", test.MustParse(`SetValue(col=10, f="hello")`), nil, nil); err == nil || err != pilosa.ErrInvalidBSIGroupValueType {
t.Fatalf("unexpected error: %s", err)
}
@ -359,7 +359,7 @@ func TestExecutor_Execute_SetRowAttrs(t *testing.T) {
// Set two attrs on f/10.
// Also set attrs on other bitmaps and fields to test isolation.
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
if _, err := e.Execute(context.Background(), "i", test.MustParse(`SetRowAttrs(row=10, field=f, foo="bar")`), nil, nil); err != nil {
t.Fatal(err)
}
@ -385,7 +385,7 @@ func TestExecutor_Execute_SetRowAttrs(t *testing.T) {
func TestExecutor_Execute_TopN(t *testing.T) {
hldr := test.MustOpenHolder()
defer hldr.Close()
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
// Set columns for rows 0, 10, & 20 across two slices.
if idx, err := hldr.CreateIndex("i", pilosa.IndexOptions{}); err != nil {
@ -437,7 +437,7 @@ func TestExecutor_Execute_TopN_fill(t *testing.T) {
hldr.SetBit("i", "f", 1, SliceWidth)
// Execute query.
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
if result, err := e.Execute(context.Background(), "i", test.MustParse(`TopN(field=f, n=1)`), nil, nil); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(result, []interface{}{[]pilosa.Pair{
@ -471,7 +471,7 @@ func TestExecutor_Execute_TopN_fill_small(t *testing.T) {
hldr.SetBit("i", "f", 4, 3*SliceWidth+1)
// Execute query.
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
if result, err := e.Execute(context.Background(), "i", test.MustParse(`TopN(field=f, n=1)`), nil, nil); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(result, []interface{}{[]pilosa.Pair{
@ -506,7 +506,7 @@ func TestExecutor_Execute_TopN_Src(t *testing.T) {
hldr.MustCreateRankedFragmentIfNotExists("i", "other", pilosa.ViewStandard, 1).RecalculateCache()
// Execute query.
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
if result, err := e.Execute(context.Background(), "i", test.MustParse(`TopN(Bitmap(row=100, field=other), field=f, n=3)`), nil, nil); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(result, []interface{}{[]pilosa.Pair{
@ -530,7 +530,7 @@ func TestExecutor_Execute_TopN_Attr(t *testing.T) {
if err := hldr.Field("i", "f").RowAttrStore().SetAttrs(10, map[string]interface{}{"category": int64(123)}); err != nil {
t.Fatal(err)
}
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
if result, err := e.Execute(context.Background(), "i", test.MustParse(`TopN(field="f", n=1, attrName="category", attrValues=[123])`), nil, nil); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(result, []interface{}{[]pilosa.Pair{
@ -553,7 +553,7 @@ func TestExecutor_Execute_TopN_Attr_Src(t *testing.T) {
if err := hldr.Field("i", "f").RowAttrStore().SetAttrs(10, map[string]interface{}{"category": uint64(123)}); err != nil {
t.Fatal(err)
}
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
if result, err := e.Execute(context.Background(), "i", test.MustParse(`TopN(Bitmap(row=10,field=f),field="f", n=1, attrName="category", attrValues=[123])`), nil, nil); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(result, []interface{}{[]pilosa.Pair{
@ -567,7 +567,7 @@ func TestExecutor_Execute_TopN_Attr_Src(t *testing.T) {
func TestExecutor_Execute_MinMax(t *testing.T) {
hldr := test.MustOpenHolder()
defer hldr.Close()
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
idx, err := hldr.CreateIndex("i", pilosa.IndexOptions{})
if err != nil {
@ -662,7 +662,7 @@ func TestExecutor_Execute_MinMax(t *testing.T) {
func TestExecutor_Execute_Sum(t *testing.T) {
hldr := test.MustOpenHolder()
defer hldr.Close()
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
idx, err := hldr.CreateIndex("i", pilosa.IndexOptions{})
if err != nil {
@ -733,7 +733,7 @@ func TestExecutor_Execute_Sum(t *testing.T) {
func TestExecutor_Execute_BSIGroupRange(t *testing.T) {
hldr := test.MustOpenHolder()
defer hldr.Close()
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
// Create index.
index := hldr.MustCreateIndexIfNotExists("i", pilosa.IndexOptions{})
@ -775,7 +775,7 @@ func TestExecutor_Execute_BSIGroupRange(t *testing.T) {
func TestExecutor_Execute_Range(t *testing.T) {
hldr := test.MustOpenHolder()
defer hldr.Close()
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
idx, err := hldr.CreateIndex("i", pilosa.IndexOptions{})
if err != nil {
@ -955,7 +955,7 @@ func TestExecutor_Execute_Range(t *testing.T) {
// Ensure a remote query can return a row.
func TestExecutor_Execute_Remote_Row(t *testing.T) {
c := test.NewCluster(2)
c := pilosa.NewTestCluster(2)
// Create secondary server and update second cluster node.
s := test.NewServer()
@ -1003,7 +1003,7 @@ func TestExecutor_Execute_Remote_Row(t *testing.T) {
// Ensure a remote query can return a count.
func TestExecutor_Execute_Remote_Count(t *testing.T) {
c := test.NewCluster(2)
c := pilosa.NewTestCluster(2)
// Create secondary server and update second cluster node.
s := test.NewServer()
@ -1038,7 +1038,7 @@ func TestExecutor_Execute_Remote_Count(t *testing.T) {
// Ensure a remote query can set columns on multiple nodes.
func TestExecutor_Execute_Remote_SetBit(t *testing.T) {
c := test.NewCluster(2)
c := pilosa.NewTestCluster(2)
c.ReplicaN = 2
// Create secondary server and update second cluster node.
@ -1090,7 +1090,7 @@ func TestExecutor_Execute_Remote_SetBit(t *testing.T) {
// Ensure a remote query can set columns on multiple nodes.
func TestExecutor_Execute_Remote_SetBit_With_Timestamp(t *testing.T) {
c := test.NewCluster(2)
c := pilosa.NewTestCluster(2)
c.ReplicaN = 2
// Create secondary server and update second cluster node.
@ -1144,7 +1144,7 @@ func TestExecutor_Execute_Remote_SetBit_With_Timestamp(t *testing.T) {
// Ensure a remote query can return a top-n query.
func TestExecutor_Execute_Remote_TopN(t *testing.T) {
c := test.NewCluster(2)
c := pilosa.NewTestCluster(2)
// Create secondary server and update second cluster node.
s := test.NewServer()
@ -1213,7 +1213,7 @@ func TestExecutor_Execute_Remote_TopN(t *testing.T) {
func TestExecutor_Execute_ErrMaxWritesPerRequest(t *testing.T) {
hldr := test.MustOpenHolder()
defer hldr.Close()
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
e.MaxWritesPerRequest = 3
if _, err := e.Execute(context.Background(), "i", test.MustParse(`SetBit() ClearBit() SetBit() SetBit()`), nil, nil); err != pilosa.ErrTooManyWrites {
t.Fatalf("unexpected error: %s", err)
@ -1229,7 +1229,7 @@ func TestExectutor_SetColumnAttrs_ExcludeField(t *testing.T) {
targetAttrs := map[string]interface{}{
"foo": "bar",
}
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
// SetColumnAttrs call should exclude the field attribute
_, err := e.Execute(context.Background(), "i", test.MustParse("SetBit(field='f', row=1, col=10)"), nil, nil)

4
fragment.go
View file

@ -1740,7 +1740,7 @@ func (s *FragmentSyncer) isClosing() bool {
// then merges any blocks which have differences.
func (s *FragmentSyncer) syncFragment() error {
// Determine replica set.
nodes := s.Cluster.SliceNodes(s.Fragment.index, s.Fragment.slice)
nodes := s.Cluster.sliceNodes(s.Fragment.index, s.Fragment.slice)
if len(nodes) == 1 {
return nil
}
@ -1821,7 +1821,7 @@ func (s *FragmentSyncer) syncBlock(id int) error {
// Read pairs from each remote block.
var uris []*URI
var pairSets []pairSet
for _, node := range s.Cluster.SliceNodes(f.index, f.slice) {
for _, node := range s.Cluster.sliceNodes(f.index, f.slice) {
if s.Node.ID == node.ID {
continue
}

4
holder.go
View file

@ -619,7 +619,7 @@ func (s *HolderSyncer) SyncHolder() error {
for slice := uint64(0); slice <= s.Holder.Index(di.Name).MaxSlice(); slice++ {
// Ignore slices that this host doesn't own.
if !s.Cluster.OwnsSlice(s.Node.ID, di.Name, slice) {
if !s.Cluster.ownsSlice(s.Node.ID, di.Name, slice) {
continue
}
@ -799,7 +799,7 @@ func (c *HolderCleaner) CleanHolder() error {
}
// Get the fragments that node is responsible for (based on hash(index, node)).
containedSlices := c.Cluster.ContainsSlices(index.Name(), index.MaxSlice(), c.Node)
containedSlices := c.Cluster.containsSlices(index.Name(), index.MaxSlice(), c.Node)
// Get the fragments registered in memory.
for _, field := range index.Fields() {

6
holder_test.go
View file

@ -383,7 +383,7 @@ func TestHolderSyncer_SyncHolder(t *testing.T) {
// Mock 2-node, fully replicated cluster.
cluster.ReplicaN = 2
cluster.Nodes[0].URI = test.NewURIFromHostPort("localhost", 0)
cluster.Nodes[0].URI = pilosa.NewTestURIFromHostPort("localhost", 0)
cluster.Nodes[1].URI = *uri
// Create fields on nodes.
@ -456,7 +456,7 @@ func TestHolderSyncer_SyncHolder(t *testing.T) {
// Ensure holder can clean up orphaned fragments.
func TestHolderCleaner_CleanHolder(t *testing.T) {
cluster := test.NewCluster(2)
cluster := pilosa.NewTestCluster(2)
// Create a local holder.
hldr0 := test.MustOpenHolder()
@ -465,7 +465,7 @@ func TestHolderCleaner_CleanHolder(t *testing.T) {
// Mock 2-node, fully replicated cluster.
cluster.ReplicaN = 2
cluster.Nodes[0].URI = test.NewURIFromHostPort("localhost", 0)
cluster.Nodes[0].URI = pilosa.NewTestURIFromHostPort("localhost", 0)
// Create fields on nodes.
for _, hldr := range []*test.Holder{hldr0} {

11
http/client_test.go
View file

@ -87,10 +87,17 @@ func TestClient_MultiNode(t *testing.T) {
// Create a dispersed set of bitmaps across 3 nodes such that each individual node and slice width increment would reveal a different TopN.
sliceNums := []uint64{1, 2, 6}
// This was generated with: `owns := s[i].Handler.Handler.API.Cluster.OwnsSlices("i", 20, s[i].HostURI())`
owns := [][]uint64{
{1, 3, 4, 8, 10, 13, 17, 19},
{2, 5, 7, 11, 12, 14, 18},
{0, 6, 9, 15, 16, 20},
}
for i, num := range sliceNums {
owns := s[i].Handler.Handler.API.Cluster.OwnsSlices("i", 20, s[i].HostURI())
ownsNum := false
for _, ownNum := range owns {
for _, ownNum := range owns[i] {
if ownNum == num {
ownsNum = true
break

22
server.go
View file

@ -332,7 +332,7 @@ func (s *Server) Open() error {
}
// Open Cluster management.
if err := s.Cluster.Open(); err != nil {
if err := s.Cluster.open(); err != nil {
return fmt.Errorf("opening Cluster: %v", err)
}
@ -340,7 +340,7 @@ func (s *Server) Open() error {
if err := s.Holder.Open(); err != nil {
return fmt.Errorf("opening Holder: %v", err)
}
if err := s.Cluster.SetNodeState(NodeStateReady); err != nil {
if err := s.Cluster.setNodeState(NodeStateReady); err != nil {
return fmt.Errorf("setting nodeState: %v", err)
}
@ -349,7 +349,7 @@ func (s *Server) Open() error {
// the cluster without waiting for data to load on the coordinator. Before
// this starts, the joins are queued up in the Cluster.joiningLeavingNodes
// buffered channel.
s.Cluster.ListenForJoins()
s.Cluster.listenForJoins()
// Start background monitoring.
s.wg.Add(3)
@ -370,7 +370,7 @@ func (s *Server) Close() error {
s.ln.Close()
}
if s.Cluster != nil {
s.Cluster.Close()
s.Cluster.close()
}
if s.Holder != nil {
s.Holder.Close()
@ -493,26 +493,26 @@ func (s *Server) ReceiveMessage(pb proto.Message) error {
return err
}
case *internal.ClusterStatus:
err := s.Cluster.MergeClusterStatus(obj)
err := s.Cluster.mergeClusterStatus(obj)
if err != nil {
return err
}
case *internal.ResizeInstruction:
err := s.Cluster.FollowResizeInstruction(obj)
err := s.Cluster.followResizeInstruction(obj)
if err != nil {
return err
}
case *internal.ResizeInstructionComplete:
err := s.Cluster.MarkResizeInstructionComplete(obj)
err := s.Cluster.markResizeInstructionComplete(obj)
if err != nil {
return err
}
case *internal.SetCoordinatorMessage:
s.Cluster.SetCoordinator(DecodeNode(obj.New))
s.Cluster.setCoordinator(DecodeNode(obj.New))
case *internal.UpdateCoordinatorMessage:
s.Cluster.UpdateCoordinator(DecodeNode(obj.New))
s.Cluster.updateCoordinator(DecodeNode(obj.New))
case *internal.NodeStateMessage:
err := s.Cluster.ReceiveNodeState(obj.NodeID, obj.State)
err := s.Cluster.receiveNodeState(obj.NodeID, obj.State)
if err != nil {
return err
}
@ -650,7 +650,7 @@ func (s *Server) monitorDiagnostics() {
s.diagnostics.Logger = s.logger
s.diagnostics.SetVersion(Version)
s.diagnostics.Set("Host", s.URI.host)
s.diagnostics.Set("Cluster", strings.Join(s.Cluster.NodeIDs(), ","))
s.diagnostics.Set("Cluster", strings.Join(s.Cluster.nodeIDs(), ","))
s.diagnostics.Set("NumNodes", len(s.Cluster.Nodes))
s.diagnostics.Set("NumCPU", runtime.NumCPU())
s.diagnostics.Set("NodeID", s.NodeID)

8
stats_test.go
View file

@ -95,7 +95,7 @@ func TestStatsCount_TopN(t *testing.T) {
// Execute query.
called := false
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
e.Holder.Stats = &MockStats{
mockCountWithTags: func(name string, value int64, rate float64, tags []string) {
if name != "TopN" {
@ -124,7 +124,7 @@ func TestStatsCount_Bitmap(t *testing.T) {
hldr.SetBit("d", "f", 0, 0)
hldr.SetBit("d", "f", 0, 1)
called := false
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
e.Holder.Stats = &MockStats{
mockCountWithTags: func(name string, value int64, rate float64, tags []string) {
if name != "Bitmap" {
@ -154,7 +154,7 @@ func TestStatsCount_SetColumnAttrs(t *testing.T) {
hldr.SetBit("d", "f", 10, 1)
called := false
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
field := e.Holder.Field("d", "f")
if field == nil {
t.Fatal("field not found")
@ -184,7 +184,7 @@ func TestStatsCount_SetProfileAttrs(t *testing.T) {
hldr.SetBit("d", "f", 10, 1)
called := false
e := test.NewExecutor(hldr.Holder, test.NewCluster(1))
e := test.NewExecutor(hldr.Holder, pilosa.NewTestCluster(1))
idx := e.Holder.Index("d")
if idx == nil {
t.Fatal("idex not found")

378
test/cluster.go
View file

@ -15,17 +15,10 @@
package test
import (
"bufio"
"bytes"
"fmt"
"io/ioutil"
"path/filepath"
"sync"
"time"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/internal"
)
// NewCluster returns a cluster with n nodes and uses a mod-based hasher.
@ -37,14 +30,14 @@ func NewCluster(n int) *pilosa.Cluster {
c := pilosa.NewCluster()
c.ReplicaN = 1
c.Hasher = NewModHasher()
c.Hasher = newModHasher()
c.Path = path
c.Topology = pilosa.NewTopology()
for i := 0; i < n; i++ {
c.Nodes = append(c.Nodes, &pilosa.Node{
ID: fmt.Sprintf("node%d", i),
URI: NewURI("http", fmt.Sprintf("host%d", i), uint16(0)),
URI: newURI("http", fmt.Sprintf("host%d", i), uint16(0)),
})
}
@ -55,372 +48,19 @@ func NewCluster(n int) *pilosa.Cluster {
return c
}
// ModHasher represents a simple, mod-based hashing.
type ModHasher struct{}
// modHasher represents a simple, mod-based hashing.
type modHasher struct{}
// NewModHasher returns a new instance of ModHasher with n buckets.
func NewModHasher() *ModHasher { return &ModHasher{} }
// newModHasher returns a new instance of ModHasher with n buckets.
func newModHasher() *modHasher { return &modHasher{} }
func (*ModHasher) Hash(key uint64, n int) int { return int(key) % n }
func (*modHasher) Hash(key uint64, n int) int { return int(key) % n }
// ConstHasher represents hash that always returns the same index.
type ConstHasher struct {
i int
}
// NewConstHasher returns a new instance of ConstHasher that always returns i.
func NewConstHasher(i int) *ConstHasher { return &ConstHasher{i: i} }
func (h *ConstHasher) Hash(key uint64, n int) int { return h.i }
// NewURI is a test URI creator that intentionally swallows errors.
func NewURI(scheme, host string, port uint16) pilosa.URI {
// newURI is a test URI creator that intentionally swallows errors.
func newURI(scheme, host string, port uint16) pilosa.URI {
uri := pilosa.DefaultURI()
uri.SetScheme(scheme)
uri.SetHost(host)
uri.SetPort(port)
return *uri
}
func NewURIFromHostPort(host string, port uint16) pilosa.URI {
uri := pilosa.DefaultURI()
uri.SetHost(host)
uri.SetPort(port)
return *uri
}
// TestCluster represents a cluster of test nodes, each of which
// has a pilosa.Cluster.
type TestCluster struct {
Clusters []*pilosa.Cluster
common *commonClusterSettings
mu sync.RWMutex
resizing bool
resizeDone chan struct{}
}
type commonClusterSettings struct {
Nodes []*pilosa.Node
}
func (t *TestCluster) CreateIndex(name string) error {
for _, c := range t.Clusters {
if _, err := c.Holder.CreateIndexIfNotExists(name, pilosa.IndexOptions{}); err != nil {
return err
}
}
return nil
}
func (t *TestCluster) CreateField(index, field string, opt pilosa.FieldOptions) error {
for _, c := range t.Clusters {
idx, err := c.Holder.CreateIndexIfNotExists(index, pilosa.IndexOptions{})
if err != nil {
return err
}
if _, err := idx.CreateField(field, opt); err != nil {
return err
}
}
return nil
}
func (t *TestCluster) SetBit(index, field, view string, rowID, colID uint64, x *time.Time) error {
// Determine which node should receive the SetBit.
c0 := t.Clusters[0] // use the first node's cluster to determine slice location.
slice := colID / pilosa.SliceWidth
nodes := c0.SliceNodes(index, slice)
for _, node := range nodes {
c := t.clusterByID(node.ID)
if c == nil {
continue
}
f := c.Holder.Field(index, field)
if f == nil {
return fmt.Errorf("index/field does not exist: %s/%s", index, field)
}
_, err := f.SetBit(view, rowID, colID, x)
if err != nil {
return err
}
}
return nil
}
func (t *TestCluster) clusterByID(id string) *pilosa.Cluster {
for _, c := range t.Clusters {
if c.Node.ID == id {
return c
}
}
return nil
}
// AddNode adds a node to the cluster and (potentially) starts a resize job.
func (t *TestCluster) AddNode(saveTopology bool) error {
id := len(t.Clusters)
c, err := t.addCluster(id, saveTopology)
if err != nil {
return err
}
// Send NodeJoin event to coordinator.
if id > 0 {
coord := t.Clusters[0]
ev := &pilosa.NodeEvent{
Event: pilosa.NodeJoin,
Node: c.Node,
}
if err := coord.ReceiveEvent(ev); err != nil {
return err
}
// Wait for the AddNode job to finish.
if c.State() != pilosa.ClusterStateNormal {
t.resizeDone = make(chan struct{})
t.mu.Lock()
t.resizing = true
t.mu.Unlock()
<-t.resizeDone
}
}
return nil
}
// WriteTopology writes the given topology to disk.
func (t *TestCluster) WriteTopology(path string, top *pilosa.Topology) error {
if buf, err := proto.Marshal(top.Encode()); err != nil {
return err
} else if err := ioutil.WriteFile(filepath.Join(path, ".topology"), buf, 0666); err != nil {
return err
}
return nil
}
func (t *TestCluster) addCluster(i int, saveTopology bool) (*pilosa.Cluster, error) {
id := fmt.Sprintf("node%d", i)
uri := NewURI("http", fmt.Sprintf("host%d", i), uint16(0))
node := &pilosa.Node{
ID: id,
URI: uri,
}
// add URI to common
//t.common.NodeIDs = append(t.common.NodeIDs, id)
//sort.Sort(t.common.NodeIDs)
// add node to common
t.common.Nodes = append(t.common.Nodes, node)
// create node-specific temp directory
path, err := ioutil.TempDir("", fmt.Sprintf("pilosa-cluster-node-%d-", i))
if err != nil {
return nil, err
}
// holder
h := pilosa.NewHolder()
h.Path = path
// cluster
c := pilosa.NewCluster()
c.ReplicaN = 1
c.Hasher = NewModHasher()
c.Path = path
c.Topology = pilosa.NewTopology()
c.Holder = h
c.MemberSet = pilosa.NewStaticMemberSet(c.Nodes)
c.Node = node
c.Coordinator = t.common.Nodes[0].ID // the first node is the coordinator
c.Broadcaster = t
// add nodes
if saveTopology {
for _, n := range t.common.Nodes {
c.AddNode(n)
}
}
// Add this node to the TestCluster.
t.Clusters = append(t.Clusters, c)
return c, nil
}
// NewTestCluster returns a new instance of test.Cluster.
func NewTestCluster(n int) *TestCluster {
tc := &TestCluster{
common: &commonClusterSettings{},
}
// add clusters
for i := 0; i < n; i++ {
_, err := tc.addCluster(i, true)
if err != nil {
panic(err)
}
}
return tc
}
// SetState sets the state of the cluster on each node.
func (t *TestCluster) SetState(state string) {
for _, c := range t.Clusters {
c.SetState(state)
}
}
// Open opens all clusters in the test cluster.
func (t *TestCluster) Open() error {
for _, c := range t.Clusters {
if err := c.Open(); err != nil {
return err
}
if err := c.Holder.Open(); err != nil {
return err
}
if err := c.SetNodeState(pilosa.NodeStateReady); err != nil {
return err
}
}
// Start the listener on the coordinator.
if len(t.Clusters) == 0 {
return nil
}
t.Clusters[0].ListenForJoins()
return nil
}
// Close closes all clusters in the test cluster.
func (t *TestCluster) Close() error {
for _, c := range t.Clusters {
err := c.Close()
if err != nil {
return err
}
}
return nil
}
// TestCluster implements Broadcaster interface.
// SendSync is a test implemenetation of Broadcaster SendSync method.
func (t *TestCluster) SendSync(pb proto.Message) error {
switch obj := pb.(type) {
case *internal.ClusterStatus:
// Apply the send message to all nodes (except the coordinator).
for _, c := range t.Clusters {
c.MergeClusterStatus(obj)
}
t.mu.RLock()
if obj.State == pilosa.ClusterStateNormal && t.resizing {
close(t.resizeDone)
}
t.mu.RUnlock()
}
return nil
}
// SendAsync is a test implemenetation of Broadcaster SendAsync method.
func (t *TestCluster) SendAsync(pb proto.Message) error {
return nil
}
// SendTo is a test implemenetation of Broadcaster SendTo method.
func (t *TestCluster) SendTo(to *pilosa.Node, pb proto.Message) error {
switch obj := pb.(type) {
case *internal.ResizeInstruction:
err := t.FollowResizeInstruction(obj)
if err != nil {
return err
}
case *internal.ResizeInstructionComplete:
coord := t.clusterByID(to.ID)
go coord.MarkResizeInstructionComplete(obj)
}
return nil
}
// FollowResizeInstruction is a version of cluster.FollowResizeInstruction used for testing.
func (t *TestCluster) FollowResizeInstruction(instr *internal.ResizeInstruction) error {
// Prepare the return message.
complete := &internal.ResizeInstructionComplete{
JobID: instr.JobID,
Node: instr.Node,
Error: "",
}
// Stop processing on any error.
if err := func() error {
// figure out which node it was meant for, then call the operation on that cluster
// basically need to mimic this: client.RetrieveSliceFromURI(context.Background(), src.Index, src.Field, src.View, src.Slice, srcURI)
instrNode := pilosa.DecodeNode(instr.Node)
destCluster := t.clusterByID(instrNode.ID)
// Sync the schema received in the resize instruction.
if err := destCluster.Holder.ApplySchema(instr.Schema); err != nil {
return err
}
for _, src := range instr.Sources {
srcNode := pilosa.DecodeNode(src.Node)
srcCluster := t.clusterByID(srcNode.ID)
srcFragment := srcCluster.Holder.Fragment(src.Index, src.Field, src.View, src.Slice)
destFragment := destCluster.Holder.Fragment(src.Index, src.Field, src.View, src.Slice)
if destFragment == nil {
// Create fragment on destination if it doesn't exist.
f := destCluster.Holder.Field(src.Index, src.Field)
v := f.View(src.View)
var err error
destFragment, err = v.CreateFragmentIfNotExists(src.Slice)
if err != nil {
return err
}
}
buf := bytes.NewBuffer(nil)
bw := bufio.NewWriter(buf)
br := bufio.NewReader(buf)
// Get the fragment from source.
if _, err := srcFragment.WriteTo(bw); err != nil {
return err
}
// Flush the bufio.buf to the io.Writer (buf).
bw.Flush()
// Write data to destination.
if _, err := destFragment.ReadFrom(br); err != nil {
return err
}
}
return nil
}(); err != nil {
complete.Error = err.Error()
}
node := pilosa.DecodeNode(instr.Coordinator)
if err := t.SendTo(node, complete); err != nil {
return err
}
return nil
}

16
utils_test.go → utils_internal_test.go
View file

@ -121,7 +121,7 @@ func (t *ClusterCluster) SetBit(index, field, view string, rowID, colID uint64,
// Determine which node should receive the SetBit.
c0 := t.Clusters[0] // use the first node's cluster to determine slice location.
slice := colID / SliceWidth
nodes := c0.SliceNodes(index, slice)
nodes := c0.sliceNodes(index, slice)
for _, node := range nodes {
c := t.clusterByID(node.ID)
@ -236,7 +236,7 @@ func (t *ClusterCluster) addCluster(i int, saveTopology bool) (*Cluster, error)
// add nodes
if saveTopology {
for _, n := range t.common.Nodes {
c.AddNode(n)
c.addNode(n)
}
}
@ -273,13 +273,13 @@ func (t *ClusterCluster) SetState(state string) {
// Open opens all clusters in the test cluster.
func (t *ClusterCluster) Open() error {
for _, c := range t.Clusters {
if err := c.Open(); err != nil {
if err := c.open(); err != nil {
return err
}
if err := c.Holder.Open(); err != nil {
return err
}
if err := c.SetNodeState(NodeStateReady); err != nil {
if err := c.setNodeState(NodeStateReady); err != nil {
return err
}
}
@ -288,7 +288,7 @@ func (t *ClusterCluster) Open() error {
if len(t.Clusters) == 0 {
return nil
}
t.Clusters[0].ListenForJoins()
t.Clusters[0].listenForJoins()
return nil
}
@ -296,7 +296,7 @@ func (t *ClusterCluster) Open() error {
// Close closes all clusters in the test cluster.
func (t *ClusterCluster) Close() error {
for _, c := range t.Clusters {
err := c.Close()
err := c.close()
if err != nil {
return err
}
@ -310,7 +310,7 @@ func (t *ClusterCluster) SendSync(pb proto.Message) error {
case *internal.ClusterStatus:
// Apply the send message to all nodes (except the coordinator).
for _, c := range t.Clusters {
c.MergeClusterStatus(obj)
c.mergeClusterStatus(obj)
}
t.mu.RLock()
if obj.State == ClusterStateNormal && t.resizing {
@ -337,7 +337,7 @@ func (t *ClusterCluster) SendTo(to *Node, pb proto.Message) error {
}
case *internal.ResizeInstructionComplete:
coord := t.clusterByID(to.ID)
go coord.MarkResizeInstructionComplete(obj)
go coord.markResizeInstructionComplete(obj)
}
return nil
}