mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-12 15:51:01 +00:00
put a mutex around Cluster.State
This commit is contained in:
parent
1b0fcb0e5a
commit
f12527d535
5 changed files with 66 additions and 54 deletions
34
cluster.go
34
cluster.go
|
|
@ -175,7 +175,7 @@ type Cluster struct {
|
|||
|
||||
// Required for cluster Resize.
|
||||
Static bool // Static is primarily used for testing in a non-gossip environment.
|
||||
State string
|
||||
state string
|
||||
Coordinator URI
|
||||
Holder *Holder
|
||||
Broadcaster Broadcaster
|
||||
|
|
@ -302,9 +302,21 @@ func (c *Cluster) setID(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) SetState(state string) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.setState(state)
|
||||
}
|
||||
|
||||
func (c *Cluster) setState(state string) {
|
||||
// Ignore cases where the state hasn't changed.
|
||||
if state == c.State {
|
||||
if state == c.state {
|
||||
return
|
||||
}
|
||||
|
||||
|
|
@ -321,12 +333,12 @@ func (c *Cluster) setState(state string) {
|
|||
// - ClusterStateStarting
|
||||
|
||||
// If state is RESIZING -> NORMAL then run cleanup.
|
||||
if c.State == ClusterStateResizing {
|
||||
if c.state == ClusterStateResizing {
|
||||
doCleanup = true
|
||||
}
|
||||
}
|
||||
|
||||
c.State = state
|
||||
c.state = state
|
||||
|
||||
// TODO: consider NOT running cleanup on an active node that has
|
||||
// been removed.
|
||||
|
|
@ -373,7 +385,7 @@ func (c *Cluster) ReceiveNodeState(uri URI, state string) error {
|
|||
}
|
||||
|
||||
// This method is really only useful during initial startup.
|
||||
if c.State != ClusterStateStarting {
|
||||
if c.State() != ClusterStateStarting {
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
@ -397,7 +409,7 @@ func (c *Cluster) ReceiveNodeState(uri URI, state string) error {
|
|||
func (c *Cluster) Status() *internal.ClusterStatus {
|
||||
return &internal.ClusterStatus{
|
||||
ClusterID: c.ID,
|
||||
State: c.State,
|
||||
State: c.state,
|
||||
NodeSet: encodeURIs(c.NodeSet()),
|
||||
}
|
||||
}
|
||||
|
|
@ -763,7 +775,7 @@ func (h *jmphasher) Hash(key uint64, n int) int {
|
|||
|
||||
func (c *Cluster) Open() error {
|
||||
// Cluster always comes up in state STARTING until cluster membership is determined.
|
||||
c.State = ClusterStateStarting
|
||||
c.state = ClusterStateStarting
|
||||
|
||||
// Load topology file if it exists.
|
||||
if err := c.loadTopology(); err != nil {
|
||||
|
|
@ -820,7 +832,7 @@ func (c *Cluster) markAsJoined() {
|
|||
}
|
||||
|
||||
func (c *Cluster) needTopologyAgreement() bool {
|
||||
return c.State == ClusterStateStarting && !URISlicesAreEqual(c.Topology.NodeSet, c.NodeSet())
|
||||
return c.State() == ClusterStateStarting && !URISlicesAreEqual(c.Topology.NodeSet, c.NodeSet())
|
||||
}
|
||||
|
||||
func (c *Cluster) haveTopologyAgreement() bool {
|
||||
|
|
@ -886,7 +898,7 @@ func (c *Cluster) handleNodeAction(nodeAction nodeAction) error {
|
|||
}
|
||||
|
||||
func (c *Cluster) setStateAndBroadcast(state string) error {
|
||||
c.setState(state)
|
||||
c.SetState(state)
|
||||
// Broadcast cluster status changes to the cluster.
|
||||
c.logger().Printf("broadcasting ClusterStatus: %s", state)
|
||||
return c.Broadcaster.SendSync(c.Status())
|
||||
|
|
@ -1618,7 +1630,7 @@ func (c *Cluster) NodeLeave(uri URI) error {
|
|||
return fmt.Errorf("Node removal requests are only valid on the Coordinator node: %s", c.Coordinator)
|
||||
}
|
||||
|
||||
if c.State != ClusterStateNormal {
|
||||
if c.State() != ClusterStateNormal {
|
||||
return fmt.Errorf("Cluster must be in state %s to remove a node. Current state: %s", ClusterStateNormal, c.State)
|
||||
}
|
||||
|
||||
|
|
@ -1683,7 +1695,7 @@ func (c *Cluster) MergeClusterStatus(cs *internal.ClusterStatus) error {
|
|||
}
|
||||
}
|
||||
|
||||
c.setState(cs.State)
|
||||
c.SetState(cs.State)
|
||||
|
||||
c.markAsJoined()
|
||||
|
||||
|
|
|
|||
|
|
@ -255,8 +255,8 @@ func TestCluster_ResizeStates(t *testing.T) {
|
|||
node := tc.Clusters[0]
|
||||
|
||||
// Ensure that node comes up in state NORMAL.
|
||||
if node.State != pilosa.ClusterStateNormal {
|
||||
t.Errorf("expected state: %v, but got: %v", pilosa.ClusterStateNormal, node.State)
|
||||
if node.State() != pilosa.ClusterStateNormal {
|
||||
t.Errorf("expected state: %v, but got: %v", pilosa.ClusterStateNormal, node.State())
|
||||
}
|
||||
|
||||
expectedTop := &pilosa.Topology{
|
||||
|
|
@ -292,8 +292,8 @@ func TestCluster_ResizeStates(t *testing.T) {
|
|||
}
|
||||
|
||||
// Ensure that node comes up in state NORMAL.
|
||||
if node.State != pilosa.ClusterStateNormal {
|
||||
t.Errorf("expected state: %v, but got: %v", pilosa.ClusterStateNormal, node.State)
|
||||
if node.State() != pilosa.ClusterStateNormal {
|
||||
t.Errorf("expected state: %v, but got: %v", pilosa.ClusterStateNormal, node.State())
|
||||
}
|
||||
|
||||
// Close TestCluster.
|
||||
|
|
@ -344,10 +344,10 @@ func TestCluster_ResizeStates(t *testing.T) {
|
|||
node1 := tc.Clusters[1]
|
||||
|
||||
// Ensure that nodes comes up in state NORMAL.
|
||||
if node0.State != pilosa.ClusterStateNormal {
|
||||
t.Errorf("expected node0 state: %v, but got: %v", pilosa.ClusterStateNormal, node0.State)
|
||||
} else if node1.State != pilosa.ClusterStateNormal {
|
||||
t.Errorf("expected node1 state: %v, but got: %v", pilosa.ClusterStateNormal, node1.State)
|
||||
if node0.State() != pilosa.ClusterStateNormal {
|
||||
t.Errorf("expected node0 state: %v, but got: %v", pilosa.ClusterStateNormal, node0.State())
|
||||
} else if node1.State() != pilosa.ClusterStateNormal {
|
||||
t.Errorf("expected node1 state: %v, but got: %v", pilosa.ClusterStateNormal, node1.State())
|
||||
}
|
||||
|
||||
expectedTop := &pilosa.Topology{
|
||||
|
|
@ -388,8 +388,8 @@ func TestCluster_ResizeStates(t *testing.T) {
|
|||
}
|
||||
|
||||
// Ensure that node is in state STARTING before the other node joins.
|
||||
if node0.State != pilosa.ClusterStateStarting {
|
||||
t.Errorf("expected node0 state: %v, but got: %v", pilosa.ClusterStateStarting, node0.State)
|
||||
if node0.State() != pilosa.ClusterStateStarting {
|
||||
t.Errorf("expected node0 state: %v, but got: %v", pilosa.ClusterStateStarting, node0.State())
|
||||
}
|
||||
|
||||
// Expect an error by adding a node not in the topology.
|
||||
|
|
@ -403,10 +403,10 @@ func TestCluster_ResizeStates(t *testing.T) {
|
|||
node2 := tc.Clusters[2]
|
||||
|
||||
// Ensure that node comes up in state NORMAL.
|
||||
if node0.State != pilosa.ClusterStateNormal {
|
||||
t.Errorf("expected node0 state: %v, but got: %v", pilosa.ClusterStateNormal, node0.State)
|
||||
} else if node2.State != pilosa.ClusterStateNormal {
|
||||
t.Errorf("expected node1 state: %v, but got: %v", pilosa.ClusterStateNormal, node2.State)
|
||||
if node0.State() != pilosa.ClusterStateNormal {
|
||||
t.Errorf("expected node0 state: %v, but got: %v", pilosa.ClusterStateNormal, node0.State())
|
||||
} else if node2.State() != pilosa.ClusterStateNormal {
|
||||
t.Errorf("expected node1 state: %v, but got: %v", pilosa.ClusterStateNormal, node2.State())
|
||||
}
|
||||
|
||||
// Close TestCluster.
|
||||
|
|
@ -470,10 +470,10 @@ func TestCluster_ResizeStates(t *testing.T) {
|
|||
node1 := tc.Clusters[1]
|
||||
|
||||
// Ensure that nodes come up in state NORMAL.
|
||||
if node0.State != pilosa.ClusterStateNormal {
|
||||
t.Errorf("expected node0 state: %v, but got: %v", pilosa.ClusterStateNormal, node0.State)
|
||||
} else if node1.State != pilosa.ClusterStateNormal {
|
||||
t.Errorf("expected node1 state: %v, but got: %v", pilosa.ClusterStateNormal, node1.State)
|
||||
if node0.State() != pilosa.ClusterStateNormal {
|
||||
t.Errorf("expected node0 state: %v, but got: %v", pilosa.ClusterStateNormal, node0.State())
|
||||
} else if node1.State() != pilosa.ClusterStateNormal {
|
||||
t.Errorf("expected node1 state: %v, but got: %v", pilosa.ClusterStateNormal, node1.State())
|
||||
}
|
||||
|
||||
expectedTop := &pilosa.Topology{
|
||||
|
|
|
|||
|
|
@ -510,7 +510,7 @@ func (s *Server) ClusterStatus() (proto.Message, error) {
|
|||
// HandleRemoteStatus receives incoming NodeStatus from remote nodes.
|
||||
func (s *Server) HandleRemoteStatus(pb proto.Message) error {
|
||||
// Ignore NodeStatus messages until the cluster is in a Normal state.
|
||||
if s.Cluster.State != ClusterStateNormal {
|
||||
if s.Cluster.State() != ClusterStateNormal {
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -209,8 +209,8 @@ func TestClusterResize_EmptyNode(t *testing.T) {
|
|||
m0 := test.MustRunMain()
|
||||
defer m0.Close()
|
||||
|
||||
if m0.Server.Cluster.State != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected cluster state: %s", m0.Server.Cluster.State)
|
||||
if m0.Server.Cluster.State() != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected cluster state: %s", m0.Server.Cluster.State())
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -236,10 +236,10 @@ func TestClusterResize_EmptyNodes(t *testing.T) {
|
|||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if m0.Server.Cluster.State != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State)
|
||||
} else if m1.Server.Cluster.State != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State)
|
||||
if m0.Server.Cluster.State() != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State())
|
||||
} else if m1.Server.Cluster.State() != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State())
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -273,10 +273,10 @@ func TestClusterResize_AddNode(t *testing.T) {
|
|||
|
||||
time.Sleep(1 * time.Second)
|
||||
|
||||
if m0.Server.Cluster.State != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State)
|
||||
} else if m1.Server.Cluster.State != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State)
|
||||
if m0.Server.Cluster.State() != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State())
|
||||
} else if m1.Server.Cluster.State() != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State())
|
||||
}
|
||||
})
|
||||
t.Run("WithIndex", func(t *testing.T) {
|
||||
|
|
@ -318,10 +318,10 @@ func TestClusterResize_AddNode(t *testing.T) {
|
|||
// Give the cluster time to settle.
|
||||
time.Sleep(1 * time.Second)
|
||||
|
||||
if m0.Server.Cluster.State != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State)
|
||||
} else if m1.Server.Cluster.State != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State)
|
||||
if m0.Server.Cluster.State() != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State())
|
||||
} else if m1.Server.Cluster.State() != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State())
|
||||
}
|
||||
})
|
||||
t.Run("ContinuousSlices", func(t *testing.T) {
|
||||
|
|
@ -373,10 +373,10 @@ func TestClusterResize_AddNode(t *testing.T) {
|
|||
// Give the cluster time to settle.
|
||||
time.Sleep(1 * time.Second)
|
||||
|
||||
if m0.Server.Cluster.State != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State)
|
||||
} else if m1.Server.Cluster.State != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State)
|
||||
if m0.Server.Cluster.State() != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State())
|
||||
} else if m1.Server.Cluster.State() != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State())
|
||||
}
|
||||
})
|
||||
t.Run("SkippedSlice", func(t *testing.T) {
|
||||
|
|
@ -428,10 +428,10 @@ func TestClusterResize_AddNode(t *testing.T) {
|
|||
// Give the cluster time to settle.
|
||||
time.Sleep(1 * time.Second)
|
||||
|
||||
if m0.Server.Cluster.State != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State)
|
||||
} else if m1.Server.Cluster.State != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State)
|
||||
if m0.Server.Cluster.State() != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node0 cluster state: %s", m0.Server.Cluster.State())
|
||||
} else if m1.Server.Cluster.State() != pilosa.ClusterStateNormal {
|
||||
t.Fatalf("unexpected node1 cluster state: %s", m1.Server.Cluster.State())
|
||||
}
|
||||
})
|
||||
}
|
||||
|
|
|
|||
|
|
@ -184,7 +184,7 @@ func (t *TestCluster) AddNode(saveTopology bool) error {
|
|||
}
|
||||
|
||||
// Wait for the AddNode job to finish.
|
||||
if c.State != pilosa.ClusterStateNormal {
|
||||
if c.State() != pilosa.ClusterStateNormal {
|
||||
t.resizeDone = make(chan struct{})
|
||||
<-t.resizeDone
|
||||
}
|
||||
|
|
@ -266,7 +266,7 @@ func NewTestCluster(n int) *TestCluster {
|
|||
// SetState sets the state of the cluster on each node.
|
||||
func (t *TestCluster) SetState(state string) {
|
||||
for _, c := range t.Clusters {
|
||||
c.State = state
|
||||
c.SetState(state)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue