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

more use of noder; remove c.nodes

Browse source 
disable some of the gossip logic

implement some of the stator logic
This commit is contained in:
Travis 2021-01-31 12:34:31 -06:00
parent 97eaff5c82
commit 855e1b35f5
No known key found for this signature in database
GPG key ID: 37080CC2042BA34E
17 changed files with 288 additions and 269 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
Makefile
View file

@ -229,7 +229,7 @@ docker-test:
# The \-\-\- FAIL avoids counting the extra two FAIL strings at then bottom of log.topt.
topt:
mv log.topt.roar log.topt.roar.prev || true
$(eval SHELL:=/bin/bash) set -o pipefail; go test -v -tags='$(BUILD_TAGS) $(TEST_TAGS)' $(TESTFLAGS) 2>&1 | tee log.topt.roar
$(eval SHELL:=/bin/bash) set -o pipefail; go test -v -timeout 60m -tags='$(BUILD_TAGS) $(TEST_TAGS)' $(TESTFLAGS) 2>&1 | tee log.topt.roar
@echo " log.topt.roar green: \c"; cat log.topt.roar | grep PASS |wc -l
@echo " log.topt.roar red: \c"; cat log.topt.roar | grep '\-\-\- FAIL' | wc -l

40
api.go
View file

@ -130,7 +130,10 @@ func appendMap(a, b map[apiMethod]struct{}) map[apiMethod]struct{} {
}
func (api *API) validate(f apiMethod) error {
state := api.cluster.State()
state, err := api.cluster.State()
if err != nil {
return errors.Wrap(err, "getting cluster state")
}
if _, ok := validAPIMethods[state][f]; ok {
return nil
}
@ -207,7 +210,11 @@ func (api *API) CreateIndex(ctx context.Context, indexName string, options Index
return nil, errors.Wrap(err, "validating api method")
}
if !api.holder.isCoordinator() {
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
if !snap.IsPrimaryFieldTranslationNode(api.Node().ID) {
fmt.Println("--- DEBUG: forward to coordinator")
if err := api.server.defaultClient.CreateIndex(ctx, indexName, options); err != nil {
return nil, errors.Wrap(err, "forwarding CreateIndex to coordinator")
}
@ -303,7 +310,10 @@ func (api *API) CreateField(ctx context.Context, indexName string, fieldName str
}
}
if !api.holder.isCoordinator() {
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
if !snap.IsPrimaryFieldTranslationNode(api.Node().ID) {
if err := api.server.defaultClient.CreateFieldWithOptions(ctx, indexName, fieldName, fo); err != nil {
return nil, errors.Wrap(err, "forwarding CreateField to coordinator")
}
@ -834,6 +844,13 @@ func (api *API) Node() *topology.Node {
return api.server.node()
}
// CoordinatorNode returns the coordinator node for the cluster.
func (api *API) CoordinatorNode() *topology.Node {
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
return snap.PrimaryFieldTranslationNode()
}
// NodeUsage represents all usage measurements for one node.
type NodeUsage struct {
Disk DiskUsage `json:"bytesOnDisk"`
@ -1791,7 +1808,7 @@ func (api *API) ResizeAbort() error {
// State returns the cluster state which is usually "NORMAL", but could be
// "STARTING", "RESIZING", or potentially others. See cluster.go for more
// details.
func (api *API) State() string {
func (api *API) State() (string, error) {
return api.cluster.State()
}
@ -2125,7 +2142,10 @@ func (api *API) ReserveIDs(key IDAllocKey, session [32]byte, offset uint64, coun
return nil, errors.Wrap(err, "validating api method")
}
if api.holder.isCoordinator() {
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
if !snap.IsPrimaryFieldTranslationNode(api.Node().ID) {
return api.holder.ida.reserve(key, session, offset, count)
}
@ -2137,7 +2157,10 @@ func (api *API) CommitIDs(key IDAllocKey, session [32]byte, count uint64) error
return errors.Wrap(err, "validating api method")
}
if api.holder.isCoordinator() {
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
if !snap.IsPrimaryFieldTranslationNode(api.Node().ID) {
return api.holder.ida.commit(key, session, count)
}
@ -2149,7 +2172,10 @@ func (api *API) ResetIDAlloc(index string) error {
return errors.Wrap(err, "validating api method")
}
if api.holder.isCoordinator() {
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN)
if !snap.IsPrimaryFieldTranslationNode(api.Node().ID) {
return api.holder.ida.reset(index)
}

1
api_test.go
View file

@ -161,7 +161,6 @@ func TestAPI_ImportColumnAttrs(t *testing.T) {
t.Fatal(err)
}
}
})
}

247
cluster.go
View file

@ -77,9 +77,8 @@ type cluster struct { // nolint: maligned
noder topology.Noder
unprotectedNoder topology.Noder
id string
Node *topology.Node
nodes []*topology.Node
id string
Node *topology.Node
// Hashing algorithm used to assign partitions to nodes.
Hasher topology.Hasher
@ -143,7 +142,7 @@ type cluster struct { // nolint: maligned
// newCluster returns a new instance of Cluster with defaults.
func newCluster() *cluster {
c := &cluster{
return &cluster{
Hasher: &topology.Jmphasher{},
partitionN: topology.DefaultPartitionN,
ReplicaN: 1,
@ -161,40 +160,10 @@ func newCluster() *cluster {
confirmDownRetries: defaultConfirmDownRetries,
confirmDownSleep: defaultConfirmDownSleep,
noder: topology.NewEmptyLocalNoder(),
stator: disco.NopStator,
}
// TODO: these are temporary until etcd fully implements noder
c.noder = c
c.unprotectedNoder = &unprotectedCluster{
c: c,
}
return c
}
// unprotectedCluster is a temporary struct used in cases of NewClusterSnapshot
// which are inside of a c.mu.Lock(). These cases can't use the normal c.noder
// (which is also temporary), because c.Nodes() aquires c.mu.Lock() as well.
type unprotectedCluster struct {
c *cluster
}
// Nodes returns a copy of the slice of nodes in the cluster.
func (uc *unprotectedCluster) Nodes() []*topology.Node {
ret := make([]*topology.Node, len(uc.c.nodes))
copy(ret, uc.c.nodes)
return ret
}
// SetNodes implements the Noder interface.
func (uc *unprotectedCluster) SetNodes(nodes []*topology.Node) {}
// AppendNode implements the Noder interface.
func (uc *unprotectedCluster) AppendNode(node *topology.Node) {}
// RemoveNode implements the Noder interface.
func (uc *unprotectedCluster) RemoveNode(nodeID string) bool {
return false
}
// initializeAntiEntropy is called by the anti entropy routine when it starts.
@ -225,25 +194,25 @@ func (c *cluster) abortAntiEntropy() {
}
func (c *cluster) coordinatorNode() *topology.Node {
c.mu.RLock()
defer c.mu.RUnlock()
return c.unprotectedCoordinatorNode()
}
// unprotectedCoordinatorNode returns the coordinator node.
func (c *cluster) unprotectedCoordinatorNode() *topology.Node {
return c.unprotectedNodeByID(c.Coordinator)
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
return snap.PrimaryFieldTranslationNode()
}
// isCoordinator is true if this node is the coordinator.
func (c *cluster) isCoordinator() bool {
c.mu.RLock()
defer c.mu.RUnlock()
return c.unprotectedIsCoordinator()
}
func (c *cluster) unprotectedIsCoordinator() bool {
return c.Coordinator == c.Node.ID
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
return snap.PrimaryFieldTranslationNode().ID == c.Node.ID
}
// setCoordinator tells the current node to become the
@ -282,7 +251,7 @@ func (c *cluster) setCoordinator(n *topology.Node) error {
// and should be refactored.
func (c *cluster) unprotectedSendSync(m Message) error {
var eg errgroup.Group
for _, node := range c.nodes {
for _, node := range c.noder.Nodes() {
node := node
// Don't send to myself.
if node.ID == c.Node.ID {
@ -309,7 +278,7 @@ func (c *cluster) unprotectedUpdateCoordinator(n *topology.Node) bool {
c.Coordinator = n.ID
changed = true
}
for _, node := range c.nodes {
for _, node := range c.noder.Nodes() {
if node.ID == n.ID {
node.IsCoordinator = true
} else {
@ -365,7 +334,7 @@ func (c *cluster) removeNode(nodeID string) error {
// nodeIDs returns the list of IDs in the cluster.
func (c *cluster) nodeIDs() []string {
return topology.Nodes(c.nodes).IDs()
return topology.Nodes(c.Nodes()).IDs()
}
func (c *cluster) unprotectedSetID(id string) {
@ -379,10 +348,12 @@ func (c *cluster) unprotectedSetID(id string) {
c.Topology.clusterID = c.id
}
func (c *cluster) State() string {
c.mu.RLock()
defer c.mu.RUnlock()
return c.state
func (c *cluster) State() (string, error) {
state, err := c.stator.ClusterState(context.Background())
if err != nil {
return string(disco.ClusterStateUnknown), err
}
return string(state), nil
}
func (c *cluster) SetState(state string) {
@ -456,33 +427,14 @@ func (c *cluster) setMyNodeState(state string) {
c.mu.Lock()
defer c.mu.Unlock()
c.Node.State = state
for i, n := range c.nodes {
nodes := c.noder.Nodes()
for i, n := range nodes {
if n.ID == c.Node.ID {
c.nodes[i].State = state
nodes[i].State = state
}
}
}
func (c *cluster) setNodeState(state string) error { // nolint: unparam
c.setMyNodeState(state)
if c.isCoordinator() {
return c.receiveNodeState(c.Node.ID, state)
}
// Send node state to coordinator.
ns := &NodeStateMessage{
NodeID: c.Node.ID,
State: state,
}
c.logger.Printf("sending state %s (%s)", state, c.Coordinator)
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
// Coordinator to keep track of, during startup, which nodes have
// finished opening their Holder.
@ -498,11 +450,12 @@ func (c *cluster) receiveNodeState(nodeID string, state string) error {
if c.Topology.nodeStates[nodeID] != state {
changed = true
c.Topology.nodeStates[nodeID] = state
for i, n := range c.nodes {
nodes := c.noder.Nodes()
for i, n := range nodes {
if n.ID == nodeID {
c.nodes[i].Mu.Lock()
c.nodes[i].State = state
c.nodes[i].Mu.Unlock()
nodes[i].Mu.Lock()
nodes[i].State = state
nodes[i].Mu.Unlock()
}
}
}
@ -547,7 +500,7 @@ func (c *cluster) unprotectedStatus() *ClusterStatus {
return &ClusterStatus{
ClusterID: c.id,
State: c.state,
Nodes: c.nodes,
Nodes: c.noder.Nodes(),
Schema: &Schema{Indexes: c.holder.Schema()},
}
}
@ -560,7 +513,7 @@ func (c *cluster) nodeByID(id string) *topology.Node {
// unprotectedNodeByID returns a node reference by ID.
func (c *cluster) unprotectedNodeByID(id string) *topology.Node {
for _, n := range c.nodes {
for _, n := range c.noder.Nodes() {
if n.ID == id {
return n
}
@ -581,7 +534,7 @@ func (c *cluster) topologyContainsNode(id string) bool {
// nodePositionByID returns the position of the node in slice c.Nodes.
func (c *cluster) nodePositionByID(nodeID string) int {
for i, n := range c.nodes {
for i, n := range c.noder.Nodes() {
if n.ID == nodeID {
return i
}
@ -609,10 +562,10 @@ func (c *cluster) addNodeBasicSorted(node *topology.Node) bool {
return false
}
c.nodes = append(c.nodes, node)
c.noder.AppendNode(node)
// All hosts must be merged in the same order on all nodes in the cluster.
sort.Sort(topology.ByID(c.nodes))
// sort.Sort(topology.ByID(c.nodes)) // TODO: this should no longer apply
return true
}
@ -620,11 +573,25 @@ func (c *cluster) addNodeBasicSorted(node *topology.Node) bool {
// Nodes returns a copy of the slice of nodes in the cluster. Safe for
// concurrent use, result may be modified.
func (c *cluster) Nodes() []*topology.Node {
c.mu.RLock()
defer c.mu.RUnlock()
ret := make([]*topology.Node, len(c.nodes))
copy(ret, c.nodes)
return ret
nodes := c.noder.Nodes()
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(topology.NewLocalNoder(nodes), c.Hasher, c.ReplicaN)
primaryNode := snap.PrimaryFieldTranslationNode()
// Set node states and IsPrimary.
for _, node := range nodes {
node.IsCoordinator = node.ID == primaryNode.ID
// s, err := c.stator.NodeState(context.Background(), node.ID)
// if err != nil {
// node.State = nodeStateDown
// continue
// }
// node.State = string(s)
}
return nodes
}
func (c *cluster) AllNodeStates() map[string]string {
@ -636,16 +603,7 @@ func (c *cluster) AllNodeStates() map[string]string {
// removeNodeBasicSorted removes a node from the cluster, maintaining the sort
// order. Returns true if the node was removed. unprotected.
func (c *cluster) removeNodeBasicSorted(nodeID string) bool {
i := c.nodePositionByID(nodeID)
if i < 0 {
return false
}
copy(c.nodes[i:], c.nodes[i+1:])
c.nodes[len(c.nodes)-1] = nil
c.nodes = c.nodes[:len(c.nodes)-1]
return true
return c.noder.RemoveNode(nodeID)
}
// frag is a struct of basic fragment information.
@ -699,7 +657,7 @@ func (c *cluster) fragsByHost(idx *Index) fragsByHost {
// for the given set of shards with data.
func (c *cluster) fragCombos(idx string, availableShards *roaring.Bitmap, fieldViews viewsByField) fragsByHost {
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(c.unprotectedNoder, c.Hasher, c.ReplicaN)
snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
t := make(fragsByHost)
_ = availableShards.ForEach(func(i uint64) error {
@ -721,8 +679,10 @@ func (c *cluster) fragCombos(idx string, availableShards *roaring.Bitmap, fieldV
// added or removed. An error is returned for any case other than where
// exactly one node is added or removed. unprotected.
func (c *cluster) diff(other *cluster) (action string, nodeID string, err error) {
lenFrom := len(c.nodes)
lenTo := len(other.nodes)
cNodes := c.noder.Nodes()
otherNodes := other.noder.Nodes()
lenFrom := len(cNodes)
lenTo := len(otherNodes)
// Determine if a node is being added or removed.
if lenFrom == lenTo {
return "", "", errors.New("clusters are the same size")
@ -734,7 +694,7 @@ func (c *cluster) diff(other *cluster) (action string, nodeID string, err error)
}
action = resizeJobActionAdd
// Determine the node ID that is being added.
for _, n := range other.nodes {
for _, n := range otherNodes {
if c.unprotectedNodeByID(n.ID) == nil {
nodeID = n.ID
break
@ -747,7 +707,7 @@ func (c *cluster) diff(other *cluster) (action string, nodeID string, err error)
}
action = resizeJobActionRemove
// Determine the node ID that is being removed.
for _, n := range c.nodes {
for _, n := range cNodes {
if other.unprotectedNodeByID(n.ID) == nil {
nodeID = n.ID
break
@ -769,7 +729,7 @@ func (c *cluster) fragSources(to *cluster, idx *Index) (map[string][]*ResizeSour
}
// Initialize the map with all the nodes in `to`.
for _, n := range to.nodes {
for _, n := range to.noder.Nodes() {
m[n.ID] = nil
}
@ -782,7 +742,7 @@ func (c *cluster) fragSources(to *cluster, idx *Index) (map[string][]*ResizeSour
srcCluster := c
if action == resizeJobActionAdd && c.ReplicaN > 1 {
srcCluster = newCluster()
srcCluster.nodes = topology.Nodes(c.nodes).Clone()
srcCluster.noder.SetNodes(topology.Nodes(c.noder.Nodes()).Clone())
srcCluster.Hasher = c.Hasher
srcCluster.partitionN = c.partitionN
srcCluster.ReplicaN = 1
@ -859,13 +819,13 @@ func (c *cluster) translationNodes(to *cluster) (map[string][]*translationResize
}
// Initialize the map with all the nodes in `to`.
for _, n := range to.nodes {
for _, n := range to.noder.Nodes() {
m[n.ID] = nil
}
// Create a snapshot of the cluster to use for node/partition calculations.
fSnap := topology.NewClusterSnapshot(c.unprotectedNoder, c.Hasher, c.ReplicaN)
toSnap := topology.NewClusterSnapshot(to.unprotectedNoder, c.Hasher, to.ReplicaN)
fSnap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
toSnap := topology.NewClusterSnapshot(to.noder, c.Hasher, to.ReplicaN)
for pid := 0; pid < c.partitionN; pid++ {
fNodes := fSnap.PartitionNodes(pid)
@ -914,7 +874,7 @@ func (c *cluster) translationNodes(to *cluster) (map[string][]*translationResize
func (c *cluster) shardDistributionByIndex(indexName string) map[string]map[string][]uint64 {
dist := make(map[string]map[string][]uint64)
for _, node := range c.nodes {
for _, node := range c.noder.Nodes() {
nodeDist := make(map[string][]uint64)
nodeDist["primary-shards"] = make([]uint64, 0)
nodeDist["replica-shards"] = make([]uint64, 0)
@ -1017,12 +977,14 @@ func (c *cluster) partitionNodes(partitionID int) []*topology.Node {
useTopology = true
}
cNodes := c.noder.Nodes()
replicaN := c.ReplicaN
var nodeN int
if useTopology {
nodeN = len(c.Topology.nodeIDs)
} else {
nodeN = len(c.nodes)
nodeN = len(cNodes)
}
if replicaN > nodeN {
replicaN = nodeN
@ -1044,11 +1006,11 @@ func (c *cluster) partitionNodes(partitionID int) []*topology.Node {
for i := 0; i < replicaN; i++ {
if useTopology {
maybeNodeID := c.Topology.nodeIDs[(nodeIndex+i)%nodeN]
if node := topology.Nodes(c.nodes).NodeByID(maybeNodeID); node != nil {
if node := topology.Nodes(cNodes).NodeByID(maybeNodeID); node != nil {
nodes = append(nodes, node)
}
} else {
nodes = append(nodes, c.nodes[(nodeIndex+i)%len(c.nodes)])
nodes = append(nodes, cNodes[(nodeIndex+i)%len(cNodes)])
}
}
@ -1078,7 +1040,7 @@ func (t *Topology) PrimaryNodeIndex(partitionID int) (nodeIndex int) {
n := len(t.nodeIDs)
if n == 0 {
if t.cluster != nil {
n = len(t.cluster.nodes)
n = len(t.cluster.noder.Nodes())
}
}
nodeIndex = t.Hasher.Hash(uint64(partitionID), n)
@ -1178,28 +1140,6 @@ func (c *cluster) open() error {
}
func (c *cluster) waitForStarted() error {
// If not coordinator then wait for ClusterStatus from coordinator.
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 coordinator needs to be alerted that this node is back up
// (and now in a state of STARTING) so that it can be put to the correct
// cluster state.
// TODO: Because the normal code path already sends a NodeJoin event (via
// memberlist), this is a bit redundant in most cases. Perhaps determine
// that the node has been restarted and don't do this step.
msg := &NodeEvent{
Event: NodeJoin,
Node: c.Node,
}
if err := c.broadcaster.SendSync(msg); err != nil {
return fmt.Errorf("sending restart NodeJoin: %v", err)
}
c.logger.Printf("%v wait for joining to complete", c.Node.ID)
<-c.joining
c.logger.Printf("joining has completed. I am NodeID '%v'", c.Node.ID)
}
return nil
}
@ -1220,7 +1160,7 @@ func (c *cluster) markAsJoined() {
// needTopologyAgreement is unprotected.
func (c *cluster) needTopologyAgreement() bool {
return (c.state == ClusterStateStarting || c.state == ClusterStateDegraded) && !stringSlicesAreEqual(c.Topology.nodeIDs, c.nodeIDs())
return false
}
// haveTopologyAgreement is unprotected.
@ -1405,7 +1345,7 @@ func (c *cluster) unprotectedGenerateResizeJob(nodeAction nodeAction) (*resizeJo
// Broadcaster is associated to the resizeJob here for use in broadcasting
// the resize instructions to other nodes in the cluster.
func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (*resizeJob, error) {
j := newResizeJob(c.nodes, nodeAction.node, nodeAction.action)
j := newResizeJob(c.noder.Nodes(), nodeAction.node, nodeAction.action)
// A *new* node which is being added needs a schema update even if
// there's no data to send it.
var sendSchemaToNewNode string
@ -1413,7 +1353,7 @@ func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (*
// toCluster is a clone of Cluster with the new node added/removed for comparison.
toCluster := newCluster()
toCluster.nodes = topology.Nodes(c.nodes).Clone()
toCluster.noder.SetNodes(topology.Nodes(c.noder.Nodes()).Clone())
toCluster.Hasher = c.Hasher
toCluster.partitionN = c.partitionN
toCluster.ReplicaN = c.ReplicaN
@ -1429,7 +1369,7 @@ func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (*
// fragmentSourcesByNode is a map of Node.ID to sources of fragment data.
// It is initialized with all the nodes in toCluster.
fragmentSourcesByNode := make(map[string][]*ResizeSource)
for _, n := range toCluster.nodes {
for _, n := range toCluster.noder.Nodes() {
fragmentSourcesByNode[n.ID] = nil
}
@ -1449,7 +1389,7 @@ func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (*
// key translation data for indexes.
// It is initialized with all the nodes in toCluster.
translationSourcesByNode := make(map[string][]*TranslationResizeSource)
for _, n := range toCluster.nodes {
for _, n := range toCluster.noder.Nodes() {
translationSourcesByNode[n.ID] = nil
}
@ -1486,7 +1426,7 @@ func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (*
}
}
for _, node := range toCluster.nodes {
for _, node := range toCluster.noder.Nodes() {
dataToSend := len(fragmentSourcesByNode[node.ID]) != 0 || len(translationSourcesByNode[node.ID]) != 0
// If we're adding a new node, that node needs to get a resize
// instruction even if there's no data it needs to read.
@ -1498,7 +1438,7 @@ func (c *cluster) unprotectedGenerateResizeJobByAction(nodeAction nodeAction) (*
}
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(c.unprotectedNoder, c.Hasher, c.ReplicaN)
snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
instr := &ResizeInstruction{
JobID: j.ID,
@ -1525,7 +1465,7 @@ func (c *cluster) completeCurrentJob(state string) error {
func (c *cluster) unprotectedCompleteCurrentJob(state string) error {
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(c.unprotectedNoder, c.Hasher, c.ReplicaN)
snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
if !snap.IsPrimaryFieldTranslationNode(c.Node.ID) {
return ErrNodeNotCoordinator
}
@ -2373,15 +2313,6 @@ func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error {
// Add all nodes from the coordinator.
for _, node := range officialNodes {
if node.ID == c.Node.ID && node.State != c.Node.State {
c.logger.Printf("mismatched state in mergeClusterStatus got %v have %v", node.State, c.Node.State)
go func(fromState, toState string) {
err := c.setNodeState(toState)
if err != nil {
c.logger.Printf("error setting node state from %v to %v: %v", fromState, toState, err)
}
}(node.State, c.Node.State)
}
if err := c.addNode(node); err != nil {
return errors.Wrap(err, "adding node")
}
@ -2391,7 +2322,7 @@ func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error {
// except for self. Generate a list to remove first
// so that nodes aren't removed mid-loop.
nodeIDsToRemove := []string{}
for _, node := range c.nodes {
for _, node := range c.noder.Nodes() {
// Don't remove this node.
if node.ID == c.Node.ID {
continue
@ -2419,7 +2350,8 @@ func (c *cluster) mergeClusterStatus(cs *ClusterStatus) error {
// If there is only one node in the cluster, returns nil.
// If the current node is the first node in the list, returns the last node.
func (c *cluster) unprotectedPreviousNode() *topology.Node {
if len(c.nodes) <= 1 {
cNodes := c.noder.Nodes()
if len(cNodes) <= 1 {
return nil
}
@ -2427,9 +2359,9 @@ func (c *cluster) unprotectedPreviousNode() *topology.Node {
if pos == -1 {
return nil
} else if pos == 0 {
return c.nodes[len(c.nodes)-1]
return cNodes[len(cNodes)-1]
} else {
return c.nodes[pos-1]
return cNodes[pos-1]
}
}
@ -2446,7 +2378,8 @@ func (c *cluster) unprotectedPrimaryReplicaNode() *topology.Node {
if pos <= 0 {
return nil
}
return c.nodes[pos-1]
cNodes := c.noder.Nodes()
return cNodes[pos-1]
}
// translateFieldKeys is basically a wrapper around
@ -2484,7 +2417,7 @@ func (c *cluster) findFieldKeys(ctx context.Context, field *Field, keys ...strin
}
if !field.Keys() {
return nil, errors.Wrap(ErrTranslatingKeyNotFound, "field is not keyed 1")
return nil, errors.Wrap(ErrTranslatingKeyNotFound, "field is not keyed")
}
// Attempt to find the keys locally.
@ -2547,7 +2480,7 @@ func (c *cluster) createFieldKeys(ctx context.Context, field *Field, keys ...str
}
if !field.Keys() {
return nil, errors.Wrap(ErrTranslatingKeyNotFound, "field is not keyed 2")
return nil, errors.Wrap(ErrTranslatingKeyNotFound, "field is not keyed")
}
// The coordinator is the only node that can create field keys, since it owns the authoritative copy.

110
cluster_internal_test.go
View file

@ -419,22 +419,24 @@ func TestResizeJob(t *testing.T) {
// Ensure the cluster can fairly distribute partitions across the nodes.
func TestCluster_Owners(t *testing.T) {
c := cluster{
nodes: []*topology.Node{
noder: topology.NewLocalNoder([]*topology.Node{
{URI: NewTestURIFromHostPort("serverA", 1000)},
{URI: NewTestURIFromHostPort("serverB", 1000)},
{URI: NewTestURIFromHostPort("serverC", 1000)},
},
}),
Hasher: NewTestModHasher(),
ReplicaN: 2,
}
cNodes := c.noder.Nodes()
// Verify nodes are distributed.
if a := c.partitionNodes(0); !reflect.DeepEqual(a, []*topology.Node{c.nodes[0], c.nodes[1]}) {
if a := c.partitionNodes(0); !reflect.DeepEqual(a, []*topology.Node{cNodes[0], cNodes[1]}) {
t.Fatalf("unexpected owners: %s", spew.Sdump(a))
}
// Verify nodes go around the ring.
if a := c.partitionNodes(2); !reflect.DeepEqual(a, []*topology.Node{c.nodes[2], c.nodes[0]}) {
if a := c.partitionNodes(2); !reflect.DeepEqual(a, []*topology.Node{cNodes[2], cNodes[0]}) {
t.Fatalf("unexpected owners: %s", spew.Sdump(a))
}
}
@ -487,7 +489,8 @@ func TestHasher(t *testing.T) {
func TestCluster_ContainsShards(t *testing.T) {
c := NewTestCluster(t, 5)
c.ReplicaN = 3
shards := c.containsShards("test", roaring.NewBitmap(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10), c.nodes[2])
cNodes := c.noder.Nodes()
shards := c.containsShards("test", roaring.NewBitmap(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10), cNodes[2])
if !reflect.DeepEqual(shards, []uint64{0, 2, 3, 5, 6, 9, 10}) {
t.Fatalf("unexpected shars for node's index: %v", shards)
@ -627,13 +630,16 @@ func TestCluster_Coordinator(t *testing.T) {
node1 := &topology.Node{ID: "node1", URI: uris[0]}
node2 := &topology.Node{ID: "node2", URI: uris[1]}
noder := topology.NewLocalNoder([]*topology.Node{node1, node2})
c1 := *newCluster()
c1.Node = node1
c1.Coordinator = node1.ID
c1.noder = noder
c2 := *newCluster()
c2.Node = node2
c2.Coordinator = node1.ID
c2.noder = noder
t.Run("IsCoordinator", func(t *testing.T) {
if !c1.isCoordinator() {
@ -697,7 +703,7 @@ func TestCluster_Topology(t *testing.T) {
// Ensure that general cluster functionality works as expected.
func TestCluster_ResizeStates(t *testing.T) {
t.Skip("these tests don't really apply anymore; they were meant to tests the cluster startup process using memberlist and a topology file")
t.Run("Single node, no data", func(t *testing.T) {
tc := NewClusterCluster(t, 1)
@ -708,9 +714,14 @@ func TestCluster_ResizeStates(t *testing.T) {
node := tc.Clusters[0]
state, err := node.State()
if 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())
if state != ClusterStateNormal {
t.Errorf("expected state: %v, but got: %v", ClusterStateNormal, state)
}
expectedTop := &Topology{
@ -749,9 +760,14 @@ func TestCluster_ResizeStates(t *testing.T) {
t.Fatal(err)
}
state, err := node.State()
if 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())
if state != ClusterStateNormal {
t.Errorf("expected state: %v, but got: %v", ClusterStateNormal, state)
}
// Close TestCluster.
@ -805,13 +821,22 @@ func TestCluster_ResizeStates(t *testing.T) {
}
node0 := tc.Clusters[0]
state0, err := node0.State()
if err != nil {
t.Fatal(err)
}
node1 := tc.Clusters[1]
state1, err := node1.State()
if err != nil {
t.Fatal(err)
}
// 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())
if state0 != ClusterStateNormal {
t.Errorf("expected node0 state: %v, but got: %v", ClusterStateNormal, state0)
} else if state1 != ClusterStateNormal {
t.Errorf("expected node1 state: %v, but got: %v", ClusterStateNormal, state1)
}
expectedTop := &Topology{
@ -851,27 +876,30 @@ func TestCluster_ResizeStates(t *testing.T) {
t.Fatalf("opening cluster: %v", 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())
state0, err := node0.State()
if err != nil {
t.Fatal(err)
}
// Expect an error by adding a node not in the topology.
expectedError := "host is not in topology: node1"
if err := tc.addNode(); err == nil || err.Error() != expectedError {
t.Errorf("did not receive expected error: %s", expectedError)
// Ensure that node is in state STARTING before the other node joins.
if state0 != ClusterStateStarting {
t.Errorf("expected node0 state: %v, but got: %v", ClusterStateStarting, state0)
}
if err := tc.addNode(); err != nil {
t.Fatalf("adding node: %v", err)
}
node2 := tc.Clusters[2]
node1 := tc.Clusters[1]
state1, err := node1.State()
if err != nil {
t.Fatal(err)
}
// 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 node2 state: %v, but got: %v", ClusterStateNormal, node2.State())
if state0 != ClusterStateNormal {
t.Errorf("expected node0 state: %v, but got: %v", ClusterStateNormal, state0)
} else if state1 != ClusterStateNormal {
t.Errorf("expected node2 state: %v, but got: %v", ClusterStateNormal, state1)
}
// Close TestCluster.
@ -933,11 +961,21 @@ func TestCluster_ResizeStates(t *testing.T) {
node1 := tc.Clusters[1]
state1, err := node1.State()
if err != nil {
t.Fatal(err)
}
state0, err := node0.State()
if err != nil {
t.Fatal(err)
}
// 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())
if state0 != ClusterStateNormal {
t.Errorf("expected node0 state: %v, but got: %v", ClusterStateNormal, state0)
} else if state1 != ClusterStateNormal {
t.Errorf("expected node1 state: %v, but got: %v", ClusterStateNormal, state1)
}
// INVAR: after node1.State() is normal, the rebalancing should have been done.
@ -1030,7 +1068,6 @@ func TestAE(t *testing.T) {
t.Fatalf("abort should not have blocked this long")
}
})
}
// Ensures that coordinator can be changed.
@ -1038,8 +1075,10 @@ func TestCluster_UpdateCoordinator(t *testing.T) {
t.Run("UpdateCoordinator", func(t *testing.T) {
c := NewTestCluster(t, 2)
oldNode := c.nodes[0]
newNode := c.nodes[1]
cNodes := c.noder.Nodes()
oldNode := cNodes[0]
newNode := cNodes[1]
// Update coordinator to the same value.
if c.updateCoordinator(oldNode) {
@ -1085,8 +1124,8 @@ func TestCluster_confirmNodeDownUp(t *testing.T) {
if c.confirmNodeDown(uri) {
t.Errorf("expected node to be up")
}
}
func TestCluster_confirmNodeDownTimeout(t *testing.T) {
t.Skip("does a listen on :0, skip for now. TODO(jea) restore this.")
sleep := 50 * time.Millisecond
@ -1143,7 +1182,6 @@ func TestCluster_confirmNodeDownDown(t *testing.T) {
}
func TestCluster_GetNonPrimaryReplicas(t *testing.T) {
c := newCluster()
c.ReplicaN = 3
topo := NewTopology(c.Hasher, c.partitionN, c.ReplicaN, c)
@ -1151,7 +1189,7 @@ func TestCluster_GetNonPrimaryReplicas(t *testing.T) {
nNodes := 4
for i := 0; i < nNodes; i++ {
nodeID := fmt.Sprintf("node%d", i)
c.nodes = append(c.nodes, &topology.Node{
c.noder.AppendNode(&topology.Node{
ID: nodeID,
URI: NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0)),
})

2
disco/disco.go
View file

@ -173,7 +173,7 @@ type nopStator struct{}
// ClusterState is a no-op implementation of the Stator ClusterState method.
func (n *nopStator) ClusterState(context.Context) (ClusterState, error) {
return "", nil
return ClusterStateUnknown, nil
}
func (n *nopStator) Started(ctx context.Context) error {

15
executor.go
View file

@ -5266,7 +5266,7 @@ func (e *executor) executeSetRowAttrs(ctx context.Context, qcx *Qcx, index strin
}
// Execute on remote nodes in parallel.
nodes := topology.Nodes(e.Cluster.nodes).FilterID(e.Node.ID)
nodes := topology.Nodes(e.Cluster.noder.Nodes()).FilterID(e.Node.ID)
resp := make(chan error, len(nodes))
for _, node := range nodes {
go func(node *topology.Node) {
@ -5378,7 +5378,7 @@ func (e *executor) executeBulkSetRowAttrs(ctx context.Context, qcx *Qcx, index s
}
// Execute on remote nodes in parallel.
nodes := topology.Nodes(e.Cluster.nodes).FilterID(e.Node.ID)
nodes := topology.Nodes(e.Cluster.noder.Nodes()).FilterID(e.Node.ID)
resp := make(chan error, len(nodes))
for _, node := range nodes {
go func(node *topology.Node) {
@ -5430,7 +5430,7 @@ func (e *executor) executeSetColumnAttrs(ctx context.Context, qcx *Qcx, index st
}
// Execute on remote nodes in parallel.
nodes := topology.Nodes(e.Cluster.nodes).FilterID(e.Node.ID)
nodes := topology.Nodes(e.Cluster.noder.Nodes()).FilterID(e.Node.ID)
resp := make(chan error, len(nodes))
for _, node := range nodes {
go func(node *topology.Node) {
@ -5484,7 +5484,12 @@ func (e *executor) shardsByNode(nodes []*topology.Node, index string, shards []u
loop:
for _, shard := range shards {
for _, node := range snap.ShardNodes(index, shard) {
if topology.Nodes(nodes).Contains(node) {
// If the node being considered is in any state other than STARTED,
// then exclude it from the map. This way, one of that node's
// healthy replicas will be included instead.
// TODO: check state once stator is implemented
//if topology.Nodes(nodes).ContainsID(node.ID) && node.State == disco.NodeStateStarted {
if topology.Nodes(nodes).ContainsID(node.ID) {
m[node] = append(m[node], shard)
continue loop
}
@ -5537,7 +5542,7 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
if resp.err != nil {
// Filter out unavailable nodes.
nodes = topology.Nodes(nodes).Filter(resp.node)
nodes = topology.Nodes(nodes).FilterID(resp.node.ID)
// Begin mapper against secondary nodes.
if err := e.mapper(ctx, cancel, ch, nodes, index, resp.shards, c, opt, mapFn, reduceFn); errors.Cause(err) == errShardUnavailable {

7
holder.go
View file

@ -1200,6 +1200,7 @@ func (h *Holder) recalculateCaches() {
}
}
// TODO: this needs to be removed
func (h *Holder) isCoordinator() bool {
if s, ok := h.broadcaster.(*Server); ok {
return s.isCoordinator
@ -1426,7 +1427,7 @@ func (s *holderSyncer) syncIndex(index string) error {
s.Stats.CountWithCustomTags(MetricColumnAttrStoreBlocks, int64(len(blks)), 1.0, []string{indexTag})
// Sync with every other host.
for _, node := range topology.Nodes(s.Cluster.nodes).FilterID(s.Node.ID) {
for _, node := range topology.Nodes(s.Cluster.noder.Nodes()).FilterID(s.Node.ID) {
// Retrieve attributes from differing blocks.
// Skip update and recomputation if no attributes have changed.
m, err := s.Cluster.InternalClient.ColumnAttrDiff(ctx, &node.URI, index, blks)
@ -1473,7 +1474,7 @@ func (s *holderSyncer) syncField(index, name string) error {
s.Stats.CountWithCustomTags(MetricRowAttrStoreBlocks, int64(len(blks)), 1.0, []string{indexTag, fieldTag})
// Sync with every other host.
for _, node := range topology.Nodes(s.Cluster.nodes).FilterID(s.Node.ID) {
for _, node := range topology.Nodes(s.Cluster.noder.Nodes()).FilterID(s.Node.ID) {
// Retrieve attributes from differing blocks.
// Skip update and recomputation if no attributes have changed.
m, err := s.Cluster.InternalClient.RowAttrDiff(ctx, &node.URI, index, name, blks)
@ -1836,7 +1837,7 @@ func (c *holderCleaner) IsClosing() bool {
// any unnecessary fragments and files.
func (c *holderCleaner) CleanHolder() error {
// Create a snapshot of the cluster to use for node/partition calculations.
snap := topology.NewClusterSnapshot(c.Cluster.unprotectedNoder, c.Cluster.Hasher, c.Cluster.ReplicaN)
snap := topology.NewClusterSnapshot(c.Cluster.noder, c.Cluster.Hasher, c.Cluster.ReplicaN)
for _, index := range c.Holder.Indexes() {
// Verify cleaner has not closed.

9
http/handler.go
View file

@ -736,8 +736,15 @@ func (h *Handler) handleGetStatus(w http.ResponseWriter, r *http.Request) {
http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable)
return
}
state, err := h.api.State()
if err != nil {
http.Error(w, "getting cluster state error: "+err.Error(), http.StatusInternalServerError)
return
}
status := getStatusResponse{
State: h.api.State(),
State: state,
Nodes: h.api.Hosts(r.Context()),
LocalID: h.api.Node().ID,
ClusterName: h.api.ClusterName(),

37
server.go
View file

@ -603,21 +603,6 @@ func (s *Server) Open() error {
s.syncer.Closing = s.closing
s.syncer.Stats = s.holder.Stats.WithTags("component:HolderSyncer")
// TODO disco
if false {
node.URI = s.uri
node.GRPCURI = s.grpcURI
// Set metadata for this node.
data, err := json.Marshal(node)
if err != nil {
return errors.Wrap(err, "marshaling json metadata")
}
if err := s.metadator.SetMetadata(context.Background(), data); err != nil {
return errors.Wrap(err, "setting metadata")
}
}
err = s.cluster.setup()
if err != nil {
return errors.Wrap(err, "setting up cluster")
@ -642,9 +627,6 @@ func (s *Server) Open() error {
// bring up the background tasks for the holder.
s.holder.SnapshotQueue = s.snapshotQueue
s.holder.Activate()
if err := s.cluster.setNodeState(nodeStateReady); err != nil {
return errors.Wrap(err, "setting nodeState")
}
// Listen for joining nodes.
// This needs to start after the Holder has opened so that nodes can join
@ -788,7 +770,14 @@ func (s *Server) monitorAntiEntropy() {
s.holder.Stats.Count(MetricAntiEntropy, 1, 1.0)
}
t := time.Now()
if s.cluster.State() == ClusterStateResizing {
state, err := s.cluster.State()
if err != nil {
s.logger.Printf("cluster state error: err=%s", err)
continue
}
if state == ClusterStateResizing {
continue // don't launch anti-entropy during resize.
// the cluster sets its state to resizing and *then* sends to
// abortAntiEntropyCh before starting to resize
@ -1021,8 +1010,14 @@ func (s *Server) node() *topology.Node {
// handleRemoteStatus receives incoming NodeStatus from remote nodes.
func (s *Server) handleRemoteStatus(pb Message) {
state, err := s.cluster.State()
if err != nil {
s.logger.Printf("getting cluster state: %s", err)
return
}
// Ignore NodeStatus messages until the cluster is in a Normal state.
if s.cluster.State() != ClusterStateNormal {
if state != ClusterStateNormal {
return
}
@ -1081,7 +1076,7 @@ func (s *Server) monitorDiagnostics() {
s.diagnostics.SetVersion(Version)
s.diagnostics.Set("Host", s.uri.Host)
s.diagnostics.Set("Cluster", strings.Join(s.cluster.nodeIDs(), ","))
s.diagnostics.Set("NumNodes", len(s.cluster.nodes))
s.diagnostics.Set("NumNodes", len(s.cluster.noder.Nodes()))
s.diagnostics.Set("NumCPU", runtime.NumCPU())
s.diagnostics.Set("NodeID", s.nodeID)
s.diagnostics.Set("ClusterID", s.cluster.id)

13
test/cluster.go
View file

@ -484,9 +484,9 @@ func (c *Cluster) AwaitCoordinatorState(expectedState string, timeout time.Durat
// in the expected state.
func (c *Cluster) ExceptionalState(expectedState string) error {
for _, node := range c.Nodes {
state := node.API.State()
if state != expectedState {
return fmt.Errorf("node %q: state %s", node.ID(), state)
state, err := node.API.State()
if err != nil || state != expectedState {
return fmt.Errorf("node %q: state %s: err %v", node.ID(), state, err)
}
}
return nil
@ -534,7 +534,12 @@ func MustNewCluster(tb testing.TB, size int, opts ...[]server.CommandOption) *Cl
// receives a matching state. It polls up to n times before returning.
func CheckClusterState(m *Command, state string, n int) bool {
for i := 0; i < n; i++ {
if m.API.State() == state {
apiState, err := m.API.State()
if err != nil {
return false
}
if apiState == state {
return true
}
time.Sleep(10 * time.Millisecond)

8
test/pilosa.go
View file

@ -192,7 +192,13 @@ func (m *Command) URL() string { return m.API.Node().URI.String() }
func (m *Command) ID() string { return m.API.Node().ID }
// IsCoordinator returns true if this is the coordinator.
func (m *Command) IsCoordinator() bool { return m.API.Node().IsCoordinator }
func (m *Command) IsCoordinator() bool {
coord := m.API.CoordinatorNode()
if coord == nil {
return false
}
return coord.ID == m.API.Node().ID
}
// Client returns a client to connect to the program.
func (m *Command) Client() *http.InternalClient {

2
topology/node.go
View file

@ -52,7 +52,7 @@ func (n *Node) Clone() *Node {
}
func (n *Node) String() string {
return fmt.Sprintf("Node:%s:%s:%s", n.URI, n.State, n.ID)
return fmt.Sprintf("Node:%s:%s:%s(%v)", n.URI, n.State, n.ID, n.IsCoordinator)
}
// Nodes represents a list of nodes.

5
topology/noder.go
View file

@ -41,6 +41,11 @@ func NewLocalNoder(nodes []*Node) *localNoder {
}
}
// NewEmptyLocalNoder is an empty Noder used for testing.
func NewEmptyLocalNoder() *localNoder {
return &localNoder{}
}
// Nodes implements the Noder interface.
func (n *localNoder) Nodes() []*Node {
return n.nodes

16
topology/snapshot.go
View file

@ -139,25 +139,13 @@ func (c *ClusterSnapshot) PartitionNodes(partitionID int) []*Node {
// field keys. The primary could be any node in the cluster, but we arbitrarily
// define it to be the node responsible for partition 0.
func (c *ClusterSnapshot) PrimaryFieldTranslationNode() *Node {
// return c.PrimaryPartitionNode(0)
for _, n := range c.Nodes {
if n.IsCoordinator {
return n
}
}
return nil
return c.PrimaryPartitionNode(0)
}
// IsPrimaryFieldTranslationNode returns true if nodeID represents the primary
// node responsible for field translation.
func (c *ClusterSnapshot) IsPrimaryFieldTranslationNode(nodeID string) bool {
// return c.PrimaryFieldTranslationNode().ID == nodeID
for i := range c.Nodes {
if c.Nodes[i].ID == nodeID && c.Nodes[i].IsCoordinator {
return true
}
}
return false
return c.PrimaryFieldTranslationNode().ID == nodeID
}
// PrimaryPartitionNode returns the primary node of the given partition.

17
translator_test.go
View file

@ -514,10 +514,14 @@ func TestTranslation_Replication(t *testing.T) {
exp := `{"results":[{"attrs":{},"columns":[],"keys":["x1","x2"]}]}`
if !test.CheckClusterState(coord, pilosa.ClusterStateNormal, 1000) {
t.Fatalf("unexpected coord cluster state: %s, got: %s", pilosa.ClusterStateNormal, coord.API.State())
} else if !test.CheckClusterState(other, pilosa.ClusterStateNormal, 1000) {
t.Fatalf("unexpected other cluster state: %s, got: %s", pilosa.ClusterStateNormal, other.API.State())
coordState, err := coord.API.State()
if err != nil || !test.CheckClusterState(coord, pilosa.ClusterStateNormal, 1000) {
t.Fatalf("unexpected coord cluster state: %s, got: %s, err: %v", pilosa.ClusterStateNormal, coordState, err)
}
otherState, err := other.API.State()
if err != nil || !test.CheckClusterState(other, pilosa.ClusterStateNormal, 1000) {
t.Fatalf("unexpected other cluster state: %s, got: %s, err: %v", pilosa.ClusterStateNormal, otherState, err)
}
// Verify the data exists
@ -528,8 +532,9 @@ func TestTranslation_Replication(t *testing.T) {
t.Fatal(err)
}
if !test.CheckClusterState(coord, pilosa.ClusterStateDegraded, 1000) {
t.Fatalf("unexpected coord cluster state: %s, got: %s", pilosa.ClusterStateDegraded, coord.API.State())
coordState, err = coord.API.State()
if err != nil || !test.CheckClusterState(coord, pilosa.ClusterStateDegraded, 1000) {
t.Fatalf("unexpected coord cluster state: %s, got: %s", pilosa.ClusterStateDegraded, coordState)
}
// Verify the data exists with one node down

26
utils_internal_test.go
View file

@ -75,14 +75,16 @@ func NewTestCluster(tb testing.TB, n int) *cluster {
c.Topology = NewTopology(c.Hasher, c.partitionN, c.ReplicaN, c)
for i := 0; i < n; i++ {
c.nodes = append(c.nodes, &topology.Node{
c.noder.AppendNode(&topology.Node{
ID: fmt.Sprintf("node%d", i),
URI: NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0)),
})
}
c.Node = c.nodes[0]
c.Coordinator = c.nodes[0].ID
cNodes := c.noder.Nodes()
c.Node = cNodes[0]
c.Coordinator = cNodes[0].ID
c.SetState(ClusterStateNormal)
return c
@ -231,8 +233,13 @@ func (t *ClusterCluster) addNode() error {
return err
}
state, err := coord.State()
if err != nil {
return err
}
// Wait for the AddNode job to finish.
if c.State() != ClusterStateNormal {
if state != ClusterStateNormal {
t.resizeDone = make(chan struct{})
t.mu.Lock()
t.resizing = true
@ -341,9 +348,6 @@ func (t *ClusterCluster) Open() error {
if err := c.holder.Open(); err != nil {
return err
}
if err := c.setNodeState(nodeStateReady); err != nil {
return err
}
}
// Start the listener on the coordinator.
@ -553,15 +557,17 @@ func NewTestClusterWithReplication(tb testing.TB, nNodes, nReplicas, partitionN
for i := 0; i < nNodes; i++ {
nodeID := fmt.Sprintf("node%d", i)
c.nodes = append(c.nodes, &topology.Node{
c.noder.AppendNode(&topology.Node{
ID: nodeID,
URI: NewTestURI("http", fmt.Sprintf("host%d", i), uint16(0)),
})
c.Topology.addID(nodeID)
}
c.Node = c.nodes[0]
c.Coordinator = c.nodes[0].ID
cNodes := c.noder.Nodes()
c.Node = cNodes[0]
c.Coordinator = cNodes[0].ID
c.SetState(ClusterStateNormal)
if err := c.holder.Open(); err != nil {