mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-07 09:05:55 +00:00
Add a Cluster.Static override for tests to treat static nodes as
Coordinator. Refactor the Server.joining channel to be in Cluster instead.
This commit is contained in:
parent
ccdd6262c5
commit
8228ef2382
3 changed files with 45 additions and 22 deletions
32
cluster.go
32
cluster.go
|
|
@ -166,6 +166,7 @@ type Cluster struct {
|
|||
Topology *Topology
|
||||
|
||||
// Required for cluster Resize.
|
||||
Static bool // Static is primarily used for testing in a non-gossip environment.
|
||||
State string
|
||||
Coordinator URI
|
||||
Holder *Holder
|
||||
|
|
@ -173,6 +174,11 @@ type Cluster struct {
|
|||
|
||||
joiningLeavingNodes chan nodeAction
|
||||
|
||||
// joining is held open until this node
|
||||
// receives ClusterStatus from the coordinator.
|
||||
joining chan struct{}
|
||||
joined bool
|
||||
|
||||
mu sync.RWMutex
|
||||
jobs map[int64]*ResizeJob
|
||||
currentJob *ResizeJob
|
||||
|
|
@ -197,6 +203,7 @@ func NewCluster() *Cluster {
|
|||
joiningLeavingNodes: make(chan nodeAction, 10), // buffered channel
|
||||
jobs: make(map[int64]*ResizeJob),
|
||||
closing: make(chan struct{}),
|
||||
joining: make(chan struct{}),
|
||||
|
||||
LogOutput: os.Stderr,
|
||||
prefect: &NopSecurityManager{},
|
||||
|
|
@ -210,7 +217,7 @@ func (c *Cluster) logger() *log.Logger {
|
|||
|
||||
// IsCoordinator is true if this node is the coordinator.
|
||||
func (c *Cluster) IsCoordinator() bool {
|
||||
return c.Coordinator == c.URI
|
||||
return c.Static || c.Coordinator == c.URI
|
||||
}
|
||||
|
||||
// SetCoordinator updates the Coordinator to new if it is
|
||||
|
|
@ -709,6 +716,12 @@ func (c *Cluster) Open() error {
|
|||
return fmt.Errorf("opening MemberSet: %v", err)
|
||||
}
|
||||
|
||||
// If not coordinator then wait for ClusterStatus from coordinator.
|
||||
if !c.IsCoordinator() {
|
||||
c.logger().Printf("wait for joining to complete")
|
||||
<-c.joining
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
@ -720,15 +733,28 @@ func (c *Cluster) Close() error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (c *Cluster) MarkAsJoined() {
|
||||
if !c.joined {
|
||||
c.joined = true
|
||||
close(c.joining)
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Cluster) needTopologyAgreement() bool {
|
||||
return c.State == ClusterStateStarting && !URISlicesAreEqual(c.Topology.NodeSet, c.NodeSet())
|
||||
}
|
||||
|
||||
func (c *Cluster) haveTopologyAgreement() bool {
|
||||
if c.Static {
|
||||
return true
|
||||
}
|
||||
return URISlicesAreEqual(c.Topology.NodeSet, c.NodeSet())
|
||||
}
|
||||
|
||||
func (c *Cluster) allNodesReady() bool {
|
||||
if c.Static {
|
||||
return true
|
||||
}
|
||||
for _, uri := range c.Topology.NodeSet {
|
||||
if c.Topology.nodeStates[uri] != NodeStateReady {
|
||||
return false
|
||||
|
|
@ -1335,6 +1361,10 @@ func decodeTopology(topology *internal.Topology) (*Topology, error) {
|
|||
}
|
||||
|
||||
func (c *Cluster) considerTopology() error {
|
||||
if c.Static {
|
||||
return nil
|
||||
}
|
||||
|
||||
// If there is no .topology file, it's safe to proceed.
|
||||
if len(c.Topology.NodeSet) == 0 {
|
||||
return nil
|
||||
|
|
|
|||
21
server.go
21
server.go
|
|
@ -51,11 +51,6 @@ type Server struct {
|
|||
wg sync.WaitGroup
|
||||
closing chan struct{}
|
||||
|
||||
// joining is held open until this node
|
||||
// receives ClusterStatus from the coordinator.
|
||||
joining chan struct{}
|
||||
joined bool
|
||||
|
||||
// Data storage and HTTP interface.
|
||||
Holder *Holder
|
||||
Handler *Handler
|
||||
|
|
@ -89,7 +84,6 @@ type Server struct {
|
|||
func NewServer() *Server {
|
||||
s := &Server{
|
||||
closing: make(chan struct{}),
|
||||
joining: make(chan struct{}),
|
||||
|
||||
Holder: NewHolder(),
|
||||
Handler: NewHandler(),
|
||||
|
|
@ -198,12 +192,6 @@ func (s *Server) Open() error {
|
|||
return fmt.Errorf("opening Cluster: %v", err)
|
||||
}
|
||||
|
||||
// If not coordinator then wait for ClusterStatus from coordinator.
|
||||
if !s.Cluster.IsCoordinator() {
|
||||
s.Logger().Printf("wait for joining to complete")
|
||||
<-s.joining
|
||||
}
|
||||
|
||||
// Open holder.
|
||||
if err := s.Holder.Open(); err != nil {
|
||||
return fmt.Errorf("opening Holder: %v", err)
|
||||
|
|
@ -228,13 +216,6 @@ func (s *Server) Open() error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (s *Server) markAsJoined() {
|
||||
if !s.joined {
|
||||
s.joined = true
|
||||
close(s.joining)
|
||||
}
|
||||
}
|
||||
|
||||
// Close closes the server and waits for it to shutdown.
|
||||
func (s *Server) Close() error {
|
||||
// Notify goroutines to stop.
|
||||
|
|
@ -370,7 +351,7 @@ func (s *Server) ReceiveMessage(pb proto.Message) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
s.markAsJoined()
|
||||
s.Cluster.MarkAsJoined()
|
||||
case *internal.ResizeInstruction:
|
||||
err := s.Cluster.FollowResizeInstruction(obj)
|
||||
if err != nil {
|
||||
|
|
|
|||
|
|
@ -63,7 +63,7 @@ type Command struct {
|
|||
// Standard input/output
|
||||
*pilosa.CmdIO
|
||||
|
||||
// running will be closed once Command.Run is finished.
|
||||
// Started will be closed once Command.Run is finished.
|
||||
Started chan struct{}
|
||||
// Done will be closed when Command.Close() is called
|
||||
Done chan struct{}
|
||||
|
|
@ -220,6 +220,18 @@ func (m *Command) SetupServer() error {
|
|||
m.Server.Broadcaster = gossipMemberSet
|
||||
m.Server.BroadcastReceiver = gossipMemberSet
|
||||
case pilosa.ClusterStatic, pilosa.ClusterNone:
|
||||
|
||||
m.Server.Cluster.Static = true
|
||||
for _, address := range m.Config.Cluster.Hosts {
|
||||
uri, err := pilosa.NewURIFromAddress(address)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
cluster.Nodes = append(cluster.Nodes, &pilosa.Node{
|
||||
URI: *uri,
|
||||
})
|
||||
}
|
||||
|
||||
m.Server.Broadcaster = pilosa.NopBroadcaster
|
||||
m.Server.Cluster.MemberSet = pilosa.NewStaticMemberSet()
|
||||
m.Server.BroadcastReceiver = pilosa.NopBroadcastReceiver
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue